thermograph/backend/api/payloads.py

"""Payload builders — pure "inputs → response dict" assembly for every view.

This is the layer between the data modules (climate/grading/grid) and any
delivery mechanism: the FastAPI endpoints, the /cell bundle, and the offline
migrate script all build payloads here, so a payload's shape has exactly one
definition and offline callers never have to import the web stack. HTTP
semantics (audit runs, derived-store lookups, ETags) stay with the callers;
``run`` is an audit.RunAudit or None.
"""
import contextlib
import datetime
import time

import polars as pl

from data import climate
from data import grading
from data import scoring

# Observed values pulled from a daily record row for grading. Includes the
# temperature-scale metrics (tmax/tmin/feels/wind/gust) plus precip; a column may
# be absent on an older cache, so only carry the ones present.
OBS_COLS = ("tmax", "tmin", "precip", "feels", "humid", "wind", "gust")

# Bump when any response payload shape changes (new metrics, renamed fields…).
# The version is part of every derived-store validity token, so one bump
# atomically orphans all pre-upgrade cached payloads instead of letting a stale
# row whose history_end happens to match keep serving the old shape.
PAYLOAD_VER = "p3"   # p3: content-city/records payloads carry an editorial `copy` field
                     # p2: place labels now carry the trailing country component

CAL_MAX_SPAN_DAYS = 732  # ~2 years — the most a single calendar request will grade
                         # (the frontend splits longer spans into 2-year chunks)


class NullRun:
    """audit.RunAudit stand-in for offline callers (the migrate script)."""

    def set(self, **kw):
        return self

    def phase(self, name):
        return contextlib.nullcontext()


def _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:
    """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 history["date"].max().isoformat()


# --- cache identity ------------------------------------------------------------
# The derived store caches payloads under (kind, cell, key) guarded by a validity
# token (see store.py). These functions are the ONLY definitions of each kind's
# key/token format: the per-view endpoints, the /cell bundle and the migrate
# script all derive cache identity here. (Formats drifting apart would silently
# split the cache — endpoints missing rows the bundle wrote, migrate
# materializing rows nobody reads.)

def history_token(history) -> str:
    """Validity for payloads derived from the archive record alone."""
    return f"{PAYLOAD_VER}:{hist_end(history)}"


def recent_token(history, cell_id: str) -> str:
    """Validity for payloads that also grade the hourly recent/forecast bundle."""
    return f"{history_token(history)}:{climate.recent_stamp(cell_id)}"


def grade_key(target, days: int, after: int) -> str:
    return f"{target.isoformat()}:{days}:{after}"


def calendar_key(start_ts, end_ts, months: int) -> str:
    return f"{start_ts.isoformat()}:{end_ts.isoformat()}:{months}"


def day_key(target) -> str:
    return target.isoformat()


def day_token(history, target) -> str:
    """A fully-archived day stays valid until the record itself advances; a newer
    day's observation comes from the hourly recent bundle, so its payload expires
    on that bundle's own cadence (hourly)."""
    token = history_token(history)
    if target > history["date"].max():
        token += f":h{int(time.time() // 3600)}"
    return token


def forecast_key(today, days: int) -> str:
    return f"{today.isoformat()}:{days}"


# The score is derived purely from the full archive record, so its validity token
# is the plain `history_token` (expires when the archive tail advances). The key
# just carries the scoring-math version, so a math change invalidates only score
# rows without disturbing any other kind.
SCORE_VER = "s5"   # s5: overall is an equal average of metrics (was weighted)


def score_key() -> str:
    return SCORE_VER


def _obs_from_row(row: dict) -> dict:
    return {k: row[k] for k in OBS_COLS if k in row}


def _grade_rows(history, rows_df) -> list[dict]:
    """Grade each daily row against its own ±7-day climatology window."""
    return [grading.grade_day(history, row["date"], _obs_from_row(row))
            for row in rows_df.iter_rows(named=True)]


