Jinemi/thermograph
4
0
Fork 0
Code Issues 2 Pull requests 3 Projects Releases Packages Wiki Activity Actions

Migrate backend dataframe layer from pandas to polars (#90)

Browse source 
* Migrate backend dataframe layer from pandas to polars

Replace pandas with polars across the backend, dropping both pandas and its
pyarrow parquet engine from the dependency set. numpy stays (the grading
percentile math is unchanged).

- climate.py: parquet IO, source→frame mappings, cache read/topup on polars.
  New _normalize_read casts the cached `date` column to pl.Date (older files
  were written by pandas as datetime64[ns]); frames now unify missing values as
  null so the grading boundary drops them consistently across sources.
- grading.py: keep the numpy percentile core; swap the frame→numpy bridge to
  .to_numpy()/.drop_nulls(), day-of-year/year to polars dt expressions, and the
  per-row loop to iter_rows(named=True).
- views.py: filter/anti-join/concat replace boolean-mask, isin and pd.concat;
  scalar dates are stdlib datetime.date; a local _months_before helper replaces
  DateOffset(months=) for the calendar-range default.
- app.py, migrate.py: request-date parsing uses datetime.date, removing pandas
  from the endpoint and migrate layers entirely.
- The date column is pl.Date end to end, eliminating the pandas normalize() calls
  and comparing cleanly against stdlib dates.

Payloads are unchanged: calendar, day, grade and forecast responses are
byte-for-byte identical to the pandas implementation on the same cached record.

Tests ported to polars fixtures, with added coverage for the combined feels-like
fallback, calendar month-offset (month-end/leap), and the concat/dedup
"fresher source wins" rule.

* Port notify.py to polars after merging dev's account system

Merge origin/dev (accounts + notification subscriptions) and carry the pandas→
polars migration into the newly added notify.py, which the merge brought in still
using pandas — with pandas removed from requirements this broke its import.

- notify.py: _candidate_rows filters/sorts the recent bundle with polars
  expressions and returns iter_rows dicts; date scalars are datetime.date;
  history/recent emptiness via is_empty().
- test_notify.py: synthetic history/rows built with polars + datetime.
This commit is contained in:
Emi Griffith 2026-07-15 12:07:38 -07:00 committed by GitHub
parent efddd15025
commit c3bacdce0c
13 changed files with 364 additions and 257 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
app.py
View file

@ -9,7 +9,6 @@ import queue
import threading import threading
import time import time
import pandas as pd
from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response
from fastapi.middleware.gzip import GZipMiddleware from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import RedirectResponse from fastapi.responses import RedirectResponse
@ -66,7 +65,7 @@ def _warm_cell(cell: dict) -> None:
"""Materialize the history-derived slices for one cell — the same rows the """Materialize the history-derived slices for one cell — the same rows the
prefetch=1 bundle serves. Never fetches weather upstream.""" prefetch=1 bundle serves. Never fetches weather upstream."""
history = climate.load_cached_history(cell) history = climate.load_cached_history(cell)
if history is None or history.empty: if history is None or history.is_empty():
return return
place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"]) place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])
token = views.history_token(history) token = views.history_token(history)
@ -75,7 +74,7 @@ def _warm_cell(cell: dict) -> None:
if store.get_payload("calendar", cell["id"], cal_key, token) is None: if store.get_payload("calendar", cell["id"], cal_key, token) is None:
store.put_payload("calendar", cell["id"], cal_key, token, store.put_payload("calendar", cell["id"], cal_key, token,
views.build_calendar(cell, history, start_ts, end_ts, 24, place)) views.build_calendar(cell, history, start_ts, end_ts, 24, place))
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
day_key = views.day_key(last) day_key = views.day_key(last)
if store.get_payload("day", cell["id"], day_key, token) is None: if store.get_payload("day", cell["id"], day_key, token) is None:
store.put_payload("day", cell["id"], day_key, token, store.put_payload("day", cell["id"], day_key, token,
@ -196,7 +195,7 @@ def _fetch_history(run, cell, recent_too=False, recent_phase="recent"):
recent = climate.get_recent_forecast(cell) recent = climate.get_recent_forecast(cell)
except Exception as e: # noqa: BLE001 except Exception as e: # noqa: BLE001
raise _weather_fetch_error(e) raise _weather_fetch_error(e)
if history.empty: if history.is_empty():
raise HTTPException(status_code=404, detail="No historical data for this cell.") raise HTTPException(status_code=404, detail="No historical data for this cell.")
return history, cache_meta, recent return history, cache_meta, recent
@ -287,18 +286,14 @@ def api_grade(
description="days to grade after the target (observed, or forecast when future; " description="days to grade after the target (observed, or forecast when future; "
"the forecast reaches ~7 days out so future days cap there)"), "the forecast reaches ~7 days out so future days cap there)"),
): ):
target = ( target = datetime.date.fromisoformat(date[:10]) if date else datetime.date.today()
pd.Timestamp(date).normalize()
if date
else pd.Timestamp(datetime.date.today())
)
cell = grid.snap(lat, lon) cell = grid.snap(lat, lon)
with audit.RunAudit( with audit.RunAudit(
endpoint="grade", endpoint="grade",
lat=round(lat, 4), lat=round(lat, 4),
lon=round(lon, 4), lon=round(lon, 4),
target_date=target.date().isoformat(), target_date=target.isoformat(),
recent_days=days, recent_days=days,
cell_id=cell["id"], cell_id=cell["id"],
) as run: ) as run:
@ -378,9 +373,9 @@ def api_day(
cell_id=cell["id"], cell_id=cell["id"],
) as run: ) as run:
history, cache_meta, _ = _fetch_history(run, cell) history, cache_meta, _ = _fetch_history(run, cell)
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
target = pd.Timestamp(date).normalize() if date else last target = datetime.date.fromisoformat(date[:10]) if date else last
run.set(target_date=target.date().isoformat()) run.set(target_date=target.isoformat())
def build(place): def build(place):
payload = views.build_day(cell, history, target, place, run) payload = views.build_day(cell, history, target, place, run)
@ -413,7 +408,7 @@ def api_forecast(
with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4), with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4),
recent_days=days, cell_id=cell["id"]) as run: recent_days=days, cell_id=cell["id"]) as run:
history, _, fc = _fetch_history(run, cell, recent_too=True, recent_phase="forecast") history, _, fc = _fetch_history(run, cell, recent_too=True, recent_phase="forecast")
today = pd.Timestamp(datetime.date.today()) today = datetime.date.today()
def build(place): def build(place):
payload = views.build_forecast(cell, days, history, fc, today, place, run) payload = views.build_forecast(cell, days, history, fc, today, place, run)
@ -452,14 +447,14 @@ def api_cell(
the client staggers neighbor prefetches to respect that service. New in API v2. the client staggers neighbor prefetches to respect that service. New in API v2.
""" """
cell = grid.snap(lat, lon) cell = grid.snap(lat, lon)
today = pd.Timestamp(datetime.date.today()) today = datetime.date.today()
with audit.RunAudit(endpoint="cell", lat=round(lat, 4), lon=round(lon, 4), with audit.RunAudit(endpoint="cell", lat=round(lat, 4), lon=round(lon, 4),
cell_id=cell["id"], prefetch=bool(prefetch)) as run: cell_id=cell["id"], prefetch=bool(prefetch)) as run:
recent = None recent = None
if prefetch: if prefetch:
history = climate.load_cached_history(cell) history = climate.load_cached_history(cell)
if history is None or history.empty: if history is None or history.is_empty():
run.set(run_type="cold-skip") run.set(run_type="cold-skip")
return Response(status_code=204) return Response(status_code=204)
cache_meta = {"cached": True} cache_meta = {"cached": True}
@ -467,7 +462,7 @@ def api_cell(
history, cache_meta, recent = _fetch_history(run, cell, recent_too=True) history, cache_meta, recent = _fetch_history(run, cell, recent_too=True)
cid = cell["id"] cid = cell["id"]
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
if neighbors: if neighbors:
_enqueue_neighbor_warming(cell) _enqueue_neighbor_warming(cell)
@ -523,7 +518,7 @@ def api_cell(
"api_version": "v2", "api_version": "v2",
"cell": cell, "cell": cell,
"place": place, "place": place,
"today": today.date().isoformat(), "today": today.isoformat(),
"slices": slices, "slices": slices,
} }
# Not stored as its own derived row — the slices already are. # Not stored as its own derived row — the slices already are.

163
climate.py
View file

@ -12,7 +12,7 @@ import time
import httpx import httpx
import numpy as np import numpy as np
import pandas as pd import polars as pl
import audit import audit
import store import store
@ -161,9 +161,30 @@ def _request(url, params, timeout, *, phase, headers=None, attempts=MAX_ATTEMPTS
raise last raise last
def _derive_humidity(df: pd.DataFrame) -> pd.DataFrame: def _normalize_read(df: pl.DataFrame) -> pl.DataFrame:
"""Normalize a frame read from the parquet cache: the daily `date` column is a
calendar date, so coerce it to ``pl.Date`` (older files were written by pandas
as a nanosecond ``Datetime``). Downstream date math and comparisons all key on
``pl.Date`` / stdlib ``datetime.date``."""
if df.schema["date"] != pl.Date:
df = df.with_columns(pl.col("date").cast(pl.Date))
return df
def _combined_feels_expr(hi: str = "fmax", lo: str = "fmin") -> pl.Expr:
"""Expression for 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 cold ones. Falls back to whichever side is present if one is
missing (the coalesce picks the non-null side when a comparison is null)."""
amax, amin = pl.col(hi), pl.col(lo)
hot, cold = amax - COMFORT_F, COMFORT_F - amin
chosen = pl.when(hot >= cold).then(amax).otherwise(amin)
return pl.coalesce([chosen, amax, amin])
def _derive_humidity(df: pl.DataFrame) -> pl.DataFrame:
"""Replace the raw mean *relative* humidity column with *absolute* humidity """Replace the raw mean *relative* humidity column with *absolute* humidity
(grams of water vapor per m³), in place. (grams of water vapor per m³).
Absolute humidity is derived from the day's mean RH and its mean temperature Absolute humidity is derived from the day's mean RH and its mean temperature
((tmax+tmin)/2) via the Magnus saturation-vapor-pressure formula. It's a far ((tmax+tmin)/2) via the Magnus saturation-vapor-pressure formula. It's a far
@ -173,25 +194,23 @@ def _derive_humidity(df: pd.DataFrame) -> pd.DataFrame:
raw RH the archive returns — no refetch needed for existing cells.""" raw RH the archive returns — no refetch needed for existing cells."""
if "humid" not in df.columns: if "humid" not in df.columns:
return df return df
tmean_c = ((df["tmax"] + df["tmin"]) / 2.0 - 32.0) * 5.0 / 9.0 tmean_c = ((pl.col("tmax") + pl.col("tmin")) / 2.0 - 32.0) * 5.0 / 9.0
rh = pd.to_numeric(df["humid"], errors="coerce") rh = pl.col("humid").cast(pl.Float64, strict=False)
es = 6.112 * np.exp(17.67 * tmean_c / (tmean_c + 243.5)) # sat. vapor pressure, hPa es = 6.112 * (17.67 * tmean_c / (tmean_c + 243.5)).exp() # sat. vapor pressure, hPa
df["humid"] = (es * rh * 2.1674 / (273.15 + tmean_c)).round(1) # absolute humidity, g/m³ return df.with_columns(
return df (es * rh * 2.1674 / (273.15 + tmean_c)).round(1).alias("humid")) # abs. humidity, g/m³
def _with_doy(df: pd.DataFrame) -> pd.DataFrame: def _with_doy(df: pl.DataFrame) -> pl.DataFrame:
"""(Re)attach the int16 day-of-year column the grading windows key on.""" """(Re)attach the int16 day-of-year column the grading windows key on."""
df["doy"] = df["date"].dt.dayofyear.astype("int16") return df.with_columns(pl.col("date").dt.ordinal_day().cast(pl.Int16).alias("doy"))
return df
def _write_cache(df: pd.DataFrame, path: str) -> None: def _write_cache(df: pl.DataFrame, path: str) -> None:
"""Persist a daily record to the parquet cache — the raw record only, never """Persist a daily record to the parquet cache — the raw record only, never
the derived doy column (it's recomputed at read time).""" the derived doy column (it's recomputed at read time)."""
os.makedirs(CACHE_DIR, exist_ok=True) os.makedirs(CACHE_DIR, exist_ok=True)
df.drop(columns=["doy"], errors="ignore").to_parquet( df.drop("doy", strict=False).write_parquet(path, compression="zstd")
path, compression="zstd", index=False)
def _om_daily_params(cell: dict, **window) -> dict: def _om_daily_params(cell: dict, **window) -> dict:
@ -210,26 +229,19 @@ def _om_daily_params(cell: dict, **window) -> dict:
} }
def _finalize_frame(df: pd.DataFrame) -> pd.DataFrame: def _finalize_frame(df: pl.DataFrame) -> pl.DataFrame:
"""Shared tail of every source→frame mapping: the combined feels-like, the """Shared tail of every source→frame mapping: unify missing values as null, add
valid-day filter, and the day-of-year column. A usable climate day needs a the combined feels-like, apply the valid-day filter, and attach day-of-year. A
real high/low; other columns may be NaN and simply grade as None.""" usable climate day needs a real high/low; other columns may be missing and
df["feels"] = _combined_feels(df["fmax"], df["fmin"]) simply grade as None. Float NaN (from numpy-derived columns) is folded into null
df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True) so the whole pipeline models "missing" one way — the grading boundary drops it."""
df = df.with_columns(pl.col(pl.Float32, pl.Float64).fill_nan(None))
df = df.with_columns(_combined_feels_expr().alias("feels"))
df = df.drop_nulls(subset=["tmax", "tmin"])
return _with_doy(df) return _with_doy(df)
def _combined_feels(amax: pd.Series, amin: pd.Series) -> pd.Series: def _to_frame(daily: dict) -> pl.DataFrame:
"""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
cold ones. Falls back to whichever side is present if one is missing."""
hot = amax - COMFORT_F
cold = COMFORT_F - amin
feels = amax.where(hot >= cold, amin) # NaN comparisons pick the min side
return feels.fillna(amax).fillna(amin)
def _to_frame(daily: dict) -> pd.DataFrame:
n = len(daily["time"]) n = len(daily["time"])
# Older upstream responses (or a narrowed variable set) may omit a series; fall # Older upstream responses (or a narrowed variable set) may omit a series; fall
# back to an all-null column of the right length so the frame shape is stable. # back to an all-null column of the right length so the frame shape is stable.
@ -237,26 +249,31 @@ def _to_frame(daily: dict) -> pd.DataFrame:
vals = daily.get(key) vals = daily.get(key)
return vals if vals else [None] * n return vals if vals else [None] * n
df = pd.DataFrame( df = pl.DataFrame(
{ {
"date": pd.to_datetime(daily["time"]), "date": daily["time"],
"tmax": daily["temperature_2m_max"], "tmax": daily["temperature_2m_max"],
"tmin": daily["temperature_2m_min"], "tmin": daily["temperature_2m_min"],
"precip": daily["precipitation_sum"], "precip": daily["precipitation_sum"],
"wind": col("wind_speed_10m_max"), "wind": col("wind_speed_10m_max"),
"gust": col("wind_gusts_10m_max"), "gust": col("wind_gusts_10m_max"),
"humid": col("relative_humidity_2m_mean"), "humid": col("relative_humidity_2m_mean"),
}
)
# The apparent (felt) high and low are kept as their own columns — graded # The apparent (felt) high and low are kept as their own columns — graded
# independently — alongside the combined `feels` (whichever side is further # independently — alongside the combined `feels` (whichever side is further
# from the fixed comfort baseline), which the weekly/day views still use. # from the fixed comfort baseline), which the weekly/day views still use.
df["fmax"] = pd.to_numeric(pd.Series(col("apparent_temperature_max")), errors="coerce") "fmax": col("apparent_temperature_max"),
df["fmin"] = pd.to_numeric(pd.Series(col("apparent_temperature_min")), errors="coerce") "fmin": col("apparent_temperature_min"),
}
)
df = df.with_columns(
pl.col("date").str.to_date(),
*[pl.col(c).cast(pl.Float64, strict=False)
for c in ("tmax", "tmin", "precip", "wind", "gust", "fmax", "fmin")],
)
return _finalize_frame(df) return _finalize_frame(df)
def _fetch_history(cell: dict) -> pd.DataFrame: def _fetch_history(cell: dict) -> pl.DataFrame:
end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).isoformat() end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).isoformat()
params = _om_daily_params(cell, start_date=START_DATE, end_date=end) params = _om_daily_params(cell, start_date=START_DATE, end_date=end)
r = _request(ARCHIVE_URL, params, 180, phase="history_fetch") r = _request(ARCHIVE_URL, params, 180, phase="history_fetch")
@ -281,35 +298,43 @@ def _wind_chill(t_f, v_mph):
return np.where((T <= 50) & (V >= 3), wc, T) return np.where((T <= 50) & (V >= 3), wc, T)
def _nasa_to_frame(param: dict) -> pd.DataFrame: def _nasa_to_frame(param: dict) -> pl.DataFrame:
"""Map a NASA POWER daily response to our (tmax/tmin/precip/wind/gust/humid/feels) """Map a NASA POWER daily response to our (tmax/tmin/precip/wind/gust/humid/feels)
schema, converting units (°C→°F, mm→in, m/s→mph) and computing feels-like.""" schema, converting units (°C→°F, mm→in, m/s→mph) and computing feels-like."""
dates = sorted(param.get("T2M_MAX", {}).keys()) dates = sorted(param.get("T2M_MAX", {}).keys())
def col(key, transform): def col(key, transform):
d = param.get(key, {}) d = param.get(key, {})
return [np.nan if (v is None or v <= NASA_FILL) else transform(v) return [None if (v is None or v <= NASA_FILL) else transform(v)
for v in (d.get(k) for k in dates)] for v in (d.get(k) for k in dates)]
c2f = lambda c: c * 9.0 / 5.0 + 32.0 c2f = lambda c: c * 9.0 / 5.0 + 32.0
df = pd.DataFrame({ df = pl.DataFrame({
"date": pd.to_datetime(dates, format="%Y%m%d"), "date": dates,
"tmax": col("T2M_MAX", c2f), "tmax": col("T2M_MAX", c2f),
"tmin": col("T2M_MIN", c2f), "tmin": col("T2M_MIN", c2f),
"precip": col("PRECTOTCORR", lambda mm: mm / 25.4), "precip": col("PRECTOTCORR", lambda mm: mm / 25.4),
"wind": col("WS10M_MAX", lambda ms: ms * 2.2369362920544), "wind": col("WS10M_MAX", lambda ms: ms * 2.2369362920544),
"gust": [np.nan] * len(dates), # POWER has no gusts "gust": [None] * len(dates), # POWER has no gusts
"humid": col("RH2M", lambda v: v), "humid": col("RH2M", lambda v: v),
}) })
df = df.with_columns(
pl.col("date").str.to_date("%Y%m%d"),
*[pl.col(c).cast(pl.Float64, strict=False)
for c in ("tmax", "tmin", "precip", "wind", "gust")],
)
# POWER has no apparent temperature; approximate the felt high with the NWS # POWER has no apparent temperature; approximate the felt high with the NWS
# heat index and the felt low with NWS wind chill (each falls back to the air # heat index and the felt low with NWS wind chill (each falls back to the air
# temperature outside its regime), mirroring the Open-Meteo columns. # temperature outside its regime), mirroring the Open-Meteo columns. numpy NaN
df["fmax"] = pd.Series(_heat_index(df["tmax"], df["humid"]), index=df.index) # from these lands in float columns and is folded to null in _finalize_frame.
df["fmin"] = pd.Series(_wind_chill(df["tmin"], df["wind"]), index=df.index) df = df.with_columns(
pl.Series("fmax", _heat_index(df["tmax"].to_numpy(), df["humid"].to_numpy())),
pl.Series("fmin", _wind_chill(df["tmin"].to_numpy(), df["wind"].to_numpy())),
)
return _finalize_frame(df) return _finalize_frame(df)
def _fetch_history_nasa(cell: dict) -> pd.DataFrame: def _fetch_history_nasa(cell: dict) -> pl.DataFrame:
"""Backup history fetch from NASA POWER (used when Open-Meteo is unavailable).""" """Backup history fetch from NASA POWER (used when Open-Meteo is unavailable)."""
end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).strftime("%Y%m%d") end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).strftime("%Y%m%d")
params = { params = {
@ -325,7 +350,7 @@ def _fetch_history_nasa(cell: dict) -> pd.DataFrame:
return _nasa_to_frame(r.json()["properties"]["parameter"]) return _nasa_to_frame(r.json()["properties"]["parameter"])
def _fetch_history_range(cell: dict, start_date: str, end_date: str) -> pd.DataFrame: def _fetch_history_range(cell: dict, start_date: str, end_date: str) -> pl.DataFrame:
"""Fetch just a date range of archive history (used to top up the recent tail).""" """Fetch just a date range of archive history (used to top up the recent tail)."""
params = _om_daily_params(cell, start_date=start_date, end_date=end_date) params = _om_daily_params(cell, start_date=start_date, end_date=end_date)
r = _request(ARCHIVE_URL, params, 60, phase="history_topup") r = _request(ARCHIVE_URL, params, 60, phase="history_topup")
@ -339,13 +364,13 @@ def _read_history_cache(path):
the only reason to refetch (to add the new metric columns).""" the only reason to refetch (to add the new metric columns)."""
if not os.path.exists(path): if not os.path.exists(path):
return None return None
df = pd.read_parquet(path) df = _normalize_read(pl.read_parquet(path))
if not all(c in df.columns for c in NEW_COLS): if not all(c in df.columns for c in NEW_COLS):
return None return None
return df, time.time() - os.path.getmtime(path) return df, time.time() - os.path.getmtime(path)
def _topup_tail(cell: dict, df: pd.DataFrame, path: str) -> pd.DataFrame: def _topup_tail(cell: dict, df: pl.DataFrame, path: str) -> pl.DataFrame:
"""Append newly-available archive days to a cached record. Best-effort and """Append newly-available archive days to a cached record. Best-effort and
serialized per cell; a small incremental fetch, not the full multi-decade pull.""" serialized per cell; a small incremental fetch, not the full multi-decade pull."""
global _archive_cooldown_until global _archive_cooldown_until
@ -358,7 +383,7 @@ def _topup_tail(cell: dict, df: pd.DataFrame, path: str) -> pd.DataFrame:
if time.time() < _archive_cooldown_until: if time.time() < _archive_cooldown_until:
return df return df
expected = datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS) expected = datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)
cached_max = pd.Timestamp(df["date"].max()).date() cached_max = df["date"].max()
if cached_max >= expected: if cached_max >= expected:
try: os.utime(path, None) # tail already current — reset the hourly timer try: os.utime(path, None) # tail already current — reset the hourly timer
except OSError: pass except OSError: pass
@ -370,22 +395,24 @@ def _topup_tail(cell: dict, df: pd.DataFrame, path: str) -> pd.DataFrame:
if is_rate_limit(e): if is_rate_limit(e):
_note_rate_limit(e) _note_rate_limit(e)
return df return df
merged = (pd.concat([df, recent.drop(columns=["doy"], errors="ignore")]) # Concatenate archive-first, recent-last so `keep="last"` prefers a freshly
.drop_duplicates(subset="date", keep="last") # fetched day over its cached duplicate; maintain_order keeps that precedence
.sort_values("date").reset_index(drop=True)) # before the final chronological sort.
merged = (pl.concat([df, recent.drop("doy", strict=False)], how="diagonal_relaxed")
.unique(subset="date", keep="last", maintain_order=True)
.sort("date"))
_write_cache(merged, path) _write_cache(merged, path)
return merged return merged
def get_history(cell: dict) -> tuple[pd.DataFrame, dict]: def get_history(cell: dict) -> tuple[pl.DataFrame, dict]:
"""Return (daily history frame, cache metadata) for a cell, with humidity as """Return (daily history frame, cache metadata) for a cell, with humidity as
absolute humidity (g/m³). Thin wrapper over the raw loader (see below).""" absolute humidity (g/m³). Thin wrapper over the raw loader (see below)."""
df, meta = _load_history(cell) df, meta = _load_history(cell)
_derive_humidity(df) return _derive_humidity(df), meta
return df, meta
def load_cached_history(cell: dict) -> pd.DataFrame | None: def load_cached_history(cell: dict) -> pl.DataFrame | None:
"""History from the parquet cache ONLY — never fetches upstream and never tops """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 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 quota) and the offline migrate script. None when the cell has no
@ -393,12 +420,10 @@ def load_cached_history(cell: dict) -> pd.DataFrame | None:
hit = _read_history_cache(_cache_path(cell["id"])) hit = _read_history_cache(_cache_path(cell["id"]))
if hit is None: if hit is None:
return None return None
df = _with_doy(hit[0].copy()) return _derive_humidity(_with_doy(hit[0]))
_derive_humidity(df)
return df
def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]: def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]:
"""Return (daily history frame, cache metadata) for a cell. """Return (daily history frame, cache metadata) for a cell.
The full archive is cached indefinitely (fetched once); only the recent tail is The full archive is cached indefinitely (fetched once); only the recent tail is
@ -411,7 +436,7 @@ def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]:
df, age_s = hit df, age_s = hit
if age_s > HISTORY_TOPUP_INTERVAL: if age_s > HISTORY_TOPUP_INTERVAL:
df = _topup_tail(cell, df, path) # refresh just the recent days df = _topup_tail(cell, df, path) # refresh just the recent days
return _with_doy(df.copy()), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)} return _with_doy(df), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
global _archive_cooldown_until global _archive_cooldown_until
@ -421,7 +446,7 @@ def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]:
df, age_s = hit df, age_s = hit
return _with_doy(df), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)} return _with_doy(df), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
stale = pd.read_parquet(path) if os.path.exists(path) else None stale = _normalize_read(pl.read_parquet(path)) if os.path.exists(path) else None
def _serve_stale(): def _serve_stale():
return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None} return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None}
@ -473,15 +498,13 @@ def recent_stamp(cell_id: str) -> int:
return 0 return 0
def get_recent_forecast(cell: dict) -> pd.DataFrame: def get_recent_forecast(cell: dict) -> pl.DataFrame:
"""Recent observations + forward forecast, with humidity as absolute humidity """Recent observations + forward forecast, with humidity as absolute humidity
(g/m³). Thin wrapper over the raw loader (see below).""" (g/m³). Thin wrapper over the raw loader (see below)."""
df = _load_recent_forecast(cell) return _derive_humidity(_load_recent_forecast(cell))
_derive_humidity(df)
return df
def _load_recent_forecast(cell: dict) -> pd.DataFrame: def _load_recent_forecast(cell: dict) -> pl.DataFrame:
"""Recent observations AND the forward forecast in ONE forecast-API call. """Recent observations AND the forward forecast in ONE forecast-API call.
Both the recent (past) view and the forecast (future) view slice from this, so Both the recent (past) view and the forecast (future) view slice from this, so
@ -492,7 +515,7 @@ def _load_recent_forecast(cell: dict) -> pd.DataFrame:
if os.path.exists(path): if os.path.exists(path):
age_h = (time.time() - os.path.getmtime(path)) / 3600.0 age_h = (time.time() - os.path.getmtime(path)) / 3600.0
if age_h < FORECAST_TTL_HOURS: if age_h < FORECAST_TTL_HOURS:
return _with_doy(pd.read_parquet(path)) return _with_doy(_normalize_read(pl.read_parquet(path)))
params = { params = {
"latitude": cell["center_lat"], "latitude": cell["center_lat"],

92
grading.py
View file

@ -5,10 +5,11 @@ 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 around the year end). An observed value is then placed on that distribution as an
empirical percentile and mapped to a human-readable grade. empirical percentile and mapped to a human-readable grade.
""" """
import datetime
import warnings import warnings
import numpy as np import numpy as np
import pandas as pd import polars as pl
HALF_WINDOW = 7 # +/- 7 days -> a 15-day seasonal window HALF_WINDOW = 7 # +/- 7 days -> a 15-day seasonal window
RAIN_THRESHOLD = 0.01 # inches; a day with < this much precip counts as "dry" RAIN_THRESHOLD = 0.01 # inches; a day with < this much precip counts as "dry"
@ -95,6 +96,23 @@ def _band(pct: float, bands) -> tuple[str, str]:
return bands[-1][1], bands[-1][2] return bands[-1][1], bands[-1][2]
def _as_date(d) -> datetime.date:
"""Coerce a date-ish value (``date``, ``datetime``, or ISO string) to a plain
``datetime.date`` — the one date type the grading layer works in."""
if isinstance(d, str):
return datetime.date.fromisoformat(d[:10])
if isinstance(d, datetime.datetime):
return d.date()
return d
def _finite(col: pl.Series) -> np.ndarray:
"""Finite float values of a polars Series as an ndarray, with nulls and NaN
dropped — the frame→numpy bridge every percentile routine grades on."""
a = np.asarray(col.to_numpy(), dtype="float64")
return a[~np.isnan(a)]
def window_mask(doys: np.ndarray, target_doy: int, half: int = HALF_WINDOW) -> np.ndarray: def window_mask(doys: np.ndarray, target_doy: int, half: int = HALF_WINDOW) -> np.ndarray:
diff = np.abs(doys.astype(int) - int(target_doy)) diff = np.abs(doys.astype(int) - int(target_doy))
circular = np.minimum(diff, 366 - diff) circular = np.minimum(diff, 366 - diff)
@ -111,11 +129,11 @@ def empirical_percentile(samples: np.ndarray, value) -> float | None:
return round(100.0 * (less + 0.5 * equal) / n, 1) return round(100.0 * (less + 0.5 * equal) / n, 1)
def climatology(df: pd.DataFrame, target_doy: int) -> dict: def climatology(df: pl.DataFrame, target_doy: int) -> dict:
"""Summarize the +/-7 day historical distribution for one day of the year.""" """Summarize the +/-7 day historical distribution for one day of the year."""
doys = df["doy"].values doys = df["doy"].to_numpy()
sub = df[window_mask(doys, target_doy)] sub = df.filter(window_mask(doys, target_doy))
years = pd.to_datetime(sub["date"]).dt.year years = sub["date"].dt.year()
out = { out = {
"target_doy": int(target_doy), "target_doy": int(target_doy),
"n_samples": int(len(sub)), "n_samples": int(len(sub)),
@ -125,7 +143,7 @@ def climatology(df: pd.DataFrame, target_doy: int) -> dict:
if var not in sub.columns: if var not in sub.columns:
out[var] = None out[var] = None
continue continue
v = sub[var].dropna().values v = _finite(sub[var])
if v.size == 0: if v.size == 0:
out[var] = None out[var] = None
continue continue
@ -193,17 +211,18 @@ def _grade_precip(samples: np.ndarray, value) -> dict | None:
def dry_streaks(dates, precips) -> dict[str, int]: def dry_streaks(dates, precips) -> dict[str, int]:
"""Map each date (ISO string) to days since the last measurable rain, walking a """Map each date (ISO string) to days since the last measurable rain, walking a
chronological precip series. NaN precip counts as a dry day (compares False).""" chronological precip series. Missing precip counts as a dry day. `dates` is an
iterable of ``datetime.date`` (a polars Date column's ``.to_list()``)."""
out: dict[str, int] = {} out: dict[str, int] = {}
streak = 0 streak = 0
for d, p in zip(dates, precips): for d, p in zip(dates, precips):
wet = p is not None and not (isinstance(p, float) and np.isnan(p)) and p >= RAIN_THRESHOLD wet = p is not None and not (isinstance(p, float) and np.isnan(p)) and p >= RAIN_THRESHOLD
streak = 0 if wet else streak + 1 streak = 0 if wet else streak + 1
out[pd.Timestamp(d).date().isoformat()] = streak out[_as_date(d).isoformat()] = streak
return out return out
def grade_range(df: pd.DataFrame, start, end) -> list[dict]: def grade_range(df: pl.DataFrame, start, end) -> list[dict]:
"""Grade every historical day in [start, end] against its own ±7-day window. """Grade every historical day in [start, end] against its own ±7-day window.
Powers the calendar view. Reuses one window per day-of-year across the whole Powers the calendar view. Reuses one window per day-of-year across the whole
@ -211,10 +230,11 @@ def grade_range(df: pd.DataFrame, start, end) -> list[dict]:
day is returned in a compact shape: value (v), percentile (pct), css class (c), day is returned in a compact shape: value (v), percentile (pct), css class (c),
grade label (g). grade label (g).
""" """
start, end = pd.Timestamp(start), pd.Timestamp(end) start, end = _as_date(start), _as_date(end)
full = df.sort_values("date") full = df.sort("date")
doys_all = full["doy"].values doys_all = full["doy"].to_numpy()
cols = {v: full[v].values for v in CLIMO_METRICS if v in full.columns} cols = {v: np.asarray(full[v].to_numpy(), dtype="float64")
for v in CLIMO_METRICS if v in full.columns}
masks: dict[int, np.ndarray] = {} masks: dict[int, np.ndarray] = {}
cache: dict[tuple[int, str], np.ndarray] = {} cache: dict[tuple[int, str], np.ndarray] = {}
@ -234,17 +254,17 @@ def grade_range(df: pd.DataFrame, start, end) -> list[dict]:
# A fixed 14-day lookback would be the bare minimum but still undercounts longer # 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 # 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. # cached per cell) is both correct for any streak length and free.
if full["date"].min() > start - pd.Timedelta(days=DSR_LOOKBACK_MIN_DAYS): if full["date"].min() > start - datetime.timedelta(days=DSR_LOOKBACK_MIN_DAYS):
# Should never happen (history starts in 1980); guards against a future # Should never happen (history starts in 1980); guards against a future
# change that trims history and would silently truncate streaks. # change that trims history and would silently truncate streaks.
warnings.warn("dry-streak lookback shorter than the 14-day minimum", stacklevel=2) warnings.warn("dry-streak lookback shorter than the 14-day minimum", stacklevel=2)
dsr_map = dry_streaks(full["date"].values, cols["precip"]) dsr_map = dry_streaks(full["date"].to_list(), cols["precip"])
sub = full[(full["date"] >= start) & (full["date"] <= end)] sub = full.filter((pl.col("date") >= start) & (pl.col("date") <= end))
out = [] out = []
for _, row in sub.iterrows(): for row in sub.iter_rows(named=True):
doy = int(row["doy"]) doy = int(row["doy"])
date = pd.Timestamp(row["date"]).date().isoformat() date = row["date"].isoformat()
rec = {"date": date, "dsr": dsr_map.get(date)} rec = {"date": date, "dsr": dsr_map.get(date)}
for var in TEMP_METRICS: for var in TEMP_METRICS:
if var not in cols: if var not in cols:
@ -295,32 +315,33 @@ def _precip_ladder(samples: np.ndarray) -> dict | None:
"min": round(float(samples.min()), 2), "max": round(float(samples.max()), 2)} "min": round(float(samples.min()), 2), "max": round(float(samples.max()), 2)}
def day_detail(df: pd.DataFrame, date: pd.Timestamp, obs: dict | None) -> dict: def day_detail(df: pl.DataFrame, date, obs: dict | None) -> dict:
"""Full percentile breakdown for one day: the value at every tier boundary in """Full percentile breakdown for one day: the value at every tier boundary in
its own ±7-day window, plus where the observed values (if any) land. its own ±7-day window, plus where the observed values (if any) land.
Powers the single-day detail page. `obs` may be None when the date isn't yet Powers the single-day detail page. `obs` may be None when the date isn't yet
in the record — then only the climatological ladders are returned.""" in the record — then only the climatological ladders are returned."""
ts = pd.Timestamp(date) d = _as_date(date)
doy = int(ts.dayofyear) doy = d.timetuple().tm_yday
sub = df[window_mask(df["doy"].values, doy)] sub = df.filter(window_mask(df["doy"].to_numpy(), doy))
years = pd.to_datetime(sub["date"]).dt.year years = sub["date"].dt.year()
metrics = {} metrics = {}
for var in TEMP_METRICS: for var in TEMP_METRICS:
if var not in sub.columns: if var not in sub.columns:
continue continue
vals = sub[var].dropna().values vals = _finite(sub[var])
metrics[var] = { metrics[var] = {
"ladder": _temp_ladder(vals), "ladder": _temp_ladder(vals),
"obs": _grade_value(vals, obs.get(var) if obs else None, TEMP_BANDS), "obs": _grade_value(vals, obs.get(var) if obs else None, TEMP_BANDS),
} }
precip = _finite(sub["precip"])
metrics["precip"] = { metrics["precip"] = {
"ladder": _precip_ladder(sub["precip"].dropna().values), "ladder": _precip_ladder(precip),
"obs": _grade_precip(sub["precip"].dropna().values, obs.get("precip") if obs else None), "obs": _grade_precip(precip, obs.get("precip") if obs else None),
} }
return { return {
"date": ts.date().isoformat(), "date": d.isoformat(),
"doy": doy, "doy": doy,
"n_samples": int(len(sub)), "n_samples": int(len(sub)),
"year_range": [int(years.min()), int(years.max())] if len(sub) else None, "year_range": [int(years.min()), int(years.max())] if len(sub) else None,
@ -328,24 +349,25 @@ def day_detail(df: pd.DataFrame, date: pd.Timestamp, obs: dict | None) -> dict:
} }
def grade_day(df: pd.DataFrame, date: pd.Timestamp, obs: dict) -> dict: def grade_day(df: pl.DataFrame, date, obs: dict) -> dict:
"""Grade one observed day against its own day-of-year +/-7 window.""" """Grade one observed day against its own day-of-year +/-7 window."""
doy = int(pd.Timestamp(date).dayofyear) d = _as_date(date)
doys = df["doy"].values doy = d.timetuple().tm_yday
sub = df[window_mask(doys, doy)] doys = df["doy"].to_numpy()
sub = df.filter(window_mask(doys, doy))
result = {"date": pd.Timestamp(date).date().isoformat(), "doy": doy} result = {"date": d.isoformat(), "doy": doy}
for var in TEMP_METRICS: for var in TEMP_METRICS:
result[var] = ( result[var] = (
_grade_value(sub[var].dropna().values, obs.get(var), TEMP_BANDS) _grade_value(_finite(sub[var]), obs.get(var), TEMP_BANDS)
if var in sub.columns else None if var in sub.columns else None
) )
result["precip"] = _grade_precip(sub["precip"].dropna().values, obs.get("precip")) result["precip"] = _grade_precip(_finite(sub["precip"]), obs.get("precip"))
# Per-day normal band (this day-of-year's ±7 window) so the trend chart can # Per-day normal band (this day-of-year's ±7 window) so the trend chart can
# draw the "normal" envelope each actual value is compared against. # draw the "normal" envelope each actual value is compared against.
result["normals"] = { result["normals"] = {
var: _band_stats(sub[var].dropna().values) var: _band_stats(_finite(sub[var]))
for var in CLIMO_METRICS if var in sub.columns for var in CLIMO_METRICS if var in sub.columns
} }

