diff --git a/.env.example b/.env.example index b149458..fafed6f 100644 --- a/.env.example +++ b/.env.example @@ -84,6 +84,15 @@ SMTP_PORT=587 SMTP_USERNAME= SMTP_PASSWORD= +# Public push notifications via self-hosted ntfy (https://ntfy.sh, single binary). +# Leave NTFY_SERVER empty to disable. Topics published: --warning, +# --danger, -warning, -danger, -p1-outlook, +# -status. See docs/NOTIFICATIONS.md. +NTFY_SERVER= +NTFY_TOPIC_PREFIX=ping +NTFY_TOKEN= +PUBLIC_URL=https://water.buildfor.life/ + # Matrix Alerting Configuration MATRIX_HOMESERVER=https://matrix.org MATRIX_ACCESS_TOKEN= diff --git a/README.md b/README.md index 1a721fc..36b5928 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,11 @@ background in [Teaching a Model to See the Ping River Rise 13 Hours Early](https - **Shows it.** A Leaflet map with the river drawn as OSM geometry and styled by live discharge, rain gauges, the Chiang Mai inundation zones, per-station history, the forecast card, a replay of the 2024 flood, English/Thai, light/dark. -- **Alerts** (optional) to a Matrix room when a gauge crosses its thresholds. +- **Notifies.** Public push alerts over a self-hosted [ntfy](https://ntfy.sh): one + message when a gauge crosses its warning or danger level, one all-clear on the + way down, an opt-in early-warning topic from the model, nothing in between. + Subscribe from the free app, no account. Matrix room alerts for a team are + also supported. ## Quick start @@ -79,6 +83,8 @@ Asia/Bangkok wall-clock without an offset suffix. | `GET /api/forecast/history/{code}?hours=N&horizon=24` | Forecasts as issued, for auditing lead time after the fact | | `GET /api/hii/rainfall/latest`, `/api/hii/waterlevel/latest` | Latest ThaiWater/HII gauge readings | | `GET /api/hii/rainfall/catchment?days=N` | HII gauge catchment-mean rain next to the Open-Meteo series the model uses | +| `GET /api/forecast/skill?station_code=P.1` | Issued forecasts vs what happened, per deployed model version | +| `GET /api/notifications` | ntfy server and topic names for the subscribe panel | | `GET /api/stats` | Row counts per source, date range, coverage | | `GET /health` | DB / upstream / memory checks | @@ -122,7 +128,8 @@ models/ trained bundles + metrics.json (gitignored) and evaluation - [docs/DATA_SOURCES.md](docs/DATA_SOURCES.md) — every ingested and candidate source, endpoints, quirks - [docs/STATION_MANAGEMENT_GUIDE.md](docs/STATION_MANAGEMENT_GUIDE.md) — adding/editing gauges - [docs/DATABASE_DEPLOYMENT_GUIDE.md](docs/DATABASE_DEPLOYMENT_GUIDE.md), [POSTGRESQL_SETUP.md](POSTGRESQL_SETUP.md) — database setup -- [docs/MATRIX_QUICK_START.md](docs/MATRIX_QUICK_START.md) — alert delivery +- [docs/NOTIFICATIONS.md](docs/NOTIFICATIONS.md) — public push alerts: topics, semantics, ntfy deployment +- [docs/MATRIX_QUICK_START.md](docs/MATRIX_QUICK_START.md) — Matrix room alerts for a team - [docs/GAP_FILLING_GUIDE.md](docs/GAP_FILLING_GUIDE.md) — data integrity tooling - [docs/references/NOTABLE_DOCUMENTS.md](docs/references/NOTABLE_DOCUMENTS.md) — official Thai government resources - Public overview: [buildfor.life/docs/tooling/ping-river-monitor](https://buildfor.life/docs/tooling/ping-river-monitor/) diff --git a/docs/NOTIFICATIONS.md b/docs/NOTIFICATIONS.md new file mode 100644 index 0000000..1454263 --- /dev/null +++ b/docs/NOTIFICATIONS.md @@ -0,0 +1,129 @@ +# Flood notifications (ntfy) + +Public push notifications for threshold crossings, without accounts, mailing +lists or app-store review: the monitor publishes to a self-hosted +[ntfy](https://ntfy.sh) server, and anyone subscribes to the topics they care +about from the free ntfy app (iOS, Android, F-Droid) or a browser tab. + +ntfy is one Go binary with a sqlite cache: ~30 MB RSS idle, negligible CPU. It +runs on the same VPS as the monitor. + +## What subscribers get + +Every message is a **transition**, never a state. Crossing up into a level sends +one message; dropping back below it (with 0.10 m hysteresis) sends one +all-clear. A river that sits at 3.9 m for three days produces two messages, not +seventy-two. In a quiet season a subscriber hears nothing. + +| Topic | Trigger | Priority | +|---|---|---| +| `ping-warning` | any gauge crosses its warning threshold; levels falling back | 4 (high) / 2 | +| `ping-danger` | any gauge crosses its danger threshold | 5 (max, breaks Do-Not-Disturb) | +| `ping--warning` | that gauge crosses warning; back to normal | 4 / 2 | +| `ping--danger` | that gauge crosses danger; back below danger | 5 / 3 | +| `ping-p1-outlook` | model P(warning within 24 h) at P.1 rises through 50 % (clears below 25 %) | 4 / 2 | +| `ping-status` | gauge feed stale ≥ 3 h; feed recovered | 3 / 2 | + +Station slugs are the code lowercased without the dot: `p1`, `p103`, `p67`. +Thresholds are the ones in `src/ml/features.py` (`THRESHOLDS`): P.1 3.70 / +4.20 m, P.103 5.95 / 6.75 m, and so on. + +The outlook topic is opt-in for a reason: it is model output, and the message +says so. Observed-crossing topics only ever report a gauge reading. + +Each message carries a click-through and an "Open dashboard" action button to +the public dashboard. + +## How it runs + +`src/notify.py` is called once per collection cycle inside the API process +(leader only), right after the forecast precompute, so it sees exactly the +readings and forecasts the dashboard shows. Per-key last-sent state is stored +in the `notification_state` table of the monitor's own database, so a restart +or redeploy never re-sends and never misses a crossing that happened while +the service was down (the next cycle compares against the persisted state). + +If ntfy is unreachable the transition is **not** recorded, so it is retried +on the next cycle rather than silently lost. Any other failure in the notify +step is logged and never reaches the collection loop. + +The dashboard's "🔔 Get alerts" button appears only when `NTFY_SERVER` is +set; it reads `GET /api/notifications` and renders subscribe links +(`ntfy://` deep links for the app, https links for the web UI). + +## Deployment + +On the monitor VPS, as root: + +```bash +cd /opt/thailand-water-monitor +NTFY_DOMAIN=ntfy.buildfor.life bash scripts/install_ntfy.sh +``` + +This installs the ntfy .deb, writes `/etc/ntfy/server.yml` (listen on +`127.0.0.1:2586`, anonymous read, token-only write, 72 h message cache, +signup/login/metrics off, tight visitor limits), enables the systemd unit, +creates the `monitor` user with **write-only access to `ping-*`**, mints a +token, and appends `NTFY_SERVER` / `NTFY_TOPIC_PREFIX` / `NTFY_TOKEN` to +`.env` if they are not there yet. Then: + +```bash +systemctl restart water-monitor +journalctl -u water-monitor -n 20 | grep ntfy # "ntfy notifications: https://... topics ping-*" +curl -s 'https://ntfy.buildfor.life/ping-status/json?poll=1' # anonymous read works +``` + +Put `https://ntfy.buildfor.life` in front of `127.0.0.1:2586` with whatever +already terminates TLS for `water.buildfor.life`. Subscribers hold a +long-lived connection, so the proxy needs websockets on and no short read +timeout: + +```caddyfile +ntfy.buildfor.life { + reverse_proxy 127.0.0.1:2586 +} +``` + +Cloudflare tunnel: add a public hostname `ntfy.buildfor.life` → +`http://127.0.0.1:2586`. Cloudflare proxies websockets by default; nothing +else to set. + +Nothing about the message pipeline needs the domain to be public before you +test: with `NTFY_SERVER=http://127.0.0.1:2586` in `.env` the monitor +publishes locally and `curl .../ping-status/json?poll=1` shows what went out. + +## Configuration + +| Variable | Default | Meaning | +|---|---|---| +| `NTFY_SERVER` | *(empty = off)* | base URL of the ntfy server the monitor publishes to | +| `NTFY_TOPIC_PREFIX` | `ping` | first segment of every topic | +| `NTFY_TOKEN` | *(empty)* | bearer token if the server requires auth to publish (it does, see above) | +| `PUBLIC_URL` | `https://water.buildfor.life/` | click-through target in messages | + +Tunables in `src/notify.py`: `CLEAR_MARGIN_M` (0.10), `OUTLOOK_ON` / `OUTLOOK_OFF` +(0.50 / 0.25), stale feed threshold (3 h, argument to `evaluate`). + +## Testing + +`tests/test_notify.py` covers the state machine: quiet river sends nothing; +crossing once, then silence while above, then all-clear; hysteresis on the way +down; escalation to danger and back; basin digest grouping; outlook on/off; +heuristic forecasts ignored; stale feed and recovery; state survives a restart +through sqlite; a failed publish is retried next cycle. + +To exercise the real path against a real ntfy locally: run `ntfy serve` (any +platform, same binary), set `NTFY_SERVER`/`NTFY_TOKEN`, seed readings, and +poll the topic JSON. `scripts/e2e_notify.py` does exactly that if you want a +template. + +## Why ntfy and not … + +- **Matrix** (`src/alerting.py`, still there): needs a homeserver account per + subscriber and a room invite; fine for a team, wrong for the public. +- **Gotify**: also self-hosted and light, but Android-only client and one + account per subscriber. +- **Email / SMS**: deliverability work, cost per message, no priority + semantics; ntfy can forward to email per subscription if someone wants it. +- **Telegram / LINE bots**: platform lock-in and a bot token in the loop; can be + added later as ntfy→webhook fan-out without touching the monitor. diff --git a/scripts/e2e_notify.py b/scripts/e2e_notify.py new file mode 100644 index 0000000..adb59f4 --- /dev/null +++ b/scripts/e2e_notify.py @@ -0,0 +1,132 @@ +"""Drive the production notify path in-process: startup init -> seeded readings +-> forecast cache -> _notify_transitions -> sqlite state -> real ntfy.""" + +import asyncio +import datetime +import json +import os +import sys + +import requests + +os.environ.update( + DB_TYPE="sqlite", + WATER_DB_PATH=os.path.join(os.environ["LOCALAPPDATA"], "Temp", "smoke3.db"), + NTFY_SERVER="http://127.0.0.1:2586", + NTFY_TOKEN=os.environ.get("NTFY_TOKEN", ""), + NTFY_TOPIC_PREFIX="ping", +) +for f in ("smoke3.db",): + p = os.path.join(os.environ["LOCALAPPDATA"], "Temp", f) + if os.path.exists(p): + os.remove(p) + +from src import web_api # noqa: E402 +from src.config import Config # noqa: E402 + +assert Config.NTFY_SERVER + + +async def main(): + # what the lifespan does at startup, minus the scheduler + from src import notify as notify_mod + from src.forecast_history import ForecastHistoryStore + from src.water_scraper_v3 import EnhancedWaterMonitorScraper + + db_config = Config.get_database_config() + web_api.app_state["scraper"] = EnhancedWaterMonitorScraper(db_config) + store = ForecastHistoryStore(db_config["connection_string"], db_config["type"]) + store.connect() + web_api.app_state["forecast_store"] = store + state = notify_mod.NotificationState(store.engine, store.db_type) + pub = notify_mod.NtfyPublisher( + Config.NTFY_SERVER, prefix=Config.NTFY_TOPIC_PREFIX, token=Config.NTFY_TOKEN + ) + web_api.app_state["notify"] = (pub, state) + + scraper = web_api.app_state["scraper"] + now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0) + + def seed(level_p1, level_p103, ts): + rows = [ + { + "station_code": "P.1", + "station_id": 1, + "timestamp": ts, + "water_level": level_p1, + "discharge": 400.0, + "station_name_en": "Nawarat Bridge", + "station_name_th": "สะพานนวรัฐ", + "discharge_percent": 30.0, + "status": "active", + }, + { + "station_code": "P.103", + "station_id": 2, + "timestamp": ts, + "water_level": level_p103, + "discharge": 300.0, + "station_name_en": "Ring Road 3", + "station_name_th": "วงแหวน 3", + "discharge_percent": 20.0, + "status": "active", + }, + ] + scraper.db_adapter.save_measurements(rows) + + def forecast(p): + with web_api.FORECAST_CACHE_LOCK: + web_api.FORECAST_CACHE["all"] = ( + 0, + [ + { + "station_code": "P.1", + "horizon_hours": 24, + "p_warning": p, + "predicted_max_level": 3.9, + "source": "model", + } + ], + ) + + def poll(topic): + out = [] + for line in ( + requests.get(f"{Config.NTFY_SERVER}/{topic}/json?poll=1", timeout=5) + .text.strip() + .splitlines() + ): + m = json.loads(line) + if m.get("event") == "message": + out.append(m.get("title") or m.get("message", "")[:40]) + return out + + # cycle 1: quiet + seed(1.6, 3.2, now - datetime.timedelta(hours=2)) + forecast(0.02) + await web_api._notify_transitions() + # cycle 2: P.1 crosses warning, model outlook on + seed(3.75, 3.3, now - datetime.timedelta(hours=1)) + forecast(0.7) + await web_api._notify_transitions() + # cycle 3: same state -> silence + seed(3.80, 3.3, now) + forecast(0.65) + await web_api._notify_transitions() + + print("ping-p1-warning:", poll("ping-p1-warning")) + print("ping-warning: ", poll("ping-warning")) + print("ping-p1-outlook:", poll("ping-p1-outlook")) + print("ping-p103-warning:", poll("ping-p103-warning")) + from sqlalchemy import text + + with store.engine.connect() as c: + print( + "state table:", + c.execute( + text("SELECT key, state, value FROM notification_state ORDER BY key") + ).fetchall(), + ) + + +asyncio.run(main()) diff --git a/scripts/install_ntfy.sh b/scripts/install_ntfy.sh new file mode 100644 index 0000000..9698fc5 --- /dev/null +++ b/scripts/install_ntfy.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# Install ntfy (https://ntfy.sh) as the public notification server for the +# Ping River Monitor. Run as root on the monitor VPS. Idempotent. +# +# NTFY_DOMAIN=ntfy.buildfor.life bash scripts/install_ntfy.sh +# +# What it does: +# - installs the ntfy .deb from the official GitHub release (single Go +# binary, ~30 MB RSS, sqlite message cache) +# - writes /etc/ntfy/server.yml: listens on 127.0.0.1:2586 only (put it +# behind your existing reverse proxy / Cloudflare tunnel), anonymous +# READ on all topics, WRITE only with a token +# - creates the `monitor` publishing user + token, writes NTFY_SERVER / +# NTFY_TOKEN into /opt/thailand-water-monitor/.env if not present +# +# Reverse proxy: forward https://$NTFY_DOMAIN -> http://127.0.0.1:2586 with +# websockets enabled and a long/no read timeout (subscribers hold the +# connection open). Caddy: `reverse_proxy 127.0.0.1:2586`. Cloudflare +# tunnel: add a public hostname pointing at http://127.0.0.1:2586. +set -euo pipefail + +NTFY_DOMAIN="${NTFY_DOMAIN:?set NTFY_DOMAIN, e.g. ntfy.buildfor.life}" +NTFY_VERSION="${NTFY_VERSION:-2.28.0}" +MONITOR_DIR="${MONITOR_DIR:-/opt/thailand-water-monitor}" +LISTEN="${NTFY_LISTEN:-127.0.0.1:2586}" + +if ! command -v ntfy >/dev/null || [[ "$(ntfy --version 2>/dev/null | awk '{print $3}')" != "$NTFY_VERSION" ]]; then + tmp=$(mktemp -d) + curl -fsSL -o "$tmp/ntfy.deb" \ + "https://github.com/binwiederhier/ntfy/releases/download/v${NTFY_VERSION}/ntfy_${NTFY_VERSION}_linux_amd64.deb" + dpkg -i "$tmp/ntfy.deb" + rm -rf "$tmp" +fi + +install -d -m 755 /var/cache/ntfy /var/lib/ntfy +cat > /etc/ntfy/server.yml </dev/null && echo "ntfy up on ${LISTEN}" + +# Publishing identity for the monitor +if ! ntfy user list 2>/dev/null | grep -q '^user monitor (role'; then + NTFY_PASSWORD="$(openssl rand -base64 24)" ntfy user add --role=user monitor +fi +ntfy access monitor 'ping-*' write-only >/dev/null +# 'ping-*' read stays anonymous via auth-default-access + +token=$(ntfy token list monitor 2>/dev/null | awk '/^- tk_/{print $2; exit}') # '- tk_xxx (label), ...' +if [[ -z "$token" ]]; then + token=$(ntfy token add --label "water-monitor" monitor | grep -oE 'tk_[A-Za-z0-9]+' | head -1) # 'token tk_xxx created for user monitor' +fi + +env_file="${MONITOR_DIR}/.env" +if [[ -f "$env_file" ]] && ! grep -q '^NTFY_SERVER=' "$env_file"; then + { + echo "" + echo "# ntfy public notifications (scripts/install_ntfy.sh)" + echo "NTFY_SERVER=https://${NTFY_DOMAIN}" + echo "NTFY_TOPIC_PREFIX=ping" + echo "NTFY_TOKEN=${token}" + } >> "$env_file" + echo "wrote NTFY_* to ${env_file}; restart water-monitor to enable" +else + echo "NTFY_TOKEN=${token}" +fi + +echo +echo "Subscribe test (anonymous read): curl -s 'http://${LISTEN}/ping-status/json?poll=1'" +echo "Publish test (needs token): curl -s -H 'Authorization: Bearer ${token}' -d 'hello' http://${LISTEN}/ping-status" diff --git a/src/config.py b/src/config.py index de24488..4c7ad97 100644 --- a/src/config.py +++ b/src/config.py @@ -38,6 +38,12 @@ class Config: TARGET_URL = "https://hyd-app-db.rid.go.th/hydro1h.html" API_URL = "https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx" THAIWATER_API_KEY = os.getenv("THAIWATER_API_KEY") + + # Public flood notifications (ntfy). Off unless NTFY_SERVER is set. + NTFY_SERVER = os.getenv("NTFY_SERVER", "").strip() + NTFY_TOPIC_PREFIX = os.getenv("NTFY_TOPIC_PREFIX", "ping").strip() + NTFY_TOKEN = os.getenv("NTFY_TOKEN", "").strip() # publish token if ACL enabled + PUBLIC_URL = os.getenv("PUBLIC_URL", "https://water.buildfor.life/").strip() REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "30")) USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " diff --git a/src/notify.py b/src/notify.py new file mode 100644 index 0000000..2668033 --- /dev/null +++ b/src/notify.py @@ -0,0 +1,422 @@ +"""Public flood notifications over ntfy. + +Runs once per collection cycle inside the API process (leader only), right +after the forecast precompute, so it sees the same readings and forecasts the +dashboard shows. Publishes to a self-hosted ntfy server; anyone subscribes to +a topic from the free app or a browser, no account needed. + +Topics (all under one configurable prefix, default "ping"): + + {prefix}-{station}-warning observed level crossed the station's warning threshold + {prefix}-{station}-danger observed level crossed the danger threshold + {prefix}-warning any station crossed warning (basin-wide digest) + {prefix}-danger any station crossed danger + {prefix}-p1-outlook model early warning for Chiang Mai city: P.1's 24 h + warning probability crossed the alert level (opt-in; + the forecast is experimental and says so) + {prefix}-status feed/monitor health: data stale, recovered + +Each notification is a TRANSITION, not a state: crossing UP into a level sends +one message; dropping back below (with hysteresis) sends an all-clear. While +the river sits above a threshold nothing is repeated, so a subscriber in a +flood gets a handful of messages, not one an hour. The per-topic state is +persisted (notification_state table) so a restart never re-sends. + +Everything is fail-safe: ntfy unreachable, table missing, malformed +reading -> a logged warning, never an exception into the collection loop. +""" + +import datetime +import logging +from dataclasses import dataclass +from typing import Dict, Iterable, List, Optional + +import requests + +from .ml import features + +logger = logging.getLogger(__name__) + +# Hysteresis: an all-clear needs the level this far BELOW the threshold, so a +# river bobbing around 3.70 m does not toggle warning/clear every hour. +CLEAR_MARGIN_M = 0.10 +# Outlook alert fires when p_warning(24h) rises through ON, clears below OFF. +OUTLOOK_ON = 0.50 +OUTLOOK_OFF = 0.25 +# Below this the outlook is not announced at all (avoid "5 % chance" noise). +OUTLOOK_HORIZON = 24 + +STATION_NAMES: Dict[str, str] = { + "P.1": "Nawarat Bridge, Chiang Mai city", + "P.103": "Ring Road Bridge 3, Chiang Mai", + "P.67": "Ban Tae (Mae Taeng)", + "P.21": "Ban Rim Tai (Mae Rim)", + "P.75": "Ban Chai Lat", + "P.92": "Ban Muang Aut", + "P.20": "Ban Chiang Dao", + "P.4A": "Ban Mae Taeng", + "P.5": "Tha Nang Bridge (downstream)", + "P.81": "Ban Pong (downstream)", + "P.82": "Ban Sob Win", + "P.84": "Ban Panton", + "P.87": "Ban Pa Sang", + "P.77": "Ban Sop Mae Sapuat", + "P.85": "Ban Lai Kaew", + "P.76": "Ban Mae I Hai", +} + + +def _slug(code: str) -> str: + return code.lower().replace(".", "") + + +@dataclass +class Notification: + topic: str + title: str + message: str + priority: int = 3 # ntfy: 1 min .. 5 max + tags: Optional[List[str]] = None + click: Optional[str] = None + + +class NtfyPublisher: + def __init__( + self, + server: str, + prefix: str = "ping", + token: Optional[str] = None, + dashboard_url: str = "https://water.buildfor.life/", + timeout: int = 10, + ): + self.server = server.rstrip("/") + self.prefix = prefix + self.token = token + self.dashboard_url = dashboard_url + self.timeout = timeout + + def topic(self, *parts: str) -> str: + return "-".join([self.prefix, *parts]) + + def publish(self, n: Notification) -> bool: + headers = { + "Title": n.title, + "Priority": str(n.priority), + "Click": n.click or self.dashboard_url, + "Actions": f"view, Open dashboard, {n.click or self.dashboard_url}", + } + if n.tags: + headers["Tags"] = ",".join(n.tags) + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + try: + r = requests.post( + f"{self.server}/{n.topic}", + data=n.message.encode("utf-8"), + headers=headers, + timeout=self.timeout, + ) + if r.status_code >= 300: + logger.warning(f"ntfy {n.topic}: HTTP {r.status_code} {r.text[:120]}") + return False + return True + except Exception as error: + logger.warning(f"ntfy {n.topic}: {error}") + return False + + +class NotificationState: + """Per-key last-sent state, in the monitor's own SQL database.""" + + def __init__(self, engine, db_type: str): + self.engine = engine + self.db_type = db_type + self._ensure() + + def _ensure(self) -> None: + from sqlalchemy import text + + with self.engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE IF NOT EXISTS notification_state (" + "key VARCHAR(64) PRIMARY KEY, state VARCHAR(16) NOT NULL, " + "value NUMERIC(8,3), updated_at TIMESTAMP NOT NULL)" + ) + ) + + def get(self, key: str) -> Optional[str]: + from sqlalchemy import text + + with self.engine.connect() as conn: + row = conn.execute( + text("SELECT state FROM notification_state WHERE key = :k"), {"k": key} + ).fetchone() + return row[0] if row else None + + def set(self, key: str, state: str, value: Optional[float] = None) -> None: + from sqlalchemy import text + + now = datetime.datetime.now() + with self.engine.begin() as conn: + if self.db_type == "mysql": + sql = ( + "INSERT INTO notification_state (key, state, value, updated_at) " + "VALUES (:k, :s, :v, :t) ON DUPLICATE KEY UPDATE " + "state = VALUES(state), value = VALUES(value), updated_at = VALUES(updated_at)" + ) + else: + sql = ( + "INSERT INTO notification_state (key, state, value, updated_at) " + "VALUES (:k, :s, :v, :t) ON CONFLICT (key) DO UPDATE SET " + "state = EXCLUDED.state, value = EXCLUDED.value, updated_at = EXCLUDED.updated_at" + ) + conn.execute(text(sql), {"k": key, "s": state, "v": value, "t": now}) + + +class InMemoryState(NotificationState): + """For tests and when no SQL engine is available (loses state on restart).""" + + def __init__(self): # noqa: D107 - intentionally skips the SQL parent + self._d: Dict[str, str] = {} + + def get(self, key: str) -> Optional[str]: + return self._d.get(key) + + def set(self, key: str, state: str, value: Optional[float] = None) -> None: + self._d[key] = state + + +def _level_state(level: float, warn: float, danger: float, prev: Optional[str]) -> str: + """'clear' | 'warning' | 'danger', with hysteresis on the way down.""" + if level >= danger: + return "danger" + if level >= warn: + # from danger: stay 'danger' until below danger - margin + if prev == "danger" and level >= danger - CLEAR_MARGIN_M: + return "danger" + return "warning" + if prev in ("warning", "danger") and level >= warn - CLEAR_MARGIN_M: + return "warning" + return "clear" + + +def evaluate( + readings: Iterable[dict], + forecasts: Iterable[dict], + state: NotificationState, + publisher: NtfyPublisher, + stale_after_h: float = 3.0, + now: Optional[datetime.datetime] = None, +) -> List[Notification]: + """Compare current readings/forecasts with last-sent state; publish transitions. + + readings: rows with station_code, water_level, timestamp (latest per station) + forecasts: /forecast rows (station_code, horizon_hours, p_warning, predicted_max_level) + Returns the notifications that were published (for logs/tests). + """ + now = now or datetime.datetime.now() + sent: List[Notification] = [] + + def emit(n: Notification) -> bool: + ok = publisher.publish(n) + if ok: + sent.append(n) + return ok + + # ---- observed levels, per station, plus basin-wide fan-out + basin_changes: Dict[str, List[str]] = {"warning": [], "danger": [], "clear": []} + latest_ts: Optional[datetime.datetime] = None + for r in readings: + code = r.get("station_code") + level = r.get("water_level") + if not code or level is None: + continue + try: + level = float(level) + except (TypeError, ValueError): + continue + ts = r.get("timestamp") + if isinstance(ts, str): + try: + ts = datetime.datetime.fromisoformat(ts) + except ValueError: + ts = None + if isinstance(ts, datetime.datetime) and (latest_ts is None or ts > latest_ts): + latest_ts = ts + warn, danger = features.get_thresholds(code) + key = f"level:{code}" + prev = state.get(key) or "clear" + cur = _level_state(level, warn, danger, prev) + if cur == prev: + continue + name = STATION_NAMES.get(code, code) + slug = _slug(code) + when = ( + ts.strftime("%d %b %H:%M") if isinstance(ts, datetime.datetime) else "now" + ) + if cur == "danger": + ok = emit( + Notification( + publisher.topic(slug, "danger"), + f"DANGER level at {code}", + f"{name}: {level:.2f} m at {when}, above the danger level of {danger:.2f} m.", + priority=5, + tags=["rotating_light", code], + ) + ) + basin_changes["danger"].append(f"{code} {level:.2f} m") + elif cur == "warning": + if prev == "danger": + ok = emit( + Notification( + publisher.topic(slug, "danger"), + f"{code} back below danger level", + f"{name}: {level:.2f} m at {when}; still above the warning level of {warn:.2f} m.", + priority=3, + tags=["arrow_down", code], + ) + ) + basin_changes["clear"].append(f"{code} below danger ({level:.2f} m)") + else: + ok = emit( + Notification( + publisher.topic(slug, "warning"), + f"Warning level at {code}", + f"{name}: {level:.2f} m at {when}, above the warning level of {warn:.2f} m.", + priority=4, + tags=["warning", code], + ) + ) + basin_changes["warning"].append(f"{code} {level:.2f} m") + else: # clear + ok = emit( + Notification( + publisher.topic(slug, "warning"), + f"{code} back to normal", + f"{name}: {level:.2f} m at {when}, below the warning level of {warn:.2f} m.", + priority=2, + tags=["white_check_mark", code], + ) + ) + basin_changes["clear"].append(f"{code} normal ({level:.2f} m)") + # Only remember the transition once it was actually delivered: if ntfy + # was down, the next cycle retries instead of silently swallowing a + # flood crossing. + if ok: + state.set(key, cur, level) + + if basin_changes["danger"]: + emit( + Notification( + publisher.topic("danger"), + "Ping River: danger level reached", + "; ".join(basin_changes["danger"]), + priority=5, + tags=["rotating_light"], + ) + ) + if basin_changes["warning"]: + emit( + Notification( + publisher.topic("warning"), + "Ping River: warning level reached", + "; ".join(basin_changes["warning"]), + priority=4, + tags=["warning"], + ) + ) + if basin_changes["clear"]: + emit( + Notification( + publisher.topic("warning"), + "Ping River: levels falling", + "; ".join(basin_changes["clear"]), + priority=2, + tags=["white_check_mark"], + ) + ) + + # ---- model outlook for the city gauge (opt-in topic, experimental) + p1 = next( + ( + f + for f in forecasts + if f.get("station_code") == "P.1" + and f.get("horizon_hours") == OUTLOOK_HORIZON + and f.get("source") == "model" + ), + None, + ) + if p1 and p1.get("p_warning") is not None: + p = float(p1["p_warning"]) + key = "outlook:P.1" + prev = state.get(key) or "off" + cur = ( + "on" if (p >= OUTLOOK_ON or (prev == "on" and p >= OUTLOOK_OFF)) else "off" + ) + if cur != prev: + peak = p1.get("predicted_max_level") + warn, _ = features.get_thresholds("P.1") + if cur == "on": + ok = emit( + Notification( + publisher.topic("p1-outlook"), + "Early warning: Chiang Mai flood risk rising", + f"The forecast model gives a {p * 100:.0f}% chance that Nawarat Bridge (P.1) " + f"reaches {warn:.2f} m within 24 h" + + ( + f" (expected peak {float(peak):.2f} m)" + if peak is not None + else "" + ) + + ". Experimental model output, not an official warning; " + "follow ThaiWater/TMD for official alerts.", + priority=4, + tags=["crystal_ball"], + ) + ) + else: + ok = emit( + Notification( + publisher.topic("p1-outlook"), + "Chiang Mai flood risk easing", + f"The model's 24 h probability of reaching {warn:.2f} m at P.1 has dropped to {p * 100:.0f}%.", + priority=2, + tags=["crystal_ball"], + ) + ) + if ok: + state.set(key, cur, p) + + # ---- feed health + if latest_ts is not None: + age_h = (now - latest_ts).total_seconds() / 3600.0 + key = "feed" + prev = state.get(key) or "ok" + cur = "stale" if age_h >= stale_after_h else "ok" + if cur != prev: + if cur == "stale": + ok = emit( + Notification( + publisher.topic("status"), + "Ping River monitor: gauge feed stale", + f"No new readings for {age_h:.0f} h (last {latest_ts:%d %b %H:%M}). " + "Levels and forecasts on the dashboard are not current.", + priority=3, + tags=["hourglass"], + ) + ) + else: + ok = emit( + Notification( + publisher.topic("status"), + "Ping River monitor: feed recovered", + f"Readings are current again (latest {latest_ts:%d %b %H:%M}).", + priority=2, + tags=["white_check_mark"], + ) + ) + if ok: + state.set(key, cur, age_h) + return sent diff --git a/src/static/dashboard.html b/src/static/dashboard.html index 2074a83..889f70a 100644 --- a/src/static/dashboard.html +++ b/src/static/dashboard.html @@ -113,6 +113,25 @@ .lang-toggle { padding: 9px 12px; font-size: .78rem; font-weight: 800; white-space: nowrap; } .lang-toggle[data-active-lang="th"] { background: var(--mint); border-color: var(--mint-border); color: var(--mint-ink); } .theme-toggle { padding: 9px 11px; font-size: .95rem; line-height: 1; } + .alerts-panel { margin-bottom: 14px; padding: 18px 20px; border-radius: 16px; background: var(--card); border: 1px solid var(--border); box-shadow: var(--shadow); } + .alerts-head { display: flex; justify-content: space-between; align-items: center; gap: 12px; } + .alerts-head h2 { margin: 0; font-size: 1.15rem; } + .alerts-close { background: transparent; border: 0; color: var(--muted); font-size: 1.1rem; cursor: pointer; padding: 4px 8px; } + .alerts-intro { color: var(--muted); font-size: .92rem; line-height: 1.5; margin: 8px 0 12px; } + .alerts-server { display: flex; align-items: center; gap: 10px; flex-wrap: wrap; font-size: .9rem; margin-bottom: 12px; } + .alerts-server code { background: var(--surface); border: 1px solid var(--border); padding: 4px 8px; border-radius: 8px; font-size: .9rem; } + .alerts-server button { padding: 4px 10px; font-size: .8rem; } + .alerts-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 10px; } + .alerts-topic { border: 1px solid var(--border); border-radius: 12px; padding: 10px 12px; background: var(--surface); display: flex; flex-direction: column; gap: 4px; } + .alerts-topic .name { font-weight: 600; font-size: .95rem; } + .alerts-topic .desc { color: var(--muted); font-size: .82rem; line-height: 1.4; } + .alerts-topic .row { display: flex; align-items: center; gap: 8px; margin-top: 4px; flex-wrap: wrap; } + .alerts-topic code { font-size: .82rem; background: var(--card); border: 1px solid var(--border); padding: 2px 6px; border-radius: 6px; } + .alerts-topic a { font-size: .82rem; } + .alerts-topic.danger { border-color: rgba(220, 38, 38, .45); } + .alerts-topic.outlook { border-style: dashed; } + .alerts-foot { color: var(--muted); font-size: .82rem; margin: 12px 0 0; line-height: 1.6; } + .alerts-disclaimer { display: block; margin-top: 4px; } .leaflet-popup-content-wrapper, .leaflet-popup-tip { background: var(--card); color: var(--ink); } .leaflet-container a.leaflet-popup-close-button { color: var(--muted); } .leaflet-bar a, .leaflet-control-attribution { background: var(--surface); color: var(--ink); border-color: var(--border); } @@ -348,10 +367,33 @@ + + +