Deduplicate the data-layer plumbing in climate.py and grading.py (#46)

climate.py repeated the same four mechanical patterns:
- the Open-Meteo daily params dict (3x) -> _om_daily_params(cell, **window)
- the doy attach line (6x) -> _with_doy
- makedirs + drop-doy + zstd to_parquet (3x) -> _write_cache
- the identical _to_frame/_nasa_to_frame tail (feels-like, valid-day
  filter, doy) -> _finalize_frame

grading.py encoded the tier tables twice — TEMP_BANDS/RAIN_BANDS plus the
hand-aligned _TEMP_LADDER/_RAIN_LADDER ('kept aligned' by comment). The
ladders are now derived from the bands (_ladder_from; verified
byte-identical to the old tables before landing), so tier boundaries have
exactly one definition. grade_range's inline dry-streak walk is replaced
with the existing dry_streaks(); its per-(doy,var) sample cache now also
memoizes the window mask per doy instead of recomputing it once per
metric (9x per day-of-year).

New tests pin the refactor: _to_frame schema/day-filter/missing-series
tolerance, the combined feels-like side selection, NASA unit conversions
and fill-sentinel handling, _om_daily_params windows, and _write_cache
stripping the derived doy column.

climate.py

@@ -180,6 +180,45 @@ def _derive_humidity(df: pd.DataFrame) -> pd.DataFrame:
     return df
 
 
+def _with_doy(df: pd.DataFrame) -> pd.DataFrame:
+    """(Re)attach the int16 day-of-year column the grading windows key on."""
+    df["doy"] = df["date"].dt.dayofyear.astype("int16")
+    return df
+
+
+def _write_cache(df: pd.DataFrame, path: str) -> None:
+    """Persist a daily record to the parquet cache — the raw record only, never
+    the derived doy column (it's recomputed at read time)."""
+    os.makedirs(CACHE_DIR, exist_ok=True)
+    df.drop(columns=["doy"], errors="ignore").to_parquet(
+        path, compression="zstd", index=False)
+
+
+def _om_daily_params(cell: dict, **window) -> dict:
+    """The Open-Meteo daily-request params every fetch shares — location, the
+    variable set, imperial units. The date/window selectors (start_date/end_date
+    or past_days/forecast_days) come in as kwargs."""
+    return {
+        "latitude": cell["center_lat"],
+        "longitude": cell["center_lon"],
+        "daily": DAILY_VARS,
+        "timezone": "auto",
+        "temperature_unit": "fahrenheit",
+        "precipitation_unit": "inch",
+        "wind_speed_unit": "mph",
+        **window,
+    }
+
+
+def _finalize_frame(df: pd.DataFrame) -> pd.DataFrame:
+    """Shared tail of every source→frame mapping: the combined feels-like, the
+    valid-day filter, and the day-of-year column. A usable climate day needs a
+    real high/low; other columns may be NaN and simply grade as None."""
+    df["feels"] = _combined_feels(df["fmax"], df["fmin"])
+    df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True)
+    return _with_doy(df)
+
+
 def _combined_feels(amax: pd.Series, amin: pd.Series) -> pd.Series:
     """One daily "feels like" value: the apparent-temperature extreme furthest from
     the comfort baseline — the heat-index high on warm days, the wind-chill low on
@@ -212,31 +251,14 @@ def _to_frame(daily: dict) -> pd.DataFrame:
     # The apparent (felt) high and low are kept as their own columns — graded
     # independently — alongside the combined `feels` (whichever side is further
     # from the fixed comfort baseline), which the weekly/day views still use.
-    amax = pd.to_numeric(pd.Series(col("apparent_temperature_max")), errors="coerce")
+    df["fmax"] = pd.to_numeric(pd.Series(col("apparent_temperature_max")), errors="coerce")
-    amin = pd.to_numeric(pd.Series(col("apparent_temperature_min")), errors="coerce")
+    df["fmin"] = pd.to_numeric(pd.Series(col("apparent_temperature_min")), errors="coerce")
-    df["fmax"] = amax
+    return _finalize_frame(df)
-    df["fmin"] = amin
-    df["feels"] = _combined_feels(amax, amin)
-    # Need a valid high/low to be a usable climate day; the rest may be NaN and are
-    # simply ungraded for that day (the percentile helpers return None).
-    df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True)
-    df["doy"] = df["date"].dt.dayofyear.astype("int16")
-    return df
 
 
 def _fetch_history(cell: dict) -> pd.DataFrame:
     end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).isoformat()
-    params = {
+    params = _om_daily_params(cell, start_date=START_DATE, end_date=end)
-        "latitude": cell["center_lat"],
-        "longitude": cell["center_lon"],
-        "start_date": START_DATE,
-        "end_date": end,
-        "daily": DAILY_VARS,
-        "timezone": "auto",
-        "temperature_unit": "fahrenheit",
-        "precipitation_unit": "inch",
-        "wind_speed_unit": "mph",
-    }
     r = _request(ARCHIVE_URL, params, 180, phase="history_fetch")
     return _to_frame(r.json()["daily"])
 
@@ -284,10 +306,7 @@ def _nasa_to_frame(param: dict) -> pd.DataFrame:
     # temperature outside its regime), mirroring the Open-Meteo columns.
     df["fmax"] = pd.Series(_heat_index(df["tmax"], df["humid"]), index=df.index)
     df["fmin"] = pd.Series(_wind_chill(df["tmin"], df["wind"]), index=df.index)
-    df["feels"] = _combined_feels(df["fmax"], df["fmin"])
+    return _finalize_frame(df)
-    df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True)
-    df["doy"] = df["date"].dt.dayofyear.astype("int16")
-    return df
 
 
 def _fetch_history_nasa(cell: dict) -> pd.DataFrame:
@@ -308,17 +327,7 @@ def _fetch_history_nasa(cell: dict) -> pd.DataFrame:
 
 def _fetch_history_range(cell: dict, start_date: str, end_date: str) -> pd.DataFrame:
     """Fetch just a date range of archive history (used to top up the recent tail)."""
-    params = {
+    params = _om_daily_params(cell, start_date=start_date, end_date=end_date)
-        "latitude": cell["center_lat"],
-        "longitude": cell["center_lon"],
-        "start_date": start_date,
-        "end_date": end_date,
-        "daily": DAILY_VARS,
-        "timezone": "auto",
-        "temperature_unit": "fahrenheit",
-        "precipitation_unit": "inch",
-        "wind_speed_unit": "mph",
-    }
     r = _request(ARCHIVE_URL, params, 60, phase="history_topup")
     return _to_frame(r.json()["daily"])
 
@@ -364,7 +373,7 @@ def _topup_tail(cell: dict, df: pd.DataFrame, path: str) -> pd.DataFrame:
         merged = (pd.concat([df, recent.drop(columns=["doy"], errors="ignore")])
                   .drop_duplicates(subset="date", keep="last")
                   .sort_values("date").reset_index(drop=True))
-        merged.to_parquet(path, compression="zstd", index=False)
+        _write_cache(merged, path)
         return merged
 
 
@@ -384,8 +393,7 @@ def load_cached_history(cell: dict) -> pd.DataFrame | None:
     hit = _read_history_cache(_cache_path(cell["id"]))
     if hit is None:
         return None
-    df = hit[0].copy()
+    df = _with_doy(hit[0].copy())
-    df["doy"] = df["date"].dt.dayofyear.astype("int16")
     _derive_humidity(df)
     return df
 
@@ -403,9 +411,7 @@ def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]:
         df, age_s = hit
         if age_s > HISTORY_TOPUP_INTERVAL:
             df = _topup_tail(cell, df, path)   # refresh just the recent days
-        df = df.copy()
+        return _with_doy(df.copy()), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
-        df["doy"] = df["date"].dt.dayofyear.astype("int16")
-        return df, {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
 
     global _archive_cooldown_until
 
@@ -413,14 +419,12 @@ def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]:
         hit = _read_history_cache(path)   # another thread may have populated it while we waited
         if hit is not None:
             df, age_s = hit
-            df["doy"] = df["date"].dt.dayofyear.astype("int16")
+            return _with_doy(df), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
-            return df, {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
 
         stale = pd.read_parquet(path) if os.path.exists(path) else None
 
         def _serve_stale():
-            stale["doy"] = stale["date"].dt.dayofyear.astype("int16")
+            return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None}
-            return stale, {"cached": True, "stale": True, "cache_age_days": None}
 
         # Fetch. Open-Meteo is primary (richer: gusts + apparent temp); NASA POWER is
         # the backup when Open-Meteo is unavailable (network error or its daily limit).
@@ -446,9 +450,8 @@ def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]:
             raise WeatherUnavailable(limit_message(_archive_limit_daily),
                                      daily=_archive_limit_daily)
 
-        os.makedirs(CACHE_DIR, exist_ok=True)
         # Store the raw record; percentiles are derived at request time.
-        df.drop(columns=["doy"]).to_parquet(path, compression="zstd", index=False)
+        _write_cache(df, path)
         return df, {"cached": False, "cache_age_days": 0, "source": source}
 
 
@@ -489,9 +492,7 @@ def _load_recent_forecast(cell: dict) -> pd.DataFrame:
     if os.path.exists(path):
         age_h = (time.time() - os.path.getmtime(path)) / 3600.0
         if age_h < FORECAST_TTL_HOURS:
-            df = pd.read_parquet(path)
+            return _with_doy(pd.read_parquet(path))
-            df["doy"] = df["date"].dt.dayofyear.astype("int16")
-            return df
 
     params = {
         "latitude": cell["center_lat"],
@@ -514,8 +515,7 @@ def _load_recent_forecast(cell: dict) -> pd.DataFrame:
             raise WeatherUnavailable(limit_message(daily), daily=daily) from e
         raise
     df = _to_frame(r.json()["daily"])
-    os.makedirs(CACHE_DIR, exist_ok=True)
+    _write_cache(df, path)
-    df.drop(columns=["doy"]).to_parquet(path, compression="zstd", index=False)
     return df
 
 

grading.py

@@ -5,6 +5,8 @@ historical day whose day-of-year is within +/- `HALF_WINDOW` days of it (wrappin
 around the year end). An observed value is then placed on that distribution as an
 empirical percentile and mapped to a human-readable grade.
 """