6
migrate.py
View file

@ -21,8 +21,6 @@ import os
import sys import sys
import time import time
import pandas as pd
import climate import climate
import grid import grid
import store import store
@ -47,7 +45,7 @@ def migrate() -> int:
skipped += 1 skipped += 1
continue continue
history = climate.load_cached_history(cell) history = climate.load_cached_history(cell)
if history is None or history.empty: if history is None or history.is_empty():
# Pre-current-schema record; the server refetches it lazily on first # Pre-current-schema record; the server refetches it lazily on first
# touch (see climate.NEW_COLS) — nothing to materialize offline. # touch (see climate.NEW_COLS) — nothing to materialize offline.
print(f" {cell_id}: no schema-complete record — skipped") print(f" {cell_id}: no schema-complete record — skipped")
@ -57,7 +55,7 @@ def migrate() -> int:
token = views.history_token(history) token = views.history_token(history)
start_ts, end_ts = views.cal_span(history, None, None, 24) start_ts, end_ts = views.cal_span(history, None, None, 24)
cal_key = views.calendar_key(start_ts, end_ts, 24) cal_key = views.calendar_key(start_ts, end_ts, 24)
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
day_key = views.day_key(last) day_key = views.day_key(last)
have_cal = store.get_payload("calendar", cell_id, cal_key, token) is not None have_cal = store.get_payload("calendar", cell_id, cal_key, token) is not None

