thermograph/web/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

from data import cities
from data import climate
from data import grading
from data import grid
from data import store
import paths
from web.views import OBS_COLS

# data/ is the only writable path under the hardened systemd unit
# (ReadWritePaths=/opt/thermograph/data …), so the feed lives there when it's a
# file (see the store.IS_POSTGRES split in refresh/load below).
FEED_PATH = os.path.join(paths.DATA_DIR, "homepage.json")

# On Postgres, persist the feed via store.py's existing derived-payload table
# instead of the local file: web replicas each have their own disk under Swarm,
# so a file only that replica's own notifier writes would leave every OTHER
# replica reading a stale (or missing) feed forever. The feed isn't cell-scoped,
# so it's stored under a fixed sentinel key rather than a real cell id; the token
# is a constant because this cache is never invalidated by content, only ever
# overwritten by the next refresh (there's nothing else to compare it against).
# dev/tests keep the plain file (store.IS_POSTGRES is False without
# THERMOGRAPH_DATABASE_URL), so today's behavior there is unchanged.
_FEED_STORE_KIND = "homepage"
_FEED_STORE_CELL = "_global"
_FEED_STORE_KEY = "feed"
_FEED_STORE_TOKEN = "v1"

# 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

# A reading older than this is not "right now" and is dropped rather than ranked.
# Cities span a day's worth of timezones, so a genuinely current observation can
# still be dated yesterday in UTC — hence 1 rather than 0.
MAX_OBSERVATION_AGE_DAYS = 1

# How many cities the strip ranks over. The strip shows ~12 cards; ranking over a
# smaller set that is actually FRESH beats ranking over every cached city when
# most of those are stale (see _refresh_recent below).
CANDIDATE_LIMIT = int(os.environ.get("THERMOGRAPH_HOMEPAGE_CITIES", "250") or 250)

# Recent/forecast fetches one refresh pass may spend, oldest cache first.
#
# This is the one place the homepage costs upstream quota, and it is not
# optional: nothing else keeps the recent/forecast cache current for the tracked
# cities. warm_cities.py skips any city whose archive is already cached, so it
# never re-fetches their forecast bundle, and the notifier only touches cells
# somebody has subscribed to. Reading that cache without refreshing it is what
# made the strip present days-old readings as "right now".
#
# One fetch covers a whole cell's recent+forecast window. Set to 0 to disable
# (the strip then only ranks cities something else happened to refresh).
MAX_RF_REFRESH_PER_PASS = int(os.environ.get("THERMOGRAPH_HOMEPAGE_REFRESH", "40") or 0)

# 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 _short_date(date_s: str) -> str:
    """'2026-07-17' -> 'Jul 17'."""
    try:
        return datetime.date.fromisoformat(date_s).strftime("%b %-d")
    except ValueError:
        return date_s


def _pct_label(pct: float) -> str:
    """Kept as a name the feed builder reads; the rule itself lives in grading."""
    return grading.pct_ordinal(pct)


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)

    # The newest cached row can be days old — the cache is never refreshed for a
    # city nobody visits. Grading it anyway is what put a reading from three days
    # ago under a heading that says "right now", disagreeing with the Day page
    # for the very same cell.
    row_date = row["date"]
    if hasattr(row_date, "toordinal"):
        if (today.toordinal() - row_date.toordinal()) > MAX_OBSERVATION_AGE_DAYS:
            return None

    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,
        # Shown on the card when the reading isn't today's, so a yesterday-in-UTC
        # observation never silently passes for "right now".
        "date_label": None if date_s == today.isoformat() else _short_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 CANDIDATE_LIMIT]:
        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_recent(limit: int | None = None, budget: int = 0) -> int:
    """Top up the recent/forecast cache for the stalest candidate cities.

    Bounded and oldest-first, so each pass advances the set a little and no pass
    can burst the forecast quota — the same shape as the notifier's archive-fetch
    budget. Returns how many cells were refreshed.
    """
    if budget <= 0:
        return 0
    candidates = []
    for city in cities.all_cities()[: limit or CANDIDATE_LIMIT]:
        cell = grid.snap(city["lat"], city["lon"])
        # Only cities whose archive is already warm can be graded at all, so
        # there is no point refreshing a forecast we can't compare to anything.
        if climate.load_cached_history(cell) is None:
            continue
        candidates.append((climate.recent_stamp(cell["id"]), cell))
    candidates.sort(key=lambda t: t[0])

    done = 0
    for _, cell in candidates[:budget]:
        try:
            climate.get_recent_forecast(cell)
            done += 1
        except Exception:  # noqa: BLE001 - one failed cell must not stop the pass
            continue
    return done


def refresh(limit: int | None = None, fetch: int | None = None) -> dict:
    """Rebuild the feed and persist it atomically, so a concurrent reader never
    sees a half-written result.

    ``fetch`` caps the recent/forecast top-ups this pass may spend; pass 0 for a
    strictly cache-only rebuild (what the tests do).
    """
    refreshed = _refresh_recent(
        limit=limit,
        budget=MAX_RF_REFRESH_PER_PASS if fetch is None else fetch,
    )
    feed = build(limit=limit)
    feed["refreshed"] = refreshed
    if store.IS_POSTGRES:
        # A single upsert is atomic in the DB sense already; store.put_payload is
        # fail-soft (a Postgres hiccup drops the write silently, matching every
        # other store.py caller), so this never turns a refresh into a hard failure.
        store.put_payload(_FEED_STORE_KIND, _FEED_STORE_CELL, _FEED_STORE_KEY,
                          _FEED_STORE_TOKEN, feed)
        return feed
    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.

    On Postgres this reads the same row every web replica shares (see refresh);
    elsewhere it reads the local file, unchanged from before this store split."""
    if store.IS_POSTGRES:
        feed = store.get_json(_FEED_STORE_KIND, _FEED_STORE_CELL, _FEED_STORE_KEY,
                              _FEED_STORE_TOKEN)
        if not isinstance(feed, dict) or "ranked" not in feed:
            return None
        return feed
    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"refreshed {out.get('refreshed', 0)} forecast caches; "
          f"graded {out['considered']} cached cities, "
          f"{len(out['ranked'])} ranked, top: "
          f"{out['ranked'][0]['display'] if out['ranked'] else '—'}")