def _attach_dry_streaks(graded: list[dict], *precip_frames) -> None:
    """Attach `dsr` (days since last measurable rain) to each graded day, computed
    over a combined, de-duplicated precip series so streaks stay continuous across
    the history / recent / forecast sources."""
    frames = [f.select("date", "precip") for f in precip_frames
              if f is not None and f.height]
    if not frames:
        return
    # Sources are passed archive-first, recent/forecast-last, so `keep="last"`
    # prefers the fresher source's row for any shared date (before the final sort).
    combined = (pl.concat(frames, how="vertical_relaxed")
                .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:
        g["dsr"] = dsr_map.get(g["date"])


def _graded_window(history, rows, precip_frames, doy) -> tuple[list[dict], dict]:
    """The shared tail of the /grade and /forecast payloads: grade `rows` against
    history, attach dry streaks over `precip_frames`, order newest-first, and pair
    the result with the target day-of-year's climatology summary. The window
    *assembly* (recent+archive merge vs future-only filter) differs per builder and
    stays there; only this post-processing is shared."""
    graded = _grade_rows(history, rows)
    _attach_dry_streaks(graded, *precip_frames)
    graded.reverse()  # newest-first for display (the chart reverses to L→R chronological)
    climo = grading.climatology(history, doy)
    return graded, climo


def build_grade(cell, target, days, history, recent, cache_meta, place, run=None, after=14) -> dict:
    """/grade payload: a window of days centered on the target — `days` of history
    before it, the target itself, then up to `after` days after it — plus the
    target day's climatology summary. Days after the target are observed when they
    are already in the past and forecast when they run into the future, so the
    weekly view can frame two weeks of history around the orange target marker and
    trail off into up to 14 days of observations / forecast after it. The forecast
    only reaches ~7 days out, so a recent target naturally yields some observed days
    plus 1-7 forecast days, while an older target fills the whole 14 with real obs."""
    run = run or NullRun()
    with run.phase("grading"):
        lo = target - datetime.timedelta(days=days)
        hi = target + datetime.timedelta(days=after)
        # Build the window from both sources. The recent+forecast bundle covers the
        # last few weeks plus the forward forecast (the only source for future days);
        # the archive reaches decades back for targets older than that bundle. Prefer
        # the bundle row for any given date, filling the rest from the archive.
        rwin = recent.filter((pl.col("date") >= lo) & (pl.col("date") <= hi))
        hwin = history.filter((pl.col("date") >= lo) & (pl.col("date") <= hi))
        hwin = hwin.join(rwin.select("date"), on="date", how="anti")
        # 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, climo = _graded_window(history, window, (history, recent),
                                       target.timetuple().tm_yday)

    # 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 = 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.isoformat(),
        "cache": cache_meta,
        "climatology": climo,
        "recent": graded,
    }


def cal_span(history, start, end, months) -> tuple[datetime.date, datetime.date]:
    """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). `start`/`end` are ISO strings or None."""
    first = history["date"].min()
    last = history["date"].max()
    end_ts = min(datetime.date.fromisoformat(end), last) if end else last
    if start:
        start_ts = datetime.date.fromisoformat(start)
    else:
        start_ts = _months_before(end_ts, months)
    # Cap the graded span at ~2 years, and keep it inside the available record.
    min_start = end_ts - datetime.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.isoformat(), "end": end_ts.isoformat()},
        "months": months,
        "days": days,
        # Per-month climatology over the full history — drives the plain-language
        # insight strip and cross-month rain comparisons in the calendar UI.
        "months_climate": grading.month_climate(history),
    }


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 = history["date"].max()

    obs = None
    hit = history.filter(pl.col("date") == target)
    if hit.height:
        obs = _obs_from_row(hit.row(0, named=True))
    elif target > last:
        try:
            if recent is None:
                with run.phase("recent"):
                    recent = climate.get_recent_forecast(cell)
            rr = recent.filter(pl.col("date") == target)
            if rr.height:
                obs = _obs_from_row(rr.row(0, named=True))
        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.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.filter(pl.col("date") > today).sort("date").head(days)
    with run.phase("grading"):
        graded, climo = _graded_window(history, future, (history, fc),
                                       today.timetuple().tm_yday)
    run.set(graded_days=len(graded), place_found=place is not None)
    return {
        "api_version": "v2",
        "cell": cell,
        "place": place,
        "target_date": today.isoformat(),
        "forecast": True,
        "climatology": climo,
        "recent": graded,
    }


def build_score(cell, history, place, run=None) -> dict:
    """/score payload: how far the cell's last 6 years have drifted from its full
    45-year baseline, per metric and season (see scoring.build_scores)."""
    run = run or NullRun()
    with run.phase("scoring"):
        scores = scoring.build_scores(history)
    run.set(place_found=place is not None)
    return {
        "api_version": "v2",
        "cell": cell,
        "place": place,
        "latest": hist_end(history),
        "scores": scores,
    }