28
notify.py
View file

@ -24,7 +24,7 @@ import os
import threading import threading
import time import time
import pandas as pd import polars as pl
from sqlalchemy import delete, select from sqlalchemy import delete, select
import climate import climate
@ -101,24 +101,26 @@ def _obs_from_row(row) -> dict:
return {k: row[k] for k in OBS_COLS if k in row} return {k: row[k] for k in OBS_COLS if k in row}
def _candidate_rows(recent: pd.DataFrame, kind: str, today: pd.Timestamp): def _candidate_rows(recent, kind: str, today: datetime.date):
"""Rows to grade for a subscription of the given kind, most-relevant first.""" """Rows to grade for a subscription of the given kind, most-relevant first.
if recent is None or recent.empty or "date" not in recent.columns: `today` is a datetime.date; `recent["date"]` is a pl.Date column."""
if recent is None or recent.is_empty() or "date" not in recent.columns:
return [] return []
dates = pd.to_datetime(recent["date"]).dt.normalize()
if kind == "forecast": if kind == "forecast":
mask = (dates > today) & (dates <= today + pd.Timedelta(days=FORECAST_HORIZON_DAYS)) window = (pl.col("date") > today) & (
ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d") # soonest first pl.col("date") <= today + datetime.timedelta(days=FORECAST_HORIZON_DAYS))
ordered = recent.filter(window).sort("date") # soonest first
else: else:
mask = (dates <= today) & (dates >= today - pd.Timedelta(days=LOOKBACK_DAYS)) window = (pl.col("date") <= today) & (
ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d", ascending=False) # newest first pl.col("date") >= today - datetime.timedelta(days=LOOKBACK_DAYS))
return [row for _, row in ordered.iterrows()] ordered = recent.filter(window).sort("date", descending=True) # newest first
return list(ordered.iter_rows(named=True))
def _first_trigger(sub: Subscription, rows, history, grade_cache): def _first_trigger(sub: Subscription, rows, history, grade_cache):
"""The first (event_date, metric, direction, graded) that crosses, or None.""" """The first (event_date, metric, direction, graded) that crosses, or None."""
for row in rows: for row in rows:
date_key = pd.Timestamp(row["date"]).date().isoformat() date_key = row["date"].isoformat()
graded = grade_cache.get(date_key) graded = grade_cache.get(date_key)
if graded is None: if graded is None:
graded = grading.grade_day(history, row["date"], _obs_from_row(row)) graded = grading.grade_day(history, row["date"], _obs_from_row(row))
@ -142,7 +144,7 @@ def _process_cell(session, cell_id, subs, today, now):
except Exception: # noqa: BLE001 - a malformed cell_id shouldn't kill the pass except Exception: # noqa: BLE001 - a malformed cell_id shouldn't kill the pass
return 0 return 0
history = climate.load_cached_history(cell) history = climate.load_cached_history(cell)
if history is None or history.empty: if history is None or history.is_empty():
return 0 # nothing cached yet — never spend archive quota from here return 0 # nothing cached yet — never spend archive quota from here
try: try:
recent = climate.get_recent_forecast(cell) recent = climate.get_recent_forecast(cell)
@ -205,7 +207,7 @@ def run_pass() -> int:
"""One full evaluation sweep over all active subscriptions. Returns the number """One full evaluation sweep over all active subscriptions. Returns the number
of notifications created.""" of notifications created."""
now = time.time() now = time.time()
today = pd.Timestamp(datetime.date.today()) today = datetime.date.today()
created = 0 created = 0
with sync_session_maker() as session: with sync_session_maker() as session:
subs = session.execute( subs = session.execute(

3
requirements.txt
View file

@ -1,8 +1,7 @@
fastapi==0.115.6 fastapi==0.115.6
uvicorn[standard]==0.34.0 uvicorn[standard]==0.34.0
httpx==0.28.1 httpx==0.28.1
pandas==2.2.3 polars==1.42.1
pyarrow==18.1.0
numpy==2.2.1 numpy==2.2.1
# Accounts + notification subscriptions (see backend/db.py, users.py, notify.py). # Accounts + notification subscriptions (see backend/db.py, users.py, notify.py).
fastapi-users[sqlalchemy]==15.0.5 fastapi-users[sqlalchemy]==15.0.5

51
tests/conftest.py
View file

@ -3,13 +3,14 @@
Everything here keeps the suite hermetic — no Open-Meteo, no Nominatim, no Everything here keeps the suite hermetic — no Open-Meteo, no Nominatim, no
GeoNames download, and no writes into the repo's data/ or logs/ folders. GeoNames download, and no writes into the repo's data/ or logs/ folders.
""" """
import datetime
import os import os
import sys import sys
import tempfile import tempfile
import threading import threading
import numpy as np import numpy as np
import pandas as pd import polars as pl
import pytest import pytest
BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@ -35,45 +36,59 @@ audit.AUDIT_DIR = os.path.join(_TMP, "logs", "audit")
audit.ERROR_DIR = os.path.join(_TMP, "logs", "errors") audit.ERROR_DIR = os.path.join(_TMP, "logs", "errors")
def make_history(years: int = 20, end: str | None = None, seed: int = 7) -> pd.DataFrame: def _years_before(d: datetime.date, years: int) -> datetime.date:
"""`d` shifted back `years` years, same month/day (Feb 29 → Feb 28)."""
try:
return d.replace(year=d.year - years)
except ValueError:
return d.replace(year=d.year - years, day=28)
def _daily(start: datetime.date, end: datetime.date) -> list[datetime.date]:
"""Inclusive daily date list from `start` to `end`."""
return [start + datetime.timedelta(days=i) for i in range((end - start).days + 1)]
def _with_doy(df: pl.DataFrame) -> pl.DataFrame:
return df.with_columns(pl.col("date").dt.ordinal_day().cast(pl.Int16).alias("doy"))
def make_history(years: int = 20, end: str | None = None, seed: int = 7) -> pl.DataFrame:
"""A plausible daily record (tmax/tmin/precip + doy), matching the columns """A plausible daily record (tmax/tmin/precip + doy), matching the columns
and dtypes climate.get_history returns for an older cache.""" and dtypes climate.get_history returns for an older cache."""
end_ts = pd.Timestamp(end) if end else pd.Timestamp.today().normalize() - pd.Timedelta(days=6) end_ts = (datetime.date.fromisoformat(end) if end
dates = pd.date_range(end_ts - pd.DateOffset(years=years), end_ts, freq="D") else datetime.date.today() - datetime.timedelta(days=6))
dates = _daily(_years_before(end_ts, years), end_ts)
rng = np.random.default_rng(seed) rng = np.random.default_rng(seed)
doy = dates.dayofyear.to_numpy() doy = np.array([d.timetuple().tm_yday for d in dates])
seasonal = 55 + 30 * np.sin((doy - 100) / 366.0 * 2 * np.pi) seasonal = 55 + 30 * np.sin((doy - 100) / 366.0 * 2 * np.pi)
tmax = seasonal + rng.normal(0, 8, len(dates)) tmax = seasonal + rng.normal(0, 8, len(dates))
tmin = tmax - 15 + rng.normal(0, 3, len(dates)) tmin = tmax - 15 + rng.normal(0, 3, len(dates))
precip = np.where(rng.random(len(dates)) < 0.3, rng.gamma(1.5, 0.2, len(dates)), 0.0) precip = np.where(rng.random(len(dates)) < 0.3, rng.gamma(1.5, 0.2, len(dates)), 0.0)
df = pd.DataFrame({ return _with_doy(pl.DataFrame({
"date": dates, "date": dates,
"tmax": np.round(tmax, 1), "tmax": np.round(tmax, 1),
"tmin": np.round(tmin, 1), "tmin": np.round(tmin, 1),
"precip": np.round(precip, 2), "precip": np.round(precip, 2),
}) }))
df["doy"] = df["date"].dt.dayofyear.astype("int16")
return df
def make_recent(history: pd.DataFrame, future_days: int = 7, seed: int = 11) -> pd.DataFrame: def make_recent(history: pl.DataFrame, future_days: int = 7, seed: int = 11) -> pl.DataFrame:
"""A recent+forecast bundle: from a couple of weeks before the archive's end """A recent+forecast bundle: from a couple of weeks before the archive's end
through `future_days` past today — the shape climate.get_recent_forecast returns.""" through `future_days` past today — the shape climate.get_recent_forecast returns."""
today = pd.Timestamp.today().normalize() today = datetime.date.today()
start = pd.Timestamp(history["date"].max()) - pd.Timedelta(days=14) start = history["date"].max() - datetime.timedelta(days=14)
dates = pd.date_range(start, today + pd.Timedelta(days=future_days), freq="D") dates = _daily(start, today + datetime.timedelta(days=future_days))
rng = np.random.default_rng(seed) rng = np.random.default_rng(seed)
doy = dates.dayofyear.to_numpy() doy = np.array([d.timetuple().tm_yday for d in dates])
seasonal = 55 + 30 * np.sin((doy - 100) / 366.0 * 2 * np.pi) seasonal = 55 + 30 * np.sin((doy - 100) / 366.0 * 2 * np.pi)
tmax = seasonal + rng.normal(0, 8, len(dates)) tmax = seasonal + rng.normal(0, 8, len(dates))
df = pd.DataFrame({ return _with_doy(pl.DataFrame({
"date": dates, "date": dates,
"tmax": np.round(tmax, 1), "tmax": np.round(tmax, 1),
"tmin": np.round(tmax - 15, 1), "tmin": np.round(tmax - 15, 1),
"precip": np.where(rng.random(len(dates)) < 0.3, 0.15, 0.0), "precip": np.where(rng.random(len(dates)) < 0.3, 0.15, 0.0),
}) }))
df["doy"] = df["date"].dt.dayofyear.astype("int16")
return df
@pytest.fixture(scope="session") @pytest.fixture(scope="session")

14
tests/test_api.py
View file

@ -1,7 +1,6 @@
"""Route-level tests over the FastAPI app with the weather/geocode layer faked — """Route-level tests over the FastAPI app with the weather/geocode layer faked —
they exercise the real routing, validation, derived-store and ETag plumbing, and they exercise the real routing, validation, derived-store and ETag plumbing, and
would catch wiring regressions (e.g. a handler calling a deleted helper).""" would catch wiring regressions (e.g. a handler calling a deleted helper)."""
import pandas as pd
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@ -12,9 +11,9 @@ import climate
@pytest.fixture @pytest.fixture
def client(monkeypatch, history, recent): def client(monkeypatch, history, recent):
monkeypatch.setattr(climate, "get_history", monkeypatch.setattr(climate, "get_history",
lambda cell: (history.copy(), {"cached": True, "cache_age_days": 3})) lambda cell: (history.clone(), {"cached": True, "cache_age_days": 3}))
monkeypatch.setattr(climate, "get_recent_forecast", lambda cell: recent.copy()) monkeypatch.setattr(climate, "get_recent_forecast", lambda cell: recent.clone())
monkeypatch.setattr(climate, "load_cached_history", lambda cell: history.copy()) monkeypatch.setattr(climate, "load_cached_history", lambda cell: history.clone())
monkeypatch.setattr(climate, "recent_stamp", lambda cell_id: "rs-test") monkeypatch.setattr(climate, "recent_stamp", lambda cell_id: "rs-test")
monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: "Testville, Washington") monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: "Testville, Washington")
return TestClient(appmod.app) return TestClient(appmod.app)
@ -67,7 +66,7 @@ def test_day_detail_and_ladders(client, history):
r = client.get("/thermograph/api/v2/day", params=Q) r = client.get("/thermograph/api/v2/day", params=Q)
assert r.status_code == 200 assert r.status_code == 200
body = r.json() body = r.json()
assert body["latest"] == pd.Timestamp(history["date"].max()).date().isoformat() assert body["latest"] == history["date"].max().isoformat()
tmax = body["detail"]["metrics"]["tmax"] tmax = body["detail"]["metrics"]["tmax"]
assert tmax["ladder"]["tiers"][0]["c"] == "rec-hot" assert tmax["ladder"]["tiers"][0]["c"] == "rec-hot"
assert tmax["obs"]["grade"] assert tmax["obs"]["grade"]
@ -142,7 +141,7 @@ def test_calendar_compact_range(client, history):
r = client.get("/thermograph/api/v2/calendar", params={**Q, "months": 2}) r = client.get("/thermograph/api/v2/calendar", params={**Q, "months": 2})
assert r.status_code == 200 assert r.status_code == 200
body = r.json() body = r.json()
assert body["range"]["end"] == pd.Timestamp(history["date"].max()).date().isoformat() assert body["range"]["end"] == history["date"].max().isoformat()
day = body["days"][0] day = body["days"][0]
assert {"date", "dsr", "tmax", "tmin", "precip"} <= set(day) assert {"date", "dsr", "tmax", "tmin", "precip"} <= set(day)
assert set(day["tmax"]) == {"v", "pct", "c", "g"} assert set(day["tmax"]) == {"v", "pct", "c", "g"}
@ -220,8 +219,7 @@ def test_warm_cell_materializes_history_slices(client, tmp_store, monkeypatch, h
start_ts, end_ts = views.cal_span(history, None, None, 24) start_ts, end_ts = views.cal_span(history, None, None, 24)
assert tmp_store.get_payload("calendar", cell["id"], assert tmp_store.get_payload("calendar", cell["id"],
views.calendar_key(start_ts, end_ts, 24), token) is not None views.calendar_key(start_ts, end_ts, 24), token) is not None
import pandas as pd last = history["date"].max()
last = pd.Timestamp(history["date"].max()).normalize()
assert tmp_store.get_payload("day", cell["id"], views.day_key(last), token) is not None assert tmp_store.get_payload("day", cell["id"], views.day_key(last), token) is not None

29
tests/test_climate.py
View file

@ -1,7 +1,8 @@
"""The source→frame mappings and cache-write path (pure parts of climate.py — """The source→frame mappings and cache-write path (pure parts of climate.py —
no network).""" no network)."""
import numpy as np import datetime
import pandas as pd
import polars as pl
import climate import climate
@ -25,7 +26,7 @@ def test_to_frame_schema_and_day_filter():
df = climate._to_frame(_om_daily()) df = climate._to_frame(_om_daily())
# The None tmax day is dropped: a usable climate day needs a real high/low. # The None tmax day is dropped: a usable climate day needs a real high/low.
assert len(df) == 2 assert len(df) == 2
assert df["doy"].dtype == np.int16 assert df["doy"].dtype == pl.Int16
assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind", "gust", assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind", "gust",
"humid", "fmax", "fmin", "feels", "doy"} "humid", "fmax", "fmin", "feels", "doy"}
@ -34,15 +35,25 @@ def test_to_frame_tolerates_missing_series():
daily = _om_daily() daily = _om_daily()
del daily["wind_gusts_10m_max"], daily["relative_humidity_2m_mean"] del daily["wind_gusts_10m_max"], daily["relative_humidity_2m_mean"]
df = climate._to_frame(daily) df = climate._to_frame(daily)
assert df["gust"].isna().all() and df["humid"].isna().all() assert df["gust"].is_null().all() and df["humid"].is_null().all()
assert df["tmax"].notna().all() # required series unaffected assert df["tmax"].is_not_null().all() # required series unaffected
def test_combined_feels_picks_the_extreme_side(): def test_combined_feels_picks_the_extreme_side():
df = climate._to_frame(_om_daily()) df = climate._to_frame(_om_daily())
# Day 2: fmax 95 is 30 past the 65°F comfort point vs fmin 68 only 3 below — # 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. # the hot side wins; both days here are hot-side days.
assert df.loc[df["date"] == "2026-06-02", "feels"].item() == 95.0 assert df.filter(pl.col("date") == datetime.date(2026, 6, 2))["feels"].item() == 95.0
def test_combined_feels_falls_back_to_the_present_side():
"""When one apparent-temperature side is missing, `feels` uses whichever side
is present (the null-vs-value coalesce must reproduce the old NaN fallback)."""
daily = _om_daily()
daily["apparent_temperature_max"] = [None, None, None] # hot side absent
df = climate._to_frame(daily)
row = df.filter(pl.col("date") == datetime.date(2026, 6, 1)).row(0, named=True)
assert row["feels"] == 58.0 # falls back to apparent low
def test_nasa_to_frame_converts_units_and_fills(): def test_nasa_to_frame_converts_units_and_fills():
@ -55,11 +66,11 @@ def test_nasa_to_frame_converts_units_and_fills():
} }
df = climate._nasa_to_frame(param) df = climate._nasa_to_frame(param)
assert len(df) == 1 # the missing-tmax day dropped assert len(df) == 1 # the missing-tmax day dropped
row = df.iloc[0] row = df.row(0, named=True)
assert row["tmax"] == 86.0 # 30°C assert row["tmax"] == 86.0 # 30°C
assert row["precip"] == 1.0 # 25.4mm = 1in assert row["precip"] == 1.0 # 25.4mm = 1in
assert round(row["wind"], 1) == 22.4 # 10 m/s in mph assert round(row["wind"], 1) == 22.4 # 10 m/s in mph
assert np.isnan(row["gust"]) # POWER has no gusts assert row["gust"] is None # POWER has no gusts (missing -> null)
assert row["fmax"] > row["tmax"] # ≥80°F: the NWS heat index applies assert row["fmax"] > row["tmax"] # ≥80°F: the NWS heat index applies
@ -77,6 +88,6 @@ def test_write_cache_strips_the_derived_doy(tmp_path):
df = climate._to_frame(_om_daily()) df = climate._to_frame(_om_daily())
path = str(tmp_path / "cell.parquet") path = str(tmp_path / "cell.parquet")
climate._write_cache(df, path) climate._write_cache(df, path)
stored = pd.read_parquet(path) stored = pl.read_parquet(path)
assert "doy" not in stored.columns assert "doy" not in stored.columns
assert len(stored) == len(df) assert len(stored) == len(df)

