Knou/Climate-Mood-Analysis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refine climate metrics and data pipeline

Browse Source
This commit is contained in:
Knou
2026-07-24 22:55:14 -04:00
parent 016f7386d8
commit 4e2d878e5b
32 changed files with 3557 additions and 3347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/apply_locally_extreme_metric_to_climate_data.py
+11 -12
View File
@@ -17,7 +17,6 @@ import argparse
import csv
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CLIMATE_DATA = REPO_ROOT / "data" / "climate-data.csv"
DEFAULT_COMPARISON = REPO_ROOT / "data" / "noaa" / "county_locally_extreme_days_comparison.csv"
@@ -84,13 +83,13 @@ def load_locally_extreme_lookup(comparison_csv: Path) -> dict[str, dict]:
def load_solar_ghi_lookup(solar_ghi_csv: Path) -> dict[str, str]:
"""Map county FIPS codes to average daily GHI values from a county CSV."""
"""Map county FIPS codes to mean daily GHI values from a county CSV."""
if not solar_ghi_csv.exists():
return {}
fieldnames, rows = read_csv_rows(solar_ghi_csv)
if "avgSolarGhiKwhM2Day" not in fieldnames:
raise ValueError(f"{solar_ghi_csv} must include avgSolarGhiKwhM2Day.")
if "meanDailyGlobalHorizontalRadiationKwhM2Day" not in fieldnames:
raise ValueError(f"{solar_ghi_csv} must include meanDailyGlobalHorizontalRadiationKwhM2Day.")
if "county_fips" in fieldnames:
fips_field = "county_fips"
@@ -102,14 +101,14 @@ def load_solar_ghi_lookup(solar_ghi_csv: Path) -> dict[str, str]:
lookup: dict[str, str] = {}
for row in rows:
county_fips = normalize_fips(row.get(fips_field, ""))
raw_value = (row.get("avgSolarGhiKwhM2Day") or "").strip()
raw_value = (row.get("meanDailyGlobalHorizontalRadiationKwhM2Day") or "").strip()
if not county_fips or not raw_value:
continue
try:
float(raw_value)
except ValueError as error:
raise ValueError(
f"Invalid avgSolarGhiKwhM2Day value for county {county_fips}: {raw_value}"
f"Invalid meanDailyGlobalHorizontalRadiationKwhM2Day value for county {county_fips}: {raw_value}"
) from error
lookup[county_fips] = raw_value
@@ -193,23 +192,23 @@ def apply_locally_extreme_metric(
polygon_solar_value = polygon_solar_lookup.get(county_fips)
representative_solar_value = representative_solar_lookup.get(county_fips)
current_solar_value = (row.get("avgSolarGhiKwhM2Day") or "").strip()
current_solar_value = (row.get("meanDailyGlobalHorizontalRadiationKwhM2Day") or "").strip()
if polygon_solar_value:
row["avgSolarGhiKwhM2Day"] = polygon_solar_value
row["meanDailyGlobalHorizontalRadiationKwhM2Day"] = polygon_solar_value
row["source"] = replace_solar_source_tag(
row.get("source", ""),
"solar-ghi-polygon-archive-area-weighted",
)
polygon_solar_count += 1
elif representative_solar_value or current_solar_value:
row["avgSolarGhiKwhM2Day"] = representative_solar_value or current_solar_value
row["meanDailyGlobalHorizontalRadiationKwhM2Day"] = representative_solar_value or current_solar_value
row["source"] = replace_solar_source_tag(
row.get("source", ""),
"solar-ghi-representative-point",
)
representative_solar_count += 1
else:
row["avgSolarGhiKwhM2Day"] = ""
row["meanDailyGlobalHorizontalRadiationKwhM2Day"] = ""
row["source"] = replace_solar_source_tag(row.get("source", ""), "missing-solar-ghi")
missing_solar_count += 1
@@ -221,9 +220,9 @@ def apply_locally_extreme_metric(
print(f"Wrote {len(climate_rows)} rows to {out}")
print(f"Updated absoluteExtremeDays for {absolute_updated_count} rows.")
print(f"Left absoluteExtremeDays blank for {absolute_missing_count} rows without NOAA data.")
print(f"Updated avgSolarGhiKwhM2Day from polygon archives for {polygon_solar_count} rows.")
print(f"Updated meanDailyGlobalHorizontalRadiationKwhM2Day from polygon archives for {polygon_solar_count} rows.")
print(f"Kept representative-point solar fallback for {representative_solar_count} rows.")
print(f"Left avgSolarGhiKwhM2Day blank for {missing_solar_count} rows without solar data.")
print(f"Left meanDailyGlobalHorizontalRadiationKwhM2Day blank for {missing_solar_count} rows without solar data.")
def parse_args() -> argparse.Namespace: