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: the loop reads history from the parquet cache only
(climate.load_cached_history) and treats a forecast-fetch failure as "skip this
cell", so it never spends archive quota or dies on an upstream rate limit.
"""
import datetime
import os
import threading
import time

import pandas as pd
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

# 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: pd.DataFrame, kind: str, today: pd.Timestamp):
    """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:
        return []
    dates = pd.to_datetime(recent["date"]).dt.normalize()
    if kind == "forecast":
        mask = (dates > today) & (dates <= today + pd.Timedelta(days=FORECAST_HORIZON_DAYS))
        ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d")  # soonest first
    else:
        mask = (dates <= today) & (dates >= today - pd.Timedelta(days=LOOKBACK_DAYS))
        ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d", ascending=False)  # newest first
    return [row for _, row in ordered.iterrows()]


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 = pd.Timestamp(row["date"]).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):
    """Evaluate every subscription on one cell, inserting any new notifications."""
    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.empty:
        return 0   # nothing cached yet — never spend archive quota from here
    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 = pd.Timestamp(datetime.date.today())
    created = 0
    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)
            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()