diff --git a/.env.example b/.env.example index e300f92..93a54b7 100644 --- a/.env.example +++ b/.env.example @@ -62,6 +62,13 @@ GAP_FILL_DAYS_ON_START=14 # share the status file between both services for /scraping/status COLLECTION_ENABLED=true # SCRAPING_STATUS_FILE=/opt/thailand-water-monitor/collector-status.json +# true: ping-collector also forecasts (models/flood_*.json.gz, rust/forecast) +# and sends the ntfy notifications; this app serves its stored forecast and +# computes itself only when that is older than FORECAST_MAX_AGE_SECONDS. +# Set by scripts/install_collector.sh --cutover-forecast. +FORECAST_IN_COLLECTOR=false +# FORECAST_MAX_AGE_SECONDS=7200 +# MODELS_DIR=models REQUEST_TIMEOUT=30 MAX_RETRIES=3 RETRY_DELAY_SECONDS=60 @@ -105,6 +112,9 @@ NTFY_PUBLISH_URL= NTFY_TOPIC_PREFIX=ping NTFY_TOKEN= PUBLIC_URL=https://water.buildfor.life/ +# ping-collector notifier: report RID lagging this many hours while ThaiWater +# telemetry is current (0 = never) +RID_STALE_ALERT_HOURS=18 # Matrix Alerting Configuration MATRIX_HOMESERVER=https://matrix.org diff --git a/.gitea/workflows/rust.yml b/.gitea/workflows/rust.yml index e40390d..c0ebe7d 100644 --- a/.gitea/workflows/rust.yml +++ b/.gitea/workflows/rust.yml @@ -1,7 +1,8 @@ name: Rust -# rust/ workspace (ping-collector): format, clippy, tests (including the -# database tests against a throwaway PostgreSQL service), release build. +# rust/ workspace (ping-collector, ping-forecast): format, clippy, tests +# (the Python parity fixtures, and the database tests against a throwaway +# PostgreSQL service), release build. # The Linux binary is uploaded as an artifact for scripts/install_collector.sh. on: diff --git a/CLAUDE.md b/CLAUDE.md index 567e7bb..0b6abc5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,6 +33,14 @@ Production: one systemd unit on a small VPS, `/opt/thailand-water-monitor`, user collector writes the same tables as the Python collectors: a change to a table, upsert or parser on one side must be made on the other until the Python side is retired (check with a live run of both into two databases). +- The forecast is served by Rust (`rust/forecast`) from models trained in Python. + A change to `src/ml/features.py`, `predict.py`, the HII fill or the dam frame + must be ported in the same change: regenerate + `rust/forecast/testdata/gen_forecast_fixtures.py` and keep + `cargo test -p ping-forecast` green. Changing thresholds, stages, leads or + notify constants: `uv run python -m src.ml.export --config` (a test fails + otherwise) and rebuild the collector. Notify texts: + `rust/collector/testdata/gen_notify_cases.py`. ## Where things are @@ -42,6 +50,8 @@ Production: one systemd unit on a small VPS, `/opt/thailand-water-monitor`, user - `scripts/retrain.sh` + `water-monitor-retrain.timer`: monthly retrain with staged promote. `scripts/dev_proxy.py`: serve the working-copy dashboard against the live API. - `rust/collector` (`ping-collector`): Rust data collector; with it deployed the - Python app runs `COLLECTION_ENABLED=false`. Rewrite plan: `docs/RUST_REWRITE.md`. + Python app runs `COLLECTION_ENABLED=false`. With `FORECAST_IN_COLLECTOR=true` it + also forecasts (`rust/forecast`) and sends ntfy; the API serves its + `forecast_current`. Rewrite plan: `docs/RUST_REWRITE.md`. - `docs/FLOOD_FORECASTING.md` is the authoritative model write-up; `docs/DATA_SOURCES.md` the source catalog. diff --git a/docs/FLOOD_FORECASTING.md b/docs/FLOOD_FORECASTING.md index 5c38013..53af623 100644 --- a/docs/FLOOD_FORECASTING.md +++ b/docs/FLOOD_FORECASTING.md @@ -616,6 +616,13 @@ fix, the served features match what this section evaluates. ### API +With `FORECAST_IN_COLLECTOR=true` the forecast is computed by `ping-collector` +(`rust/forecast`, a parity-tested port of this serving path) from +`models/flood_.json.gz`, which `train_all()` exports next to each +joblib bundle; `/forecast` serves its `forecast_current` row and computes +in-process only when that is older than `FORECAST_MAX_AGE_SECONDS` (2 h). See +`rust/README.md`. The rest of this section describes the in-process path. + `GET /forecast` (`src/web_api.py`) returns one JSON row per station × horizon with the fields listed in section 1. Results are cached in-process for `FORECAST_TTL = 900` seconds (15 minutes), which matches the data cadence — the diff --git a/docs/NOTIFICATIONS.md b/docs/NOTIFICATIONS.md index 014a7e3..52a5ce3 100644 --- a/docs/NOTIFICATIONS.md +++ b/docs/NOTIFICATIONS.md @@ -22,7 +22,7 @@ seventy-two. In a quiet season a subscriber hears nothing. | `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 | +| `ping-status` | no reading for ≥ 3 h; feed recovered; with the Rust notifier also: RID reports ≥ 18 h behind while ThaiWater is current, and back | 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 / @@ -36,7 +36,21 @@ the public dashboard. ## How it runs -`src/notify.py` is called once per collection cycle inside the API process +Since phase 2 of the Rust rewrite the notifier runs in `ping-collector` +(`rust/collector/src/notify.rs`) when `FORECAST_IN_COLLECTOR=true`, after each +forecast run. It is a port of `src/notify.py` with the same topics, texts and +state keys (checked word for word against Python on 751 generated messages), +so switching between the two sends nothing twice. One difference: the Rust +notifier also uses ThaiWater (HII) telemetry. For the stations whose HII twin +is the same sensor (P.1, P.103, P.20, P.4A, P.67, P.75, P.82, P.84, P.92, and +P.81 bias-corrected), a reading newer than RID's drives the level alerts; the +message then says "ThaiWater telemetry; the RID report is not in yet". RID's +report regularly lags 2-10 h at night, which used to send "gauge feed stale" +around midnight and "recovered" in the morning. Now the feed is stale only +when neither source has a reading for 3 h, and RID alone lagging is reported +once after `RID_STALE_ALERT_HOURS` (18, 0 = off). + +With `FORECAST_IN_COLLECTOR=false`, `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 diff --git a/docs/RUST_REWRITE.md b/docs/RUST_REWRITE.md index 430f146..3866ae9 100644 --- a/docs/RUST_REWRITE.md +++ b/docs/RUST_REWRITE.md @@ -1,6 +1,7 @@ # A Rust rewrite: assessment -Status: **in progress.** Phase 1 (collector) is in production since 2026-10-17; see +Status: **in progress.** Phase 1 (collector) is in production since 2026-10-17; +phase 2 (forecast + ntfy) is built and parity-verified, cutover pending; see [rust/README.md](../rust/README.md). Spike: [spikes/rust-forecast](../spikes/rust-forecast/README.md). ## Short version @@ -51,10 +52,10 @@ One binary, one process (tokio): the 257 KB river GeoJSON), static dashboard, CORS; `utoipa` if `/docs` stays. - sqlx against the existing schema. Python and Rust can run against the same database during migration. -- models: `scripts/retrain.sh` also exports each bundle to JSON (spike format, - ~4.9 MB per station vs 6.2 MB joblib). Rust loads them at start and reloads on - change. A bundle whose `feature_names` Rust cannot build is refused and the - station falls back to the heuristic, as Python does today. +- models: `train_all()` also exports each bundle (`models/flood_.json.gz`, + 1.2-2.3 MB), `scripts/retrain.sh` promotes it. Rust reloads changed files on + its next run. A bundle whose `feature_names` Rust cannot build falls back to + the heuristic, as Python does. ## The permanent cost: the Python/Rust model contract @@ -83,7 +84,7 @@ This slows model iteration, which is the most active part of the repo. | --- | --- | --- | | 0 | Spike: P.1 features + HGB inference | done, bit-identical | | 1 | `ping-collector`: RID, HII, dams, Open-Meteo | in production since 2026-10-17 23:09 (Python: `COLLECTION_ENABLED=false`) | -| 2 | Forecast crate + ntfy | next | +| 2 | Forecast crate + ntfy | built 2026-10-18, parity-verified; cutover with `install_collector.sh --cutover-forecast` | | 3 | Read API | | | 4 | Cleanup | | @@ -95,10 +96,16 @@ This slows model iteration, which is the most active part of the repo. RID reading in the database advances. ntfy moved to phase 2: the outlook alert needs the forecasts, which are still Python. Cutover and rollback are one command each (`scripts/install_collector.sh`). -2. **Forecast crate + ntfy** (~2k LOC): all 15 stations, dam/HII/rain inputs, - heuristic fallback, probabilities and stages, `forecast_history`, - `notification_state`; parity CI for every station on a flood window and a - dry window. The Python API then only serves. +2. **Forecast crate + ntfy** (`rust/forecast`, run by `ping-collector` with + `FORECAST_IN_COLLECTOR=true`): all stations, HII/rain/dam inputs, heuristic + fallback, probabilities and stages, `forecast_history`, `forecast_current` + (served by the API) and `notification_state`. Parity: fixtures from the + Python production path (CI: P.1 model; locally all 15), a live comparison + script, and 751 notifications replayed word for word. Porting found a + Python serving bug, fixed first: forecast rain (`rain_fc24`) was always + NaN at serving time (docs/FLOOD_FORECASTING.md section 5). The ntfy change: + ThaiWater mirror readings keep level alerts and the feed check current + while RID lags at night. The Python API then only serves. 3. **Read API** (~2k LOC): routes recorded from the Python API as golden files, then a Caddy switch. Python keeps only training/evaluation. 4. **Cleanup:** drop the multi-backend adapters, demo databases and the diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 7ea58a7..66a6dce 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -1199,7 +1199,9 @@ dependencies = [ "anyhow", "chrono", "clap", + "flate2", "log", + "ping-forecast", "reqwest", "serde", "serde_json", diff --git a/rust/README.md b/rust/README.md index 74b7608..19a1983 100644 --- a/rust/README.md +++ b/rust/README.md @@ -6,6 +6,7 @@ costs and the parity rules are in [docs/RUST_REWRITE.md](../docs/RUST_REWRITE.md | Crate | Replaces | Status | | --- | --- | --- | | `collector` (`ping-collector`) | RID, HII/ThaiWater, RID dams and Open-Meteo collection in the Python web app | phase 1: in production since 2026-10-17 | +| `forecast` (`ping-forecast`, library) | the flood forecast (`src/ml` serving path); run by `ping-collector` with ntfy | phase 2: built, parity-verified, cutover pending | ## ping-collector @@ -17,11 +18,15 @@ the database advances. ```text ping-collector run # the service: startup gap fill, RID at :05 (+10 min retries - # while RID is behind), HII/dams/rain once per clock hour + # while RID is behind), HII/dams/rain once per clock hour; + # with FORECAST_IN_COLLECTOR=true also forecast + ntfy ping-collector once [--dry-run] # one cycle of every source; --dry-run writes nothing # and compares the RID fetch with the database ping-collector fill-gaps # re-fetch RID reports with holes ping-collector check-db # connect, create missing tables, row counts +ping-collector forecast [--json] [--save] [--now T] [--rain-file F] + # forecast from the database and print it +ping-collector notify --dry-run # the readings the notifier judges, and what it would send ``` Configuration is the Python app's `.env` (read from the working directory, or @@ -35,8 +40,11 @@ only its line number, never its value. Variables already set (e.g. by systemd's `ENABLE_RESERVOIR_COLLECTION`, `ENABLE_RAIN_COLLECTION`, `HII_BASIN_CODE` (6), `STATION_CONFIG_PATH` (falls back to the bundled `src/data/stations.json`), `SCRAPING_STATUS_FILE` (status JSON served by the API's `/scraping/status`), -`LOG_LEVEL` / `RUST_LOG`. All timestamps are Asia/Bangkok wall-clock, -whatever the server's time zone. +`LOG_LEVEL` / `RUST_LOG`; for phase 2 `FORECAST_IN_COLLECTOR` (false), +`MODELS_DIR` (`models`), `NTFY_SERVER` / `NTFY_PUBLISH_URL` / `NTFY_TOPIC_PREFIX` +/ `NTFY_TOKEN` / `PUBLIC_URL` (as the Python app), `RID_STALE_ALERT_HOURS` (18, +0 = off). All timestamps are Asia/Bangkok wall-clock, whatever the server's +time zone. Not ported: the one-off archive backfills (`scripts/backfill_*.py`), which stay in Python. @@ -56,6 +64,47 @@ stay in Python. `/measurements/latest`, `/api/stats`, `/scraping/status` and model forecasts from a database filled only by `ping-collector`. +## Forecast and notifications (phase 2) + +With `FORECAST_IN_COLLECTOR=true` in the shared `.env`, `ping-collector` runs +the flood forecast after every cycle that brought newer RID or HII readings +(and at least hourly), stores it in `forecast_history` and `forecast_current` +(the Python API serves the latter as `/forecast`, falling back to computing +itself when it is older than `FORECAST_MAX_AGE_SECONDS`), and sends the ntfy +notifications. The Python app then neither forecasts nor notifies. + +- **Models** stay trained in Python. `src/ml/export.py` writes + `models/flood_.json.gz` next to each joblib bundle (`train_all()` + does it, `retrain.sh` promotes it, `python -m src.ml.export` exports + existing bundles); the collector reloads changed files on its next run. +- **Shared constants** (thresholds, flood stages, upstream leads, notify + settings) are compiled in from `forecast/flood_config.json`, generated by + `uv run python -m src.ml.export --config`. `tests/test_model_export.py` + fails while it differs from Python, and a change needs a collector rebuild. +- **Notifications**: same topics, texts, hysteresis and `notification_state` + keys as `src/notify.py`, so switching sends nothing twice. One change: for + the stations whose ThaiWater twin is the same sensor, a reading newer than + RID's (RID's report lags 2-10 h at night) drives the level alerts, marked + "ThaiWater telemetry; the RID report is not in yet". "Gauge feed stale" now + means no source has a reading for 3 h; RID alone lagging is reported once + after `RID_STALE_ALERT_HOURS`. +- Memory: 70 MB peak with all 15 models loaded (24-byte tree nodes). + +Verified: + +- `forecast/tests/parity.rs` replays fixtures from the Python production path + (`forecast/testdata/gen_forecast_fixtures.py`: the 2024 flood rising and at + its peak with RID lagging behind HII, a quiet window with dam data, a + stale-feed window, moved thresholds). HII fill, every feature and every + level are bit-identical; with all 15 local models 377 of 384 probabilities + are bit-identical and the rest within 2 ulp (`exp`). CI checks the P.1 model. +- `scripts/forecast_parity.py` does the same on the live database: Python and + `ping-collector forecast --json` at the same moment with the same rain + series. The install script runs it before `--cutover-forecast`. +- `collector/testdata/gen_notify_cases.py`: 652 hourly steps through + `src/notify.py` (thresholds, capacity guard, outlook, feed gaps, ntfy + outages); the Rust notifier sends the same 751 messages word for word. + ### Develop ```bash @@ -90,8 +139,13 @@ the same commit): ```bash cargo build --release --locked --manifest-path rust/Cargo.toml # or BIN= -sudo bash scripts/install_collector.sh # install + dry run, nothing switched -sudo bash scripts/install_collector.sh --cutover # Rust collects from now on -sudo bash scripts/install_collector.sh --rollback # back to Python collection +sudo bash scripts/install_collector.sh # install + checks, nothing switched +sudo bash scripts/install_collector.sh --cutover # Rust collects from now on +sudo bash scripts/install_collector.sh --cutover-forecast # Rust also forecasts and notifies +sudo bash scripts/install_collector.sh --rollback-forecast # forecasts + ntfy back to Python +sudo bash scripts/install_collector.sh --rollback # Python does everything again journalctl -u ping-collector -f ``` + +Every install exports the models and runs `scripts/forecast_parity.py`; +`--cutover-forecast` refuses to switch if it fails (`FORCE=1` overrides). diff --git a/rust/collector/Cargo.toml b/rust/collector/Cargo.toml index 729af76..ca042f6 100644 --- a/rust/collector/Cargo.toml +++ b/rust/collector/Cargo.toml @@ -13,6 +13,7 @@ anyhow.workspace = true chrono.workspace = true clap.workspace = true log.workspace = true +ping-forecast = { path = "../forecast" } reqwest.workspace = true serde.workspace = true serde_json.workspace = true @@ -20,3 +21,6 @@ sqlx.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true + +[dev-dependencies] +flate2.workspace = true diff --git a/rust/collector/src/config.rs b/rust/collector/src/config.rs index 5841705..d5a8a29 100644 --- a/rust/collector/src/config.rs +++ b/rust/collector/src/config.rs @@ -35,6 +35,19 @@ pub struct Config { pub station_config_path: PathBuf, /// Status JSON read by the Python API's /scraping/status. pub status_file: Option, + /// FORECAST_IN_COLLECTOR: forecast and notify here; the Python API then + /// serves the stored forecast and does neither. + pub forecast_enabled: bool, + /// Exported bundles (`flood_.json.gz`, src/ml/export.py). + pub models_dir: PathBuf, + /// Where to POST ntfy messages; `None` (NTFY_SERVER unset) = off. + pub ntfy_publish_url: Option, + pub ntfy_token: Option, + pub ntfy_topic_prefix: String, + /// Dashboard link in notifications. + pub public_url: String, + /// Report RID lagging this long while ThaiWater is current (0 = never). + pub rid_stale_alert_hours: f64, } fn var(name: &str) -> Option { @@ -125,6 +138,15 @@ impl Config { .unwrap_or_else(|| "stations.json".into()) .into(), status_file: var("SCRAPING_STATUS_FILE").map(PathBuf::from), + forecast_enabled: flag("FORECAST_IN_COLLECTOR", false), + models_dir: var("MODELS_DIR").unwrap_or_else(|| "models".into()).into(), + // as src/config.py: off unless NTFY_SERVER is set + ntfy_publish_url: var("NTFY_SERVER") + .map(|server| var("NTFY_PUBLISH_URL").unwrap_or(server)), + ntfy_token: var("NTFY_TOKEN"), + ntfy_topic_prefix: var("NTFY_TOPIC_PREFIX").unwrap_or_else(|| "ping".into()), + public_url: var("PUBLIC_URL").unwrap_or_else(|| "https://water.buildfor.life/".into()), + rid_stale_alert_hours: number("RID_STALE_ALERT_HOURS", 18.0)?, }; if cfg.poll_offset_minutes >= 60 { bail!("RID_POLL_OFFSET_MINUTES must be between 0 and 59"); diff --git a/rust/collector/src/db.rs b/rust/collector/src/db.rs index 90ccbfc..63aa074 100644 --- a/rust/collector/src/db.rs +++ b/rust/collector/src/db.rs @@ -131,6 +131,31 @@ const SCHEMA: &[&str] = &[ mae_rim NUMERIC(6,2), chiang_mai NUMERIC(6,2), catchment_mean NUMERIC(6,2), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)", + // src/forecast_history.py + "CREATE TABLE IF NOT EXISTS forecast_history ( + as_of TIMESTAMP NOT NULL, + station_code VARCHAR(10) NOT NULL, + horizon_hours INTEGER NOT NULL, + predicted_max_level NUMERIC(8,3), + p_warning NUMERIC(7,5), + p_danger NUMERIC(7,5), + current_level NUMERIC(8,3), + model_version VARCHAR(64), + source VARCHAR(16), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (as_of, station_code, horizon_hours))", + "CREATE INDEX IF NOT EXISTS idx_forecast_history_station ON forecast_history(station_code, as_of)", + // src/notify.py + "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)", + // The current /forecast payload, read by the Python API when + // FORECAST_IN_COLLECTOR=true (one row, id 'all'). + "CREATE TABLE IF NOT EXISTS forecast_current ( + id VARCHAR(16) PRIMARY KEY, + computed_at TIMESTAMP NOT NULL, + model_versions TEXT, + payload TEXT NOT NULL)", ]; pub async fn ensure_schema(pool: &PgPool) -> Result<()> { @@ -481,3 +506,277 @@ pub async fn table_counts(pool: &PgPool) -> Result> } Ok(out) } + +// --------------------------------------------------------------------------- +// Forecast inputs and outputs (FORECAST_IN_COLLECTOR) +// +// NUMERIC columns are read as text and parsed: Python's `float(Decimal)` is a +// correctly rounded parse of the same digits, so both sides get the same f64. +// Forecast values are written as text (shortest round-trip digits, as Python's +// repr) so PostgreSQL rounds the same decimal as it does for psycopg2. +// --------------------------------------------------------------------------- + +/// A row of (key, key, NUMERIC as text, NUMERIC as text). +type TextRow = (A, B, Option, Option); +/// (date, storage_pct, inflow_mcm, outflow_mcm) with NUMERIC as text. +type DamRow = (NaiveDate, Option, Option, Option); + +fn parse_num(v: Option) -> f64 { + v.and_then(|s| s.parse().ok()).unwrap_or(f64::NAN) +} + +fn num_text(v: f64) -> Option { + v.is_finite().then(|| format!("{v}")) +} + +/// RID readings in [start, end], as `data._fetch_from_db`. +pub async fn load_rid_window( + pool: &PgPool, + start: NaiveDateTime, + end: NaiveDateTime, +) -> Result> { + let rows: Vec> = sqlx::query_as( + "SELECT m.timestamp, s.station_code, m.water_level::text, m.discharge::text + FROM water_measurements m JOIN stations s ON m.station_id = s.id + WHERE m.timestamp >= $1 AND m.timestamp <= $2 ORDER BY m.timestamp", + ) + .bind(start) + .bind(end) + .fetch_all(pool) + .await + .context("loading the RID window")?; + Ok(rows + .into_iter() + .map(|(t, code, level, q)| ping_forecast::Row { + timestamp: ping_forecast::grid::floor_hour(t), + station_code: code, + water_level: parse_num(level), + discharge: parse_num(q), + }) + .collect()) +} + +/// HII mirror readings for `codes` in [start, end], as `data._fetch_hii_levels`. +pub async fn load_hii_window( + pool: &PgPool, + codes: &[String], + start: NaiveDateTime, + end: NaiveDateTime, +) -> Result> { + let rows: Vec> = sqlx::query_as( + "SELECT m.timestamp, s.rid_code, m.wl_msl::text, m.discharge::text + FROM hii_waterlevel m JOIN hii_wl_stations s ON s.id = m.station_id + WHERE s.rid_code IS NOT NULL AND m.timestamp >= $1 AND m.timestamp <= $2 + AND s.rid_code = ANY($3) + ORDER BY m.timestamp, m.station_id", + ) + .bind(start) + .bind(end) + .bind(codes) + .fetch_all(pool) + .await + .context("loading HII mirror levels")?; + Ok(rows + .into_iter() + .map(|(t, code, wl, q)| ping_forecast::hii::HiiRow { + timestamp: t, + station_code: code, + wl_msl: parse_num(wl), + discharge: parse_num(q), + }) + .collect()) +} + +/// Newest HII mirror reading for `codes` (a forecast trigger). +pub async fn latest_hii_timestamp( + pool: &PgPool, + codes: &[String], +) -> Result> { + Ok(sqlx::query_scalar( + "SELECT MAX(m.timestamp) FROM hii_waterlevel m JOIN hii_wl_stations s ON s.id = m.station_id + WHERE s.rid_code = ANY($1) AND m.wl_msl IS NOT NULL", + ) + .bind(codes) + .fetch_one(pool) + .await?) +} + +/// Daily reservoir rows from `start`, as `dam.load_daily`. +pub async fn load_dam_days( + pool: &PgPool, + dam_id: &str, + start: NaiveDate, +) -> Result> { + let rows: Vec = sqlx::query_as( + "SELECT date, storage_pct::text, inflow_mcm::text, outflow_mcm::text + FROM rid_reservoir_daily WHERE dam_id = $1 AND date >= $2 ORDER BY date", + ) + .bind(dam_id) + .bind(start) + .fetch_all(pool) + .await + .context("loading the dam series")?; + Ok(rows + .into_iter() + .map(|(date, s, i, o)| ping_forecast::dam::DamDay { + date, + storage_pct: parse_num(s), + inflow_mcm: parse_num(i), + outflow_mcm: parse_num(o), + }) + .collect()) +} + +/// The newest RID row per station (`get_latest_measurements(200)`). +#[derive(Clone, Debug, PartialEq)] +pub struct LatestReading { + pub timestamp: NaiveDateTime, + pub station_code: String, + pub water_level: Option, + pub discharge_percent: Option, +} + +pub async fn latest_rid_readings(pool: &PgPool) -> Result> { + let rows: Vec> = sqlx::query_as( + "SELECT m.timestamp, s.station_code, m.water_level::text, m.discharge_percent::text + FROM water_measurements m + JOIN stations s ON m.station_id = s.id + INNER JOIN ( + SELECT station_id, MAX(timestamp) AS max_timestamp + FROM water_measurements GROUP BY station_id + ) latest ON m.station_id = latest.station_id AND m.timestamp = latest.max_timestamp + ORDER BY s.station_code + LIMIT 200", + ) + .fetch_all(pool) + .await + .context("loading the latest readings")?; + let num = |v: Option| v.and_then(|s| s.parse().ok()); + Ok(rows + .into_iter() + .map(|(t, code, level, pct)| LatestReading { + timestamp: t, + station_code: code, + water_level: num(level), + discharge_percent: num(pct), + }) + .collect()) +} + +/// `ForecastHistoryStore.save_rows`: one row per (as_of, station, horizon). +pub async fn save_forecast_history( + pool: &PgPool, + rows: &[ping_forecast::Forecast], +) -> Result { + let rows = dedupe_last(rows, |r| (r.as_of, r.station_code.clone(), r.horizon_hours)); + if rows.is_empty() { + return Ok(0); + } + sqlx::query( + "INSERT INTO forecast_history (as_of, station_code, horizon_hours, predicted_max_level, + p_warning, p_danger, current_level, model_version, source) + SELECT a, c, h, pm::numeric, pw::numeric, pd::numeric, cl::numeric, mv, src + FROM UNNEST($1::timestamp[], $2::text[], $3::int4[], $4::text[], $5::text[], $6::text[], + $7::text[], $8::text[], $9::text[]) + AS t(a, c, h, pm, pw, pd, cl, mv, src) + ON CONFLICT (as_of, station_code, horizon_hours) DO UPDATE SET + predicted_max_level = EXCLUDED.predicted_max_level, p_warning = EXCLUDED.p_warning, + p_danger = EXCLUDED.p_danger, current_level = EXCLUDED.current_level, + model_version = EXCLUDED.model_version, source = EXCLUDED.source", + ) + .bind(rows.iter().map(|r| r.as_of).collect::>()) + .bind( + rows.iter() + .map(|r| r.station_code.clone()) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| r.horizon_hours as i32) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| num_text(r.predicted_max_level)) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| num_text(r.p_warning)) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| num_text(r.p_danger)) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| num_text(r.current_level)) + .collect::>(), + ) + .bind( + rows.iter() + .map(|r| r.model_version.clone()) + .collect::>(), + ) + .bind(rows.iter().map(|r| r.source.clone()).collect::>()) + .execute(pool) + .await + .context("upserting forecast_history")?; + Ok(rows.len()) +} + +/// The payload the Python API serves as /forecast. +pub async fn save_forecast_current( + pool: &PgPool, + computed_at: NaiveDateTime, + rows: &[ping_forecast::Forecast], +) -> Result<()> { + let versions: std::collections::BTreeSet<&str> = + rows.iter().map(|r| r.model_version.as_str()).collect(); + sqlx::query( + "INSERT INTO forecast_current (id, computed_at, model_versions, payload) + VALUES ('all', $1, $2, $3) + ON CONFLICT (id) DO UPDATE SET computed_at = EXCLUDED.computed_at, + model_versions = EXCLUDED.model_versions, payload = EXCLUDED.payload", + ) + .bind(computed_at) + .bind(versions.into_iter().collect::>().join(",")) + .bind(serde_json::to_string(rows)?) + .execute(pool) + .await + .context("writing forecast_current")?; + Ok(()) +} + +/// notification_state as a key -> state map. +pub async fn notification_states(pool: &PgPool) -> Result> { + let rows: Vec<(String, String)> = sqlx::query_as("SELECT key, state FROM notification_state") + .fetch_all(pool) + .await + .context("reading notification_state")?; + Ok(rows.into_iter().collect()) +} + +pub async fn set_notification_state( + pool: &PgPool, + key: &str, + state: &str, + value: Option, + now: NaiveDateTime, +) -> Result<()> { + sqlx::query( + "INSERT INTO notification_state (key, state, value, updated_at) + VALUES ($1, $2, $3::numeric, $4) ON CONFLICT (key) DO UPDATE SET + state = EXCLUDED.state, value = EXCLUDED.value, updated_at = EXCLUDED.updated_at", + ) + .bind(key) + .bind(state) + .bind(value.and_then(num_text)) + .bind(now) + .execute(pool) + .await + .with_context(|| format!("saving notification state {key}"))?; + Ok(()) +} diff --git a/rust/collector/src/db_tests.rs b/rust/collector/src/db_tests.rs index 00deca2..bf30fdb 100644 --- a/rust/collector/src/db_tests.rs +++ b/rust/collector/src/db_tests.rs @@ -29,10 +29,13 @@ impl TestDb { eprintln!("TEST_DATABASE_URL not set: skipping database test"); return None; }; + // the counter: tests run in parallel and the clock is coarse on Windows + static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); let schema = format!( - "t_{}_{}", + "t_{}_{}_{}", std::process::id(), - chrono::Utc::now().timestamp_nanos_opt().unwrap() + chrono::Utc::now().timestamp_nanos_opt().unwrap(), + N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) ); let admin = PgPool::connect(&url).await.expect("connect"); sqlx::query(&format!("CREATE SCHEMA {schema}")) @@ -330,3 +333,189 @@ async fn hii_dams_and_rain_round_trip() { assert_eq!((mean, missing), (1.11, None)); db.drop().await; } + +fn forecast_row(code: &str, h: u32, p: f64) -> ping_forecast::Forecast { + ping_forecast::Forecast { + station_code: code.into(), + horizon_hours: h, + p_warning: p, + p_danger: p / 2.0, + predicted_max_level: 3.123456, + current_level: 1.23, + as_of: t(2026, 10, 18, 13, 0), + model_version: "hgb-v3+test".into(), + trained_at: Some("2026-10-01T03:30:00".into()), + source: "model".into(), + threshold_warning: 3.7, + threshold_danger: 4.2, + stages: None, + } +} + +#[tokio::test] +async fn forecast_outputs_round_trip() { + let Some(db) = TestDb::new().await else { + return; + }; + // 0.123455 + 1 ulp: Python's repr gives 17 digits and NUMERIC(7,5) + // rounds that up; a float8 -> numeric cast (15 digits) would round down + let tricky = f64::from_bits(0.123455f64.to_bits() + 1); + let rows = [ + forecast_row("P.1", 24, tricky), + forecast_row("P.1", 24, 0.2), + forecast_row("P.20", 6, 1e-7), + ]; + assert_eq!(db::save_forecast_history(&db.pool, &rows).await.unwrap(), 2); + db::save_forecast_history(&db.pool, &rows[..1]) + .await + .unwrap(); + let got: Vec<(String, i32, f64, f64, f64)> = sqlx::query_as( + "SELECT station_code, horizon_hours, p_warning::float8, predicted_max_level::float8, + current_level::float8 FROM forecast_history ORDER BY station_code", + ) + .fetch_all(&db.pool) + .await + .unwrap(); + assert_eq!( + got, + vec![ + ("P.1".into(), 24, 0.12346, 3.123, 1.23), + ("P.20".into(), 6, 0.0, 3.123, 1.23) + ] + ); + + let now = t(2026, 10, 18, 13, 7); + db::save_forecast_current(&db.pool, now, &rows[1..]) + .await + .unwrap(); + db::save_forecast_current(&db.pool, now, &rows[..1]) + .await + .unwrap(); + let (at, versions, payload): (NaiveDateTime, String, String) = + sqlx::query_as("SELECT computed_at, model_versions, payload FROM forecast_current") + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!((at, versions.as_str()), (now, "hgb-v3+test")); + let v: serde_json::Value = serde_json::from_str(&payload).unwrap(); + assert_eq!(v[0]["as_of"], "2026-10-18T13:00:00"); + assert_eq!(v[0]["p_warning"].as_f64(), Some(tricky)); + assert!(v[0].get("stages").is_none()); + db.drop().await; +} + +#[tokio::test] +async fn notification_state_round_trip() { + let Some(db) = TestDb::new().await else { + return; + }; + let now = t(2026, 10, 18, 2, 0); + db::set_notification_state(&db.pool, "level:P.1", "warning", Some(3.7512), now) + .await + .unwrap(); + db::set_notification_state(&db.pool, "feed", "stale", None, now) + .await + .unwrap(); + db::set_notification_state(&db.pool, "level:P.1", "clear", Some(3.5), now) + .await + .unwrap(); + let states = db::notification_states(&db.pool).await.unwrap(); + assert_eq!(states.len(), 2); + assert_eq!(states["level:P.1"], "clear"); + assert_eq!(states["feed"], "stale"); + db.drop().await; +} + +#[tokio::test] +async fn forecast_inputs_load_like_python() { + let Some(db) = TestDb::new().await else { + return; + }; + let stations = Stations::bundled(); + let r = |h: u32, level: f64| Reading { + timestamp: t(2026, 10, 18, h, 0), + station_num: 8, // P.1 + water_level: level, + discharge: Some(51.25), + discharge_percent: Some(40.5), + }; + db::save_rid(&db.pool, &[r(10, 1.1), r(11, 1.2), r(12, 1.3)], &stations) + .await + .unwrap(); + let rows = db::load_rid_window(&db.pool, t(2026, 10, 18, 11, 0), t(2026, 10, 18, 12, 30)) + .await + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!( + (rows[0].station_code.as_str(), rows[0].water_level), + ("P.1", 1.2) + ); + assert_eq!(rows[1].discharge, 51.25); + + let latest = db::latest_rid_readings(&db.pool).await.unwrap(); + assert_eq!(latest.len(), 1); + assert_eq!( + ( + latest[0].timestamp, + latest[0].water_level, + latest[0].discharge_percent + ), + (t(2026, 10, 18, 12, 0), Some(1.3), Some(40.5)) + ); + + let wl = |minute: u32, v: f64| WaterLevelRecord { + station_id: 5, + oldcode: None, + rid_code: Some("P.1".into()), + name_th: None, + name_en: None, + latitude: None, + longitude: None, + agency: None, + river_name: None, + offset_msl: None, + ground_level_msl: None, + min_bank_msl: None, + critical_level_msl: None, + critical_level_m: None, + qmax: None, + is_key_station: true, + timestamp: t(2026, 10, 18, 13, minute), + wl_msl: Some(v), + wl_m: None, + discharge: Some(52.0), + flow_rate: None, + storage_percent: None, + situation_level: None, + diff_wl_bank: None, + }; + db::save_hii_waterlevel( + &db.pool, + &[wl(0, 301.4), wl(40, 301.45)], + t(2026, 10, 18, 13, 45), + ) + .await + .unwrap(); + let codes = vec!["P.1".to_string(), "P.103".to_string()]; + let hii = db::load_hii_window( + &db.pool, + &codes, + t(2026, 10, 18, 0, 0), + t(2026, 10, 18, 14, 0), + ) + .await + .unwrap(); + assert_eq!(hii.len(), 2); + assert_eq!(ping_forecast::hii::prepare(&hii)[0].wl_msl, 301.45); + assert_eq!( + db::latest_hii_timestamp(&db.pool, &codes).await.unwrap(), + Some(t(2026, 10, 18, 13, 40)) + ); + assert!( + db::load_dam_days(&db.pool, "200103", d(2026, 10, 1)) + .await + .unwrap() + .is_empty() + ); + db.drop().await; +} diff --git a/rust/collector/src/forecasting.rs b/rust/collector/src/forecasting.rs new file mode 100644 index 0000000..f8ba940 --- /dev/null +++ b/rust/collector/src/forecasting.rs @@ -0,0 +1,262 @@ +//! Forecasting and notifications in the collector (FORECAST_IN_COLLECTOR=true), +//! replacing the Python API's forecast_notify_task. +//! +//! After each collection cycle that brought newer RID or HII readings, and at +//! least once per clock hour: load the last 336 h (RID plus the HII mirror +//! fill), fetch the rain forecast, run every exported model, store the result +//! (`forecast_history`, and `forecast_current` which the API serves as +//! /forecast) and push ntfy transitions. + +use crate::config::Config; +use crate::db; +use crate::notify::{self, Notifier, Settings}; +use crate::rain::OpenMeteoClient; +use crate::util::floor_hour; +use anyhow::{Result, bail}; +use chrono::{Duration, NaiveDateTime}; +use ping_forecast::config::config; +use ping_forecast::dam::hourly_frame; +use ping_forecast::hii::{self, FillInfo}; +use ping_forecast::rain::RainSeries; +use ping_forecast::{Forecast, ModelSet, ModelStore}; +use sqlx::PgPool; + +/// Everything one forecast run used and produced. +pub struct Snapshot { + pub now: NaiveDateTime, + pub fills: Vec, + pub rain_hours: usize, + pub dam_days: usize, + pub forecasts: Vec, +} + +/// `predict.get_latest_forecasts` with this process's data access. Source +/// failures degrade like Python: no HII fill, NaN rain, NaN dam features. +pub async fn compute( + pool: &PgPool, + rain_client: &OpenMeteoClient, + models: &ModelSet, + now: NaiveDateTime, + rain_override: Option, +) -> Result { + let cfg = config(); + let start = now - Duration::hours(cfg.latest_window_h); + let rid = db::load_rid_window(pool, start, now).await?; + let codes = hii::fill_candidates(&rid); + let hii_rows = if codes.is_empty() { + vec![] + } else { + db::load_hii_window(pool, &codes, start, now) + .await + .unwrap_or_else(|e| { + tracing::warn!("HII gap-fill skipped (fetch failed): {e:#}"); + vec![] + }) + }; + let (merged, fills) = hii::fill(&rid, &hii::prepare(&hii_rows)); + for f in &fills { + tracing::debug!( + "HII gap-fill {}: +{} hours (offset {:.3} m)", + f.station_code, + f.filled_hours, + f.offset + ); + } + let rain: RainSeries = match rain_override { + Some(r) => r, + None => match rain_client.fetch_forecast(5, 2).await { + Ok(rows) => rows + .into_iter() + .map(|r| (r.timestamp, r.catchment_mean.unwrap_or(f64::NAN))) + .collect(), + Err(e) => { + tracing::warn!("live rain unavailable, rain features will be NaN: {e:#}"); + RainSeries::new() + } + }, + }; + let dam_start = now.date() - Duration::days(cfg.dam.serving_days); + let days = db::load_dam_days(pool, &cfg.dam.dam_id, dam_start) + .await + .unwrap_or_else(|e| { + tracing::warn!("dam state unavailable, dam features will be NaN: {e:#}"); + vec![] + }); + let dam = hourly_frame(&days).unwrap_or_default(); + let forecasts = ping_forecast::forecasts(&merged, models, now, Some(&rain), Some(&dam)); + Ok(Snapshot { + now, + fills, + rain_hours: rain.values().filter(|v| !v.is_nan()).count(), + dam_days: days.len(), + forecasts, + }) +} + +pub fn summarize(s: &Snapshot) -> String { + let model = s.forecasts.iter().filter(|f| f.source == "model").count(); + let p1 = s + .forecasts + .iter() + .find(|f| f.station_code == "P.1" && f.horizon_hours == 24); + let filled: Vec = s + .fills + .iter() + .filter(|f| { + f.latest + .is_some_and(|(t, _)| s.now - t < Duration::hours(24)) + }) + .map(|f| format!("{} +{}h", f.station_code, f.filled_hours)) + .collect(); + format!( + "{} rows ({model} model, {} heuristic){}{}; rain {} h, dam {} days", + s.forecasts.len(), + s.forecasts.len() - model, + p1.map(|f| format!( + "; P.1 {} {:.2} m, 24 h peak {:.2} m, p_warning {:.3}", + f.as_of.format("%d %H:%M"), + f.current_level, + f.predicted_max_level, + f.p_warning + )) + .unwrap_or_default(), + if filled.is_empty() { + String::new() + } else { + format!("; HII fill {}", filled.join(", ")) + }, + s.rain_hours, + s.dam_days + ) +} + +pub fn notifier(cfg: &Config, http: reqwest::Client) -> Option { + let url = cfg.ntfy_publish_url.clone()?; + Some(Notifier { + http, + publish_url: url, + token: cfg.ntfy_token.clone(), + dashboard_url: cfg.public_url.clone(), + settings: Settings { + prefix: cfg.ntfy_topic_prefix.clone(), + rid_stale_alert_hours: cfg.rid_stale_alert_hours, + }, + }) +} + +/// What the status file reports about the last run. +#[derive(Clone, Debug)] +pub struct RunInfo { + pub at: NaiveDateTime, + pub rows: usize, + pub model_rows: usize, + pub models: usize, + pub notified: usize, +} + +pub struct Forecaster { + store: ModelStore, + notifier: Option, + last_inputs: Option<(Option, Option)>, + last_slot: Option, +} + +impl Forecaster { + pub fn new(cfg: &Config, http: reqwest::Client) -> Self { + let notifier = notifier(cfg, http); + match ¬ifier { + Some(n) => tracing::info!( + "ntfy notifications: publish to {}, topics {}-*", + n.publish_url, + n.settings.prefix + ), + None => tracing::info!("ntfy notifications off (NTFY_SERVER not set)"), + } + Self { + store: ModelStore::new(&cfg.models_dir), + notifier, + last_inputs: None, + last_slot: None, + } + } + + /// Run when RID or the HII mirrors brought newer readings, and once per + /// clock hour regardless (so a stopped feed is noticed and reported). + pub async fn maybe_run( + &mut self, + pool: &PgPool, + rain: &OpenMeteoClient, + now: NaiveDateTime, + ) -> Option { + let codes = &config().hii_fill_stations; + let inputs = ( + db::latest_rid_timestamp(pool).await.ok().flatten(), + db::latest_hii_timestamp(pool, codes).await.ok().flatten(), + ); + let slot = floor_hour(now); + if self.last_inputs == Some(inputs) && self.last_slot == Some(slot) { + return None; + } + match self.run(pool, rain, now).await { + Ok(info) => { + self.last_inputs = Some(inputs); + self.last_slot = Some(slot); + Some(info) + } + Err(e) => { + tracing::error!("forecast run failed: {e:#}"); + None + } + } + } + + pub async fn run( + &mut self, + pool: &PgPool, + rain: &OpenMeteoClient, + now: NaiveDateTime, + ) -> Result { + let models = self.store.refresh()?; + if models.is_empty() { + bail!( + "no exported models in {} (run `python -m src.ml.export`)", + self.store.dir().display() + ); + } + let snap = compute(pool, rain, &models, now, None).await?; + let saved = db::save_forecast_history(pool, &snap.forecasts).await?; + db::save_forecast_current(pool, now, &snap.forecasts).await?; + tracing::info!("forecast: {}; {saved} stored", summarize(&snap)); + let mut notified = 0; + if let Some(n) = &self.notifier { + let readings = + notify::readings_with_hii(&db::latest_rid_readings(pool).await?, &snap.fills); + match n.run(pool, &readings, &snap.forecasts, now).await { + Ok(sent) => { + notified = sent.len(); + if !sent.is_empty() { + tracing::info!( + "ntfy: published {}", + sent.iter() + .map(|s| format!("{}: {}", s.topic, s.title)) + .collect::>() + .join(", ") + ); + } + } + Err(e) => tracing::warn!("ntfy notify cycle failed: {e:#}"), + } + } + Ok(RunInfo { + at: now, + rows: snap.forecasts.len(), + model_rows: snap + .forecasts + .iter() + .filter(|f| f.source == "model") + .count(), + models: models.len(), + notified, + }) + } +} diff --git a/rust/collector/src/main.rs b/rust/collector/src/main.rs index 3f1db61..f7e683a 100644 --- a/rust/collector/src/main.rs +++ b/rust/collector/src/main.rs @@ -1,9 +1,10 @@ //! ping-collector: the Ping River Monitor's data collection service. //! //! Replaces the collection half of the Python web app (RID hourly gauges, -//! HII/ThaiWater rain + water level, RID dams, Open-Meteo rain). Writes the -//! same PostgreSQL tables; the Python API keeps forecasting and notifying -//! with `COLLECTION_ENABLED=false`. See docs/RUST_REWRITE.md. +//! HII/ThaiWater rain + water level, RID dams, Open-Meteo rain) and, with +//! FORECAST_IN_COLLECTOR=true, the flood forecast (rust/forecast) and the +//! ntfy notifications. Writes the same PostgreSQL tables as the Python app, +//! which then runs with `COLLECTION_ENABLED=false`. See docs/RUST_REWRITE.md. mod config; mod dams; @@ -11,7 +12,9 @@ mod db; #[cfg(test)] mod db_tests; mod envfile; +mod forecasting; mod hii; +mod notify; mod rain; mod rid; mod service; @@ -47,6 +50,51 @@ enum Command { FillGaps { days: String }, /// Connect, create missing tables, print row counts and the newest RID reading CheckDb, + /// Forecast from the database now and print it; stores nothing without --save + Forecast { + /// Print the /forecast JSON instead of a table + #[arg(long)] + json: bool, + /// Store the result (forecast_history and the API's forecast_current) + #[arg(long)] + save: bool, + /// Forecast as of this local time, e.g. 2026-10-18T14:23:05.5 (comparisons) + #[arg(long)] + now: Option, + /// Rain series from a JSON file {"timestamps": [...], "values": [...]} + /// instead of a live Open-Meteo fetch (comparisons) + #[arg(long)] + rain_file: Option, + }, + /// Show the readings the notifier judges and the ntfy transitions due now + Notify { + /// Show what would be published; publish nothing, change no state + #[arg(long)] + dry_run: bool, + }, +} + +fn read_rain_file(path: &std::path::Path) -> Result { + #[derive(serde::Deserialize)] + struct File { + timestamps: Vec, + values: Vec>, + } + let f: File = serde_json::from_str(&std::fs::read_to_string(path)?)?; + if f.timestamps.len() != f.values.len() { + bail!("{}: timestamps and values differ in length", path.display()); + } + f.timestamps + .iter() + .zip(f.values) + .map(|(t, v)| Ok((parse_local(t)?, v.unwrap_or(f64::NAN)))) + .collect() +} + +fn parse_local(s: &str) -> Result { + chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S%.f") + .or_else(|_| chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M")) + .map_err(|e| anyhow::anyhow!("{s:?}: {e}")) } fn main() -> Result<()> { @@ -73,7 +121,7 @@ fn main() -> Result<()> { .unwrap_or_else(|_| "info".into()); tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::new(format!( - "warn,ping_collector={level}" + "warn,ping_collector={level},ping_forecast={level}" ))) .with_target(false) .init(); @@ -117,6 +165,17 @@ async fn run(cli: Cli) -> Result<()> { println!("{filled} readings re-fetched"); Ok(()) } + Command::Forecast { + json, + save, + now, + rain_file, + } => { + let now = now.as_deref().map(parse_local).transpose()?; + let rain = rain_file.as_deref().map(read_rain_file).transpose()?; + collector.forecast(json, save, now, rain).await + } + Command::Notify { dry_run } => collector.notify(dry_run).await, Command::CheckDb => { db::ensure_schema(&collector.pool).await?; for (table, n) in db::table_counts(&collector.pool).await? { diff --git a/rust/collector/src/notify.rs b/rust/collector/src/notify.rs new file mode 100644 index 0000000..88a39c2 --- /dev/null +++ b/rust/collector/src/notify.rs @@ -0,0 +1,879 @@ +//! Public flood notifications over ntfy: the port of src/notify.py, run by +//! the collector when FORECAST_IN_COLLECTOR=true. +//! +//! Same topics, texts, hysteresis, capacity guard, outlook and persisted +//! state keys (`notification_state`) as Python, so a cutover neither repeats +//! nor drops a message. One change: for stations whose HII/ThaiWater twin is +//! the same sensor, a reading newer than RID's (RID's report often lags 2-10 h +//! at night) is used for the level alerts and the feed check. The feed is +//! "stale" only when neither source has a reading for 3 h; RID lagging alone +//! is reported separately after RID_STALE_ALERT_HOURS. + +use crate::db::{self, LatestReading}; +use anyhow::Result; +use chrono::{Duration, NaiveDateTime}; +use ping_forecast::Forecast; +use ping_forecast::config::config; +use ping_forecast::hii::FillInfo; +use sqlx::PgPool; +use std::collections::HashMap; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Source { + Rid, + /// HII/ThaiWater telemetry of the same sensor, offset to the RID datum. + Hii, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct Reading { + pub station_code: String, + pub timestamp: NaiveDateTime, + pub water_level: Option, + pub discharge_percent: Option, + pub source: Source, + /// The station's newest RID reading with a level (for the RID lag check). + pub rid_timestamp: Option, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct Notification { + pub topic: String, + pub title: String, + pub message: String, + pub priority: u8, + pub tags: Vec, +} + +/// A notification and the state to record once it was delivered. +#[derive(Clone, Debug, PartialEq)] +pub struct Planned { + pub notification: Notification, + pub state: Option<(String, String, Option)>, +} + +/// The readings the notifier judges: the newest RID row per station, replaced +/// by the newest HII fill when that is newer (or RID's newest row has no level). +pub fn readings_with_hii(rid: &[LatestReading], fills: &[FillInfo]) -> Vec { + let by_code: HashMap<&str, &FillInfo> = + fills.iter().map(|f| (f.station_code.as_str(), f)).collect(); + rid.iter() + .map(|r| { + let fill = by_code.get(r.station_code.as_str()).and_then(|f| f.latest); + match fill { + Some((t, level)) + if t > r.timestamp || (t == r.timestamp && r.water_level.is_none()) => + { + Reading { + station_code: r.station_code.clone(), + timestamp: t, + water_level: Some(level), + // capacity drifts slowly: RID's figure if it is recent + discharge_percent: r + .discharge_percent + .filter(|_| t - r.timestamp <= Duration::hours(6)), + source: Source::Hii, + rid_timestamp: r.water_level.map(|_| r.timestamp), + } + } + _ => Reading { + station_code: r.station_code.clone(), + timestamp: r.timestamp, + water_level: r.water_level, + discharge_percent: r.discharge_percent, + source: Source::Rid, + rid_timestamp: r.water_level.map(|_| r.timestamp), + }, + } + }) + .collect() +} + +/// `(a - b).total_seconds() / 3600` as Python computes it. +fn hours_between(a: NaiveDateTime, b: NaiveDateTime) -> f64 { + (a - b) + .num_microseconds() + .map_or(f64::INFINITY, |us| us as f64 / 1e6 / 3600.0) +} + +fn slug(code: &str) -> String { + code.to_lowercase().replace('.', "") +} + +/// 'clear' | 'warning' | 'danger', with hysteresis on the way down. +fn level_state(level: f64, warn: f64, danger: f64, prev: &str) -> &'static str { + let margin = config().notify.clear_margin_m; + if level >= danger { + return "danger"; + } + if level >= warn { + if prev == "danger" && level >= danger - margin { + return "danger"; + } + return "warning"; + } + if (prev == "warning" || prev == "danger") && level >= warn - margin { + return "warning"; + } + "clear" +} + +pub struct Settings { + pub prefix: String, + /// 0 = off + pub rid_stale_alert_hours: f64, +} + +impl Settings { + fn topic(&self, parts: &[&str]) -> String { + std::iter::once(self.prefix.as_str()) + .chain(parts.iter().copied()) + .collect::>() + .join("-") + } +} + +fn note(n: Notification) -> Planned { + Planned { + notification: n, + state: None, + } +} + +fn n( + topic: String, + title: impl Into, + message: String, + priority: u8, + tags: &[&str], +) -> Notification { + Notification { + topic, + title: title.into(), + message, + priority, + tags: tags.iter().map(|t| t.to_string()).collect(), + } +} + +/// `notify.evaluate` as a plan: every transition against the stored states. +pub fn evaluate( + readings: &[Reading], + forecasts: &[Forecast], + states: &HashMap, + settings: &Settings, + now: NaiveDateTime, +) -> Vec { + let cfg = config(); + let nc = &cfg.notify; + let state = |k: &str| states.get(k).map(String::as_str); + let mut plan: Vec = Vec::new(); + let (mut b_danger, mut b_warning, mut b_clear) = (vec![], vec![], vec![]); + let mut latest: Option = None; + let mut latest_rid: Option = None; + + for r in readings { + latest_rid = latest_rid.max(r.rid_timestamp); + let Some(level) = r.water_level else { + continue; + }; + let ts = r.timestamp; + latest = latest.max(Some(ts)); + let code = r.station_code.as_str(); + let (warn, danger) = cfg.thresholds(code); + let key = format!("level:{code}"); + let prev = state(&key).unwrap_or("clear"); + let cur = level_state(level, warn, danger, prev); + if cur != "clear" + && prev == "clear" + && !nc.capacity_guard_exempt.iter().any(|c| c == code) + && let Some(pct) = r.discharge_percent + && pct < nc.capacity_guard_min_pct + { + tracing::info!( + "{code}: level {level:.2} m >= {warn:.2} but only {pct:.0}% capacity; \ + threshold looks stale, not alerting" + ); + continue; + } + if cur == prev { + continue; + } + let name = nc.station_names.get(code).map_or(code, String::as_str); + let slug = slug(code); + let mut when = ts.format("%d %b %H:%M").to_string(); + if r.source == Source::Hii { + when.push_str(" (ThaiWater telemetry; the RID report is not in yet)"); + } + let (topic_kind, title, message, priority, tag) = match (cur, prev) { + ("danger", _) => { + b_danger.push(format!("{code} {level:.2} m")); + ( + "danger", + format!("DANGER level at {code}"), + format!( + "{name}: {level:.2} m at {when}, above the danger level of {danger:.2} m." + ), + 5, + "rotating_light", + ) + } + ("warning", "danger") => { + b_clear.push(format!("{code} below danger ({level:.2} m)")); + ( + "danger", + format!("{code} back below danger level"), + format!( + "{name}: {level:.2} m at {when}; still above the warning level of {warn:.2} m." + ), + 3, + "arrow_down", + ) + } + ("warning", _) => { + b_warning.push(format!("{code} {level:.2} m")); + ( + "warning", + format!("Warning level at {code}"), + format!( + "{name}: {level:.2} m at {when}, above the warning level of {warn:.2} m." + ), + 4, + "warning", + ) + } + _ => { + b_clear.push(format!("{code} normal ({level:.2} m)")); + ( + "warning", + format!("{code} back to normal"), + format!( + "{name}: {level:.2} m at {when}, below the warning level of {warn:.2} m." + ), + 2, + "white_check_mark", + ) + } + }; + plan.push(Planned { + notification: n( + settings.topic(&[&slug, topic_kind]), + title, + message, + priority, + &[tag, code], + ), + // remembered only once delivered, so ntfy being down means a retry + state: Some((key, cur.to_string(), Some(level))), + }); + } + + if !b_danger.is_empty() { + plan.push(note(n( + settings.topic(&["danger"]), + "Ping River: danger level reached", + b_danger.join("; "), + 5, + &["rotating_light"], + ))); + } + if !b_warning.is_empty() { + plan.push(note(n( + settings.topic(&["warning"]), + "Ping River: warning level reached", + b_warning.join("; "), + 4, + &["warning"], + ))); + } + if !b_clear.is_empty() { + plan.push(note(n( + settings.topic(&["warning"]), + "Ping River: levels falling", + b_clear.join("; "), + 2, + &["white_check_mark"], + ))); + } + + // model outlook for the city gauge (opt-in topic, experimental) + if let Some(p1) = forecasts.iter().find(|f| { + f.station_code == "P.1" && f.horizon_hours == nc.outlook_horizon && f.source == "model" + }) { + let p = p1.p_warning; + let key = "outlook:P.1"; + let prev = state(key).unwrap_or("off"); + let cur = if p >= nc.outlook_on || (prev == "on" && p >= nc.outlook_off) { + "on" + } else { + "off" + }; + if cur != prev { + let (warn, _) = cfg.thresholds("P.1"); + let notification = if cur == "on" { + n( + settings.topic(&["p1-outlook"]), + "Early warning: Chiang Mai flood risk rising", + format!( + "The forecast model gives a {:.0}% chance that Nawarat Bridge (P.1) reaches \ + {warn:.2} m within 24 h (expected peak {:.2} m). Experimental model output, \ + not an official warning; follow ThaiWater/TMD for official alerts.", + p * 100.0, + p1.predicted_max_level + ), + 4, + &["crystal_ball"], + ) + } else { + n( + settings.topic(&["p1-outlook"]), + "Chiang Mai flood risk easing", + format!( + "The model's 24 h probability of reaching {warn:.2} m at P.1 has dropped to {:.0}%.", + p * 100.0 + ), + 2, + &["crystal_ball"], + ) + }; + plan.push(Planned { + notification, + state: Some((key.into(), cur.into(), Some(p))), + }); + } + } + + // feed health: stale only when no source has a recent reading + let mut feed_stale = false; + if let Some(latest) = latest { + let age_h = hours_between(now, latest); + let prev = state("feed").unwrap_or("ok"); + feed_stale = age_h >= nc.feed_stale_after_h; + let cur = if feed_stale { "stale" } else { "ok" }; + if cur != prev { + let notification = if feed_stale { + n( + settings.topic(&["status"]), + "Ping River monitor: gauge feed stale", + format!( + "No new readings for {age_h:.0} h (last {}). Levels and forecasts on the \ + dashboard are not current.", + latest.format("%d %b %H:%M") + ), + 3, + &["hourglass"], + ) + } else { + n( + settings.topic(&["status"]), + "Ping River monitor: feed recovered", + format!( + "Readings are current again (latest {}).", + latest.format("%d %b %H:%M") + ), + 2, + &["white_check_mark"], + ) + }; + plan.push(Planned { + notification, + state: Some(("feed".into(), cur.into(), Some(age_h))), + }); + } + } + + // RID itself far behind while ThaiWater keeps the levels current + if settings.rid_stale_alert_hours > 0.0 + && let Some(rid_ts) = latest_rid + { + let age_h = hours_between(now, rid_ts); + let prev = state("feed:rid").unwrap_or("ok"); + let cur = if age_h >= settings.rid_stale_alert_hours { + "stale" + } else { + "ok" + }; + // while the whole feed is stale, that message covers RID too + if cur != prev && !(cur == "stale" && feed_stale) { + let notification = if cur == "stale" { + n( + settings.topic(&["status"]), + "Ping River monitor: RID reports delayed", + format!( + "No RID gauge report for {age_h:.0} h (last {}). Levels and alerts continue \ + from ThaiWater telemetry for the stations it mirrors.", + rid_ts.format("%d %b %H:%M") + ), + 3, + &["hourglass"], + ) + } else { + n( + settings.topic(&["status"]), + "Ping River monitor: RID reports current again", + format!("Latest RID report {}.", rid_ts.format("%d %b %H:%M")), + 2, + &["white_check_mark"], + ) + }; + plan.push(Planned { + notification, + state: Some(("feed:rid".into(), cur.into(), Some(age_h))), + }); + } + } + plan +} + +/// Posts to ntfy and records delivered transitions. +pub struct Notifier { + pub http: reqwest::Client, + pub publish_url: String, + pub token: Option, + pub dashboard_url: String, + pub settings: Settings, +} + +impl Notifier { + pub async fn publish(&self, n: &Notification) -> bool { + let mut req = self + .http + .post(format!( + "{}/{}", + self.publish_url.trim_end_matches('/'), + n.topic + )) + .timeout(std::time::Duration::from_secs(10)) + .header("Title", &n.title) + .header("Priority", n.priority.to_string()) + .header("Click", &self.dashboard_url) + .header( + "Actions", + format!("view, Open dashboard, {}", self.dashboard_url), + ) + .body(n.message.clone()); + if !n.tags.is_empty() { + req = req.header("Tags", n.tags.join(",")); + } + if let Some(t) = &self.token { + req = req.bearer_auth(t); + } + match req.send().await { + Ok(r) if r.status().as_u16() < 300 => true, + Ok(r) => { + let status = r.status(); + let body = r.text().await.unwrap_or_default(); + tracing::warn!( + "ntfy {}: HTTP {status} {}", + n.topic, + body.chars().take(120).collect::() + ); + false + } + Err(e) => { + tracing::warn!("ntfy {}: {e}", n.topic); + false + } + } + } + + /// Evaluate and publish; returns the notifications delivered. + pub async fn run( + &self, + pool: &PgPool, + readings: &[Reading], + forecasts: &[Forecast], + now: NaiveDateTime, + ) -> Result> { + let states = db::notification_states(pool).await?; + let mut sent = vec![]; + for p in evaluate(readings, forecasts, &states, &self.settings, now) { + if !self.publish(&p.notification).await { + continue; + } + if let Some((key, state, value)) = &p.state { + db::set_notification_state(pool, key, state, *value, now).await?; + } + sent.push(p.notification); + } + Ok(sent) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::NaiveDate; + + fn now() -> NaiveDateTime { + NaiveDate::from_ymd_opt(2026, 10, 1) + .unwrap() + .and_hms_opt(12, 0, 0) + .unwrap() + } + + fn settings() -> Settings { + Settings { + prefix: "ping".into(), + rid_stale_alert_hours: 18.0, + } + } + + fn reading(code: &str, level: f64, hours_ago: i64) -> Reading { + Reading { + station_code: code.into(), + timestamp: now() - Duration::hours(hours_ago), + water_level: Some(level), + discharge_percent: None, + source: Source::Rid, + rid_timestamp: Some(now() - Duration::hours(hours_ago)), + } + } + + /// Run `evaluate` and apply the state of every planned transition, as a + /// successful publish would. + fn step( + rs: &[Reading], + fcs: &[Forecast], + st: &mut HashMap, + at: NaiveDateTime, + ) -> Vec { + let plan = evaluate(rs, fcs, st, &settings(), at); + for p in &plan { + if let Some((k, s, _)) = &p.state { + st.insert(k.clone(), s.clone()); + } + } + plan.into_iter().map(|p| p.notification).collect() + } + + fn titles(ns: &[Notification]) -> Vec<(&str, &str)> { + ns.iter() + .map(|n| (n.topic.as_str(), n.title.as_str())) + .collect() + } + + #[test] + fn quiet_river_sends_nothing() { + let mut st = HashMap::new(); + assert!(step(&[reading("P.1", 1.2, 1)], &[], &mut st, now()).is_empty()); + } + + #[test] + fn warning_once_then_silence_then_clear_with_hysteresis() { + let mut st = HashMap::new(); + let sent = step(&[reading("P.1", 3.75, 0)], &[], &mut st, now()); + assert_eq!( + titles(&sent), + [ + ("ping-p1-warning", "Warning level at P.1"), + ("ping-warning", "Ping River: warning level reached") + ] + ); + assert!( + sent[0] + .message + .starts_with("Nawarat Bridge, Chiang Mai city: 3.75 m at 01 Oct 12:00") + ); + assert!(step(&[reading("P.1", 3.8, 0)], &[], &mut st, now()).is_empty()); + // within the 0.10 m margin: still warning + assert!(step(&[reading("P.1", 3.65, 0)], &[], &mut st, now()).is_empty()); + let sent = step(&[reading("P.1", 3.5, 0)], &[], &mut st, now()); + assert_eq!(titles(&sent)[0], ("ping-p1-warning", "P.1 back to normal")); + assert_eq!( + titles(&sent)[1], + ("ping-warning", "Ping River: levels falling") + ); + } + + #[test] + fn danger_escalation_and_deescalation() { + let mut st = HashMap::new(); + step(&[reading("P.1", 3.8, 0)], &[], &mut st, now()); + let sent = step(&[reading("P.1", 4.3, 0)], &[], &mut st, now()); + assert_eq!(titles(&sent)[0], ("ping-p1-danger", "DANGER level at P.1")); + assert_eq!(sent[0].priority, 5); + assert!(step(&[reading("P.1", 4.15, 0)], &[], &mut st, now()).is_empty()); + let sent = step(&[reading("P.1", 4.0, 0)], &[], &mut st, now()); + assert_eq!( + titles(&sent)[0], + ("ping-p1-danger", "P.1 back below danger level") + ); + } + + #[test] + fn capacity_guard_blocks_a_stale_threshold_but_not_p1() { + let mut st = HashMap::new(); + let mut r = reading("P.77", 4.5, 0); + r.discharge_percent = Some(22.0); + assert!(step(&[r.clone()], &[], &mut st, now()).is_empty()); + r.discharge_percent = Some(80.0); + assert_eq!(step(&[r], &[], &mut st, now()).len(), 2); + let mut p1 = reading("P.1", 3.8, 0); + p1.discharge_percent = Some(10.0); + assert_eq!(step(&[p1], &[], &mut st, now()).len(), 2); + } + + fn outlook(p: f64) -> Forecast { + Forecast { + station_code: "P.1".into(), + horizon_hours: 24, + p_warning: p, + p_danger: 0.0, + predicted_max_level: 3.9, + current_level: 3.0, + as_of: now(), + model_version: "hgb-v3+test".into(), + trained_at: None, + source: "model".into(), + threshold_warning: 3.7, + threshold_danger: 4.2, + stages: None, + } + } + + #[test] + fn outlook_on_off_with_hysteresis_and_heuristic_ignored() { + let mut st = HashMap::new(); + let r = [reading("P.1", 3.0, 0)]; + let sent = step(&r, &[outlook(0.62)], &mut st, now()); + assert_eq!( + titles(&sent), + [( + "ping-p1-outlook", + "Early warning: Chiang Mai flood risk rising" + )] + ); + assert!(sent[0].message.contains("62% chance") && sent[0].message.contains("3.90 m")); + assert!(step(&r, &[outlook(0.3)], &mut st, now()).is_empty()); + assert_eq!( + step(&r, &[outlook(0.2)], &mut st, now())[0].title, + "Chiang Mai flood risk easing" + ); + let mut h = outlook(0.9); + h.source = "heuristic".into(); + assert!(step(&r, &[h], &mut st, now()).is_empty()); + } + + fn rid_row(code: &str, hours_ago: i64, level: Option, pct: Option) -> LatestReading { + LatestReading { + timestamp: now() - Duration::hours(hours_ago), + station_code: code.into(), + water_level: level, + discharge_percent: pct, + } + } + + fn fill(code: &str, hours_ago: i64, level: f64) -> FillInfo { + FillInfo { + station_code: code.into(), + offset: 300.0, + filled_hours: 1, + latest: Some((now() - Duration::hours(hours_ago), level)), + } + } + + #[test] + fn hii_reading_replaces_a_lagging_rid_reading() { + let rid = [ + rid_row("P.1", 5, Some(3.2), Some(40.0)), + rid_row("P.21", 5, Some(1.0), None), + rid_row("P.67", 1, None, None), + ]; + let rs = readings_with_hii( + &rid, + &[ + fill("P.1", 0, 3.8), + fill("P.67", 1, 2.0), + fill("P.103", 0, 1.0), + ], + ); + assert_eq!(rs[0].source, Source::Hii); + assert_eq!(rs[0].water_level, Some(3.8)); + assert_eq!(rs[0].discharge_percent, Some(40.0)); // RID figure 5 h old + assert_eq!(rs[1].source, Source::Rid); // no twin + assert_eq!((rs[2].source, rs[2].water_level), (Source::Hii, Some(2.0))); // RID row had no level + let mut st = HashMap::new(); + let sent = step(&rs, &[], &mut st, now()); + assert!( + sent[0] + .message + .contains("(ThaiWater telemetry; the RID report is not in yet)") + ); + // an older HII value never replaces a newer RID one + let rs = readings_with_hii( + &[rid_row("P.1", 0, Some(1.0), None)], + &[fill("P.1", 2, 3.9)], + ); + assert_eq!(rs[0].source, Source::Rid); + } + + #[test] + fn feed_is_stale_only_when_both_sources_are() { + let mut st = HashMap::new(); + // RID 5 h behind, ThaiWater current: no stale alert (the 2026-10 nights) + let rs = readings_with_hii( + &[rid_row("P.1", 5, Some(1.2), None)], + &[fill("P.1", 0, 1.25)], + ); + assert!(step(&rs, &[], &mut st, now()).is_empty()); + // both behind: stale, then recovered + let rs = readings_with_hii( + &[rid_row("P.1", 5, Some(1.2), None)], + &[fill("P.1", 4, 1.25)], + ); + let sent = step(&rs, &[], &mut st, now()); + assert_eq!( + titles(&sent), + [("ping-status", "Ping River monitor: gauge feed stale")] + ); + assert!( + sent[0] + .message + .starts_with("No new readings for 4 h (last 01 Oct 08:00)") + ); + let rs = readings_with_hii( + &[rid_row("P.1", 5, Some(1.2), None)], + &[fill("P.1", 0, 1.25)], + ); + assert_eq!( + step(&rs, &[], &mut st, now())[0].title, + "Ping River monitor: feed recovered" + ); + } + + #[test] + fn rid_lagging_a_day_is_reported_once_while_hii_is_current() { + let mut st = HashMap::new(); + let rs = readings_with_hii( + &[rid_row("P.1", 20, Some(1.2), None)], + &[fill("P.1", 0, 1.25)], + ); + let sent = step(&rs, &[], &mut st, now()); + assert_eq!( + titles(&sent), + [("ping-status", "Ping River monitor: RID reports delayed")] + ); + assert!(step(&rs, &[], &mut st, now()).is_empty()); + let rs = readings_with_hii(&[rid_row("P.1", 0, Some(1.2), None)], &[]); + assert_eq!( + step(&rs, &[], &mut st, now())[0].title, + "Ping River monitor: RID reports current again" + ); + // whole feed stale: only the general message + let mut st = HashMap::new(); + let sent = step(&[reading("P.1", 1.2, 30)], &[], &mut st, now()); + assert_eq!( + titles(&sent), + [("ping-status", "Ping River monitor: gauge feed stale")] + ); + } + + #[test] + fn undelivered_transitions_are_retried() { + let mut st = HashMap::new(); + let plan = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now()); + assert_eq!(plan[0].state.as_ref().unwrap().1, "warning"); + // nothing recorded (publish failed): the next run plans it again + let again = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now()); + assert_eq!(plan, again); + st.insert("level:P.1".into(), "warning".into()); + assert!( + evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now()) + .iter() + .all(|p| p.notification.topic == "ping-warning") + ); + } + + /// 652 hourly steps from src/notify.py itself (testdata/gen_notify_cases.py): + /// the same notifications, word for word, in the same order. + #[test] + fn matches_python_notify_on_generated_sequences() { + use std::io::Read; + let raw = include_bytes!("../testdata/notify_python_cases.json.gz"); + let mut text = String::new(); + flate2::read::GzDecoder::new(&raw[..]) + .read_to_string(&mut text) + .unwrap(); + let fixture: serde_json::Value = serde_json::from_str(&text).unwrap(); + let ts = |v: &serde_json::Value| { + NaiveDateTime::parse_from_str(v.as_str().unwrap(), "%Y-%m-%dT%H:%M:%S").unwrap() + }; + let mut compared = 0; + for seq in fixture["sequences"].as_array().unwrap() { + let mut states: HashMap = HashMap::new(); + for step in seq.as_array().unwrap() { + let readings: Vec = step["readings"] + .as_array() + .unwrap() + .iter() + .map(|r| { + let level = r["water_level"].as_f64(); + Reading { + station_code: r["station_code"].as_str().unwrap().into(), + timestamp: ts(&r["timestamp"]), + water_level: level, + discharge_percent: r["discharge_percent"].as_f64(), + source: Source::Rid, + rid_timestamp: level.map(|_| ts(&r["timestamp"])), + } + }) + .collect(); + let forecasts: Vec = step["forecasts"] + .as_array() + .unwrap() + .iter() + .map(|f| Forecast { + horizon_hours: f["horizon_hours"].as_u64().unwrap() as u32, + p_warning: f["p_warning"].as_f64().unwrap(), + predicted_max_level: f["predicted_max_level"].as_f64().unwrap(), + source: f["source"].as_str().unwrap().into(), + ..outlook(0.0) + }) + .collect(); + let down: Vec<&str> = step["down"] + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap()) + .collect(); + let mut delivered = vec![]; + for p in evaluate( + &readings, + &forecasts, + &states, + &settings(), + ts(&step["now"]), + ) { + if down.contains(&p.notification.topic.as_str()) { + continue; + } + if let Some((k, s, _)) = &p.state { + states.insert(k.clone(), s.clone()); + } + delivered.push(p.notification); + } + let want: Vec = step["sent"] + .as_array() + .unwrap() + .iter() + .map(|n| Notification { + topic: n["topic"].as_str().unwrap().into(), + title: n["title"].as_str().unwrap().into(), + message: n["message"].as_str().unwrap().into(), + priority: n["priority"].as_u64().unwrap() as u8, + tags: n["tags"] + .as_array() + .unwrap() + .iter() + .map(|t| t.as_str().unwrap().into()) + .collect(), + }) + .collect(); + assert_eq!(delivered, want, "at {}", step["now"]); + compared += want.len(); + } + } + assert!(compared > 700, "only {compared} notifications compared"); + } +} diff --git a/rust/collector/src/rain.rs b/rust/collector/src/rain.rs index 431d381..616d6c3 100644 --- a/rust/collector/src/rain.rs +++ b/rust/collector/src/rain.rs @@ -52,15 +52,10 @@ pub fn parse(payload: &Value) -> Vec { } by_time .into_iter() - .map(|(timestamp, points)| { - let present: Vec = points.iter().flatten().copied().collect(); - let catchment_mean = - (!present.is_empty()).then(|| present.iter().sum::() / present.len() as f64); - RainRow { - timestamp, - points, - catchment_mean, - } + .map(|(timestamp, points)| RainRow { + timestamp, + points, + catchment_mean: ping_forecast::rain::catchment_mean(&points), }) .collect() } diff --git a/rust/collector/src/service.rs b/rust/collector/src/service.rs index d4b00c4..a81149a 100644 --- a/rust/collector/src/service.rs +++ b/rust/collector/src/service.rs @@ -3,13 +3,17 @@ use crate::config::{Config, USER_AGENT}; use crate::dams::DamsClient; use crate::db; +use crate::forecasting::{self, Forecaster, RunInfo}; use crate::hii::HiiClient; +use crate::notify; use crate::rain::OpenMeteoClient; use crate::rid::{self, CycleOutcome, RidClient}; use crate::stations::Stations; use crate::util::{bangkok_now, floor_hour}; -use anyhow::Result; +use anyhow::{Result, bail}; use chrono::{Duration, NaiveDateTime}; +use ping_forecast::ModelStore; +use ping_forecast::rain::RainSeries; use serde_json::json; use sqlx::PgPool; @@ -21,6 +25,7 @@ pub struct Collector { hii: HiiClient, dams: DamsClient, rain: OpenMeteoClient, + http: reqwest::Client, } #[derive(Default)] @@ -33,6 +38,7 @@ struct Stats { feed_current: Option, latest_reading: Option, readings_saved: usize, + forecast: Option, } fn iso(t: Option) -> serde_json::Value { @@ -51,7 +57,8 @@ impl Collector { rid: RidClient::new(http.clone(), cfg.rid_api_url.clone()), hii: HiiClient::new(http.clone(), cfg.hii_api_base.clone()), dams: DamsClient::new(http.clone(), cfg.dams_url.clone()), - rain: OpenMeteoClient::new(http, cfg.openmeteo_url.clone()), + rain: OpenMeteoClient::new(http.clone(), cfg.openmeteo_url.clone()), + http, cfg, pool, stations, @@ -114,6 +121,13 @@ impl Collector { "latest_reading": iso(stats.latest_reading), "readings_saved": stats.readings_saved, "collector": concat!("ping-collector ", env!("CARGO_PKG_VERSION")), + "forecast": stats.forecast.as_ref().map(|f| json!({ + "last_run": iso(Some(f.at)), + "rows": f.rows, + "model_rows": f.model_rows, + "models": f.models, + "notifications_sent": f.notified, + })), }); let tmp = path.with_extension("json.tmp"); let result = @@ -143,6 +157,13 @@ impl Collector { let mut stats = Stats::default(); let mut last_hourly_slot: Option = None; + let mut forecaster = self.cfg.forecast_enabled.then(|| { + tracing::info!( + "FORECAST_IN_COLLECTOR: forecasting from {} and notifying here", + self.cfg.models_dir.display() + ); + Forecaster::new(&self.cfg, self.http.clone()) + }); loop { let start = bangkok_now(); self.write_status(&stats, true); @@ -166,6 +187,11 @@ impl Collector { last_hourly_slot = Some(slot); self.collect_hourly(bangkok_now()).await; } + if let Some(f) = forecaster.as_mut() + && let Some(info) = f.maybe_run(&self.pool, &self.rain, bangkok_now()).await + { + stats.forecast = Some(info); + } let now = bangkok_now(); let next = rid::next_poll_time( @@ -256,6 +282,117 @@ impl Collector { } } +impl Collector { + /// `ping-collector forecast`: compute from the database now (or at + /// `now`), print a table or the /forecast JSON, store only with `save`. + pub async fn forecast( + &self, + json_out: bool, + save: bool, + now: Option, + rain: Option, + ) -> Result<()> { + let mut store = ModelStore::new(&self.cfg.models_dir); + let models = store.refresh()?; + if models.is_empty() { + bail!( + "no exported models in {} (run `python -m src.ml.export`)", + self.cfg.models_dir.display() + ); + } + let now = now.unwrap_or_else(bangkok_now); + let snap = forecasting::compute(&self.pool, &self.rain, &models, now, rain).await?; + if json_out { + println!("{}", serde_json::to_string(&snap.forecasts)?); + } else { + println!( + "{:<6} {:>3} {:<9} {:<16} {:>7} {:>7} {:>9} {:>8}", + "code", "h", "source", "as_of", "level", "peak", "p_warning", "p_danger" + ); + for f in &snap.forecasts { + println!( + "{:<6} {:>3} {:<9} {:<16} {:>7.2} {:>7.2} {:>9.3} {:>8.3}", + f.station_code, + f.horizon_hours, + f.source, + f.as_of.format("%Y-%m-%d %H:%M"), + f.current_level, + f.predicted_max_level, + f.p_warning, + f.p_danger + ); + } + eprintln!( + "{} models from {}; {}", + models.len(), + self.cfg.models_dir.display(), + forecasting::summarize(&snap) + ); + } + if save { + db::ensure_schema(&self.pool).await?; + let n = db::save_forecast_history(&self.pool, &snap.forecasts).await?; + db::save_forecast_current(&self.pool, now, &snap.forecasts).await?; + eprintln!("stored: {n} forecast_history rows, forecast_current"); + } + Ok(()) + } + + /// `ping-collector notify`: what would be (or, without `dry_run`, is) + /// published right now against the stored notification state. + pub async fn notify(&self, dry_run: bool) -> Result<()> { + if !dry_run && !self.cfg.forecast_enabled { + bail!( + "the Python app notifies while FORECAST_IN_COLLECTOR is not true; \ + use --dry-run" + ); + } + let Some(notifier) = forecasting::notifier(&self.cfg, self.http.clone()) else { + bail!("ntfy is off: NTFY_SERVER is not set"); + }; + let mut store = ModelStore::new(&self.cfg.models_dir); + let models = store.refresh()?; + let now = bangkok_now(); + let snap = forecasting::compute(&self.pool, &self.rain, &models, now, None).await?; + let readings = + notify::readings_with_hii(&db::latest_rid_readings(&self.pool).await?, &snap.fills); + for r in &readings { + println!( + "{:<6} {} {:>7} {:?}{}", + r.station_code, + r.timestamp.format("%Y-%m-%d %H:%M"), + r.water_level.map_or("-".into(), |l| format!("{l:.2}")), + r.source, + r.discharge_percent + .map_or(String::new(), |p| format!(" {p:.0}%")) + ); + } + db::ensure_schema(&self.pool).await?; + if dry_run { + let states = db::notification_states(&self.pool).await?; + let plan = + notify::evaluate(&readings, &snap.forecasts, &states, ¬ifier.settings, now); + if plan.is_empty() { + println!("no transitions: nothing would be published"); + } + for p in plan { + println!( + "would publish {}: {} -- {}", + p.notification.topic, p.notification.title, p.notification.message + ); + } + } else { + for n in notifier + .run(&self.pool, &readings, &snap.forecasts, now) + .await? + { + println!("published {}: {}", n.topic, n.title); + } + } + Ok(()) + } +} + async fn shutdown_signal() { #[cfg(unix)] { diff --git a/rust/collector/testdata/gen_notify_cases.py b/rust/collector/testdata/gen_notify_cases.py new file mode 100644 index 0000000..7f8f3ba --- /dev/null +++ b/rust/collector/testdata/gen_notify_cases.py @@ -0,0 +1,127 @@ +"""Notification cases for rust/collector/src/notify.rs, from src/notify.py itself. + + uv run python rust/collector/testdata/gen_notify_cases.py + +Random hourly sequences around the thresholds (capacity guard, hysteresis, +danger escalation, the P.1 outlook, feed gaps, ntfy outages for some topics) +run through notify.evaluate with an in-memory state; every delivered +notification is recorded. The Rust test replays the same steps (RID readings +only, so the ThaiWater additions stay out of play) and must deliver exactly +the same messages. +""" + +import datetime +import gzip +import json +import random +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[2])) + +from src import notify # noqa: E402 + + +class Publisher(notify.NtfyPublisher): + def __init__(self): + super().__init__("http://ntfy.invalid", prefix="ping") + self.down = set() + + def publish(self, n): + return n.topic not in self.down + + +STATIONS = { # around (warning, danger) + "P.1": (3.7, 4.2), + "P.77": (4.3, 4.9), + "P.21": (3.2, 3.6), + "P.20": (2.35, 2.8), +} + + +def main(): + rng = random.Random(20261018) + sequences = [] + for _ in range(40): + state = notify.InMemoryState() + pub = Publisher() + t = datetime.datetime(2026, 9, 1) + datetime.timedelta( + hours=rng.randrange(2000) + ) + codes = rng.sample(sorted(STATIONS), rng.randint(1, 4)) + levels = {c: STATIONS[c][0] + rng.uniform(-0.6, 0.6) for c in codes} + p_out = rng.random() + steps = [] + for _ in range(rng.randint(8, 25)): + gap = rng.choice([1, 1, 1, 1, 2, 4, 6]) + t += datetime.timedelta(hours=gap) + # the clock sometimes runs ahead of the data (a stale feed) + now = t + datetime.timedelta(minutes=rng.choice([5, 5, 30, 200, 300])) + readings = [] + for c in codes: + levels[c] += rng.uniform(-0.25, 0.25) + pct = rng.choice([None, rng.uniform(5, 120)]) + level = round(levels[c], 2) if rng.random() > 0.05 else None + readings.append( + { + "station_code": c, + "timestamp": t, + "water_level": level, + "discharge_percent": None if pct is None else round(pct, 2), + } + ) + p_out = min(1.0, max(0.0, p_out + rng.uniform(-0.3, 0.3))) + forecasts = [ + { + "station_code": "P.1", + "horizon_hours": rng.choice([24, 24, 24, 12]), + "p_warning": p_out, + "predicted_max_level": round(rng.uniform(2.5, 5.0), 3), + "source": rng.choice(["model", "model", "model", "heuristic"]), + } + ] + pub.down = set( + rng.sample( + [ + "ping-p1-warning", + "ping-warning", + "ping-status", + "ping-p1-outlook", + ], + rng.choice([0, 0, 0, 1]), + ) + ) + sent = notify.evaluate(readings, forecasts, state, pub, now=now) + steps.append( + { + "now": now.isoformat(), + "readings": [ + {**r, "timestamp": r["timestamp"].isoformat()} for r in readings + ], + "forecasts": forecasts, + "down": sorted(pub.down), + "sent": [ + { + "topic": n.topic, + "title": n.title, + "message": n.message, + "priority": n.priority, + "tags": n.tags or [], + } + for n in sent + ], + } + ) + sequences.append(steps) + out = HERE / "notify_python_cases.json.gz" + with open(out, "wb") as raw, gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as gz: + gz.write(json.dumps({"sequences": sequences}).encode()) + total = sum(len(s["sent"]) for seq in sequences for s in seq) + print( + f"{len(sequences)} sequences, {sum(map(len, sequences))} steps, {total} notifications" + ) + + +if __name__ == "__main__": + main() diff --git a/rust/collector/testdata/notify_python_cases.json.gz b/rust/collector/testdata/notify_python_cases.json.gz new file mode 100644 index 0000000..d2cdf22 Binary files /dev/null and b/rust/collector/testdata/notify_python_cases.json.gz differ diff --git a/rust/forecast/src/hgb.rs b/rust/forecast/src/hgb.rs index 04374b3..2efb8e0 100644 --- a/rust/forecast/src/hgb.rs +++ b/rust/forecast/src/hgb.rs @@ -7,19 +7,23 @@ use anyhow::{Result, bail}; use serde::Deserialize; -fn thresholds<'de, D: serde::Deserializer<'de>>(d: D) -> Result, D::Error> { - let raw: Vec> = Deserialize::deserialize(d)?; - Ok(raw - .into_iter() - .map(|t| t.unwrap_or(f64::INFINITY)) - .collect()) +/// One node, packed (24 bytes): `x` is the split threshold of an internal +/// node and the output of a leaf. +#[derive(Debug, Clone, Copy)] +struct Node { + x: f64, + left: u32, + right: u32, + feature: u16, + leaf: bool, + missing_left: bool, } -#[derive(Debug, Deserialize)] -pub struct Tree { +/// The exported arrays, as in sklearn's predictor node table. +#[derive(Deserialize)] +struct RawTree { feature: Vec, - #[serde(deserialize_with = "thresholds")] - threshold: Vec, + threshold: Vec>, missing_left: Vec, left: Vec, right: Vec, @@ -27,44 +31,81 @@ pub struct Tree { value: Vec, } -impl Tree { - #[inline] - fn predict(&self, x: &[f64]) -> f64 { - let mut n = 0; - while !self.leaf[n] { - let v = x[self.feature[n]]; - let go_left = if v.is_nan() { - self.missing_left[n] - } else { - v <= self.threshold[n] - }; - n = if go_left { self.left[n] } else { self.right[n] }; - } - self.value[n] - } +#[derive(Debug)] +pub struct Tree { + nodes: Vec, +} - fn validate(&self, n_features: usize) -> Result<()> { - let n = self.value.len(); +impl<'de> Deserialize<'de> for Tree { + fn deserialize>(d: D) -> Result { + use serde::de::Error; + let r = RawTree::deserialize(d)?; + let n = r.value.len(); if n == 0 || [ - self.feature.len(), - self.threshold.len(), - self.missing_left.len(), - self.left.len(), - self.right.len(), - self.leaf.len(), + r.feature.len(), + r.threshold.len(), + r.missing_left.len(), + r.left.len(), + r.right.len(), + r.leaf.len(), ] .iter() .any(|&l| l != n) { - bail!("tree arrays have different lengths"); + return Err(D::Error::custom("tree arrays have different lengths")); } - for i in 0..n { - if !self.leaf[i] - && (self.feature[i] >= n_features || self.left[i] >= n || self.right[i] >= n) - { - bail!("tree node {i} points outside the tree or the feature vector"); - } + let nodes = (0..n) + .map(|i| { + let leaf = r.leaf[i]; + if !leaf && (r.left[i] >= n || r.right[i] >= n || r.feature[i] > u16::MAX as usize) + { + return Err(D::Error::custom(format!( + "tree node {i} points outside the tree" + ))); + } + Ok(Node { + // +inf thresholds ("missing vs everything") are exported as null + x: if leaf { + r.value[i] + } else { + r.threshold[i].unwrap_or(f64::INFINITY) + }, + left: r.left[i] as u32, + right: r.right[i] as u32, + feature: r.feature[i] as u16, + leaf, + missing_left: r.missing_left[i], + }) + }) + .collect::>()?; + Ok(Tree { nodes }) + } +} + +impl Tree { + #[inline] + fn predict(&self, x: &[f64]) -> f64 { + let mut node = &self.nodes[0]; + while !node.leaf { + let v = x[node.feature as usize]; + let go_left = if v.is_nan() { + node.missing_left + } else { + v <= node.x + }; + node = &self.nodes[if go_left { node.left } else { node.right } as usize]; + } + node.x + } + + fn validate(&self, n_features: usize) -> Result<()> { + if let Some(i) = self + .nodes + .iter() + .position(|n| !n.leaf && n.feature as usize >= n_features) + { + bail!("tree node {i} reads feature outside the feature vector"); } Ok(()) } diff --git a/rust/forecast/src/hii.rs b/rust/forecast/src/hii.rs index 22abb09..fcde817 100644 --- a/rust/forecast/src/hii.rs +++ b/rust/forecast/src/hii.rs @@ -55,6 +55,8 @@ pub struct FillInfo { pub station_code: String, pub offset: f64, pub filled_hours: usize, + /// The newest filled hour and its level. + pub latest: Option<(NaiveDateTime, f64)>, } /// pandas `Series.median` (numpy: mean of the two middle values when even). @@ -117,6 +119,11 @@ pub fn fill(rid: &[Row], hii: &[HiiRow]) -> (Vec, Vec) { station_code: code.to_string(), offset, filled_hours: fills.len() - before, + latest: fills + .iter() + .skip(before) + .max_by_key(|r| r.timestamp) + .map(|r| (r.timestamp, r.water_level)), }); } } diff --git a/scripts/forecast_parity.py b/scripts/forecast_parity.py new file mode 100644 index 0000000..115b725 --- /dev/null +++ b/scripts/forecast_parity.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Compare the Python forecast with ping-collector's on the live database. + +Run before setting FORECAST_IN_COLLECTOR=true, and after any change to src/ml +or rust/forecast. On the server, as the service user from the app dir: + + sudo -u water-monitor .venv/bin/python scripts/forecast_parity.py + +Both sides get the same moment and the same Open-Meteo rain series (fetched +once here and handed to the collector as a file), so a difference is a porting +bug, not timing. Exit 0 when every row matches: station, horizon, as_of, +source, model and every level exactly; probabilities within a few units in the +last place (numpy's and libm's exp may differ in the last bit). +""" + +import argparse +import datetime +import json +import math +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO)) + +ULPS = 4 +EXACT = ( + "station_code", + "horizon_hours", + "as_of", + "source", + "model_version", + "trained_at", + "current_level", + "predicted_max_level", + "threshold_warning", + "threshold_danger", +) + + +def ulps(a: float, b: float) -> float: + if a == b: + return 0 + return abs(a - b) / math.ulp(max(abs(a), abs(b))) + + +def compare(py: list, rs: list) -> list: + problems = [] + if len(py) != len(rs): + problems.append(f"row count: python {len(py)}, collector {len(rs)}") + for p, r in zip(py, rs): + where = f"{p.get('station_code')} {p.get('horizon_hours')}h" + for k in EXACT: + if p.get(k) != r.get(k): + problems.append( + f"{where} {k}: python {p.get(k)!r}, collector {r.get(k)!r}" + ) + for k in ("p_warning", "p_danger"): + d = ulps(float(p[k]), float(r[k])) + if d > ULPS: + problems.append( + f"{where} {k}: python {p[k]!r}, collector {r[k]!r} ({d:.0f} ulp)" + ) + ps, rs_ = p.get("stages"), r.get("stages") + if (ps is None) != (rs_ is None) or (ps and len(ps) != len(rs_)): + problems.append(f"{where} stages differ") + elif ps: + for a, b in zip(ps, rs_): + if a["stage"] != b["stage"] or a["level"] != b["level"]: + problems.append(f"{where} stage {a['stage']} differs") + elif ulps(a["p_exceed"], b["p_exceed"]) > ULPS: + problems.append( + f"{where} stage {a['stage']} p_exceed {a['p_exceed']} vs {b['p_exceed']}" + ) + return problems + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--collector", default=str(REPO / "bin" / "ping-collector")) + parser.add_argument("--models-dir", default=str(REPO / "models")) + parser.add_argument( + "--db-url", default=None, help="default: the app's configuration" + ) + parser.add_argument("--now", default=None, help="local time, default the clock") + args = parser.parse_args() + + from src.ml import predict + from src.ml import rain as rain_mod + from src.ml.export import json_path_for + + models = Path(args.models_dir) + stale = [ + p.name + for p in models.glob("flood_*.joblib") + if not json_path_for(p).exists() + or json_path_for(p).stat().st_mtime < p.stat().st_mtime + ] + if stale: + print( + f"not exported for the collector: {', '.join(stale)}; run `python -m src.ml.export`" + ) + return 2 + + now = ( + datetime.datetime.fromisoformat(args.now) + if args.now + else datetime.datetime.now() + ) + rain = rain_mod.serving_series() + if rain is None: + import pandas as pd + + rain = pd.Series(dtype=float) + py = predict.get_latest_forecasts( + db_url=args.db_url, models_dir=models, now=now, rain=rain + ) + py = json.loads(json.dumps(py, default=float)) + + with tempfile.TemporaryDirectory() as tmp: + rain_file = Path(tmp) / "rain.json" + rain_file.write_text( + json.dumps( + { + "timestamps": [t.strftime("%Y-%m-%dT%H:%M:%S") for t in rain.index], + "values": [None if math.isnan(v) else float(v) for v in rain], + } + ) + ) + env = dict(os.environ, MODELS_DIR=str(models), RUST_LOG="error") + if args.db_url: + env["POSTGRES_CONNECTION_STRING"] = args.db_url + out = subprocess.run( + [ + args.collector, + "forecast", + "--json", + "--now", + now.isoformat(), + "--rain-file", + str(rain_file), + ], + capture_output=True, + text=True, + env=env, + cwd=REPO, + check=False, + ) + if out.returncode != 0: + print(f"ping-collector forecast failed:\n{out.stderr}") + return 2 + rs = json.loads(out.stdout) + + problems = compare(py, rs) + models_used = sorted({r["model_version"] for r in py}) + n_model = sum(r["source"] == "model" for r in py) + print( + f"as of {now:%Y-%m-%d %H:%M:%S}: {len(py)} rows ({n_model} from models {models_used}), " + f"rain {int(rain.notna().sum())} h" + ) + if problems: + print(f"{len(problems)} DIFFERENCES:") + for p in problems[:40]: + print(" " + p) + return 1 + exact = sum(p[k] == r[k] for p, r in zip(py, rs) for k in ("p_warning", "p_danger")) + print( + f"identical: every level and field; probabilities {exact}/{2 * len(py)} bit-identical, " + f"the rest within {ULPS} ulp" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/install_collector.sh b/scripts/install_collector.sh index 0035fac..0c7bcf4 100755 --- a/scripts/install_collector.sh +++ b/scripts/install_collector.sh @@ -1,12 +1,18 @@ #!/usr/bin/env bash -# Install the Rust collector (rust/collector, `ping-collector`) and switch -# data collection over to it, or back. +# Install the Rust collector (rust/collector, `ping-collector`) and move work +# from the Python app to it, or back. # -# sudo bash scripts/install_collector.sh # install binary + unit, dry-run check -# sudo bash scripts/install_collector.sh --cutover # install, then Rust collects: -# # COLLECTION_ENABLED=false, start -# # ping-collector, restart water-monitor -# sudo bash scripts/install_collector.sh --rollback # Python collects again +# sudo bash scripts/install_collector.sh # install binary + unit; checks only +# sudo bash scripts/install_collector.sh --cutover # Rust collects (COLLECTION_ENABLED=false) +# sudo bash scripts/install_collector.sh --cutover-forecast # Rust also forecasts and sends the +# # ntfy notifications (FORECAST_IN_COLLECTOR=true) +# sudo bash scripts/install_collector.sh --rollback-forecast # forecasts + notifications back to Python +# sudo bash scripts/install_collector.sh --rollback # Python does everything again +# +# Every install also exports the trained models for the Rust forecaster and +# compares its forecast with Python's on the live data +# (scripts/forecast_parity.py); --cutover-forecast refuses to switch when they +# differ (FORCE=1 overrides). # # The binary comes from BIN=/path/to/ping-collector (e.g. the "ping-collector- # linux-x86_64" CI artifact) or from rust/target/release/ping-collector, built @@ -20,10 +26,12 @@ APP_DIR="${APP_DIR:-/opt/thailand-water-monitor}" SERVICE_USER="${SERVICE_USER:-water-monitor}" # as in scripts/ping-collector.service ENV_FILE="${APP_DIR}/.env" STATUS_FILE="${APP_DIR}/collector-status.json" +PYTHON="${APP_DIR}/.venv/bin/python" REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" MODE="${1:-install}" log() { printf '\033[1;32m==>\033[0m %s\n' "$*"; } +warn() { printf '\033[1;33m[warn]\033[0m %s\n' "$*" >&2; } die() { printf '\033[1;31m[error]\033[0m %s\n' "$*" >&2; exit 1; } [ "$(id -u)" -eq 0 ] || die "run as root (sudo)" @@ -38,15 +46,44 @@ set_env() { fi } -if [ "$MODE" = "--rollback" ]; then - log "Stopping ping-collector; the Python app collects again" - systemctl disable --now ping-collector.service 2>/dev/null || true - set_env COLLECTION_ENABLED true - systemctl restart water-monitor.service - log "Done. Check: journalctl -u water-monitor -n 50" - exit 0 -fi -case "$MODE" in install | --cutover) ;; *) die "unknown option: $MODE" ;; esac +# Run a command the way the services run: same user, working directory and +# EnvironmentFile parsing (systemd's, not a .env library's). +as_service() { + if systemd-run --version >/dev/null 2>&1; then + systemd-run --quiet --pipe --wait --collect \ + -p User="$SERVICE_USER" -p Group="$(id -gn "$SERVICE_USER")" \ + -p WorkingDirectory="$APP_DIR" -p EnvironmentFile="$ENV_FILE" "$@" + else + (cd "$APP_DIR" && runuser -u "$SERVICE_USER" -- "$@") + fi +} + +show() { + sleep "$1" + systemctl --no-pager --lines=0 status ping-collector.service || true + journalctl -u ping-collector -n 25 --no-pager || true +} + +case "$MODE" in + --rollback) + log "Stopping ping-collector; the Python app collects, forecasts and notifies again" + systemctl disable --now ping-collector.service 2>/dev/null || true + set_env FORECAST_IN_COLLECTOR false + set_env COLLECTION_ENABLED true + systemctl restart water-monitor.service + log "Done. Check: journalctl -u water-monitor -n 50" + exit 0 + ;; + --rollback-forecast) + log "Forecasts and notifications back to the Python app; ping-collector keeps collecting" + set_env FORECAST_IN_COLLECTOR false + systemctl restart ping-collector.service water-monitor.service + log "Done. Check: journalctl -u water-monitor -n 50" + exit 0 + ;; + install | --cutover | --cutover-forecast) ;; + *) die "unknown option: $MODE" ;; +esac BIN="${BIN:-${REPO_DIR}/rust/target/release/ping-collector}" [ -x "$BIN" ] || die "no binary at $BIN: build it (see header) or pass BIN=/path/to/ping-collector" @@ -63,26 +100,47 @@ touch "$STATUS_FILE" chown "${SERVICE_USER}:$(id -gn "$SERVICE_USER")" "$STATUS_FILE" log "Dry run against the configured database (writes nothing)" -# Run it the way the service will: same user, working directory and -# EnvironmentFile parsing (systemd's, not a .env library's). -if systemd-run --version >/dev/null 2>&1; then - systemd-run --quiet --pipe --wait --collect \ - -p User="$SERVICE_USER" -p Group="$(id -gn "$SERVICE_USER")" \ - -p WorkingDirectory="$APP_DIR" -p EnvironmentFile="$ENV_FILE" \ - "${APP_DIR}/bin/ping-collector" once --dry-run +as_service "${APP_DIR}/bin/ping-collector" once --dry-run + +forecast_ok=1 +if compgen -G "${APP_DIR}/models/flood_*.joblib" >/dev/null; then + log "Exporting the trained models for the Rust forecaster (models/flood_*.json.gz)" + as_service "$PYTHON" -m src.ml.export + log "Forecast check: Python and ping-collector on the same data, moment and rain" + as_service "$PYTHON" scripts/forecast_parity.py || forecast_ok=0 else - (cd "$APP_DIR" && "${APP_DIR}/bin/ping-collector" --env-file "$ENV_FILE" once --dry-run) + warn "no trained models in ${APP_DIR}/models: forecast check skipped" + forecast_ok=0 fi -if [ "$MODE" = "--cutover" ]; then - log "Cutover: ping-collector collects, the Python app forecasts and notifies" - set_env COLLECTION_ENABLED false - systemctl enable --now ping-collector.service - systemctl restart water-monitor.service - sleep 20 - systemctl --no-pager --lines=0 status ping-collector.service || true - journalctl -u ping-collector -n 20 --no-pager || true - log "Rollback if needed: sudo bash scripts/install_collector.sh --rollback" -else - log "Installed, not started. Switch over with: sudo bash scripts/install_collector.sh --cutover" -fi +case "$MODE" in + --cutover) + log "Cutover: ping-collector collects; the Python app forecasts and notifies" + set_env COLLECTION_ENABLED false + systemctl enable ping-collector.service + systemctl restart ping-collector.service water-monitor.service + show 20 + log "Next step: --cutover-forecast. Rollback: sudo bash scripts/install_collector.sh --rollback" + ;; + --cutover-forecast) + grep -q '^COLLECTION_ENABLED=false' "$ENV_FILE" || + die "ping-collector must collect first: run --cutover" + if [ "$forecast_ok" != 1 ] && [ "${FORCE:-0}" != 1 ]; then + die "the forecast check did not pass; not switching (FORCE=1 overrides)" + fi + log "Cutover: ping-collector also forecasts and notifies; the Python app serves its forecast" + set_env FORECAST_IN_COLLECTOR true + systemctl enable ping-collector.service + systemctl restart ping-collector.service water-monitor.service + show 30 + log "Rollback if needed: sudo bash scripts/install_collector.sh --rollback-forecast" + ;; + *) + if [ "$(systemctl is-active ping-collector.service 2>/dev/null)" = active ]; then + systemctl restart ping-collector.service + log "Installed; ping-collector restarted on the new binary" + else + log "Installed, not started. Switch over with: sudo bash scripts/install_collector.sh --cutover" + fi + ;; +esac diff --git a/scripts/ping-collector.service b/scripts/ping-collector.service index 6c79fd4..fe70eeb 100644 --- a/scripts/ping-collector.service +++ b/scripts/ping-collector.service @@ -1,5 +1,5 @@ [Unit] -Description=Ping River Monitor data collector (Rust: RID, HII, dams, Open-Meteo) +Description=Ping River Monitor collector (Rust: RID, HII, dams, Open-Meteo; forecast + ntfy) Documentation=https://git.b4l.co.th/B4L/Northern-Thailand-Ping-River-Monitor/src/branch/master/rust After=network-online.target Wants=network-online.target @@ -15,7 +15,9 @@ RestartSec=30 TimeoutStopSec=20 # Same .env as water-monitor.service (POSTGRES_*, RID_*, GAP_FILL_*, -# SCRAPING_STATUS_FILE). The Python app must run with COLLECTION_ENABLED=false. +# SCRAPING_STATUS_FILE, FORECAST_IN_COLLECTOR, NTFY_*). The Python app must run +# with COLLECTION_ENABLED=false. Models are read from models/ (relative to +# WorkingDirectory). EnvironmentFile=/opt/thailand-water-monitor/.env # Security settings diff --git a/src/config.py b/src/config.py index e1191ad..2bdeb71 100644 --- a/src/config.py +++ b/src/config.py @@ -156,6 +156,17 @@ class Config: "yes", ) FORECAST_POLL_SECONDS = int(os.getenv("FORECAST_POLL_SECONDS", "60")) + # true: ping-collector also forecasts (rust/forecast) and sends the ntfy + # notifications; this app serves its stored forecast (forecast_current) + # and does neither. Read by both services from the shared .env. + FORECAST_IN_COLLECTOR = os.getenv("FORECAST_IN_COLLECTOR", "false").lower() in ( + "1", + "true", + "yes", + ) + # A stored forecast older than this is not served; /forecast then computes + # in-process as before (the collector is down or has no models). + FORECAST_MAX_AGE_SECONDS = int(os.getenv("FORECAST_MAX_AGE_SECONDS", "7200")) # Logging settings LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") @@ -306,6 +317,7 @@ class Config: "RID_RETRY_MINUTES": cls.RID_RETRY_MINUTES, "GAP_FILL_DAYS_ON_START": cls.GAP_FILL_DAYS_ON_START, "COLLECTION_ENABLED": cls.COLLECTION_ENABLED, + "FORECAST_IN_COLLECTOR": cls.FORECAST_IN_COLLECTOR, "LOG_LEVEL": cls.LOG_LEVEL, "LOG_FILE": cls.LOG_FILE, "DATA_RETENTION_DAYS": cls.DATA_RETENTION_DAYS, diff --git a/src/web_api.py b/src/web_api.py index f65a03e..6cfdb89 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -217,10 +217,16 @@ async def lifespan(app: FastAPI): ) app_state["notify"] = None if app_state["leader_lock"]: - app_state["notify"] = _init_notifications() + if not Config.FORECAST_IN_COLLECTOR: + app_state["notify"] = _init_notifications() if Config.COLLECTION_ENABLED: app_state["scraping_task"] = asyncio.create_task(background_scraping_task()) logger.info("This worker is the background-collection leader") + elif Config.FORECAST_IN_COLLECTOR: + logger.info( + "COLLECTION_ENABLED=false, FORECAST_IN_COLLECTOR=true: ping-collector " + "collects, forecasts and notifies; this app serves its results" + ) else: app_state["scraping_task"] = asyncio.create_task(forecast_notify_task()) logger.info( @@ -353,7 +359,7 @@ def _init_notifications(): async def _notify_transitions(): """Publish flood/outlook/feed transitions to ntfy (leader only, fail-safe).""" cfg = app_state.get("notify") - if not cfg: + if not cfg or Config.FORECAST_IN_COLLECTOR: return publisher, state = cfg try: @@ -379,6 +385,8 @@ async def _notify_transitions(): async def _precompute_forecasts(): """Refresh the forecast cache and persist the issued forecasts (leader only).""" + if Config.FORECAST_IN_COLLECTOR: + return # ping-collector forecasts and persists try: from .ml.predict import get_latest_forecasts except ImportError: @@ -1400,10 +1408,57 @@ async def get_postgres_history( raise HTTPException(status_code=502, detail="Measurement history unavailable") +STORED_FORECAST_TTL = 60 # seconds; the collector refreshes at most every few minutes + + +def _stored_forecasts(): + """The collector's latest forecast (forecast_current), or None when it is + missing or older than FORECAST_MAX_AGE_SECONDS.""" + from sqlalchemy import text + + engine = getattr(app_state["scraper"].db_adapter, "engine", None) + if engine is None: + return None + try: + with engine.connect() as conn: + row = conn.execute( + text( + "SELECT computed_at, payload FROM forecast_current WHERE id = 'all'" + ) + ).fetchone() + except Exception as error: + logger.warning(f"stored forecast unavailable: {error}") + return None + if row is None: + return None + computed_at, payload = row[0], row[1] + if isinstance(computed_at, str): + computed_at = datetime.fromisoformat(computed_at) + age = (datetime.now() - computed_at).total_seconds() + if age > Config.FORECAST_MAX_AGE_SECONDS: + logger.warning( + f"stored forecast is {age / 3600:.1f} h old (ping-collector down?); " + "computing in-process" + ) + return None + return json.loads(payload) + + @app.get("/forecast") async def get_flood_forecasts(): """Flood-risk forecasts per station for the 6/12/24 h horizons.""" increment_counter("api_requests", labels={"endpoint": "forecast"}) + if Config.FORECAST_IN_COLLECTOR: + cached = _cache_fresh( + FORECAST_CACHE, FORECAST_CACHE_LOCK, "stored", STORED_FORECAST_TTL + ) + if cached is not None: + return cached + stored = await asyncio.to_thread(_stored_forecasts) + if stored: + with FORECAST_CACHE_LOCK: + FORECAST_CACHE["stored"] = (time.monotonic(), stored) + return stored now = time.monotonic() with FORECAST_CACHE_LOCK: cached = FORECAST_CACHE.get("all") diff --git a/tests/test_forecast_in_collector.py b/tests/test_forecast_in_collector.py new file mode 100644 index 0000000..97543b8 --- /dev/null +++ b/tests/test_forecast_in_collector.py @@ -0,0 +1,107 @@ +"""FORECAST_IN_COLLECTOR=true: ping-collector forecasts and notifies; the API +serves its stored forecast and falls back to computing when that is stale.""" + +import asyncio +import datetime +import json +import types + +import pytest +from sqlalchemy import create_engine, text + +from src import web_api +from src.config import Config + +ROWS = [ + { + "station_code": "P.1", + "horizon_hours": 24, + "p_warning": 0.012, + "p_danger": 0.001, + "predicted_max_level": 1.31, + "current_level": 1.23, + "as_of": "2026-10-18T13:00:00", + "model_version": "hgb-v3+abc", + "trained_at": "2026-10-01T03:42:00", + "source": "model", + "threshold_warning": 3.7, + "threshold_danger": 4.2, + } +] + + +@pytest.fixture +def collector_db(tmp_path, monkeypatch): + engine = create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") + with engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE forecast_current (id VARCHAR(16) PRIMARY KEY, " + "computed_at TIMESTAMP NOT NULL, model_versions TEXT, payload TEXT NOT NULL)" + ) + ) + scraper = types.SimpleNamespace(db_adapter=types.SimpleNamespace(engine=engine)) + monkeypatch.setitem(web_api.app_state, "scraper", scraper) + monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) + monkeypatch.setattr(web_api, "FORECAST_CACHE", {}) + + def store(age_minutes): + computed = datetime.datetime.now() - datetime.timedelta(minutes=age_minutes) + with engine.begin() as conn: + conn.execute(text("DELETE FROM forecast_current")) + conn.execute( + text( + "INSERT INTO forecast_current VALUES ('all', :t, 'hgb-v3+abc', :p)" + ), + {"t": computed, "p": json.dumps(ROWS)}, + ) + + return store + + +def test_serves_the_collectors_forecast(collector_db, monkeypatch): + collector_db(age_minutes=20) + + def must_not_compute(): + raise AssertionError("computed in-process although a fresh forecast is stored") + + monkeypatch.setattr("src.ml.predict.get_latest_forecasts", must_not_compute) + assert asyncio.run(web_api.get_flood_forecasts()) == ROWS + # served from the short cache on the next request + assert "stored" in web_api.FORECAST_CACHE + + +def test_stale_stored_forecast_falls_back_to_computing(collector_db, monkeypatch): + collector_db(age_minutes=Config.FORECAST_MAX_AGE_SECONDS / 60 + 5) + computed = [dict(ROWS[0], source="heuristic")] + monkeypatch.setattr("src.ml.predict.get_latest_forecasts", lambda: computed) + assert asyncio.run(web_api.get_flood_forecasts()) == computed + + +def test_missing_table_falls_back_to_computing(tmp_path, monkeypatch): + engine = create_engine(f"sqlite:///{tmp_path / 'empty.sqlite'}") + scraper = types.SimpleNamespace(db_adapter=types.SimpleNamespace(engine=engine)) + monkeypatch.setitem(web_api.app_state, "scraper", scraper) + monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) + monkeypatch.setattr(web_api, "FORECAST_CACHE", {}) + monkeypatch.setattr("src.ml.predict.get_latest_forecasts", lambda: ROWS) + assert asyncio.run(web_api.get_flood_forecasts()) == ROWS + + +def test_this_app_neither_forecasts_nor_notifies(monkeypatch): + published = [] + + class Publisher: + def publish(self, n): + published.append(n) + return True + + monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) + monkeypatch.setitem(web_api.app_state, "notify", (Publisher(), object())) + monkeypatch.setattr( + "src.ml.predict.get_latest_forecasts", + lambda: pytest.fail("forecast precompute ran in the API"), + ) + asyncio.run(web_api._precompute_forecasts()) + asyncio.run(web_api._notify_transitions()) + assert published == []