From 38a39df6abfd299826fa1d6ec7ab6590f0451369 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Sat, 11 Jul 2026 12:43:41 -0700 Subject: [PATCH] Extract payload builders into views.py; load places index at startup (#42) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit app.py had grown three co-resident strata: HTTP/caching plumbing, payload assembly, and search policy. This moves the payload layer — the four build_* functions, cal_span clamping, hist_end, PAYLOAD_VER, NullRun and their helpers — into a new views.py with no web dependencies (pure moves, public names). app.py keeps routing, ETag/derived-store plumbing, and page serving; migrate.py imports views instead of reaching into app's privates. That import previously constructed the whole FastAPI app and — because places.start_loading() ran at import time — kicked off a background GeoNames download from an offline batch script. The index load now hangs off the app's lifespan hook, so it fires when a server starts, not when the module is imported (tests, migrate). New tests: cal_span clamping (months-back default, record bounds, 2-year cap, inversion), builder shapes (grade window ordering, day obs from the recent bundle, forecast future-only), a regression test that build_day degrades to climatology when the recent fetch fails, and a subprocess layering guard that importing views/migrate pulls in neither FastAPI nor the app module and starts no index download. --- app.py | 260 ++++++-------------------------------------- migrate.py | 10 +- tests/test_views.py | 96 ++++++++++++++++ views.py | 215 ++++++++++++++++++++++++++++++++++++ 4 files changed, 347 insertions(+), 234 deletions(-) create mode 100644 tests/test_views.py create mode 100644 views.py diff --git a/app.py b/app.py index 7d5a2f1..a8c40d4 100644 --- a/app.py +++ b/app.py @@ -15,39 +15,19 @@ from fastapi.staticfiles import StaticFiles import audit import climate -import grading import grid import places import store +import views FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend") -# Warm the local place-name index (/suggest's typo tolerance) in the -# background; the app boots and serves fine without it — suggestions just fall -# back to the upstream geocoder until it's ready. -places.start_loading() - # Everything (pages, assets, API) is served under this base path so the app can # live at https:///thermograph/ behind a shared domain. Override with the # THERMOGRAPH_BASE env var. The frontend uses base-relative URLs, so it follows # this automatically without hardcoding the prefix. BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/") -# Observed values pulled from a daily record row for grading. Includes the -# temperature-scale metrics (tmax/tmin/feels/wind/gust) plus precip; a column may -# be absent on an older cache, so only carry the ones present. -OBS_COLS = ("tmax", "tmin", "precip", "feels", "humid", "wind", "gust") - -# Bump when any response payload shape changes (new metrics, renamed fields…). -# The version is part of every derived-store validity token, so one bump -# atomically orphans all pre-upgrade cached payloads instead of letting a stale -# row whose history_end happens to match keep serving the old shape. -PAYLOAD_VER = "p1" - - -def _obs_from_row(row) -> dict: - return {k: row[k] for k in OBS_COLS if k in row} - def _weather_fetch_error(e) -> HTTPException: """A clean, retryable message for a rate limit; the raw error otherwise.""" @@ -68,28 +48,18 @@ def _weather_fetch_error(e) -> HTTPException: return HTTPException(status_code=502, detail=f"weather data fetch failed: {e}") -def _grade_rows(history, rows_df) -> list[dict]: - """Grade each daily row against its own ±7-day climatology window.""" - return [grading.grade_day(history, row["date"], _obs_from_row(row)) - for _, row in rows_df.iterrows()] +@contextlib.asynccontextmanager +async def _lifespan(app): + # Warm the local place-name index (/suggest's typo tolerance) in the + # background; the app boots and serves fine without it — suggestions just + # fall back to the upstream geocoder until it's ready. A server-startup + # hook (not import time) so offline importers (tests, the migrate script's + # dependencies) don't kick off a GeoNames download. + places.start_loading() + yield -def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None: - """Attach `dsr` (days since last measurable rain) to each graded day, computed - over a combined, de-duplicated precip series so streaks stay continuous across - the history / recent / forecast sources.""" - frames = [f[["date", "precip"]] for f in precip_frames if f is not None and not f.empty] - if not frames: - return - combined = (pd.concat(frames) - .drop_duplicates(subset="date", keep="last") - .sort_values("date")) - dsr_map = grading.dry_streaks(combined["date"].values, combined["precip"].values) - for g in graded: - g["dsr"] = dsr_map.get(g["date"]) - - -app = FastAPI(title="Thermograph", version="0.2.0") +app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan) # Compress every sizeable response (the 2-year calendar JSON shrinks ~6-8×). # Applies to API JSON and static assets alike; tiny responses are left alone. @@ -120,23 +90,8 @@ async def revalidate_static(request, call_next): # from what those return, so a cached payload expires exactly when its inputs # change and never before. The same tokens double as ETags: a client sending # If-None-Match gets an empty 304 without the payload even being loaded. - -class _NullRun: - """audit.RunAudit stand-in for offline callers (the migrate script).""" - - def set(self, **kw): - return self - - def phase(self, name): - return contextlib.nullcontext() - - -def _hist_end(history) -> str: - """Last day in the cell's archive record — the freshness token for everything - derived purely from history. Advances via the hourly tail top-up inside - climate.get_history, which cached payloads follow automatically.""" - return pd.Timestamp(history["date"].max()).date().isoformat() - +# The payloads themselves are assembled in views.py, shared with the offline +# migrate script. def _etag_for(kind: str, cell_id: str, key: str, token: str) -> str: """Deterministic weak ETag from a payload's identity + validity token. Weak @@ -166,159 +121,6 @@ def _json_response(body: bytes, etag: str | None = None) -> Response: return Response(content=body, media_type="application/json", headers=headers) -# --- payload builders -------------------------------------------------------- -# Pure "inputs → response dict" functions shared by the per-view endpoints, the -# /cell bundle, and the offline migrate script. HTTP semantics (audit runs, cache -# lookups, etags) stay in the endpoints; `run` is the audit run or None. - -def _build_grade(cell, target, days, history, recent, cache_meta, place, run=None, after=14) -> dict: - """/grade payload: a window of days centered on the target — `days` of history - before it, the target itself, then up to `after` days after it — plus the - target day's climatology summary. Days after the target are observed when they - are already in the past and forecast when they run into the future, so the - weekly view can frame two weeks of history around the orange target marker and - trail off into up to 14 days of observations / forecast after it. The forecast - only reaches ~7 days out, so a recent target naturally yields some observed days - plus 1-7 forecast days, while an older target fills the whole 14 with real obs.""" - run = run or _NullRun() - with run.phase("grading"): - lo = target - pd.Timedelta(days=days) - hi = target + pd.Timedelta(days=after) - # Build the window from both sources. The recent+forecast bundle covers the - # last few weeks plus the forward forecast (the only source for future days); - # the archive reaches decades back for targets older than that bundle. Prefer - # the bundle row for any given date, filling the rest from the archive. - rwin = recent[(recent["date"] >= lo) & (recent["date"] <= hi)] - hwin = history[(history["date"] >= lo) & (history["date"] <= hi)] - hwin = hwin[~hwin["date"].isin(set(rwin["date"]))] - window = pd.concat([rwin, hwin]).sort_values("date") - graded = _grade_rows(history, window) - _attach_dry_streaks(graded, history, recent) - graded.reverse() # newest first for display - climo = grading.climatology(history, int(target.dayofyear)) - - # Summarize what this run actually covered. A fresh history fetch pulls the - # full ~45-year archive (run_type "full"); a cache hit only fetched the - # recent window (run_type "partial"). - years = pd.to_datetime(history["date"]).dt.year - full = not cache_meta.get("cached", False) - run.set( - run_type="full" if full else "partial", - history_source="fetch" if full else "cache", - history_rows=int(len(history)), - history_years=int(years.max() - years.min() + 1), - history_span=[int(years.min()), int(years.max())], - cache_age_days=cache_meta.get("cache_age_days"), - graded_days=len(graded), - place_found=place is not None, - ) - return { - "cell": cell, - "place": place, - "target_date": target.date().isoformat(), - "cache": cache_meta, - "climatology": climo, - "recent": graded, - } - - -CAL_MAX_SPAN_DAYS = 732 # ~2 years — the most a single calendar request will grade - # (the frontend splits longer spans into 2-year chunks) - - -def _cal_span(history, start, end, months) -> tuple[pd.Timestamp, pd.Timestamp]: - """Clamp a requested calendar range to the available record (≤ ~2 years). - - Without a start, defaults to whole months back landing on the SAME - day-of-month as the end, so the two range endpoints line up - (e.g. 2024-06-29 → 2026-06-29).""" - first = pd.Timestamp(history["date"].min()).normalize() - last = pd.Timestamp(history["date"].max()).normalize() - end_ts = min(pd.Timestamp(end).normalize(), last) if end else last - if start: - start_ts = pd.Timestamp(start).normalize() - else: - start_ts = (end_ts - pd.DateOffset(months=months)).normalize() - # Cap the graded span at ~2 years, and keep it inside the available record. - min_start = end_ts - pd.Timedelta(days=CAL_MAX_SPAN_DAYS - 1) - start_ts = max(start_ts, min_start, first) - start_ts = min(start_ts, end_ts) - return start_ts, end_ts - - -def _build_calendar(cell, history, start_ts, end_ts, months, place, run=None) -> dict: - """/calendar payload: every day in [start_ts, end_ts] graded, compact shape.""" - run = run or _NullRun() - with run.phase("grading"): - days = grading.grade_range(history, start_ts, end_ts) - run.set(graded_days=len(days)) - return { - "api_version": "v2", - "cell": cell, - "place": place, - "range": {"start": start_ts.date().isoformat(), "end": end_ts.date().isoformat()}, - "months": months, - "days": days, - } - - -def _build_day(cell, history, target, place, run=None, recent=None) -> dict: - """/day payload: the full percentile breakdown for one day. Observed values - prefer the archive record; a date newer than it comes from the recent window - (passed pre-loaded by the bundle path, fetched here otherwise).""" - run = run or _NullRun() - last = pd.Timestamp(history["date"].max()).normalize() - - obs = None - hit = history[history["date"] == target] - if not hit.empty: - obs = _obs_from_row(hit.iloc[0]) - elif target > last: - try: - if recent is None: - with run.phase("recent"): - recent = climate.get_recent_forecast(cell) - rr = recent[recent["date"] == target] - if not rr.empty: - obs = _obs_from_row(rr.iloc[0]) - except Exception: # noqa: BLE001 - detail page still works from climatology alone - obs = None - - with run.phase("detail"): - detail = grading.day_detail(history, target, obs) - run.set(has_observation=obs is not None) - return { - "api_version": "v2", - "cell": cell, - "place": place, - "latest": last.date().isoformat(), - "detail": detail, - } - - -def _build_forecast(cell, days, history, fc, today, place, run=None) -> dict: - """/forecast payload: the next `days` forecast days graded — same shape as - /grade so the frontend renders it identically, furthest-out day first.""" - run = run or _NullRun() - # Strictly future days (tomorrow onward), earliest→latest, capped at `days`. - future = fc[fc["date"] > today].sort_values("date").head(days) - with run.phase("grading"): - graded = _grade_rows(history, future) - _attach_dry_streaks(graded, history, fc) - graded.reverse() # furthest-out first, so the chart reverses to L→R chronological - climo = grading.climatology(history, int(today.dayofyear)) - run.set(graded_days=len(graded), place_found=place is not None) - return { - "api_version": "v2", - "cell": cell, - "place": place, - "target_date": today.date().isoformat(), - "forecast": True, - "climatology": climo, - "recent": graded, - } - - # --- endpoints ---------------------------------------------------------------- def api_geocode(q: str = Query(..., min_length=1)): @@ -465,7 +267,7 @@ def api_grade( raise HTTPException(status_code=404, detail="No historical data for this cell.") key = f"{target.date().isoformat()}:{days}:{after}" - token = f"{PAYLOAD_VER}:{_hist_end(history)}:{climate.recent_stamp(cell['id'])}" + token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}:{climate.recent_stamp(cell['id'])}" etag = _etag_for("grade", cell["id"], key, token) if _not_modified(request, etag): run.set(run_type="cache", not_modified=True) @@ -477,7 +279,7 @@ def api_grade( with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = _build_grade(cell, target, days, history, recent, cache_meta, place, run, after=after) + payload = views.build_grade(cell, target, days, history, recent, cache_meta, place, run, after=after) return _json_response(store.put_payload("grade", cell["id"], key, token, payload), etag) @@ -516,9 +318,9 @@ def api_calendar( if history.empty: raise HTTPException(status_code=404, detail="No historical data for this cell.") - start_ts, end_ts = _cal_span(history, start, end, months) + start_ts, end_ts = views.cal_span(history, start, end, months) key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:{months}" - token = f"{PAYLOAD_VER}:{_hist_end(history)}" + token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" etag = _etag_for("calendar", cell["id"], key, token) if _not_modified(request, etag): run.set(run_type="cache", not_modified=True) @@ -530,7 +332,7 @@ def api_calendar( with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = _build_calendar(cell, history, start_ts, end_ts, months, place, run) + payload = views.build_calendar(cell, history, start_ts, end_ts, months, place, run) full = not cache_meta.get("cached", False) run.set(run_type="full" if full else "partial", history_source="fetch" if full else "cache") @@ -578,7 +380,7 @@ def api_day( run.set(target_date=target.date().isoformat()) key = target.date().isoformat() - token = f"{PAYLOAD_VER}:{_hist_end(history)}" + token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" if target > last: token += f":h{int(time.time() // 3600)}" # obs from the hourly recent bundle etag = _etag_for("day", cell["id"], key, token) @@ -592,7 +394,7 @@ def api_day( with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = _build_day(cell, history, target, place, run) + payload = views.build_day(cell, history, target, place, run) full = not cache_meta.get("cached", False) run.set(run_type="full" if full else "partial", history_source="fetch" if full else "cache") @@ -630,7 +432,7 @@ def api_forecast( today = pd.Timestamp(datetime.date.today()) key = f"{today.date().isoformat()}:{days}" - token = f"{PAYLOAD_VER}:{_hist_end(history)}:{climate.recent_stamp(cell['id'])}" + token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}:{climate.recent_stamp(cell['id'])}" etag = _etag_for("forecast", cell["id"], key, token) if _not_modified(request, etag): run.set(run_type="cache", not_modified=True) @@ -642,7 +444,7 @@ def api_forecast( with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - payload = _build_forecast(cell, days, history, fc, today, place, run) + payload = views.build_forecast(cell, days, history, fc, today, place, run) run.set(run_type="partial") return _json_response(store.put_payload("forecast", cell["id"], key, token, payload), etag) @@ -694,7 +496,7 @@ def api_cell( raise HTTPException(status_code=404, detail="No historical data for this cell.") cid = cell["id"] - hist_end = _hist_end(history) + hend = views.hist_end(history) last = pd.Timestamp(history["date"].max()).normalize() with run.phase("reverse_geocode"): @@ -711,35 +513,35 @@ def api_cell( return {"etag": etag, "data": data} slices = {} - hist_token = f"{PAYLOAD_VER}:{hist_end}" + hist_token = f"{views.PAYLOAD_VER}:{hend}" # Calendar: the default last-24-months span — what the calendar view (and # the cross-view prefetch) requests first. - start_ts, end_ts = _cal_span(history, None, None, 24) + start_ts, end_ts = views.cal_span(history, None, None, 24) cal_key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:24" slices["calendar"] = slice_for( "calendar", cal_key, hist_token, - lambda: _build_calendar(cell, history, start_ts, end_ts, 24, place, run)) + lambda: views.build_calendar(cell, history, start_ts, end_ts, 24, place, run)) if prefetch: # Latest archived day: its observation comes from history alone, so # it's buildable without the (skipped) recent bundle. slices["day"] = slice_for( "day", last.date().isoformat(), hist_token, - lambda: _build_day(cell, history, last, place, run)) + lambda: views.build_day(cell, history, last, place, run)) else: - rf_token = f"{PAYLOAD_VER}:{hist_end}:{climate.recent_stamp(cid)}" + rf_token = f"{views.PAYLOAD_VER}:{hend}:{climate.recent_stamp(cid)}" slices["grade"] = slice_for( "grade", f"{today.date().isoformat()}:14:14", rf_token, - lambda: _build_grade(cell, today, 14, history, recent, cache_meta, place, run, after=14)) + lambda: views.build_grade(cell, today, 14, history, recent, cache_meta, place, run, after=14)) slices["forecast"] = slice_for( "forecast", f"{today.date().isoformat()}:7", rf_token, - lambda: _build_forecast(cell, 7, history, recent, today, place, run)) + lambda: views.build_forecast(cell, 7, history, recent, today, place, run)) day_token = hist_token if today > last: day_token += f":h{int(time.time() // 3600)}" # obs from the hourly recent bundle slices["day"] = slice_for( "day", today.date().isoformat(), day_token, - lambda: _build_day(cell, history, today, place, run, recent=recent)) + lambda: views.build_day(cell, history, today, place, run, recent=recent)) # The bundle's identity is the combination of its slices' identities. etag = _etag_for("cell", cid, "bundle" + (":p" if prefetch else ""), diff --git a/migrate.py b/migrate.py index 9b2a1a1..f879275 100644 --- a/migrate.py +++ b/migrate.py @@ -26,7 +26,7 @@ import pandas as pd import climate import grid import store -from app import PAYLOAD_VER, _build_calendar, _build_day, _cal_span, _hist_end +import views def migrate() -> int: @@ -54,8 +54,8 @@ def migrate() -> int: skipped += 1 continue - token = f"{PAYLOAD_VER}:{_hist_end(history)}" - start_ts, end_ts = _cal_span(history, None, None, 24) + token = f"{views.PAYLOAD_VER}:{views.hist_end(history)}" + start_ts, end_ts = views.cal_span(history, None, None, 24) cal_key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:24" last = pd.Timestamp(history["date"].max()).normalize() day_key = last.date().isoformat() @@ -73,10 +73,10 @@ def migrate() -> int: if not have_cal: store.put_payload("calendar", cell_id, cal_key, token, - _build_calendar(cell, history, start_ts, end_ts, 24, place)) + views.build_calendar(cell, history, start_ts, end_ts, 24, place)) if not have_day: store.put_payload("day", cell_id, day_key, token, - _build_day(cell, history, last, place)) + views.build_day(cell, history, last, place)) built += 1 print(f" {cell_id}: materialized ({place or 'no label'})") diff --git a/tests/test_views.py b/tests/test_views.py new file mode 100644 index 0000000..c4d54e7 --- /dev/null +++ b/tests/test_views.py @@ -0,0 +1,96 @@ +"""The payload layer: pure builders, span clamping, and the layering guarantee +that offline callers (migrate) can use it without dragging in the web stack.""" +import os +import subprocess +import sys + +import pandas as pd +import pytest + +import climate +import views + +CELL = {"id": "1642_-4223", "center_lat": 47.6087, "center_lon": -122.29377} + + +def test_views_and_migrate_import_without_the_web_stack(): + """migrate.py must stay runnable offline: importing the payload layer may + not construct the FastAPI app or start the places-index download.""" + code = ("import sys; import views, migrate; " + "assert 'fastapi' not in sys.modules, 'views/migrate pulled in FastAPI'; " + "assert 'app' not in sys.modules, 'views/migrate imported the web app'; " + "import places; assert places._load_started is False") + backend = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + subprocess.run([sys.executable, "-c", code], cwd=backend, check=True) + + +# ---- cal_span clamping --------------------------------------------------------- + +def _hist(start, end): + df = pd.DataFrame({"date": pd.date_range(start, end, freq="D")}) + return df + + +def test_cal_span_defaults_to_months_back_same_day_of_month(): + start_ts, end_ts = views.cal_span(_hist("2020-01-01", "2026-06-29"), None, None, 24) + assert end_ts == pd.Timestamp("2026-06-29") + assert start_ts == pd.Timestamp("2024-06-29") + + +def test_cal_span_clamps_to_the_record(): + h = _hist("2025-03-01", "2026-06-29") # record shorter than the 2-year cap + start_ts, end_ts = views.cal_span(h, "2020-01-01", None, 24) + assert start_ts == pd.Timestamp("2025-03-01") # can't start before the record + _, end_ts = views.cal_span(h, None, "2030-01-01", 24) + assert end_ts == pd.Timestamp("2026-06-29") # nor end past it + + +def test_cal_span_caps_at_two_years(): + start_ts, end_ts = views.cal_span(_hist("2018-01-01", "2026-06-29"), + "2018-01-01", "2026-06-29", 24) + assert (end_ts - start_ts).days == views.CAL_MAX_SPAN_DAYS - 1 + + +def test_cal_span_never_inverts(): + start_ts, end_ts = views.cal_span(_hist("2020-01-01", "2026-06-29"), + "2026-06-01", "2021-01-01", 24) + assert start_ts == end_ts + + +# ---- builders ------------------------------------------------------------------- + +def test_build_grade_window_and_shape(history, recent): + target = pd.Timestamp.today().normalize() + payload = views.build_grade(CELL, target, 14, history, recent, + {"cached": True}, "Testville") + assert payload["target_date"] == target.date().isoformat() + days = [d["date"] for d in payload["recent"]] + assert days == sorted(days, reverse=True) # newest first + assert payload["climatology"]["tmax"] is not None + assert all(d["dsr"] is not None for d in payload["recent"]) + + +def test_build_day_pulls_future_obs_from_recent(history, recent): + today = pd.Timestamp.today().normalize() + payload = views.build_day(CELL, history, today, "Testville", recent=recent) + assert payload["detail"]["date"] == today.date().isoformat() + assert payload["detail"]["metrics"]["tmax"]["obs"] is not None + + +def test_build_day_survives_recent_fetch_failure(history, monkeypatch): + def boom(cell): + raise RuntimeError("upstream down") + monkeypatch.setattr(climate, "get_recent_forecast", boom) + today = pd.Timestamp.today().normalize() + payload = views.build_day(CELL, history, today, "Testville") + assert payload["detail"]["metrics"]["tmax"]["obs"] is None # climatology only + assert payload["detail"]["metrics"]["tmax"]["ladder"] is not None + + +def test_build_forecast_only_future_days(history, recent): + today = pd.Timestamp.today().normalize() + payload = views.build_forecast(CELL, 7, history, recent, today, None) + days = [d["date"] for d in payload["recent"]] + assert days == sorted(days, reverse=True) # furthest-out first + assert min(days) > today.date().isoformat() + assert payload["forecast"] is True diff --git a/views.py b/views.py new file mode 100644 index 0000000..767f55a --- /dev/null +++ b/views.py @@ -0,0 +1,215 @@ +"""Payload builders — pure "inputs → response dict" assembly for every view. + +This is the layer between the data modules (climate/grading/grid) and any +delivery mechanism: the FastAPI endpoints, the /cell bundle, and the offline +migrate script all build payloads here, so a payload's shape has exactly one +definition and offline callers never have to import the web stack. HTTP +semantics (audit runs, derived-store lookups, ETags) stay with the callers; +``run`` is an audit.RunAudit or None. +""" +import contextlib + +import pandas as pd + +import climate +import grading + +# Observed values pulled from a daily record row for grading. Includes the +# temperature-scale metrics (tmax/tmin/feels/wind/gust) plus precip; a column may +# be absent on an older cache, so only carry the ones present. +OBS_COLS = ("tmax", "tmin", "precip", "feels", "humid", "wind", "gust") + +# Bump when any response payload shape changes (new metrics, renamed fields…). +# The version is part of every derived-store validity token, so one bump +# atomically orphans all pre-upgrade cached payloads instead of letting a stale +# row whose history_end happens to match keep serving the old shape. +PAYLOAD_VER = "p1" + +CAL_MAX_SPAN_DAYS = 732 # ~2 years — the most a single calendar request will grade + # (the frontend splits longer spans into 2-year chunks) + + +class NullRun: + """audit.RunAudit stand-in for offline callers (the migrate script).""" + + def set(self, **kw): + return self + + def phase(self, name): + return contextlib.nullcontext() + + +def hist_end(history) -> str: + """Last day in the cell's archive record — the freshness token for everything + derived purely from history. Advances via the hourly tail top-up inside + climate.get_history, which cached payloads follow automatically.""" + return pd.Timestamp(history["date"].max()).date().isoformat() + + +def _obs_from_row(row) -> dict: + return {k: row[k] for k in OBS_COLS if k in row} + + +def _grade_rows(history, rows_df) -> list[dict]: + """Grade each daily row against its own ±7-day climatology window.""" + return [grading.grade_day(history, row["date"], _obs_from_row(row)) + for _, row in rows_df.iterrows()] + + +def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None: + """Attach `dsr` (days since last measurable rain) to each graded day, computed + over a combined, de-duplicated precip series so streaks stay continuous across + the history / recent / forecast sources.""" + frames = [f[["date", "precip"]] for f in precip_frames if f is not None and not f.empty] + if not frames: + return + combined = (pd.concat(frames) + .drop_duplicates(subset="date", keep="last") + .sort_values("date")) + dsr_map = grading.dry_streaks(combined["date"].values, combined["precip"].values) + for g in graded: + g["dsr"] = dsr_map.get(g["date"]) + + +def build_grade(cell, target, days, history, recent, cache_meta, place, run=None, after=14) -> dict: + """/grade payload: a window of days centered on the target — `days` of history + before it, the target itself, then up to `after` days after it — plus the + target day's climatology summary. Days after the target are observed when they + are already in the past and forecast when they run into the future, so the + weekly view can frame two weeks of history around the orange target marker and + trail off into up to 14 days of observations / forecast after it. The forecast + only reaches ~7 days out, so a recent target naturally yields some observed days + plus 1-7 forecast days, while an older target fills the whole 14 with real obs.""" + run = run or NullRun() + with run.phase("grading"): + lo = target - pd.Timedelta(days=days) + hi = target + pd.Timedelta(days=after) + # Build the window from both sources. The recent+forecast bundle covers the + # last few weeks plus the forward forecast (the only source for future days); + # the archive reaches decades back for targets older than that bundle. Prefer + # the bundle row for any given date, filling the rest from the archive. + rwin = recent[(recent["date"] >= lo) & (recent["date"] <= hi)] + hwin = history[(history["date"] >= lo) & (history["date"] <= hi)] + hwin = hwin[~hwin["date"].isin(set(rwin["date"]))] + window = pd.concat([rwin, hwin]).sort_values("date") + graded = _grade_rows(history, window) + _attach_dry_streaks(graded, history, recent) + graded.reverse() # newest first for display + climo = grading.climatology(history, int(target.dayofyear)) + + # Summarize what this run actually covered. A fresh history fetch pulls the + # full ~45-year archive (run_type "full"); a cache hit only fetched the + # recent window (run_type "partial"). + years = pd.to_datetime(history["date"]).dt.year + full = not cache_meta.get("cached", False) + run.set( + run_type="full" if full else "partial", + history_source="fetch" if full else "cache", + history_rows=int(len(history)), + history_years=int(years.max() - years.min() + 1), + history_span=[int(years.min()), int(years.max())], + cache_age_days=cache_meta.get("cache_age_days"), + graded_days=len(graded), + place_found=place is not None, + ) + return { + "cell": cell, + "place": place, + "target_date": target.date().isoformat(), + "cache": cache_meta, + "climatology": climo, + "recent": graded, + } + + +def cal_span(history, start, end, months) -> tuple[pd.Timestamp, pd.Timestamp]: + """Clamp a requested calendar range to the available record (≤ ~2 years). + + Without a start, defaults to whole months back landing on the SAME + day-of-month as the end, so the two range endpoints line up + (e.g. 2024-06-29 → 2026-06-29).""" + first = pd.Timestamp(history["date"].min()).normalize() + last = pd.Timestamp(history["date"].max()).normalize() + end_ts = min(pd.Timestamp(end).normalize(), last) if end else last + if start: + start_ts = pd.Timestamp(start).normalize() + else: + start_ts = (end_ts - pd.DateOffset(months=months)).normalize() + # Cap the graded span at ~2 years, and keep it inside the available record. + min_start = end_ts - pd.Timedelta(days=CAL_MAX_SPAN_DAYS - 1) + start_ts = max(start_ts, min_start, first) + start_ts = min(start_ts, end_ts) + return start_ts, end_ts + + +def build_calendar(cell, history, start_ts, end_ts, months, place, run=None) -> dict: + """/calendar payload: every day in [start_ts, end_ts] graded, compact shape.""" + run = run or NullRun() + with run.phase("grading"): + days = grading.grade_range(history, start_ts, end_ts) + run.set(graded_days=len(days)) + return { + "api_version": "v2", + "cell": cell, + "place": place, + "range": {"start": start_ts.date().isoformat(), "end": end_ts.date().isoformat()}, + "months": months, + "days": days, + } + + +def build_day(cell, history, target, place, run=None, recent=None) -> dict: + """/day payload: the full percentile breakdown for one day. Observed values + prefer the archive record; a date newer than it comes from the recent window + (passed pre-loaded by the bundle path, fetched here otherwise).""" + run = run or NullRun() + last = pd.Timestamp(history["date"].max()).normalize() + + obs = None + hit = history[history["date"] == target] + if not hit.empty: + obs = _obs_from_row(hit.iloc[0]) + elif target > last: + try: + if recent is None: + with run.phase("recent"): + recent = climate.get_recent_forecast(cell) + rr = recent[recent["date"] == target] + if not rr.empty: + obs = _obs_from_row(rr.iloc[0]) + except Exception: # noqa: BLE001 - detail page still works from climatology alone + obs = None + + with run.phase("detail"): + detail = grading.day_detail(history, target, obs) + run.set(has_observation=obs is not None) + return { + "api_version": "v2", + "cell": cell, + "place": place, + "latest": last.date().isoformat(), + "detail": detail, + } + + +def build_forecast(cell, days, history, fc, today, place, run=None) -> dict: + """/forecast payload: the next `days` forecast days graded — same shape as + /grade so the frontend renders it identically, furthest-out day first.""" + run = run or NullRun() + # Strictly future days (tomorrow onward), earliest→latest, capped at `days`. + future = fc[fc["date"] > today].sort_values("date").head(days) + with run.phase("grading"): + graded = _grade_rows(history, future) + _attach_dry_streaks(graded, history, fc) + graded.reverse() # furthest-out first, so the chart reverses to L→R chronological + climo = grading.climatology(history, int(today.dayofyear)) + run.set(graded_days=len(graded), place_found=place is not None) + return { + "api_version": "v2", + "cell": cell, + "place": place, + "target_date": today.date().isoformat(), + "forecast": True, + "climatology": climo, + "recent": graded, + }