18
tests/test_grading.py
View file

@ -1,5 +1,7 @@
import datetime
import numpy as np import numpy as np
import pandas as pd import polars as pl
import pytest import pytest
import grading import grading
@ -82,17 +84,17 @@ def test_rain_with_no_historical_rain_days_is_extreme():
# ---- dry streaks --------------------------------------------------------------- # ---- dry streaks ---------------------------------------------------------------
def test_dry_streaks_walk(): def test_dry_streaks_walk():
dates = pd.date_range("2024-01-01", periods=5, freq="D") dates = [datetime.date(2024, 1, 1) + datetime.timedelta(days=i) for i in range(5)]
precips = [0.5, 0.0, float("nan"), 0.02, 0.005] precips = [0.5, 0.0, float("nan"), 0.02, 0.005]
out = grading.dry_streaks(dates.values, precips) out = grading.dry_streaks(dates, precips)
assert list(out.values()) == [0, 1, 2, 0, 1] # NaN counts as dry assert list(out.values()) == [0, 1, 2, 0, 1] # NaN counts as dry
# ---- range + day grading over a synthetic record -------------------------------- # ---- range + day grading over a synthetic record --------------------------------
def test_grade_range_compact_shape(history): def test_grade_range_compact_shape(history):
end = pd.Timestamp(history["date"].max()) end = history["date"].max()
start = end - pd.Timedelta(days=30) start = end - datetime.timedelta(days=30)
days = grading.grade_range(history, start, end) days = grading.grade_range(history, start, end)
assert len(days) == 31 assert len(days) == 31
first = days[0] first = days[0]
@ -108,8 +110,8 @@ def test_grade_range_compact_shape(history):
def test_grade_day_normals_and_departure(history): def test_grade_day_normals_and_departure(history):
target = pd.Timestamp(history["date"].max()) target = history["date"].max()
row = history[history["date"] == target].iloc[0] row = history.filter(pl.col("date") == target).row(0, named=True)
result = grading.grade_day(history, target, {"tmax": row["tmax"], "tmin": row["tmin"], result = grading.grade_day(history, target, {"tmax": row["tmax"], "tmin": row["tmin"],
"precip": row["precip"]}) "precip": row["precip"]})
assert set(result["normals"]["tmax"]) == {"p1", "p10", "p25", "p40", "p50", "p60", assert set(result["normals"]["tmax"]) == {"p1", "p10", "p25", "p40", "p50", "p60",
@ -119,7 +121,7 @@ def test_grade_day_normals_and_departure(history):
def test_day_detail_with_and_without_observation(history): def test_day_detail_with_and_without_observation(history):
target = pd.Timestamp(history["date"].max()) target = history["date"].max()
detail = grading.day_detail(history, target, {"tmax": 60.0, "precip": 0.0}) detail = grading.day_detail(history, target, {"tmax": 60.0, "precip": 0.0})
assert detail["metrics"]["tmax"]["obs"]["value"] == 60.0 assert detail["metrics"]["tmax"]["obs"]["value"] == 60.0
assert detail["metrics"]["tmax"]["ladder"]["tiers"][0]["c"] == "rec-hot" assert detail["metrics"]["tmax"]["ladder"]["tiers"][0]["c"] == "rec-hot"

23
tests/test_notify.py
View file

@ -1,23 +1,26 @@
"""Unit tests for the subscription evaluation engine's pure logic (no DB, no """Unit tests for the subscription evaluation engine's pure logic (no DB, no
network): trigger detection against a synthetic climatology and message wording.""" network): trigger detection against a synthetic climatology and message wording."""
import datetime
import types import types
import numpy as np import numpy as np
import pandas as pd import polars as pl
import notify import notify
def _history() -> pd.DataFrame: def _history() -> pl.DataFrame:
dates = pd.date_range("1990-01-01", "2020-12-31", freq="D") start, end = datetime.date(1990, 1, 1), datetime.date(2020, 12, 31)
dates = [start + datetime.timedelta(days=i) for i in range((end - start).days + 1)]
rng = np.random.default_rng(1) rng = np.random.default_rng(1)
n = len(dates) n = len(dates)
df = pd.DataFrame({"date": dates}) return pl.DataFrame({
df["doy"] = df["date"].dt.dayofyear "date": dates,
df["tmax"] = 70 + rng.normal(0, 8, n) "doy": np.array([d.timetuple().tm_yday for d in dates], dtype="int16"),
df["tmin"] = 48 + rng.normal(0, 7, n) "tmax": 70 + rng.normal(0, 8, n),
df["precip"] = rng.exponential(0.05, n) "tmin": 48 + rng.normal(0, 7, n),
return df "precip": rng.exponential(0.05, n),
})
def _sub(**kw): def _sub(**kw):
@ -28,7 +31,7 @@ def _sub(**kw):
def _row(date, **vals): def _row(date, **vals):
r = {"date": pd.Timestamp(date)} r = {"date": datetime.date.fromisoformat(date)}
r.update(vals) r.update(vals)
return r return r

57
tests/test_views.py
View file

@ -1,10 +1,11 @@
"""The payload layer: pure builders, span clamping, and the layering guarantee """The payload layer: pure builders, span clamping, and the layering guarantee
that offline callers (migrate) can use it without dragging in the web stack.""" that offline callers (migrate) can use it without dragging in the web stack."""
import datetime
import os import os
import subprocess import subprocess
import sys import sys
import pandas as pd import polars as pl
import pytest import pytest
import climate import climate
@ -30,9 +31,9 @@ def test_views_and_migrate_import_without_the_web_stack():
# every existing derived row — change them only alongside a PAYLOAD_VER bump). # every existing derived row — change them only alongside a PAYLOAD_VER bump).
def test_cache_identity_formats_are_pinned(history): def test_cache_identity_formats_are_pinned(history):
t = pd.Timestamp("2026-06-15") t = datetime.date(2026, 6, 15)
assert views.grade_key(t, 14, 7) == "2026-06-15:14:7" assert views.grade_key(t, 14, 7) == "2026-06-15:14:7"
assert views.calendar_key(t, pd.Timestamp("2026-06-20"), 24) == "2026-06-15:2026-06-20:24" assert views.calendar_key(t, datetime.date(2026, 6, 20), 24) == "2026-06-15:2026-06-20:24"
assert views.day_key(t) == "2026-06-15" assert views.day_key(t) == "2026-06-15"
assert views.forecast_key(t, 7) == "2026-06-15:7" assert views.forecast_key(t, 7) == "2026-06-15:7"
assert views.history_token(history) == f"{views.PAYLOAD_VER}:{views.hist_end(history)}" assert views.history_token(history) == f"{views.PAYLOAD_VER}:{views.hist_end(history)}"
@ -44,31 +45,32 @@ def test_recent_token_composes_history_and_stamp(history, monkeypatch):
def test_day_token_expires_hourly_only_beyond_the_archive(history): def test_day_token_expires_hourly_only_beyond_the_archive(history):
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
assert views.day_token(history, last) == views.history_token(history) assert views.day_token(history, last) == views.history_token(history)
future = views.day_token(history, last + pd.Timedelta(days=1)) future = views.day_token(history, last + datetime.timedelta(days=1))
assert future.startswith(views.history_token(history) + ":h") assert future.startswith(views.history_token(history) + ":h")
# ---- cal_span clamping --------------------------------------------------------- # ---- cal_span clamping ---------------------------------------------------------
def _hist(start, end): def _hist(start, end):
df = pd.DataFrame({"date": pd.date_range(start, end, freq="D")}) # cal_span only reads the record's min/max date, so two rows pin the span.
return df return pl.DataFrame({"date": [datetime.date.fromisoformat(start),
datetime.date.fromisoformat(end)]})
def test_cal_span_defaults_to_months_back_same_day_of_month(): 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) 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 end_ts == datetime.date(2026, 6, 29)
assert start_ts == pd.Timestamp("2024-06-29") assert start_ts == datetime.date(2024, 6, 29)
def test_cal_span_clamps_to_the_record(): def test_cal_span_clamps_to_the_record():
h = _hist("2025-03-01", "2026-06-29") # record shorter than the 2-year cap 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) 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 assert start_ts == datetime.date(2025, 3, 1) # can't start before the record
_, end_ts = views.cal_span(h, None, "2030-01-01", 24) _, end_ts = views.cal_span(h, None, "2030-01-01", 24)
assert end_ts == pd.Timestamp("2026-06-29") # nor end past it assert end_ts == datetime.date(2026, 6, 29) # nor end past it
def test_cal_span_caps_at_two_years(): def test_cal_span_caps_at_two_years():
@ -83,13 +85,32 @@ def test_cal_span_never_inverts():
assert start_ts == end_ts assert start_ts == end_ts
def test_months_before_clamps_month_end_and_leap():
"""Calendar-aware month subtraction (replaces pandas DateOffset): land on the
same day-of-month, clamped to the target month's last valid day."""
assert views._months_before(datetime.date(2026, 3, 31), 1) == datetime.date(2026, 2, 28)
assert views._months_before(datetime.date(2024, 3, 31), 1) == datetime.date(2024, 2, 29)
assert views._months_before(datetime.date(2026, 1, 15), 14) == datetime.date(2024, 11, 15)
def test_attach_dry_streaks_prefers_the_fresher_source_on_shared_dates():
"""De-dup keeps the recent/forecast row over the archive one for a shared date
(concat archive-first, keep='last') — the fresher precip drives the streak."""
d = datetime.date(2026, 1, 1)
hist = pl.DataFrame({"date": [d], "precip": [1.0]}) # archive: it rained (streak resets)
rec = pl.DataFrame({"date": [d], "precip": [0.0]}) # fresher: dry (streak counts)
graded = [{"date": d.isoformat()}]
views._attach_dry_streaks(graded, hist, rec) # archive first, recent last
assert graded[0]["dsr"] == 1 # recent (dry) won
# ---- builders ------------------------------------------------------------------- # ---- builders -------------------------------------------------------------------
def test_build_grade_window_and_shape(history, recent): def test_build_grade_window_and_shape(history, recent):
target = pd.Timestamp.today().normalize() target = datetime.date.today()
payload = views.build_grade(CELL, target, 14, history, recent, payload = views.build_grade(CELL, target, 14, history, recent,
{"cached": True}, "Testville") {"cached": True}, "Testville")
assert payload["target_date"] == target.date().isoformat() assert payload["target_date"] == target.isoformat()
days = [d["date"] for d in payload["recent"]] days = [d["date"] for d in payload["recent"]]
assert days == sorted(days, reverse=True) # newest first assert days == sorted(days, reverse=True) # newest first
assert payload["climatology"]["tmax"] is not None assert payload["climatology"]["tmax"] is not None
@ -97,9 +118,9 @@ def test_build_grade_window_and_shape(history, recent):
def test_build_day_pulls_future_obs_from_recent(history, recent): def test_build_day_pulls_future_obs_from_recent(history, recent):
today = pd.Timestamp.today().normalize() today = datetime.date.today()
payload = views.build_day(CELL, history, today, "Testville", recent=recent) payload = views.build_day(CELL, history, today, "Testville", recent=recent)
assert payload["detail"]["date"] == today.date().isoformat() assert payload["detail"]["date"] == today.isoformat()
assert payload["detail"]["metrics"]["tmax"]["obs"] is not None assert payload["detail"]["metrics"]["tmax"]["obs"] is not None
@ -107,16 +128,16 @@ def test_build_day_survives_recent_fetch_failure(history, monkeypatch):
def boom(cell): def boom(cell):
raise RuntimeError("upstream down") raise RuntimeError("upstream down")
monkeypatch.setattr(climate, "get_recent_forecast", boom) monkeypatch.setattr(climate, "get_recent_forecast", boom)
today = pd.Timestamp.today().normalize() today = datetime.date.today()
payload = views.build_day(CELL, history, today, "Testville") payload = views.build_day(CELL, history, today, "Testville")
assert payload["detail"]["metrics"]["tmax"]["obs"] is None # climatology only assert payload["detail"]["metrics"]["tmax"]["obs"] is None # climatology only
assert payload["detail"]["metrics"]["tmax"]["ladder"] is not None assert payload["detail"]["metrics"]["tmax"]["ladder"] is not None
def test_build_forecast_only_future_days(history, recent): def test_build_forecast_only_future_days(history, recent):
today = pd.Timestamp.today().normalize() today = datetime.date.today()
payload = views.build_forecast(CELL, 7, history, recent, today, None) payload = views.build_forecast(CELL, 7, history, recent, today, None)
days = [d["date"] for d in payload["recent"]] days = [d["date"] for d in payload["recent"]]
assert days == sorted(days, reverse=True) # furthest-out first assert days == sorted(days, reverse=True) # furthest-out first
assert min(days) > today.date().isoformat() assert min(days) > today.isoformat()
assert payload["forecast"] is True assert payload["forecast"] is True

