thermograph/notify.py

"""Subscription evaluation engine — the background worker that turns unusual
weather into notifications.

A daemon thread (modeled on app.py's neighbor warmer) wakes on an interval and, for
every active subscription, checks whether the subscribed cell's recent/forecast
weather crossed the user's percentile threshold on any watched metric. Crossings
become rows in the notifications table.

Two guards keep it quiet and cheap:

* **Per-event dedup** — the notifications table has a UNIQUE(subscription_id,
  event_date, metric, direction, kind); an INSERT OR IGNORE means re-running the
  pass never re-notifies the same event.
* **Weekly cap** — a subscription that has notified within the last 7 days is
  skipped entirely (Subscription.last_notified_at), so one alert = at most one
  notification per week, as specified.

Quota safety: a subscribed cell's ~45-year archive is read from the parquet cache
(climate.load_cached_history) and never re-fetched once present. A cell that has no
cached archive yet is fetched ONCE (budget-capped per pass) so a freshly-subscribed
city starts working; the recent/forecast bundle refreshes on its own hourly cadence.
A forecast-fetch failure just skips the cell, so a pass never dies on a rate limit.
"""
import datetime
import os
import threading
import time

import polars as pl
from sqlalchemy import delete, select

import climate
import grading
import grid
from db import sync_session_maker
from models import AccessToken, Notification, Subscription
from views import OBS_COLS

WEEK_SECONDS = 7 * 86400
INTERVAL = int(os.environ.get("THERMOGRAPH_NOTIFY_INTERVAL", "900"))  # 15 min default
SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400
LOOKBACK_DAYS = 3        # re-check the last few observed days (dedup makes it safe)
FORECAST_HORIZON_DAYS = 7
# The ~45-year archive is otherwise never fetched from the background loop. When a
# subscribed cell has no cached archive yet, fetch it ONCE (it then stays cached and
# is only ever read afterwards). This caps how many missing archives a single pass
# will fetch, so a burst of new subscriptions can't hammer the strict archive quota
# — the rest are picked up on later passes.
MAX_ARCHIVE_FETCHES_PER_PASS = int(os.environ.get("THERMOGRAPH_NOTIFY_ARCHIVE_FETCHES", "4"))

# Metrics whose low tail also counts as unusual when a subscription is two-sided
# (cold snaps, unusually calm/dry). Precipitation is one-directional (high only).
TWO_SIDED_METRICS = set(grading.TEMP_METRICS)

METRIC_NOUN = {
    "tmax": "daytime high",
    "tmin": "overnight low",
    "feels": "feels-like temperature",
    "humid": "humidity",
    "wind": "wind speed",
    "gust": "wind gusts",
    "precip": "rainfall",
}
METRIC_UNIT = {
    "tmax": "°F", "tmin": "°F", "feels": "°F",
    "humid": " g/m³", "wind": " mph", "gust": " mph", "precip": " in",
}
# A readable descriptor per (metric, direction) so titles don't read like
# "unusually high high temperature". Precipitation is high-only.
METRIC_PHRASE = {
    ("tmax", "high"): "unusually hot day", ("tmax", "low"): "unusually cool day",
    ("tmin", "high"): "unusually warm night", ("tmin", "low"): "unusually cold night",
    ("feels", "high"): "unusually hot conditions", ("feels", "low"): "unusually cold conditions",
    ("humid", "high"): "unusually humid air", ("humid", "low"): "unusually dry air",
    ("wind", "high"): "unusually strong wind", ("wind", "low"): "unusually calm wind",
    ("gust", "high"): "unusually strong gusts", ("gust", "low"): "unusually light gusts",
    ("precip", "high"): "unusually heavy rain",
}

_STOP = threading.Event()
_thread = None


# --- message building --------------------------------------------------------
def _place(sub: Subscription) -> str:
    return sub.label or f"{sub.lat:.2f}, {sub.lon:.2f}"


def _compose(sub, event_date, metric, direction, graded):
    noun = METRIC_NOUN.get(metric, metric)
    pct = graded["percentile"]
    grade = graded.get("grade")
    value = graded.get("value")
    unit = METRIC_UNIT.get(metric, "")
    phrase = METRIC_PHRASE.get((metric, direction), f"unusual {noun}")
    title = f"{_place(sub)}: {phrase}"
    val_txt = f" ({value}{unit})" if value is not None else ""
    if sub.kind == "forecast":
        body = (f"Forecast for {event_date}: the {noun} is projected at the "
                f"{pct:g}th percentile{val_txt} — {grade}.")
    else:
        body = (f"On {event_date}, the {noun} hit the {pct:g}th "
                f"percentile{val_txt} — {grade}.")
    return title, body


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


