387 lines
16 KiB
Python
387 lines
16 KiB
Python
"""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
|
|
|
|
from data import climate
|
|
from notifications import discord
|
|
from data import grading
|
|
from data import grid
|
|
from web import homepage
|
|
from core import metrics
|
|
from notifications import push
|
|
from accounts.db import sync_session_maker
|
|
from accounts.models import AccessToken, Notification, PushSubscription, Subscription, User
|
|
from web.views import OBS_COLS
|
|
|
|
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
|
|
BASE = f"/{_BASE}" if _BASE else ""
|
|
|
|
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
|
|
|
|
|
|
# --- push delivery -----------------------------------------------------------
|
|
def _deep_link(sub: Subscription, event_date: str) -> str:
|
|
"""Where a tapped notification lands — the single-day breakdown for the
|
|
subscribed cell on the event date."""
|
|
return f"{BASE}/day#lat={sub.lat:.5f}&lon={sub.lon:.5f}&date={event_date}"
|
|
|
|
|
|
def _dispatch_push(session, sub, title, body, event_date) -> None:
|
|
"""Best-effort Web Push of a just-created notification to all of the user's
|
|
devices. Prunes endpoints the push service reports gone. Runs in the notifier
|
|
daemon thread (no event loop), so the blocking send is fine here. Never raises
|
|
— the in-app notification is already committed regardless."""
|
|
rows = session.execute(
|
|
select(PushSubscription).where(PushSubscription.user_id == sub.user_id)
|
|
).scalars().all()
|
|
if not rows:
|
|
return
|
|
payload = {"title": title, "body": body, "url": _deep_link(sub, event_date),
|
|
"tag": f"sub-{sub.id}"}
|
|
for row in rows:
|
|
info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
|
|
if push.send(info, payload) == "gone":
|
|
session.delete(row) # committed with the rest of the pass
|
|
|
|
|
|
def _dispatch_discord(session, sub, title, body, event_date) -> None:
|
|
"""Best-effort Discord DM of a just-created notification, when the subscriber
|
|
has linked Discord and opted in. Isolated like the push side-channel — the
|
|
in-app row is already committed, and push/email remain the fallback for anyone
|
|
not reachable on Discord. Runs in the notifier thread (no event loop)."""
|
|
if not discord.dm_enabled():
|
|
return
|
|
user = session.get(User, sub.user_id)
|
|
if not user or not user.discord_id or not user.discord_dm:
|
|
return
|
|
discord.send_dm(user.discord_id, title, body, _deep_link(sub, event_date))
|
|
|
|
|
|
# --- 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
|
|
# Additionally deliver over Web Push (the in-app row above is the record of
|
|
# truth for the bell; push is a side-channel that must never break it).
|
|
try:
|
|
_dispatch_push(session, sub, title, body, event_date)
|
|
except Exception: # noqa: BLE001 - push failures stay isolated from the DB write
|
|
pass
|
|
# And, independently, a Discord DM for linked opted-in users.
|
|
try:
|
|
_dispatch_discord(session, sub, title, body, event_date)
|
|
except Exception: # noqa: BLE001 - Discord failures stay isolated too
|
|
pass
|
|
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
|
|
|
|
|
|
# The homepage's "unusual right now" feed rides along on this loop rather than
|
|
# running its own thread: this one already wakes on a timer and already carries
|
|
# the single-leader invariant a second daemon would have to re-establish. The
|
|
# sweep is cache-only, so it costs no upstream quota.
|
|
HOMEPAGE_REFRESH_INTERVAL = 3600.0
|
|
_last_homepage_refresh = 0.0
|
|
|
|
|
|
def _maybe_refresh_homepage() -> None:
|
|
global _last_homepage_refresh
|
|
now = time.time()
|
|
if now - _last_homepage_refresh < HOMEPAGE_REFRESH_INTERVAL:
|
|
return
|
|
_last_homepage_refresh = now
|
|
try:
|
|
homepage.refresh()
|
|
except Exception: # noqa: BLE001 - the feed is optional; the homepage renders without it
|
|
pass
|
|
|
|
|
|
# The daily "most unusual right now" post to Discord rides this loop too: once per
|
|
# day, after the feed is refreshed, leader-only like everything in run_loop. No
|
|
# webhook configured => no-op.
|
|
_last_discord_post_day: str | None = None
|
|
|
|
|
|
def _maybe_post_discord() -> None:
|
|
global _last_discord_post_day
|
|
if not discord.enabled():
|
|
return
|
|
today = datetime.date.today().isoformat()
|
|
if _last_discord_post_day == today:
|
|
return
|
|
# Only record the day as done once a post actually lands, so a transient
|
|
# failure retries on the next wake instead of silently skipping the day.
|
|
if discord.post_daily_feed():
|
|
_last_discord_post_day = today
|
|
|
|
|
|
def run_loop():
|
|
# Beat once at startup so the dashboard shows the notifier alive immediately after a
|
|
# (re)start, without waiting a full INTERVAL for the first pass. Then beat every wake:
|
|
# healthy => a fresh beat at most INTERVAL apart, which is how the dashboard tells this
|
|
# single-leader daemon apart from a genuinely dead one (per-worker thread checks can't).
|
|
metrics.record_heartbeat("subscription-notifier", INTERVAL)
|
|
# Wait first, then run — gives the app a moment to finish booting. The whole
|
|
# body is guarded: this is a daemon thread, so ANY escaping exception (a bad
|
|
# pass, a homepage refresh, a Discord post) would silently kill it for the
|
|
# process's lifetime — no notifier, no heartbeat, no restart. Swallow per
|
|
# iteration instead; the next wake retries and keeps the heartbeat fresh.
|
|
while not _STOP.wait(INTERVAL):
|
|
metrics.record_heartbeat("subscription-notifier", INTERVAL)
|
|
try:
|
|
run_pass()
|
|
_maybe_refresh_homepage()
|
|
_maybe_post_discord()
|
|
except Exception: # noqa: BLE001 - a bad iteration 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()
|