""" Parse USDA NASS CSVs and generate Iowa Corn Dashboard HTML. """ import csv import json import re from collections import defaultdict CORN_CSV = r"C:\Users\wsu\Downloads\88D08B80-1879-34AE-A13B-7D5D70C76F9E.csv" SOIL_CSV = r"C:\Users\wsu\Downloads\nass_soil_iowa.csv" OUT_HTML = r"C:\Users\wsu\Downloads\viz\iowa_corn_dashboard.html" YEARS = [2021, 2022, 2023, 2024, 2025] # ── Parse corn CSV ────────────────────────────────────────────── condition_data = defaultdict(lambda: defaultdict(dict)) # {year: {week: {rating: val}}} progress_data = defaultdict(lambda: defaultdict(dict)) # {year: {week: {stage: val}}} with open(CORN_CSV, encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if row['Program'] != 'SURVEY': continue year = int(row['Year']) if year not in YEARS: continue period = row['Period'] if not period.startswith('WEEK'): continue week = int(period.replace('WEEK #', '')) item = row['Data Item'] val_str = row['Value'].replace(',', '').strip() if val_str in ('', '(D)', '(Z)', '(H)', '(L)', '(S)'): continue try: val = float(val_str) except ValueError: continue # Condition data - exclude PREVIOUS YEAR and 5 YEAR AVG if 'CONDITION, MEASURED IN PCT' in item: for rating in ['EXCELLENT', 'GOOD', 'FAIR', 'POOR', 'VERY POOR']: if item.endswith(f'PCT {rating}'): condition_data[year][week][rating] = val break # Progress data - exclude PREVIOUS YEAR and 5 YEAR AVG if 'PROGRESS, MEASURED IN PCT' in item: for stage in ['PLANTED', 'EMERGED', 'SILKING', 'DOUGH', 'DENTED', 'MATURE', 'HARVESTED']: if item.endswith(f'PCT {stage}'): progress_data[year][week][stage] = val break # ── Parse soil CSV ────────────────────────────────────────────── topsoil_data = defaultdict(lambda: defaultdict(dict)) # {year: {week: {metric: val}}} subsoil_data = defaultdict(lambda: defaultdict(dict)) with open(SOIL_CSV, encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if row['Program'] != 'SURVEY': continue year = int(row['Year']) if year not in YEARS: continue period = row['Period'] if not period.startswith('WEEK'): continue week = int(period.replace('WEEK #', '')) item = row['Data Item'] val_str = row['Value'].replace(',', '').strip() if val_str in ('', '(D)', '(Z)', '(H)', '(L)', '(S)'): continue # Skip PREVIOUS YEAR and 5 YEAR AVG if 'PREVIOUS YEAR' in item or '5 YEAR AVG' in item: continue try: val = float(val_str) except ValueError: continue for level in ['ADEQUATE', 'SURPLUS', 'SHORT', 'VERY SHORT']: if item.endswith(f'PCT {level}'): if 'TOPSOIL' in item: topsoil_data[year][week][level] = val elif 'SUBSOIL' in item: subsoil_data[year][week][level] = val break # ── Build JSON for JS ─────────────────────────────────────────── def build_condition_json(): """Good + Excellent combined % by week for each year.""" result = {} for year in YEARS: weeks_sorted = sorted(condition_data[year].keys()) weeks = [] values = [] for w in weeks_sorted: d = condition_data[year][w] good = d.get('GOOD', 0) exc = d.get('EXCELLENT', 0) weeks.append(w) values.append(good + exc) result[str(year)] = {'weeks': weeks, 'values': values} return result def build_progress_json(): """Progress stages by week for each year.""" stages = ['PLANTED', 'EMERGED', 'SILKING', 'DOUGH', 'DENTED', 'MATURE', 'HARVESTED'] result = {} for year in YEARS: weeks_sorted = sorted(progress_data[year].keys()) year_data = {'weeks': weeks_sorted, 'stages': {}} for stage in stages: vals = [] for w in weeks_sorted: vals.append(progress_data[year][w].get(stage, None)) # Only include stage if it has any non-None data if any(v is not None for v in vals): year_data['stages'][stage] = vals result[str(year)] = year_data return result def build_soil_json(data): """Adequate and Surplus % by week for each year.""" result = {} for year in YEARS: weeks_sorted = sorted(data[year].keys()) adequate = [] surplus = [] short_vals = [] vshort = [] for w in weeks_sorted: d = data[year][w] adequate.append(d.get('ADEQUATE', None)) surplus.append(d.get('SURPLUS', None)) short_vals.append(d.get('SHORT', None)) vshort.append(d.get('VERY SHORT', None)) result[str(year)] = { 'weeks': weeks_sorted, 'adequate': adequate, 'surplus': surplus, 'short': short_vals, 'very_short': vshort, } return result # Compute summary stats def compute_peak_ge(): """Peak Good+Excellent for each year and in which week.""" peaks = {} for year in YEARS: max_val = 0 max_week = 0 for w, d in condition_data[year].items(): ge = d.get('GOOD', 0) + d.get('EXCELLENT', 0) if ge > max_val: max_val = ge max_week = w peaks[str(year)] = {'peak': max_val, 'week': max_week} return peaks condition_json = json.dumps(build_condition_json()) progress_json = json.dumps(build_progress_json()) topsoil_json = json.dumps(build_soil_json(topsoil_data)) subsoil_json = json.dumps(build_soil_json(subsoil_data)) peak_ge_json = json.dumps(compute_peak_ge()) # Quick check print("Condition years:", list(condition_data.keys())) for y in YEARS: cw = sorted(condition_data[y].keys()) pw = sorted(progress_data[y].keys()) tw = sorted(topsoil_data[y].keys()) sw = sorted(subsoil_data[y].keys()) print(f" {y}: condition weeks={len(cw)}, progress weeks={len(pw)}, topsoil weeks={len(tw)}, subsoil weeks={len(sw)}") # ── Generate HTML ─────────────────────────────────────────────── html = f""" Iowa Corn Crop Progress & Condition 2021-2025

🌽Iowa Corn Crop Progress & Condition 2021-2025

Source: USDA NASS Quick Stats • State Level Data • Weekly Survey Reports

2021 2022 2023 2024 2025

Corn Condition: Good + Excellent %

Weekly combined Good & Excellent rating by year

Corn Progress by Growth Stage

Cumulative % of crop reaching each stage — select year

Topsoil Moisture: Adequate %

Weekly adequate moisture level by year

Subsoil Moisture: Adequate %

Weekly adequate moisture level by year

Generated from USDA NASS Quick Stats data • Iowa State Level • Crop Years 2021–2025
""" with open(OUT_HTML, 'w', encoding='utf-8') as f: f.write(html) print(f"\nDashboard saved to: {OUT_HTML}")