thermograph/homepage.py

"""Homepage "unusual right now" feed.

The homepage wants to answer "where is the weather most unusual today?" across
every city we track. That question has no cheap answer at request time: the
derived store keeps graded payloads as zlib-compressed JSON blobs keyed by
(kind, cell, key), with no percentile column to sort on, so finding today's
extremes means grading every cached city from scratch.

So we precompute it. A sweep grades every city whose parquet cache is already
warm and writes the ranked result to data/homepage.json; the homepage template
reads that file. The sweep is deliberately **cache-only** — it calls
climate.load_cached_history and climate.load_cached_recent_forecast, never their
fetching counterparts, so a sweep over ~1000 cities costs zero upstream requests
and cannot burst the archive quota. Cities without a warm cache are skipped; the
deploy-time warm_cities.py run is what makes them eligible.

Runs from the notifier daemon (hourly guard) and at the tail of warm_cities, so
it refreshes both on a live server and right after a deploy.
"""
from __future__ import annotations

import datetime
import json
import os
import tempfile
import time

import polars as pl

import cities
import climate
import grading
import grid
from views import OBS_COLS

# data/ is the only writable path under the hardened systemd unit
# (ReadWritePaths=/opt/thermograph/data …), so the feed lives there.
FEED_PATH = os.path.join(os.path.dirname(__file__), os.pardir, "data", "homepage.json")

# How many cards the "Unusual right now" strip can show.
RANK_LIMIT = 12

# A city must be at least this far from the median to be worth calling unusual.
MIN_DEPARTURE = 15.0

# A cold-tail card is force-included when one is at or below this percentile,
# even if every warmer city outranks it. Both tails or it isn't honest.
COLD_TAIL_MAX_PCT = 10.0

# The feed is stale (label it "as of <date>" rather than claiming today) beyond this.
STALE_AFTER_S = 6 * 3600

# Ranking looks at the two metrics whose percentile people read as "how unusual
# was today" — the same pair grade_day's own `departure` score uses.
RANK_METRICS = ("tmax", "tmin")

_METRIC_LABEL = {"tmax": "High temp", "tmin": "Low temp"}


def _seasonal_window_label(d: datetime.date) -> str:
    """'mid-July' — the ±7-day seasonal window a percentile is measured against,
    said the way a person would say it."""
    part = "early" if d.day <= 10 else ("mid" if d.day <= 20 else "late")
    return f"{part}-{d.strftime('%B')}"


