diff --git a/.env.example b/.env.example index a8a0b61..179008d 100644 --- a/.env.example +++ b/.env.example @@ -53,6 +53,9 @@ NTFY_TOKEN= PUBLIC_URL=https://water.buildfor.life/ # Report RID lagging this many hours while ThaiWater telemetry is current (0 = never) RID_STALE_ALERT_HOURS=18 +# RID publishes nothing overnight: from 21:00 until this hour a quiet river's +# feed is not reported stale (0 = off; docs/NOTIFICATIONS.md) +FEED_NIGHT_UNTIL_HOUR=9 # Matrix room for a team: the same transitions (docs/MATRIX_QUICK_START.md); # off while token or room is empty MATRIX_HOMESERVER=https://matrix.org diff --git a/docs/NOTIFICATIONS.md b/docs/NOTIFICATIONS.md index e423183..3a3c51f 100644 --- a/docs/NOTIFICATIONS.md +++ b/docs/NOTIFICATIONS.md @@ -43,11 +43,22 @@ keys (checked word for word on 751 generated messages), so the switch sent nothing twice. One difference: it 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 `MATRIX_*` set, the +message then says "ThaiWater telemetry; the RID report is not in yet". The feed +counts as stale when no source has a reading for 3 h, and RID alone lagging is +reported once after `RID_STALE_ALERT_HOURS` (18, 0 = off). + +**RID's nightly pause.** RID publishes no hourly report overnight: on +2026-10-17, -18 and -19 the 21:00 reading was the last one until the hours +22:00-06:00 arrived together between 06:35 and 07:22. ThaiWater's copies of the +RID gauges come from the same feed and arrive with the same delay (P.1's +mirror had no value for 22:00-00:00 even the next morning), so they do not +bridge the night. Without a rule for it, every night sent "gauge feed stale" +around midnight and "feed recovered" around 07:00. So between 21:00 and +`FEED_NIGHT_UNTIL_HOUR` (9, 0 = off) a quiet river's feed is not reported +stale; a feed still without new readings at 09:00 is. While any station is at +warning or danger, or the P.1 outlook is on, the 3 h rule holds all night: +then a stalled dashboard matters. A feed that was already stale stays stale +through the night, and a recovery is announced whenever it happens. With `MATRIX_*` set, the same transitions also go to a Matrix room, each once (per station, outlook, feed; not the per-topic summaries), with their own `matrix:` state keys. @@ -120,6 +131,7 @@ holds back an alert. Test the pipeline before the domain is live with | `PUBLIC_URL` | `https://water.buildfor.life/` | click-through target in messages | | `RID_STALE_ALERT_HOURS` | `18` | report RID lagging this long while ThaiWater is current (0 = never) | +| `FEED_NIGHT_UNTIL_HOUR` | `9` | end of the night window from 21:00 in which a quiet river's feed is not reported stale (0 = off) | | `MATRIX_HOMESERVER` / `MATRIX_ACCESS_TOKEN` / `MATRIX_ROOM_ID` | *(empty = off)* | the team room (docs/MATRIX_QUICK_START.md) | Tunables in `src/ml/export.py` (compiled into ping-collector through diff --git a/rust/README.md b/rust/README.md index 916892f..322eead 100644 --- a/rust/README.md +++ b/rust/README.md @@ -62,9 +62,10 @@ readings and forecasts with `notification_state`. fails while it differs from Python; a change needs a rebuild. - **ntfy** topics, texts and hysteresis: [docs/NOTIFICATIONS.md](../docs/NOTIFICATIONS.md). For the stations whose ThaiWater twin is the same sensor, a reading newer - than RID's (RID lags 2-10 h at night) drives the level alerts, marked as - ThaiWater telemetry. "Gauge feed stale" means no source has a reading for - 3 h; RID alone lagging is reported once after `RID_STALE_ALERT_HOURS`. + than RID's drives the level alerts, marked as ThaiWater telemetry. "Gauge + feed stale" means no source has a reading for 3 h, except during RID's + nightly pause (21:00 to `FEED_NIGHT_UNTIL_HOUR`, 9) while the river is + quiet; RID alone lagging is reported once after `RID_STALE_ALERT_HOURS`. - **Matrix** (`MATRIX_*`): the room gets the same transitions (per station, outlook, feed; not ntfy's per-topic summaries), each once, tracked under its own `matrix:` keys in `notification_state`, so a Matrix outage is diff --git a/rust/collector/src/config.rs b/rust/collector/src/config.rs index f28df8e..71b64ba 100644 --- a/rust/collector/src/config.rs +++ b/rust/collector/src/config.rs @@ -50,6 +50,9 @@ pub struct Config { pub public_url: String, /// Report RID lagging this long while ThaiWater is current (0 = never). pub rid_stale_alert_hours: f64, + /// FEED_NIGHT_UNTIL_HOUR: from 21:00 until this hour a quiet river's feed + /// is not reported stale (RID publishes nothing overnight); 0 = off. + pub feed_night_until_hour: u32, /// Public ntfy URL (what subscribers use), shown by /api/notifications. pub ntfy_server: Option, /// MATRIX_*: the notifications also go to this room (off when unset). @@ -211,6 +214,10 @@ impl Config { .var("PUBLIC_URL") .unwrap_or_else(|| "https://water.buildfor.life/".into()), rid_stale_alert_hours: v.number("RID_STALE_ALERT_HOURS", 18.0)?, + feed_night_until_hour: match v.number("FEED_NIGHT_UNTIL_HOUR", 9u32)? { + h if h < 21 => h, + h => bail!("FEED_NIGHT_UNTIL_HOUR must be 0 (off) to 20, got {h}"), + }, ntfy_server: v.var("NTFY_SERVER"), matrix: match (v.var("MATRIX_ACCESS_TOKEN"), v.var("MATRIX_ROOM_ID")) { (Some(access_token), Some(room_id)) => Some(MatrixConfig { diff --git a/rust/collector/src/forecasting.rs b/rust/collector/src/forecasting.rs index 609fd7a..7940870 100644 --- a/rust/collector/src/forecasting.rs +++ b/rust/collector/src/forecasting.rs @@ -151,6 +151,7 @@ pub fn notifier(cfg: &Config, http: reqwest::Client) -> Option { settings: Settings { prefix: cfg.ntfy_topic_prefix.clone(), rid_stale_alert_hours: cfg.rid_stale_alert_hours, + feed_night_until_hour: cfg.feed_night_until_hour, }, }) } diff --git a/rust/collector/src/notify.rs b/rust/collector/src/notify.rs index 1b80476..7c631e9 100644 --- a/rust/collector/src/notify.rs +++ b/rust/collector/src/notify.rs @@ -7,16 +7,16 @@ //! //! Same topics, texts, hysteresis, capacity guard, outlook and persisted //! state keys (`notification_state`) as Python had, so the cutover neither -//! repeated nor dropped 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. +//! repeated nor dropped a message. Changes since: for stations whose HII/ThaiWater +//! twin is the same sensor, a reading newer than RID's is used for the level +//! alerts and the feed check; RID lagging alone is reported after +//! RID_STALE_ALERT_HOURS; and RID's nightly pause is not reported (see +//! [`NIGHT_FROM_HOUR`]). use crate::config::MatrixConfig; use crate::db::{self, LatestReading}; use anyhow::Result; -use chrono::{Duration, NaiveDateTime}; +use chrono::{Duration, NaiveDateTime, Timelike}; use ping_forecast::Forecast; use ping_forecast::config::config; use ping_forecast::hii::FillInfo; @@ -123,10 +123,28 @@ fn level_state(level: f64, warn: f64, danger: f64, prev: &str) -> &'static str { "clear" } +/// RID publishes no hourly report overnight: on 2026-10-17, -18 and -19 the +/// 21:00 reading was the last until the 22:00..06:00 hours arrived together +/// between 06:35 and 07:22 (ThaiWater's copies of the RID gauges arrive with +/// the same delay, so they cannot bridge it). Between this hour and +/// `Settings::feed_night_until_hour` a quiet river's feed is therefore not +/// reported stale; a feed still stale at that hour is. While any station is at +/// warning or danger, or the P.1 outlook is on, the 3 h rule holds all night. +pub const NIGHT_FROM_HOUR: u32 = 21; + pub struct Settings { pub prefix: String, /// 0 = off pub rid_stale_alert_hours: f64, + /// End of the night window (hour of day); 0 = no night window. + pub feed_night_until_hour: u32, +} + +impl Settings { + fn in_night_window(&self, now: NaiveDateTime) -> bool { + let h = now.hour(); + self.feed_night_until_hour > 0 && (h >= NIGHT_FROM_HOUR || h < self.feed_night_until_hour) + } } impl Settings { @@ -176,6 +194,8 @@ pub fn evaluate( let (mut b_danger, mut b_warning, mut b_clear) = (vec![], vec![], vec![]); let mut latest: Option = None; let mut latest_rid: Option = None; + // any station at warning/danger or the outlook on: the night window is off + let mut river_high = false; for r in readings { latest_rid = latest_rid.max(r.rid_timestamp); @@ -201,6 +221,7 @@ pub fn evaluate( ); continue; } + river_high |= cur != "clear"; if cur == prev { continue; } @@ -302,6 +323,7 @@ pub fn evaluate( } // model outlook for the city gauge (opt-in topic, experimental) + river_high |= state("outlook:P.1") == Some("on"); if let Some(p1) = forecasts.iter().find(|f| { f.station_code == "P.1" && f.horizon_hours == nc.outlook_horizon && f.source == "model" }) { @@ -313,6 +335,7 @@ pub fn evaluate( } else { "off" }; + river_high |= cur == "on"; if cur != prev { let (warn, _) = cfg.thresholds("P.1"); let notification = if cur == "on" { @@ -348,13 +371,22 @@ pub fn evaluate( } } - // feed health: stale only when no source has a recent reading + // feed health: stale only when no source has a recent reading; during + // RID's nightly pause a quiet river keeps its previous state (a recovery + // is still announced, a new stale message waits for the morning) 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" }; + let deferred = !river_high && settings.in_night_window(now); + let cur = if age_h < nc.feed_stale_after_h { + "ok" + } else if deferred { + prev + } else { + "stale" + }; + feed_stale = cur == "stale"; if cur != prev { let notification = if feed_stale { n( @@ -743,6 +775,7 @@ mod tests { Settings { prefix: "ping".into(), rid_stale_alert_hours: 18.0, + feed_night_until_hour: 9, } } @@ -996,6 +1029,81 @@ mod tests { ); } + fn at(day: u32, hour: u32, min: u32) -> NaiveDateTime { + NaiveDate::from_ymd_opt(2026, 10, day) + .unwrap() + .and_hms_opt(hour, min, 0) + .unwrap() + } + + fn reading_at(code: &str, level: f64, t: NaiveDateTime) -> Reading { + Reading { + timestamp: t, + rid_timestamp: Some(t), + ..reading(code, level, 0) + } + } + + #[test] + fn rids_nightly_pause_is_not_reported_on_a_quiet_river() { + // 2026-10-19: last reading 21:00, nothing until the 06:00 batch at 06:55 + let mut st = HashMap::new(); + let evening = [reading_at("P.1", 1.26, at(18, 21, 0))]; + for t in [at(18, 23, 5), at(19, 0, 5), at(19, 3, 5), at(19, 6, 5)] { + assert!(step(&evening, &[], &mut st, t).is_empty(), "at {t}"); + } + let morning = [reading_at("P.1", 1.33, at(19, 6, 0))]; + assert!(step(&morning, &[], &mut st, at(19, 6, 55)).is_empty()); + // still nothing new at 09:00: reported then + let sent = step(&evening, &[], &mut st, at(19, 9, 5)); + assert_eq!( + titles(&sent), + [("ping-status", "Ping River monitor: gauge feed stale")] + ); + assert!( + sent[0] + .message + .starts_with("No new readings for 12 h (last 18 Oct 21:00)") + ); + // the window can be turned off + let off = Settings { + feed_night_until_hour: 0, + ..settings() + }; + let plan = evaluate(&evening, &[], &HashMap::new(), &off, at(19, 0, 5)); + assert_eq!( + plan[0].notification.title, + "Ping River monitor: gauge feed stale" + ); + } + + #[test] + fn night_window_keeps_a_stale_feed_stale_and_announces_recovery() { + let mut st = HashMap::new(); + let old = [reading_at("P.1", 1.2, at(18, 14, 0))]; + assert_eq!(step(&old, &[], &mut st, at(18, 18, 5)).len(), 1); // stale, daytime + assert!(step(&old, &[], &mut st, at(18, 22, 5)).is_empty()); // not "recovered" + let back = [reading_at("P.1", 1.2, at(18, 22, 0))]; + assert_eq!( + step(&back, &[], &mut st, at(18, 23, 5))[0].title, + "Ping River monitor: feed recovered" + ); + } + + #[test] + fn the_three_hour_rule_holds_at_night_while_the_river_is_high() { + let mut st = HashMap::from([("level:P.1".to_string(), "warning".to_string())]); + let high = [reading_at("P.1", 3.8, at(18, 21, 0))]; + assert_eq!( + titles(&step(&high, &[], &mut st, at(19, 0, 5))), + [("ping-status", "Ping River monitor: gauge feed stale")] + ); + // the outlook being on counts too + let mut st = HashMap::from([("outlook:P.1".to_string(), "on".to_string())]); + let quiet = [reading_at("P.1", 1.3, at(18, 21, 0))]; + assert_eq!(step(&quiet, &[], &mut st, at(19, 0, 5)).len(), 1); + } + #[test] fn undelivered_transitions_are_retried() { let mut st = HashMap::new(); @@ -1067,13 +1175,12 @@ mod tests { .map(|v| v.as_str().unwrap()) .collect(); let mut delivered = vec![]; - for p in evaluate( - &readings, - &forecasts, - &states, - &settings(), - ts(&step["now"]), - ) { + // the Python notifier had no night window + let python = Settings { + feed_night_until_hour: 0, + ..settings() + }; + for p in evaluate(&readings, &forecasts, &states, &python, ts(&step["now"])) { if down.contains(&p.notification.topic.as_str()) { continue; } diff --git a/rust/collector/src/web/system.rs b/rust/collector/src/web/system.rs index 92ef7fc..5c880a0 100644 --- a/rust/collector/src/web/system.rs +++ b/rust/collector/src/web/system.rs @@ -58,8 +58,8 @@ fn check(status: &'static str, message: impl Into) -> Check { /// Data freshness. As the Python check it judges the newest reading of the /// first station (`get_latest_measurements(limit=1)`: P.1) against 2 h. One -/// change, as in the notifier: while RID's report lags (2-10 h at night) but -/// the ThaiWater telemetry of the same gauges is current, the data is current. +/// change, as in the notifier: while RID's report lags but the ThaiWater +/// telemetry of the same gauges is current, the data is current. async fn database_check(st: &Shared) -> Check { let latest: Result, _> = sqlx::query_as( "SELECT m.timestamp diff --git a/rust/forecast/src/hii.rs b/rust/forecast/src/hii.rs index fcde817..7894ebd 100644 --- a/rust/forecast/src/hii.rs +++ b/rust/forecast/src/hii.rs @@ -2,8 +2,9 @@ //! //! For stations whose HII twin is the same sensor, hours missing from the RID //! series (or carrying no level) are filled from HII, shifted by the median -//! offset between the two series over their overlap. This is also what keeps -//! the forecast current at night while RID's report lags. +//! offset between the two series over their overlap. ThaiWater relays RID's +//! own telemetry, so this repairs holes in RID's report but cannot bridge +//! RID's overnight pause (the mirror arrives with the same delay). use crate::config::config; use crate::grid::{Row, floor_hour};