+import warnings
+
 import numpy as np
 import pandas as pd
 
@@ -58,34 +60,25 @@ RAIN_BANDS = [
 ]
 
 
-# Tier ladders for the single-day detail view: each tier and the two percentile
+def _ladder_from(bands, bottom_lo=None):
-# marks that bound it. Kept aligned with TEMP_BANDS / RAIN_BANDS (same labels and
+    """Derive the single-day detail view's tier ladder from a band table, so the
-# css classes) so the UI names and colors stay consistent across every view.
+    two can never drift apart: each tier as (class, label, printable percentile
-# (class, label, percentile-range label, lower-bound pct, upper-bound pct)
+    range, lower-bound pct, upper-bound pct). The top tier is open-ended (>99);
-_TEMP_LADDER = [
+    the bottom runs down from the 1st percentile — ``bottom_lo`` marks its lower
-    ("rec-hot", "Near Record", ">99%", 99, None),
+    bound (None for temperature; 0 for rain, whose lightest tier bottoms out at
-    ("very-hot", "Very High", "90–99%", 90, 99),
+    the smallest measured rain day)."""
-    ("hot", "High", "75–90%", 75, 90),
+    top_thr, top_label, top_css = bands[0]
-    ("warm", "Above Normal", "60–75%", 60, 75),
+    out = [(top_css, top_label, f">{top_thr}%", top_thr, None)]
-    ("normal", "Normal", "40–60%", 40, 60),
+    for (thr, label, css), (prev_thr, _, _) in zip(bands[1:-1], bands[:-2]):
-    ("cool", "Below Normal", "25–40%", 25, 40),
+        out.append((css, label, f"{thr}–{prev_thr}%", thr, prev_thr))
-    ("cold", "Low", "10–25%", 10, 25),
+    edge = bands[-2][0]
-    ("very-cold", "Very Low", "1–10%", 1, 10),
+    out.append((bands[-1][2], bands[-1][1], f"<{edge}%", bottom_lo, edge))
-    ("rec-cold", "Near Record", "<1%", None, 1),
+    return out
-]
+
-# Rain tiers are on the rain-day-only percentile scale (see _grade_precip). The
+
-# lightest tier's lower bound is the smallest measured rain day (marked with 0).
+_TEMP_LADDER = _ladder_from(TEMP_BANDS)
-_RAIN_LADDER = [
+# Rain tiers are on the rain-day-only percentile scale (see _grade_precip).
-    ("wet-9", "Extreme", ">99%", 99, None),
+_RAIN_LADDER = _ladder_from(RAIN_BANDS, bottom_lo=0)
-    ("wet-8", "Very Heavy", "90–99%", 90, 99),
-    ("wet-7", "Heavy", "75–90%", 75, 90),
-    ("wet-6", "Mod–Heavy", "60–75%", 60, 75),
-    ("wet-5", "Moderate", "40–60%", 40, 60),
-    ("wet-4", "Light–Mod", "25–40%", 25, 40),
-    ("wet-3", "Light", "10–25%", 10, 25),
-    ("wet-2", "Very Light", "1–10%", 1, 10),
-    ("wet-1", "Trace", "<1%", 0, 1),
-]
 
 
 def _band(pct: float, bands) -> tuple[str, str]:
@@ -222,12 +215,16 @@ def grade_range(df: pd.DataFrame, start, end) -> list[dict]:
     full = df.sort_values("date")
     doys_all = full["doy"].values
     cols = {v: full[v].values for v in CLIMO_METRICS if v in full.columns}
+    masks: dict[int, np.ndarray] = {}
     cache: dict[tuple[int, str], np.ndarray] = {}
 
     def samples(doy: int, var: str) -> np.ndarray:
         key = (doy, var)
         if key not in cache:
-            arr = cols[var][window_mask(doys_all, doy)]
+            mask = masks.get(doy)
+            if mask is None:
+                mask = masks[doy] = window_mask(doys_all, doy)
+            arr = cols[var][mask]
             cache[key] = arr[~np.isnan(arr)]
         return cache[key]
 
@@ -236,18 +233,12 @@ def grade_range(df: pd.DataFrame, start, end) -> list[dict]:
     # the first shown day is exact even when a dry spell straddles the window start.
     # A fixed 14-day lookback would be the bare minimum but still undercounts longer
     # droughts (the record has 25-day dry streaks); using the full history (already
-    # cached per cell) is both correct for any streak length and free. NaN precip
+    # cached per cell) is both correct for any streak length and free.
-    # compares False against the threshold, so it counts as a dry day.
     if full["date"].min() > start - pd.Timedelta(days=DSR_LOOKBACK_MIN_DAYS):
         # Should never happen (history starts in 1980); guards against a future
         # change that trims history and would silently truncate streaks.