def _candidate_rows(recent, kind: str, today: datetime.date):
    """Rows to grade for a subscription of the given kind, most-relevant first.
    `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 []
    if kind == "forecast":
        window = (pl.col("date") > today) & (
            pl.col("date") <= today + datetime.timedelta(days=FORECAST_HORIZON_DAYS))
        ordered = recent.filter(window).sort("date")                     # soonest first
    else:
        window = (pl.col("date") <= today) & (
            pl.col("date") >= today - datetime.timedelta(days=LOOKBACK_DAYS))
        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):
    """The first (event_date, metric, direction, graded) that crosses, or None."""
    for row in rows:
        date_key = row["date"].isoformat()
        graded = grade_cache.get(date_key)
        if graded is None:
            graded = grading.grade_day(history, row["date"], _obs_from_row(row))
            grade_cache[date_key] = graded
        for metric in sub.metrics:
            g = graded.get(metric)
            if not g or g.get("percentile") is None:
                continue
            pct = g["percentile"]
            if pct >= sub.threshold:
                return (date_key, metric, "high", g)
            if sub.two_sided and metric in TWO_SIDED_METRICS and pct <= 100 - sub.threshold:
                return (date_key, metric, "low", g)
    return None


def _process_cell(session, cell_id, subs, today, now, archive_budget):
    """Evaluate every subscription on one cell, inserting any new notifications.

    `archive_budget` is a one-element list holding the remaining number of missing
    archives this pass may fetch; it's decremented when this cell fetches one.
    """
    try:
        cell = grid.from_id(cell_id)
    except Exception:  # noqa: BLE001 - a malformed cell_id shouldn't kill the pass
        return 0
    history = climate.load_cached_history(cell)
    if history is None or history.is_empty():
        # No cached archive for this subscribed cell. Grab it ONCE (get_history
        # fetches + caches the full archive); every later pass reads it from cache
        # above and never re-fetches. Budget-gated so a burst of new subscriptions
        # can't exhaust the archive quota in a single pass.
        if archive_budget[0] <= 0:
            return 0
        archive_budget[0] -= 1
        try:
            history, _ = climate.get_history(cell)
        except climate.WeatherUnavailable:
            return 0   # archive rate-limited — retry on a later pass
        except Exception:  # noqa: BLE001
            return 0
        if history is None or history.is_empty():
            return 0
    try:
        recent = climate.get_recent_forecast(cell)
    except climate.WeatherUnavailable:
        return 0   # upstream rate-limited — skip this cell this pass
    except Exception:  # noqa: BLE001
        return 0

    grade_cache = {}
    created = 0
    for sub in subs:
        # Weekly cap: one notification per subscription per 7 days.
        if sub.last_notified_at and now - sub.last_notified_at < WEEK_SECONDS:
            continue
        rows = _candidate_rows(recent, sub.kind, today)
        hit = _first_trigger(sub, rows, history, grade_cache)
        if hit is None:
            continue
        event_date, metric, direction, graded = hit
        title, body = _compose(sub, event_date, metric, direction, graded)
        notif = Notification(
            user_id=sub.user_id,
            subscription_id=sub.id,
            event_date=event_date,
            metric=metric,
            direction=direction,
            kind=sub.kind,
            percentile=graded["percentile"],
            value=graded.get("value"),
            grade=graded.get("grade"),
            title=title,
            body=body,
            channel="inapp",
            created_at=now,
        )
        session.add(notif)
        try:
            session.flush()   # trips the UNIQUE dedup constraint if already seen
        except Exception:  # noqa: BLE001 - already notified for this exact event
            session.rollback()
            continue
        sub.last_notified_at = now
        created += 1
    session.commit()
    return created


# --- housekeeping ------------------------------------------------------------
def _cleanup_sessions(session) -> None:
    """Drop expired login sessions (access tokens past their lifetime)."""
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
        seconds=SESSION_TTL_SECONDS
    )
    session.execute(delete(AccessToken).where(AccessToken.created_at < cutoff))
    session.commit()


# --- pass + loop -------------------------------------------------------------
def run_pass() -> int:
    """One full evaluation sweep over all active subscriptions. Returns the number
    of notifications created."""
    now = time.time()
    today = datetime.date.today()
    created = 0
    archive_budget = [MAX_ARCHIVE_FETCHES_PER_PASS]   # missing archives to fetch this pass
    with sync_session_maker() as session:
        subs = session.execute(
            select(Subscription).where(Subscription.active.is_(True))
        ).scalars().all()
        by_cell: dict[str, list] = {}
        for sub in subs:
            by_cell.setdefault(sub.cell_id, []).append(sub)
        for cell_id, cell_subs in by_cell.items():
            try:
                created += _process_cell(session, cell_id, cell_subs, today, now, archive_budget)
            except Exception:  # noqa: BLE001 - one bad cell must not abort the pass
                session.rollback()
        try:
            _cleanup_sessions(session)   # opportunistic expired-token sweep
        except Exception:  # noqa: BLE001
            session.rollback()
    return created


def run_loop():
    # Wait first, then run — gives the app a moment to finish booting.
    while not _STOP.wait(INTERVAL):
        try:
            run_pass()
        except Exception:  # noqa: BLE001 - a bad pass must never kill the loop
            pass


def start():
    """Start the notifier daemon thread (unless disabled via env)."""
    global _thread
    if os.environ.get("THERMOGRAPH_ENABLE_NOTIFIER", "1") == "0":
        return
    if _thread is not None:
        return
    _STOP.clear()
    _thread = threading.Thread(target=run_loop, name="subscription-notifier", daemon=True)
    _thread.start()


def stop():
    _STOP.set()