104
views.py
View file

@ -8,9 +8,10 @@ semantics (audit runs, derived-store lookups, ETags) stay with the callers;
``run`` is an audit.RunAudit or None. ``run`` is an audit.RunAudit or None.
""" """
import contextlib import contextlib
import datetime
import time import time
import pandas as pd import polars as pl
import climate import climate
import grading import grading
@ -40,11 +41,22 @@ class NullRun:
return contextlib.nullcontext() return contextlib.nullcontext()
def _months_before(d: datetime.date, months: int) -> datetime.date:
"""`d` shifted back `months` calendar months, landing on the same day-of-month
(clamped to the last valid day, e.g. Mar 31 → Feb 28). Mirrors pandas
``DateOffset(months=...)`` for the default calendar-range start."""
m = d.month - 1 - months
year, month = d.year + m // 12, m % 12 + 1
nxt = datetime.date(year + 1, 1, 1) if month == 12 else datetime.date(year, month + 1, 1)
last_day = (nxt - datetime.timedelta(days=1)).day
return datetime.date(year, month, min(d.day, last_day))
def hist_end(history) -> str: def hist_end(history) -> str:
"""Last day in the cell's archive record — the freshness token for everything """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 derived purely from history. Advances via the hourly tail top-up inside
climate.get_history, which cached payloads follow automatically.""" climate.get_history, which cached payloads follow automatically."""
return pd.Timestamp(history["date"].max()).date().isoformat() return history["date"].max().isoformat()
# --- cache identity ------------------------------------------------------------ # --- cache identity ------------------------------------------------------------
@ -66,15 +78,15 @@ def recent_token(history, cell_id: str) -> str:
def grade_key(target, days: int, after: int) -> str: def grade_key(target, days: int, after: int) -> str:
return f"{target.date().isoformat()}:{days}:{after}" return f"{target.isoformat()}:{days}:{after}"
def calendar_key(start_ts, end_ts, months: int) -> str: def calendar_key(start_ts, end_ts, months: int) -> str:
return f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:{months}" return f"{start_ts.isoformat()}:{end_ts.isoformat()}:{months}"
def day_key(target) -> str: def day_key(target) -> str:
return target.date().isoformat() return target.isoformat()
def day_token(history, target) -> str: def day_token(history, target) -> str:
@ -82,36 +94,39 @@ def day_token(history, target) -> str:
day's observation comes from the hourly recent bundle, so its payload expires day's observation comes from the hourly recent bundle, so its payload expires
on that bundle's own cadence (hourly).""" on that bundle's own cadence (hourly)."""
token = history_token(history) token = history_token(history)
if target > pd.Timestamp(history["date"].max()).normalize(): if target > history["date"].max():
token += f":h{int(time.time() // 3600)}" token += f":h{int(time.time() // 3600)}"
return token return token
def forecast_key(today, days: int) -> str: def forecast_key(today, days: int) -> str:
return f"{today.date().isoformat()}:{days}" return f"{today.isoformat()}:{days}"
def _obs_from_row(row) -> dict: def _obs_from_row(row: dict) -> dict:
return {k: row[k] for k in OBS_COLS if k in row} return {k: row[k] for k in OBS_COLS if k in row}
def _grade_rows(history, rows_df) -> list[dict]: def _grade_rows(history, rows_df) -> list[dict]:
"""Grade each daily row against its own ±7-day climatology window.""" """Grade each daily row against its own ±7-day climatology window."""
return [grading.grade_day(history, row["date"], _obs_from_row(row)) return [grading.grade_day(history, row["date"], _obs_from_row(row))
for _, row in rows_df.iterrows()] for row in rows_df.iter_rows(named=True)]
def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None: def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None:
"""Attach `dsr` (days since last measurable rain) to each graded day, computed """Attach `dsr` (days since last measurable rain) to each graded day, computed
over a combined, de-duplicated precip series so streaks stay continuous across over a combined, de-duplicated precip series so streaks stay continuous across
the history / recent / forecast sources.""" the history / recent / forecast sources."""
frames = [f[["date", "precip"]] for f in precip_frames if f is not None and not f.empty] frames = [f.select("date", "precip") for f in precip_frames
if f is not None and f.height]
if not frames: if not frames:
return return
combined = (pd.concat(frames) # Sources are passed archive-first, recent/forecast-last, so `keep="last"`
.drop_duplicates(subset="date", keep="last") # prefers the fresher source's row for any shared date (before the final sort).
.sort_values("date")) combined = (pl.concat(frames, how="vertical_relaxed")
dsr_map = grading.dry_streaks(combined["date"].values, combined["precip"].values) .unique(subset="date", keep="last", maintain_order=True)
.sort("date"))
dsr_map = grading.dry_streaks(combined["date"].to_list(), combined["precip"].to_numpy())
for g in graded: for g in graded:
g["dsr"] = dsr_map.get(g["date"]) g["dsr"] = dsr_map.get(g["date"])
@ -127,25 +142,28 @@ def build_grade(cell, target, days, history, recent, cache_meta, place, run=None
plus 1-7 forecast days, while an older target fills the whole 14 with real obs.""" plus 1-7 forecast days, while an older target fills the whole 14 with real obs."""
run = run or NullRun() run = run or NullRun()
with run.phase("grading"): with run.phase("grading"):
lo = target - pd.Timedelta(days=days) lo = target - datetime.timedelta(days=days)
hi = target + pd.Timedelta(days=after) hi = target + datetime.timedelta(days=after)
# Build the window from both sources. The recent+forecast bundle covers the # 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); # 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 archive reaches decades back for targets older than that bundle. Prefer
# the bundle row for any given date, filling the rest from the archive. # the bundle row for any given date, filling the rest from the archive.
rwin = recent[(recent["date"] >= lo) & (recent["date"] <= hi)] rwin = recent.filter((pl.col("date") >= lo) & (pl.col("date") <= hi))
hwin = history[(history["date"] >= lo) & (history["date"] <= hi)] hwin = history.filter((pl.col("date") >= lo) & (pl.col("date") <= hi))
hwin = hwin[~hwin["date"].isin(set(rwin["date"]))] hwin = hwin.join(rwin.select("date"), on="date", how="anti")
window = pd.concat([rwin, hwin]).sort_values("date") # diagonal_relaxed aligns by column name and null-fills gaps, matching the
# old pandas concat when the two sources' columns differ (e.g. a reduced
# recent bundle); grading treats an absent column's value as None.
window = pl.concat([rwin, hwin], how="diagonal_relaxed").sort("date")
graded = _grade_rows(history, window) graded = _grade_rows(history, window)
_attach_dry_streaks(graded, history, recent) _attach_dry_streaks(graded, history, recent)
graded.reverse() # newest first for display graded.reverse() # newest first for display
climo = grading.climatology(history, int(target.dayofyear)) climo = grading.climatology(history, target.timetuple().tm_yday)
# Summarize what this run actually covered. A fresh history fetch pulls the # 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 # full ~45-year archive (run_type "full"); a cache hit only fetched the
# recent window (run_type "partial"). # recent window (run_type "partial").
years = pd.to_datetime(history["date"]).dt.year years = history["date"].dt.year()
full = not cache_meta.get("cached", False) full = not cache_meta.get("cached", False)
run.set( run.set(
run_type="full" if full else "partial", run_type="full" if full else "partial",
@ -160,28 +178,28 @@ def build_grade(cell, target, days, history, recent, cache_meta, place, run=None
return { return {
"cell": cell, "cell": cell,
"place": place, "place": place,
"target_date": target.date().isoformat(), "target_date": target.isoformat(),
"cache": cache_meta, "cache": cache_meta,
"climatology": climo, "climatology": climo,
"recent": graded, "recent": graded,
} }
def cal_span(history, start, end, months) -> tuple[pd.Timestamp, pd.Timestamp]: def cal_span(history, start, end, months) -> tuple[datetime.date, datetime.date]:
"""Clamp a requested calendar range to the available record (≤ ~2 years). """Clamp a requested calendar range to the available record (≤ ~2 years).
Without a start, defaults to whole months back landing on the SAME 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 day-of-month as the end, so the two range endpoints line up
(e.g. 2024-06-29 → 2026-06-29).""" (e.g. 2024-06-29 → 2026-06-29). `start`/`end` are ISO strings or None."""
first = pd.Timestamp(history["date"].min()).normalize() first = history["date"].min()
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
end_ts = min(pd.Timestamp(end).normalize(), last) if end else last end_ts = min(datetime.date.fromisoformat(end), last) if end else last
if start: if start:
start_ts = pd.Timestamp(start).normalize() start_ts = datetime.date.fromisoformat(start)
else: else:
start_ts = (end_ts - pd.DateOffset(months=months)).normalize() start_ts = _months_before(end_ts, months)
# Cap the graded span at ~2 years, and keep it inside the available record. # 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) min_start = end_ts - datetime.timedelta(days=CAL_MAX_SPAN_DAYS - 1)
start_ts = max(start_ts, min_start, first) start_ts = max(start_ts, min_start, first)
start_ts = min(start_ts, end_ts) start_ts = min(start_ts, end_ts)
return start_ts, end_ts return start_ts, end_ts
@ -197,7 +215,7 @@ def build_calendar(cell, history, start_ts, end_ts, months, place, run=None) ->
"api_version": "v2", "api_version": "v2",
"cell": cell, "cell": cell,
"place": place, "place": place,
"range": {"start": start_ts.date().isoformat(), "end": end_ts.date().isoformat()}, "range": {"start": start_ts.isoformat(), "end": end_ts.isoformat()},
"months": months, "months": months,
"days": days, "days": days,
} }
@ -208,20 +226,20 @@ def build_day(cell, history, target, place, run=None, recent=None) -> dict:
prefer the archive record; a date newer than it comes from the recent window prefer the archive record; a date newer than it comes from the recent window
(passed pre-loaded by the bundle path, fetched here otherwise).""" (passed pre-loaded by the bundle path, fetched here otherwise)."""
run = run or NullRun() run = run or NullRun()
last = pd.Timestamp(history["date"].max()).normalize() last = history["date"].max()
obs = None obs = None
hit = history[history["date"] == target] hit = history.filter(pl.col("date") == target)
if not hit.empty: if hit.height:
obs = _obs_from_row(hit.iloc[0]) obs = _obs_from_row(hit.row(0, named=True))
elif target > last: elif target > last:
try: try:
if recent is None: if recent is None:
with run.phase("recent"): with run.phase("recent"):
recent = climate.get_recent_forecast(cell) recent = climate.get_recent_forecast(cell)
rr = recent[recent["date"] == target] rr = recent.filter(pl.col("date") == target)
if not rr.empty: if rr.height:
obs = _obs_from_row(rr.iloc[0]) obs = _obs_from_row(rr.row(0, named=True))
except Exception: # noqa: BLE001 - detail page still works from climatology alone except Exception: # noqa: BLE001 - detail page still works from climatology alone
obs = None obs = None
@ -232,7 +250,7 @@ def build_day(cell, history, target, place, run=None, recent=None) -> dict:
"api_version": "v2", "api_version": "v2",
"cell": cell, "cell": cell,
"place": place, "place": place,
"latest": last.date().isoformat(), "latest": last.isoformat(),
"detail": detail, "detail": detail,
} }
@ -242,18 +260,18 @@ def build_forecast(cell, days, history, fc, today, place, run=None) -> dict:
/grade so the frontend renders it identically, furthest-out day first.""" /grade so the frontend renders it identically, furthest-out day first."""
run = run or NullRun() run = run or NullRun()
# Strictly future days (tomorrow onward), earliest→latest, capped at `days`. # Strictly future days (tomorrow onward), earliest→latest, capped at `days`.
future = fc[fc["date"] > today].sort_values("date").head(days) future = fc.filter(pl.col("date") > today).sort("date").head(days)
with run.phase("grading"): with run.phase("grading"):
graded = _grade_rows(history, future) graded = _grade_rows(history, future)
_attach_dry_streaks(graded, history, fc) _attach_dry_streaks(graded, history, fc)
graded.reverse() # furthest-out first, so the chart reverses to L→R chronological graded.reverse() # furthest-out first, so the chart reverses to L→R chronological
climo = grading.climatology(history, int(today.dayofyear)) climo = grading.climatology(history, today.timetuple().tm_yday)
run.set(graded_days=len(graded), place_found=place is not None) run.set(graded_days=len(graded), place_found=place is not None)
return { return {
"api_version": "v2", "api_version": "v2",
"cell": cell, "cell": cell,
"place": place, "place": place,
"target_date": today.date().isoformat(), "target_date": today.isoformat(),
"forecast": True, "forecast": True,
"climatology": climo, "climatology": climo,
"recent": graded, "recent": graded,