-        import warnings
         warnings.warn("dry-streak lookback shorter than the 14-day minimum", stacklevel=2)
-    dsr_map: dict[str, int] = {}
+    dsr_map = dry_streaks(full["date"].values, cols["precip"])
-    streak = 0
-    for d, p in zip(full["date"].values, cols["precip"]):
-        streak = 0 if p >= RAIN_THRESHOLD else streak + 1
-        dsr_map[pd.Timestamp(d).date().isoformat()] = streak
 
     sub = full[(full["date"] >= start) & (full["date"] <= end)]
     out = []

tests/test_climate.py

@@ -0,0 +1,82 @@
+"""The source→frame mappings and cache-write path (pure parts of climate.py —
+no network)."""
+import numpy as np
+import pandas as pd
+
+import climate
+
+
+def _om_daily(n=3):
+    """A minimal Open-Meteo daily response."""
+    return {
+        "time": [f"2026-06-{d:02d}" for d in range(1, n + 1)],
+        "temperature_2m_max": [80.0, 90.0, None],
+        "temperature_2m_min": [60.0, 70.0, 55.0],
+        "precipitation_sum": [0.0, 0.25, 0.1],
+        "wind_speed_10m_max": [10.0, 20.0, 5.0],
+        "wind_gusts_10m_max": [15.0, 30.0, 8.0],
+        "apparent_temperature_max": [82.0, 95.0, 70.0],
+        "apparent_temperature_min": [58.0, 68.0, 50.0],
+        "relative_humidity_2m_mean": [50.0, 60.0, 70.0],
+    }
+
+
+def test_to_frame_schema_and_day_filter():
+    df = climate._to_frame(_om_daily())
+    # The None tmax day is dropped: a usable climate day needs a real high/low.
+    assert len(df) == 2
+    assert df["doy"].dtype == np.int16
+    assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind", "gust",
+                               "humid", "fmax", "fmin", "feels", "doy"}
+
+
+def test_to_frame_tolerates_missing_series():
+    daily = _om_daily()
+    del daily["wind_gusts_10m_max"], daily["relative_humidity_2m_mean"]
+    df = climate._to_frame(daily)
+    assert df["gust"].isna().all() and df["humid"].isna().all()
+    assert df["tmax"].notna().all()   # required series unaffected
+
+
+def test_combined_feels_picks_the_extreme_side():
+    df = climate._to_frame(_om_daily())
+    # Day 2: fmax 95 is 30 past the 65°F comfort point vs fmin 68 only 3 below —
+    # the hot side wins; both days here are hot-side days.
+    assert df.loc[df["date"] == "2026-06-02", "feels"].item() == 95.0
+
+
+def test_nasa_to_frame_converts_units_and_fills():
+    param = {
+        "T2M_MAX": {"20260601": 30.0, "20260602": -999.0},   # °C; -999 = missing
+        "T2M_MIN": {"20260601": 20.0, "20260602": 15.0},
+        "PRECTOTCORR": {"20260601": 25.4, "20260602": 0.0},  # mm
+        "WS10M_MAX": {"20260601": 10.0, "20260602": 5.0},    # m/s
+        "RH2M": {"20260601": 50.0, "20260602": 60.0},
+    }
+    df = climate._nasa_to_frame(param)
+    assert len(df) == 1                                   # the missing-tmax day dropped
+    row = df.iloc[0]
+    assert row["tmax"] == 86.0                            # 30°C
+    assert row["precip"] == 1.0                           # 25.4mm = 1in
+    assert round(row["wind"], 1) == 22.4                  # 10 m/s in mph
+    assert np.isnan(row["gust"])                          # POWER has no gusts
+    assert row["fmax"] > row["tmax"]                      # ≥80°F: the NWS heat index applies
+
+
+def test_om_daily_params_carries_the_window():
+    cell = {"center_lat": 47.6, "center_lon": -122.3}
+    p = climate._om_daily_params(cell, start_date="2026-01-01", end_date="2026-02-01")
+    assert p["latitude"] == 47.6 and p["daily"] == climate.DAILY_VARS
+    assert p["temperature_unit"] == "fahrenheit"
+    assert p["start_date"] == "2026-01-01" and p["end_date"] == "2026-02-01"
+    p2 = climate._om_daily_params(cell, past_days=25, forecast_days=8)
+    assert p2["past_days"] == 25 and "start_date" not in p2
+
+
+def test_write_cache_strips_the_derived_doy(tmp_path):
+    df = climate._to_frame(_om_daily())
+    path = str(tmp_path / "cell.parquet")
+    climate._write_cache(df, path)
+    stored = pd.read_parquet(path)
+    assert "doy" not in stored.columns
+    assert len(stored) == len(df)