def _pct_label(pct: float) -> str:
    """A percentile as an ordinal, clamped to 1..99.

    An empirical percentile can round to 100 (or 0), and "100th percentile" reads
    as a measurement error rather than "as hot as it has ever been". Flooring into
    the real 1-99 range keeps it honest — the value and the band label carry the
    "how extreme" part anyway.
    """
    n = min(99, max(1, int(round(pct))))
    suffix = "th" if 10 <= n % 100 <= 20 else {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th")
    return f"{n}{suffix}"


def _grade_label(grade: str | None, tail: str) -> str:
    """The band label, disambiguated when it is the same at both ends."""
    if not grade:
        return ""
    if grade == "Near Record":
        return f"Near Record {'cold' if tail == 'cold' else 'hot'}"
    return grade


def _grade_city(city: dict, today: datetime.date) -> dict | None:
    """Grade one city's latest observed day from cache alone. None when the city
    has no warm cache, no observed row, or nothing gradeable."""
    cell = grid.snap(city["lat"], city["lon"])

    history = climate.load_cached_history(cell)
    if history is None:
        return None
    recent = climate.load_cached_recent_forecast(cell)
    if recent is None or recent.is_empty() or "date" not in recent.columns:
        return None

    observed = recent.filter(pl.col("date") <= today).sort("date")
    if observed.is_empty():
        return None
    row = observed.row(observed.height - 1, named=True)
    obs = {k: row[k] for k in OBS_COLS if k in row}

    try:
        graded = grading.grade_day(history, row["date"], obs)
    except Exception:  # noqa: BLE001 - one bad cell must not abort the sweep
        return None

    # Pick the metric that strayed furthest from the median; that is the one the
    # card should lead with.
    best = None
    for metric in RANK_METRICS:
        g = graded.get(metric)
        if not g or g.get("percentile") is None:
            continue
        departure = abs(g["percentile"] - 50)
        if best is None or departure > best[0]:
            best = (departure, metric, g)
    if best is None:
        return None

    departure, metric, g = best
    date = row["date"]
    date_s = date.isoformat() if hasattr(date, "isoformat") else str(date)
    pct = g["percentile"]
    tail = "cold" if pct < 50 else "warm"
    # The median for this metric's ±7-day window — "104°F, normal 97°F" is the
    # line that makes a percentile mean something.
    normals = (graded.get("normals") or {}).get(metric) or {}
    return {
        "slug": city["slug"],
        "name": city["name"],
        # "Tehran, Iran", not "Dar es Salaam, Dar es Salaam Region, Tanzania":
        # the card is one line wide and the region almost always restates the
        # city on the metros we track.
        "display": f"{city['name']}, {city['country']}" if city.get("country") else city["name"],
        "lat": round(city["lat"], 4),
        "lon": round(city["lon"], 4),
        "cell_id": cell["id"],
        "date": date_s,
        "metric": metric,
        "metric_label": _METRIC_LABEL.get(metric, metric),
        "window_label": _seasonal_window_label(
            date if hasattr(date, "timetuple") else today
        ),
        "value": g.get("value"),
        "normal": normals.get("p50"),
        "percentile": pct,
        "pct_label": _pct_label(pct),
        "grade": g.get("grade"),
        # "Near Record" is the band label at BOTH ends of the scale, so on its own
        # a card can't say whether it's a hot or a cold extreme — colour would be
        # the only signal. Qualify it; the other tiers already read directionally
        # ("Very High", "Low"), so they're left alone.
        "grade_label": _grade_label(g.get("grade"), tail),
        # grading's css class is already the style.css token name minus the
        # leading '--', so the template emits it with no mapping table.
        "cls": g.get("class"),
        "departure": round(departure, 1),
        "tail": tail,
    }


def build(limit: int | None = None) -> dict:
    """Sweep the warm cache and rank today's most unusual cities. Pure — does no
    I/O beyond reading the parquet cache, and never touches the network."""
    today = datetime.date.today()
    graded: list[dict] = []
    considered = 0

    for city in cities.all_cities()[: limit or None]:
        row = _grade_city(city, today)
        if row is None:
            continue
        considered += 1
        if row["departure"] >= MIN_DEPARTURE:
            graded.append(row)

    graded.sort(key=lambda r: r["departure"], reverse=True)
    ranked = graded[:RANK_LIMIT]

    # Two-sided honesty: if any tracked city sits in the cold tail, one must
    # appear in the strip even when warm anomalies dominate the ranking.
    if not any(r["tail"] == "cold" for r in ranked):
        cold = next(
            (r for r in graded if r["tail"] == "cold"
             and r["percentile"] <= COLD_TAIL_MAX_PCT),
            None,
        )
        if cold is not None:
            ranked = ranked[: RANK_LIMIT - 1] + [cold]

    picks = {}
    if graded:
        picks["extreme"] = graded[0]
        cold = next((r for r in graded if r["tail"] == "cold"), None)
        if cold is not None:
            picks["cold"] = cold

    return {
        "generated_at": time.time(),
        "date": today.isoformat(),
        "considered": considered,
        "picks": picks,
        "ranked": ranked,
    }


def refresh(limit: int | None = None) -> dict:
    """Rebuild the feed and write it atomically, so a concurrent reader never
    sees a half-written file."""
    feed = build(limit=limit)
    path = os.path.abspath(FEED_PATH)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(feed, f, separators=(",", ":"))
        os.replace(tmp, path)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    return feed


def load() -> dict | None:
    """Read the feed. Returns None when it is missing or unreadable — the
    homepage renders its frame without live numbers rather than failing, so a
    cold checkout with no feed still serves a complete page."""
    try:
        with open(os.path.abspath(FEED_PATH), encoding="utf-8") as f:
            feed = json.load(f)
    except (OSError, ValueError):
        return None
    if not isinstance(feed, dict) or "ranked" not in feed:
        return None
    return feed


def is_stale(feed: dict) -> bool:
    """Older than STALE_AFTER_S, or built for a day that is no longer today."""
    if feed.get("date") != datetime.date.today().isoformat():
        return True
    return (time.time() - float(feed.get("generated_at") or 0)) > STALE_AFTER_S


if __name__ == "__main__":
    import sys
    n = int(sys.argv[1]) if len(sys.argv) > 1 else None
    out = refresh(limit=n)
    print(f"graded {out['considered']} cached cities, "
          f"{len(out['ranked'])} ranked, top: "
          f"{out['ranked'][0]['display'] if out['ranked'] else '—'}")