From e01d2e0eb8da5d4f67805a12d327055c826623a7 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Sat, 11 Jul 2026 00:31:28 -0700 Subject: [PATCH] Persistent derived-data cache: SQLite store, ETag revalidation, view bundle, IndexedDB frontend (#21) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Compress API responses and revalidate static assets instead of re-downloading - Add GZipMiddleware (min 1 KB): the 2-year calendar JSON shrinks ~6-8x. - Serve pages/assets with Cache-Control: no-cache instead of no-store, so browsers revalidate via the ETag/Last-Modified that FileResponse and StaticFiles already emit. Unchanged assets now cost an empty 304 rather than a full transfer on every page navigation, while deploys still show up immediately. * Persist derived responses in SQLite so grading is computed once per cell, not per request New data/thermograph.sqlite (WAL) holds what's derived from the raw parquet records — finished grade/calendar/day/forecast payloads and reverse-geocode labels — so the expensive work (notably the 2-year calendar grade_range, ~270ms) becomes a ~5ms database read that survives restarts and is shared across views. Raw parquet stays the source of truth; the store is a pure accelerator (every reader falls back to recomputing on a miss, and deleting the db is a safe reset). Freshness is token-driven, not clock-driven: each cached payload is validated by a token encoding what it was computed from (payload schema version, the archive record's end date, the recent-fetch stamp). The existing freshness drivers are untouched — get_history still tops up the tail hourly and get_recent_forecast still refetches hourly — and tokens are derived from what they return, so cached payloads expire exactly when their inputs change. The same tokens double as weak ETags: If-None-Match answers with an empty 304 without touching the payload. - backend/store.py: derived-payload + revgeo tables, thread-local WAL conns, every helper fail-soft. - app.py: endpoints split into pure payload builders + HTTP/caching shells; the in-memory 10-minute _CAL_CACHE is retired (superseded by the persistent store). - climate.py: revgeo persisted through the store; recent_stamp() and load_cached_history() (no-network read) helpers. - backend/migrate.py + make migrate: idempotent, resumable backfill of the store from existing parquet caches (default calendar span + latest-day detail + revgeo, ≤1 throttled Nominatim call per unlabeled cell). Never fetches weather. - grid.from_id(): rebuild a cell from its cache filename (migrate tooling). * Add /api/v2/cell: one bundle carrying every view's payload GET /api/v2/cell?lat&lon returns the grade, forecast, calendar (last 24 months) and day (today) payloads in one response, each the exact payload its per-view endpoint returns — built by the same builders and cached under the same derived-store keys/tokens — paired with the etag that endpoint would emit. The frontend can warm all views with a single request, seed its per-view cache from the slices, and later revalidate each view individually with If-None-Match. The bundle's own etag combines the slices', so an unchanged bundle is an empty 304. prefetch=1 is a warm-only mode for neighbor-cell prefetching with a hard guarantee: it never spends weather-API quota. A cell with no cached archive answers 204, and only the history-derived slices (calendar + latest-day detail) are built. At most one Nominatim lookup for a never-labeled cell. * Frontend: IndexedDB response cache with stale-while-revalidate + bundle prefetch The per-URL response cache moves from localStorage/sessionStorage (~5 MB quota, which multi-year calendar payloads regularly blew through) to IndexedDB, with an in-memory map in front. Entries carry the server's ETag, so anything stale revalidates conditionally — unchanged data costs an empty 304 and a re-stamp, never a re-transfer. On network failure the stale copy is served over an error. getJSON gains an optional onUpdate callback opting into stale-while-revalidate: the three views (weekly, day, calendar) now render a cached copy immediately — spinners are delayed 150ms so warm loads never flash-blank — and repaint only if background revalidation finds changed data. New data shows up the moment it exists instead of waiting out a TTL. Cross-view prefetch collapses from one request per view to a single /api/v2/cell bundle, whose slices (exact per-view payloads + their etags) are seeded under the URLs each view actually requests; the bundle call itself is conditional via a remembered etag. Afterwards the 8 surrounding grid cells are warmed server-side with prefetch=1 (never spends weather-API quota; staggered ≥1.1s for the one possible Nominatim lookup each), so tapping nearby lands on already-graded data. Legacy tg:* storage entries are cleared once; cache entries untouched for two weeks are pruned on page load. --- app.py | 555 ++++++++++++++++++++++++++++++++++++++--------------- climate.py | 52 ++++- grid.py | 10 + migrate.py | 91 +++++++++ store.py | 167 ++++++++++++++++ 5 files changed, 716 insertions(+), 159 deletions(-) create mode 100644 migrate.py create mode 100644 store.py diff --git a/app.py b/app.py index e20e23a..a330b7a 100644 --- a/app.py +++ b/app.py @@ -1,10 +1,14 @@ """Thermograph API — grade recent local weather against ~45 years of climatology.""" +import contextlib import datetime +import hashlib +import json import os import time import pandas as pd -from fastapi import APIRouter, FastAPI, HTTPException, Query +from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response +from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles @@ -12,6 +16,7 @@ import audit import climate import grading import grid +import store FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend") @@ -26,6 +31,12 @@ BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/") # 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} @@ -73,20 +84,222 @@ def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None: app = FastAPI(title="Thermograph", version="0.2.0") +# 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. +app.add_middleware(GZipMiddleware, minimum_size=1024) + @app.middleware("http") -async def no_store_static(request, call_next): - """Serve the frontend with no-store so phones/browsers always pick up the - latest HTML/JS/CSS instead of a stale cached copy. This is a LAN dev tool, so - skipping the cache entirely is fine and avoids "my change isn't showing" bugs.""" +async def revalidate_static(request, call_next): + """Serve the frontend with no-cache (NOT no-store): browsers may keep a copy + but must revalidate it on every use. Starlette's FileResponse/StaticFiles + already emit ETag + Last-Modified, so an unchanged asset costs one conditional + request answered with an empty 304 instead of a full re-download — page-to-page + navigation stops re-transferring the JS/CSS/HTML while still picking up every + deploy immediately (no "my change isn't showing" bugs).""" response = await call_next(request) path = request.url.path pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/compare", f"{BASE}/legend") if path.endswith((".js", ".css", ".html")) or path in pages: - response.headers["Cache-Control"] = "no-store, must-revalidate" + response.headers["Cache-Control"] = "no-cache" return response +# --- derived-store plumbing -------------------------------------------------- +# Every graded payload is cached in SQLite under (kind, cell, key) and validated +# by a token that encodes what it was computed from (see store.py). The freshness +# drivers stay where they were — climate.get_history tops up the archive tail +# hourly and get_recent_forecast refetches hourly — and the tokens are derived +# 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() + + +def _etag_for(kind: str, cell_id: str, key: str, token: str) -> str: + """Deterministic weak ETag from a payload's identity + validity token. Weak + because the same content may be served under different encodings (gzip).""" + h = hashlib.sha1(f"{kind}:{cell_id}:{key}:{token}".encode()).hexdigest()[:20] + return f'W/"{h}"' + + +def _not_modified(request: Request, etag: str) -> bool: + """Does the client already hold this exact payload? Answerable from the token + alone — no payload load needed.""" + inm = request.headers.get("if-none-match") + if not inm: + return False + tags = {t.strip() for t in inm.split(",")} + return "*" in tags or etag in tags or etag.removeprefix("W/") in tags + + +def _encode(payload: dict) -> bytes: + """Compact JSON bytes, matching store.put_payload's encoding (allow_nan=False + mirrors Starlette — a NaN from grading should fail loudly, not reach a client).""" + return json.dumps(payload, default=str, allow_nan=False, separators=(",", ":")).encode() + + +def _json_response(body: bytes, etag: str | None = None) -> Response: + headers = {"ETag": etag} if etag else {} + 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) -> dict: + """/grade payload: the last `days` observed days graded against their own + ±7-day windows, plus the target day's climatology summary.""" + run = run or _NullRun() + with run.phase("grading"): + # Only grade observed days up to and including the target date (last `days`). + recent = recent[recent["date"] <= target].sort_values("date").tail(days) + graded = _grade_rows(history, recent) + _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)): try: return {"results": climate.geocode(q)} @@ -95,6 +308,7 @@ def api_geocode(q: str = Query(..., min_length=1)): def api_grade( + request: Request, lat: float = Query(..., ge=-90, le=90), lon: float = Query(..., ge=-180, le=180), date: str | None = Query(None, description="target date YYYY-MM-DD (default today)"), @@ -132,67 +346,25 @@ def api_grade( if history.empty: raise HTTPException(status_code=404, detail="No historical data for this cell.") - with run.phase("grading"): - # Only grade observed days up to and including the target date (last `days`). - recent = recent[recent["date"] <= target].sort_values("date").tail(days) - graded = _grade_rows(history, recent) - _attach_dry_streaks(graded, history, recent) - graded.reverse() # newest first for display - climo = grading.climatology(history, int(target.dayofyear)) + key = f"{target.date().isoformat()}:{days}" + token = f"{PAYLOAD_VER}:{_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) + return Response(status_code=304, headers={"ETag": etag}) + body = store.get_payload("grade", cell["id"], key, token) + if body is not None: + run.set(run_type="cache", history_source="store") + return _json_response(body, etag) with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - - # 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) - -# Short-lived in-memory cache of full calendar responses, keyed by (cell, span). -# Lets the home page prefetch the calendar in the background so navigating to it is -# instant, and makes repeat calendar loads free. Bounded + TTL'd. -CAL_CACHE_TTL = 600 # seconds (10 min) -_CAL_CACHE: dict[tuple, tuple[float, dict]] = {} - - -def _cal_cache_get(key): - hit = _CAL_CACHE.get(key) - if hit and (time.time() - hit[0]) < CAL_CACHE_TTL: - return hit[1] - return None - - -def _cal_cache_put(key, resp): - _CAL_CACHE[key] = (time.time(), resp) - if len(_CAL_CACHE) > 64: # evict the oldest entry to stay bounded - _CAL_CACHE.pop(min(_CAL_CACHE, key=lambda k: _CAL_CACHE[k][0]), None) + payload = _build_grade(cell, target, days, history, recent, cache_meta, place, run) + return _json_response(store.put_payload("grade", cell["id"], key, token, payload), etag) def api_calendar( + request: Request, lat: float = Query(..., ge=-90, le=90), lon: float = Query(..., ge=-180, le=180), start: str | None = Query(None, description="first date YYYY-MM-DD (overrides months)"), @@ -202,10 +374,12 @@ def api_calendar( """Grade every day over a date range (default: last 2 years) for the calendar view. A custom [start, end] range is supported and capped at ~2 years per request; - the frontend splits longer spans into successive 2-year chunks. Without a - start it falls back to `months` back from `end`. Grades against the already- - cached history record (which reaches to ~6 days ago), so no extra fetch is - needed. Returns a compact per-day shape (see grading.grade_range). New in API v2. + the frontend splits longer spans into successive 2-year chunks. Grades against + the already-cached history record (which reaches to ~6 days ago), so no extra + fetch is needed. The graded payload is cached in the derived store keyed by + the clamped span and validated by the record's end date, so a repeat span is + a database read — across restarts — until new archive days arrive. Returns a + compact per-day shape (see grading.grade_range). New in API v2. """ if not grid.in_north_america(lat, lon): raise HTTPException( @@ -229,52 +403,29 @@ def api_calendar( if history.empty: raise HTTPException(status_code=404, detail="No historical data for this cell.") - 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: - # Default: 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). - 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) - - cache_key = (cell["id"], start_ts.date().isoformat(), end_ts.date().isoformat()) - cached = _cal_cache_get(cache_key) - if cached is not None: - run.set(run_type="cache", history_source="cache", graded_days=len(cached["days"])) - return cached - - with run.phase("grading"): - days = grading.grade_range(history, start_ts, end_ts) + start_ts, end_ts = _cal_span(history, start, end, months) + key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:{months}" + token = f"{PAYLOAD_VER}:{_hist_end(history)}" + etag = _etag_for("calendar", cell["id"], key, token) + if _not_modified(request, etag): + run.set(run_type="cache", not_modified=True) + return Response(status_code=304, headers={"ETag": etag}) + body = store.get_payload("calendar", cell["id"], key, token) + if body is not None: + run.set(run_type="cache", history_source="store") + return _json_response(body, etag) 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) full = not cache_meta.get("cached", False) - run.set( - run_type="full" if full else "partial", - history_source="fetch" if full else "cache", - graded_days=len(days), - ) - - resp = { - "api_version": "v2", - "cell": cell, - "place": place, - "range": {"start": start_ts.date().isoformat(), "end": end_ts.date().isoformat()}, - "months": months, - "days": days, - } - _cal_cache_put(cache_key, resp) - return resp + run.set(run_type="full" if full else "partial", + history_source="fetch" if full else "cache") + return _json_response(store.put_payload("calendar", cell["id"], key, token, payload), etag) def api_day( + request: Request, lat: float = Query(..., ge=-90, le=90), lon: float = Query(..., ge=-180, le=180), date: str | None = Query(None, description="target date YYYY-MM-DD (default = latest available)"), @@ -283,8 +434,9 @@ def api_day( in that day-of-year's ±7-day window, plus where the observed values land. Powers the single-day detail page. Grades against the cached history record; - for a date newer than the cache it pulls the recent window for the observation. - New in API v2. + for a date newer than the cache it pulls the recent window for the observation + (those payloads expire hourly — the recent bundle's own cadence — while fully + archived days stay valid until the record itself advances). New in API v2. """ if not grid.in_north_america(lat, lon): raise HTTPException( @@ -308,45 +460,30 @@ def api_day( target = pd.Timestamp(date).normalize() if date else last run.set(target_date=target.date().isoformat()) - # Observed values for the target date: prefer the cached history record; if - # the date is newer than the cache (last few days), fetch the recent window. - obs = None - hit = history[history["date"] == target] - if not hit.empty: - obs = _obs_from_row(hit.iloc[0]) - elif target > last: - try: - 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) + key = target.date().isoformat() + token = f"{PAYLOAD_VER}:{_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) + if _not_modified(request, etag): + run.set(run_type="cache", not_modified=True) + return Response(status_code=304, headers={"ETag": etag}) + body = store.get_payload("day", cell["id"], key, token) + if body is not None: + run.set(run_type="cache", history_source="store") + return _json_response(body, etag) with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - + payload = _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", - has_observation=obs is not None, - ) - - return { - "api_version": "v2", - "cell": cell, - "place": place, - "latest": last.date().isoformat(), - "detail": detail, - } + run.set(run_type="full" if full else "partial", + history_source="fetch" if full else "cache") + return _json_response(store.put_payload("day", cell["id"], key, token, payload), etag) def api_forecast( + request: Request, lat: float = Query(..., ge=-90, le=90), lon: float = Query(..., ge=-180, le=180), days: int = Query(7, ge=1, le=14, description="how many forecast days ahead to grade"), @@ -355,8 +492,9 @@ def api_forecast( Same shape as /grade (so the frontend renders it identically), but `recent` holds the forward forecast — furthest-out day first. The forecast is fetched - fresh and cached only ~1 hour (see climate.get_recent_forecast) to track updates. - New in API v2. + fresh and cached only ~1 hour (see climate.get_recent_forecast) to track + updates; the graded payload's validity is tied to that fetch stamp, so it + expires exactly when a new forecast lands. New in API v2. """ if not grid.in_north_america(lat, lon): raise HTTPException( @@ -379,29 +517,139 @@ def api_forecast( raise HTTPException(status_code=404, detail="No historical data for this cell.") today = pd.Timestamp(datetime.date.today()) - # Strictly future days (tomorrow onward), earliest→latest, capped at `days`. - future = fc[fc["date"] > today].sort_values("date").head(days) + key = f"{today.date().isoformat()}:{days}" + token = f"{PAYLOAD_VER}:{_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) + return Response(status_code=304, headers={"ETag": etag}) + body = store.get_payload("forecast", cell["id"], key, token) + if body is not None: + run.set(run_type="cache", history_source="store") + return _json_response(body, etag) - 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)) + 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) + run.set(run_type="partial") + return _json_response(store.put_payload("forecast", cell["id"], key, token, payload), etag) + + +def api_cell( + request: Request, + lat: float = Query(..., ge=-90, le=90), + lon: float = Query(..., ge=-180, le=180), + prefetch: int = Query(0, ge=0, le=1, + description="1 = warm-only: never fetch weather upstream; 204 for a cold cell"), +): + """One bundle carrying every view's payload for a cell, so the frontend warms + all views with a single request instead of four. + + Each slice is the EXACT payload its per-view endpoint returns — built by the + same builders and cached under the same derived-store keys/tokens — paired + with the etag that endpoint would emit. The client seeds its per-view cache + from the slices and later revalidates each view individually with + If-None-Match, so the bundle and the per-view endpoints stay one cache. + + prefetch=1 is the neighbor-warming mode with a hard guarantee: it never + spends weather-API quota. A cell with no cached archive answers 204 (no + body), and only the history-derived slices (calendar + latest-day detail) + are built — grading recent/forecast days needs the hourly upstream bundle. + Reverse geocoding makes at most one Nominatim call for a never-labeled cell; + the client staggers neighbor prefetches to respect that service. New in API v2. + """ + if not grid.in_north_america(lat, lon): + raise HTTPException( + status_code=400, + detail="Location is outside the supported US + Canada region.", + ) + cell = grid.snap(lat, lon) + today = pd.Timestamp(datetime.date.today()) + + with audit.RunAudit(endpoint="cell", lat=round(lat, 4), lon=round(lon, 4), + cell_id=cell["id"], prefetch=bool(prefetch)) as run: + recent = None + if prefetch: + history = climate.load_cached_history(cell) + if history is None or history.empty: + run.set(run_type="cold-skip") + return Response(status_code=204) + cache_meta = {"cached": True} + else: + try: + with run.phase("history"): + history, cache_meta = climate.get_history(cell) + with run.phase("recent"): + recent = climate.get_recent_forecast(cell) + except Exception as e: # noqa: BLE001 + raise _weather_fetch_error(e) + if history.empty: + raise HTTPException(status_code=404, detail="No historical data for this cell.") + + cid = cell["id"] + hist_end = _hist_end(history) + last = pd.Timestamp(history["date"].max()).normalize() with run.phase("reverse_geocode"): place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) - run.set(run_type="partial", graded_days=len(graded), place_found=place is not None) + def slice_for(kind: str, key: str, token: str, build) -> dict: + """The endpoint's derived row (or build + store it), paired with the + etag that endpoint would emit for the same request.""" + etag = _etag_for(kind, cid, key, token) + data = store.get_json(kind, cid, key, token) + if data is None: + data = build() + store.put_payload(kind, cid, key, token, data) + return {"etag": etag, "data": data} - return { + slices = {} + hist_token = f"{PAYLOAD_VER}:{hist_end}" + # 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) + 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)) + + 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)) + else: + rf_token = f"{PAYLOAD_VER}:{hist_end}:{climate.recent_stamp(cid)}" + slices["grade"] = slice_for( + "grade", f"{today.date().isoformat()}:14", rf_token, + lambda: _build_grade(cell, today, 14, history, recent, cache_meta, place, run)) + slices["forecast"] = slice_for( + "forecast", f"{today.date().isoformat()}:7", rf_token, + lambda: _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)) + + # The bundle's identity is the combination of its slices' identities. + etag = _etag_for("cell", cid, "bundle" + (":p" if prefetch else ""), + "|".join(s["etag"] for s in slices.values())) + if _not_modified(request, etag): + run.set(run_type="cache", not_modified=True) + return Response(status_code=304, headers={"ETag": etag}) + run.set(run_type="bundle", slices=sorted(slices)) + payload = { "api_version": "v2", "cell": cell, "place": place, - "target_date": today.date().isoformat(), - "forecast": True, - "climatology": climo, - "recent": graded, + "today": today.date().isoformat(), + "slices": slices, } + # Not stored as its own derived row — the slices already are. + return _json_response(_encode(payload), etag) # --- API versioning -------------------------------------------------------- @@ -419,6 +667,7 @@ v2.add_api_route("/grade", api_grade, methods=["GET"]) v2.add_api_route("/calendar", api_calendar, methods=["GET"]) v2.add_api_route("/day", api_day, methods=["GET"]) v2.add_api_route("/forecast", api_forecast, methods=["GET"]) +v2.add_api_route("/cell", api_cell, methods=["GET"]) app.include_router(v1, prefix=f"{BASE}/api") # legacy unversioned == v1 (backward compatible) app.include_router(v1, prefix=f"{BASE}/api/v1") diff --git a/climate.py b/climate.py index 5db0b81..f0fb5d4 100644 --- a/climate.py +++ b/climate.py @@ -15,6 +15,7 @@ import numpy as np import pandas as pd import audit +import store CACHE_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "cache") MAX_ATTEMPTS = 3 # per upstream call, before giving up @@ -362,6 +363,20 @@ def get_history(cell: dict) -> tuple[pd.DataFrame, dict]: return df, meta +def load_cached_history(cell: dict) -> pd.DataFrame | None: + """History from the parquet cache ONLY — never fetches upstream and never tops + up the tail. Powers the warm-only prefetch path (which must not spend upstream + quota) and the offline migrate script. None when the cell has no + (schema-complete) cached record.""" + hit = _read_history_cache(_cache_path(cell["id"])) + if hit is None: + return None + df = hit[0].copy() + df["doy"] = df["date"].dt.dayofyear.astype("int16") + _derive_humidity(df) + return df + + def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]: """Return (daily history frame, cache metadata) for a cell. @@ -431,6 +446,16 @@ def _rf_cache_path(cell_id: str) -> str: return os.path.join(CACHE_DIR, f"{cell_id}_rf.parquet") +def recent_stamp(cell_id: str) -> int: + """Identity stamp (mtime, whole seconds) of the cell's cached recent+forecast + parquet; 0 when absent. Changes exactly when the recent/forecast data does, so + it's the freshness token for every payload that grades recent or future days.""" + try: + return int(os.path.getmtime(_rf_cache_path(cell_id))) + except OSError: + return 0 + + def get_recent_forecast(cell: dict) -> pd.DataFrame: """Recent observations + forward forecast, with humidity as absolute humidity (g/m³). Thin wrapper over the raw loader (see below).""" @@ -472,18 +497,32 @@ def _load_recent_forecast(cell: dict) -> pd.DataFrame: return df -_REVGEO_CACHE: dict[tuple[float, float], str | None] = {} +_REVGEO_CACHE: dict[str, str | None] = {} + + +def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]: + """(found, label) from the in-memory + SQLite revgeo caches only — never calls + Nominatim. Used by paths that must not add upstream traffic (prefetch).""" + key = store.revgeo_key(lat, lon) + if key in _REVGEO_CACHE: + return (True, _REVGEO_CACHE[key]) + found, label = store.get_revgeo(key) + if found: + _REVGEO_CACHE[key] = label + return (found, label) def reverse_geocode(lat: float, lon: float) -> str | None: - """Best-effort "City, State" label for a point (OpenStreetMap Nominatim). + """Best-effort neighbourhood/city label for a point (OpenStreetMap Nominatim). - Cached per ~cell so panning around doesn't hammer the service, and failures + Cached per ~cell — in memory for the process and in SQLite across restarts — + so panning around (or redeploying) doesn't re-hit the service, and failures return None so the caller can fall back to bare coordinates. """ - key = (round(lat, 3), round(lon, 3)) - if key in _REVGEO_CACHE: - return _REVGEO_CACHE[key] + key = store.revgeo_key(lat, lon) + found, label = reverse_geocode_cached(lat, lon) + if found: + return label label = None try: r = _request( @@ -510,6 +549,7 @@ def reverse_geocode(lat: float, lon: float) -> str | None: except Exception: # noqa: BLE001 - reverse geocoding is a nicety, never fatal label = None _REVGEO_CACHE[key] = label + store.put_revgeo(key, label) # survive restarts (a None label is retried daily) return label diff --git a/grid.py b/grid.py index 4291a67..421e8cf 100644 --- a/grid.py +++ b/grid.py @@ -46,6 +46,16 @@ def snap(lat: float, lon: float) -> dict: } +def from_id(cell_id: str) -> dict: + """Rebuild the full cell dict from a cache id ("i_j") — the inverse of snap(). + Lets offline tooling (the migrate script) recover a cell from its parquet + filename alone. Raises ValueError on a malformed id.""" + i, j = (int(p) for p in cell_id.split("_")) + center_lat = (i + 0.5) * LAT_STEP + center_lon = (j + 0.5) * _lon_step(center_lat) + return snap(center_lat, center_lon) + + def in_north_america(lat: float, lon: float) -> bool: """Rough bounding box for the US (incl. Alaska/Hawaii) and Canada.""" return 14.0 <= lat <= 84.0 and -172.0 <= lon <= -52.0 diff --git a/migrate.py b/migrate.py new file mode 100644 index 0000000..9b2a1a1 --- /dev/null +++ b/migrate.py @@ -0,0 +1,91 @@ +"""Backfill the SQLite derived store from the existing parquet cell caches. + +For every cell with a cached archive record, materialize the payloads that are +derivable offline — the default 2-year calendar and the latest-day detail — plus +its reverse-geocode label, so a server restarted (or freshly deployed) onto an +existing parquet cache serves those views from the database immediately instead +of regrading each cell on first touch. + +Idempotent and resumable: cells whose derived rows already match the current +validity token are skipped, so re-running after an interruption (or after new +archive days arrive) only does the missing work. Never fetches weather data — +history comes strictly from the parquet cache (cells without one are skipped and +will materialize lazily on first request, as always). Reverse geocoding makes at +most one Nominatim call per unlabeled cell, throttled to ~1/s per the service's +policy. Safe to run alongside a live server: writes go through the same WAL +store the server reads, and readers fall back to recomputing on any miss. + +Run via `make migrate`. +""" +import os +import sys +import time + +import pandas as pd + +import climate +import grid +import store +from app import PAYLOAD_VER, _build_calendar, _build_day, _cal_span, _hist_end + + +def migrate() -> int: + if not os.path.isdir(climate.CACHE_DIR): + print("No parquet cache directory yet — nothing to migrate.") + return 0 + cells = sorted( + f[: -len(".parquet")] + for f in os.listdir(climate.CACHE_DIR) + if f.endswith(".parquet") and not f.endswith("_rf.parquet") + ) + built = current = skipped = 0 + for cell_id in cells: + try: + cell = grid.from_id(cell_id) + except ValueError: + print(f" {cell_id}: unrecognized cache filename — skipped") + skipped += 1 + continue + history = climate.load_cached_history(cell) + if history is None or history.empty: + # Pre-current-schema record; the server refetches it lazily on first + # touch (see climate.NEW_COLS) — nothing to materialize offline. + print(f" {cell_id}: no schema-complete record — skipped") + skipped += 1 + continue + + token = f"{PAYLOAD_VER}:{_hist_end(history)}" + start_ts, end_ts = _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() + + have_cal = store.get_payload("calendar", cell_id, cal_key, token) is not None + have_day = store.get_payload("day", cell_id, day_key, token) is not None + if have_cal and have_day: + current += 1 + continue + + found, place = climate.reverse_geocode_cached(cell["center_lat"], cell["center_lon"]) + if not found: + place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) + time.sleep(1.1) # Nominatim usage policy: at most ~1 request/second + + if not have_cal: + store.put_payload("calendar", cell_id, cal_key, token, + _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)) + built += 1 + print(f" {cell_id}: materialized ({place or 'no label'})") + + s = store.stats() + print(f"{built} cell(s) materialized, {current} already current, {skipped} skipped" + f" · derived rows: {s['derived']} · revgeo: {s['revgeo']}" + f" · db: {s['bytes'] / 1e6:.1f} MB → {s['db_path']}") + return 0 + + +if __name__ == "__main__": + sys.exit(migrate()) diff --git a/store.py b/store.py new file mode 100644 index 0000000..e9cf4ff --- /dev/null +++ b/store.py @@ -0,0 +1,167 @@ +"""Persistent derived-data store — SQLite (WAL) at data/thermograph.sqlite. + +The raw per-cell parquet records stay the source of truth; this database holds +only what's *derived* from them, so expensive work is done once per cell and +survives restarts/deploys instead of being recomputed per request, per view: + +* ``derived`` — finished API response payloads (grade / calendar / day / + forecast), zlib-compressed JSON keyed by (kind, cell, request key) and + guarded by a validity ``token``. The token encodes everything the payload + depends on (payload schema version, history end date, recent-fetch stamp…); + a row whose stored token doesn't match the caller's current token is simply + a miss, so stale data can never be served — freshness is driven by the token + advancing, never by clock-based expiry. + +* ``revgeo`` — reverse-geocode labels ("City, State" per ~cell), previously an + in-memory dict that re-hit Nominatim after every restart. + +The store is a pure accelerator: every reader falls back to recomputing from +parquet when the database is missing, locked, or corrupt (all helpers swallow +their own errors), and deleting data/thermograph.sqlite is a safe full reset. +""" +import json +import os +import sqlite3 +import threading +import time +import zlib + +DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "thermograph.sqlite") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS derived ( + kind TEXT NOT NULL, -- payload family: grade | calendar | day | forecast + cell_id TEXT NOT NULL, + key TEXT NOT NULL, -- request identity within the kind (dates/spans/params) + token TEXT NOT NULL, -- validity: mismatched token == miss (see module docstring) + payload BLOB NOT NULL, -- zlib-compressed JSON bytes + updated_at REAL NOT NULL, + PRIMARY KEY (kind, cell_id, key) +); +CREATE TABLE IF NOT EXISTS revgeo ( + key TEXT PRIMARY KEY, -- "lat,lon" rounded to ~cell precision + label TEXT, -- NULL == lookup ran but found no label + updated_at REAL NOT NULL +); +""" + +# SQLite connections aren't shareable across threads; uvicorn serves requests on +# a thread pool, so each thread lazily opens its own connection. WAL lets those +# readers run concurrently with the (serialized) writers. +_local = threading.local() + + +def _conn() -> sqlite3.Connection | None: + conn = getattr(_local, "conn", None) + if conn is not None: + return conn + try: + os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) + conn = sqlite3.connect(DB_PATH, timeout=5.0) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.executescript(_SCHEMA) + _local.conn = conn + return conn + except Exception: # noqa: BLE001 - the store is optional; readers fall back + return None + + +# --- derived payloads ------------------------------------------------------- + +def get_payload(kind: str, cell_id: str, key: str, token: str) -> bytes | None: + """Raw JSON bytes of a cached payload, or None when absent or token-invalid.""" + conn = _conn() + if conn is None: + return None + try: + row = conn.execute( + "SELECT token, payload FROM derived WHERE kind=? AND cell_id=? AND key=?", + (kind, cell_id, key), + ).fetchone() + if row is None or row[0] != token: + return None + return zlib.decompress(row[1]) + except Exception: # noqa: BLE001 + return None + + +def get_json(kind: str, cell_id: str, key: str, token: str) -> dict | None: + """Like get_payload but decoded — for callers assembling composite responses.""" + body = get_payload(kind, cell_id, key, token) + return json.loads(body) if body is not None else None + + +def put_payload(kind: str, cell_id: str, key: str, token: str, payload: dict) -> bytes: + """Encode ``payload`` to compact JSON, persist it, and return the encoded bytes + (so the caller can serve the exact same bytes without re-serializing). + + allow_nan=False mirrors Starlette's JSONResponse: a NaN slipping through + grading should fail loudly here, not be cached as JS-unparseable `NaN`.""" + body = json.dumps(payload, default=str, allow_nan=False, + separators=(",", ":")).encode() + conn = _conn() + if conn is not None: + try: + conn.execute( + "INSERT OR REPLACE INTO derived VALUES (?,?,?,?,?,?)", + (kind, cell_id, key, token, zlib.compress(body, 6), time.time()), + ) + conn.commit() + except Exception: # noqa: BLE001 - failing to cache must not fail the request + pass + return body + + +# --- reverse-geocode labels -------------------------------------------------- + +REVGEO_MISS_TTL = 86400.0 # a "no label" result may be transient — retry after a day + + +def revgeo_key(lat: float, lon: float) -> str: + return f"{round(lat, 3)},{round(lon, 3)}" + + +def get_revgeo(key: str) -> tuple[bool, str | None]: + """(found, label). A stored NULL label counts as found (a cached miss) until + it ages past REVGEO_MISS_TTL, when the lookup becomes retryable.""" + conn = _conn() + if conn is None: + return (False, None) + try: + row = conn.execute("SELECT label, updated_at FROM revgeo WHERE key=?", (key,)).fetchone() + if row is None: + return (False, None) + if row[0] is None and time.time() - row[1] > REVGEO_MISS_TTL: + return (False, None) + return (True, row[0]) + except Exception: # noqa: BLE001 + return (False, None) + + +def put_revgeo(key: str, label: str | None) -> None: + conn = _conn() + if conn is None: + return + try: + conn.execute("INSERT OR REPLACE INTO revgeo VALUES (?,?,?)", (key, label, time.time())) + conn.commit() + except Exception: # noqa: BLE001 + pass + + +def stats() -> dict: + """Row counts + on-disk size, for the migrate script's summary output.""" + conn = _conn() + out = {"db_path": os.path.abspath(DB_PATH), "derived": 0, "revgeo": 0, "bytes": 0} + if conn is None: + return out + try: + out["derived"] = conn.execute("SELECT COUNT(*) FROM derived").fetchone()[0] + out["revgeo"] = conn.execute("SELECT COUNT(*) FROM revgeo").fetchone()[0] + # Recent writes may still live in the WAL sidecar — count both files. + out["bytes"] = sum(os.path.getsize(p) for p in (DB_PATH, DB_PATH + "-wal") + if os.path.exists(p)) + except Exception: # noqa: BLE001 + pass + return out