From 9bf904b23c4db8ca3405fef5f6c521288b2d98cb Mon Sep 17 00:00:00 2001 From: grabowski Date: Mon, 26 Oct 2026 15:04:01 +0700 Subject: [PATCH] feat(rid): screen one-hour spikes; refetch and redo-from to repair the store RID's report sometimes carries a wrong value for one hour and corrects it later. P.1 read 10.6 m at 2026-10-25 00:00 between 1.04 and 1.07 (RID now says 1.06); the store kept it, because a report is re-fetched only while it has holes, and the model issued a 10.8 m forecast for that hour. - rid::judge / rid::screen: a reading that leaves the previous one (within 3 h) by more than 1 m (2 m on the flashy tributaries P.76, P.77, P.81, P.82, P.84, P.85, P.87) while the next reading comes back near the previous level is a spike: not stored, a stored copy is deleted. Limbs and steps the next reading confirms are kept; the newest reading waits for the next one when it jumps more than twice the limit. On 2018-2026 history: 16 spikes found (slipped decimals, sign flips, -5/-6/-9 m placeholders), no genuine change rejected. - Every store goes through it (rid::store: run_cycle, fill-gaps, refetch). - `ping-collector refetch FROM [TO]`: re-fetch reports regardless of holes (corrects revised readings, fills the discharge of level-only evening hours, removes stored spikes) and print the catch-up command when readings changed. - `catch-up-forecasts --redo-from T`: delete the model forecasts issued from T on, then recompute them. - Tests: judge/screen unit tests; a DB test of refetch, the live hold and the forecast deletion. Checked end to end against RID on a scratch DB. - Docs: DATA_SOURCES (spikes, level-only hours), FLOOD_FORECASTING, rust/README. --- docs/DATA_SOURCES.md | 21 +- docs/FLOOD_FORECASTING.md | 5 +- rust/README.md | 9 +- rust/collector/src/db.rs | 56 +++++ rust/collector/src/db_tests.rs | 141 ++++++++++++ rust/collector/src/main.rs | 51 ++++- rust/collector/src/rid.rs | 396 ++++++++++++++++++++++++++++++++- rust/collector/src/service.rs | 20 +- 8 files changed, 684 insertions(+), 15 deletions(-) diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md index 794f750..be91916 100644 --- a/docs/DATA_SOURCES.md +++ b/docs/DATA_SOURCES.md @@ -23,10 +23,29 @@ The only source persisted to the database and used by the ML pipeline. | Params | `DW[UtokID]=1`, `DW[BasinID]=6` (Ping), `DW[TimeCurrent]=`, `rows=100` | | Variables | Water level (m, gauge datum), discharge (m³/s, `'***'` = malformed), discharge % of channel capacity | | Stations | 16 P-series gauges (P.1 anchor at Nawarat Bridge; see `src/data/stations.json`) | -| Client | `rust/collector/src/rid.rs` (`ping-collector`; `fill-gaps` re-fetches old reports) | +| Client | `rust/collector/src/rid.rs` (`ping-collector`; `fill-gaps` re-fetches old reports with holes, `refetch FROM [TO]` any reports) | Human-facing page: +**Spikes.** The report sometimes carries a wrong value for one hour and corrects +it later: a slipped decimal point (P.1 10.6 m between 1.04 and 1.07 on 2026-10-25 +00:00, later 1.06; P.85 0.101 for 1.1), a flipped sign (P.4A ±1.18, P.67 ±0.6) or +a placeholder (−5, −6, −9, 50 m). The collector re-fetches a report only while it +has holes, so a later correction never arrived. Since 2026-10-26 every batch is +screened before it is stored (`rid::judge`, `rid::screen`): a reading that leaves +the previous one (within 3 h) by more than 1 m (2 m on the flashy tributaries +P.76, P.77, P.81, P.82, P.84, P.85, P.87) while the next reading comes back near +the previous level is not stored, and a stored copy is deleted. A rising or falling +limb, or a step the next reading confirms, is kept. The newest reading of a gauge +waits for the next one when it jumps more than twice the limit. On the 2018–2026 +history the rule finds 16 spikes and would have rejected no genuine change (the +largest genuine 1-h changes: 1.1 m on the mainstem, 3.1 m at P.76). Two wrong hours +in a row are not caught. + +Evening hours often arrive level-only (no discharge) and gain the discharge +later; the API's history fills a missing discharge with 20 × level (an old +placeholder, kept for compatibility). `refetch` brings in RID's discharge. + --- ## 2. ThaiWater / HII ecosystem diff --git a/docs/FLOOD_FORECASTING.md b/docs/FLOOD_FORECASTING.md index 504e8e0..1517531 100644 --- a/docs/FLOOD_FORECASTING.md +++ b/docs/FLOOD_FORECASTING.md @@ -643,7 +643,10 @@ Open-Meteo's estimate of rain that has since fallen, as in training, where the forward rain is observed rain. Such rows are written only where no model row exists; `created_at` tells them apart (more than an hour after `as_of`). `ping-collector catch-up-forecasts --hours N` does the same further back (up -to 60 days). +to 60 days); with `--redo-from T` it first deletes the model forecasts issued +from T on, to recompute them after `ping-collector refetch` corrected the +readings they came from (a bad reading reaches the features for 72 h through +`roll_max_72`). A model row in `forecast_history` is never replaced by a heuristic one. While RID is silent overnight each hourly run re-issues the 21:00 reading, and once diff --git a/rust/README.md b/rust/README.md index 47e5afa..4d29b53 100644 --- a/rust/README.md +++ b/rust/README.md @@ -22,12 +22,17 @@ ping-collector serve [--listen A] [--no-analytics] 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 refetch FROM [TO] # re-fetch the RID reports of FROM..=TO (YYYY-MM-DD) even + # without holes, through the spike screening: corrects + # readings RID revised, deletes stored spikes; prints the + # catch-up-forecasts command when readings changed 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 catch-up-forecasts [--hours N] +ping-collector catch-up-forecasts [--hours N] [--redo-from T] # forecast the hours whose readings came late (RID's - # overnight batch) as a run then would have; default 336 h + # overnight batch) as a run then would have; default 336 h; + # --redo-from first deletes the model forecasts from T on ping-collector notify --dry-run # the readings the notifier judges, and what it would send ``` diff --git a/rust/collector/src/db.rs b/rust/collector/src/db.rs index f4803df..a186a85 100644 --- a/rust/collector/src/db.rs +++ b/rust/collector/src/db.rs @@ -237,6 +237,62 @@ pub async fn save_rid(pool: &PgPool, readings: &[Reading], stations: &Stations) Ok(rows.len()) } +/// Stored RID levels of report columns `nums` in [start, end], for the +/// screening (rid::screen). +pub async fn rid_readings_between( + pool: &PgPool, + nums: &[u32], + start: NaiveDateTime, + end: NaiveDateTime, +) -> Result> { + let rows: Vec<(NaiveDateTime, i32, f64, Option)> = sqlx::query_as( + "SELECT timestamp, station_id, water_level::float8, discharge::float8 + FROM water_measurements + WHERE station_id = ANY($1) AND timestamp >= $2 AND timestamp <= $3 + AND water_level IS NOT NULL", + ) + .bind(nums.iter().map(|&n| n as i32).collect::>()) + .bind(start) + .bind(end) + .fetch_all(pool) + .await + .context("reading stored RID levels")?; + Ok(rows + .into_iter() + .map(|(t, n, l, q)| (t, n as u32, l, q)) + .collect()) +} + +/// Delete stored RID readings (timestamp, report column). +pub async fn delete_rid_readings(pool: &PgPool, keys: &[(NaiveDateTime, u32)]) -> Result { + if keys.is_empty() { + return Ok(0); + } + let r = sqlx::query( + "DELETE FROM water_measurements w + USING UNNEST($1::timestamp[], $2::int4[]) AS k(ts, sid) + WHERE w.timestamp = k.ts AND w.station_id = k.sid", + ) + .bind(keys.iter().map(|(t, _)| *t).collect::>()) + .bind(keys.iter().map(|(_, n)| *n as i32).collect::>()) + .execute(pool) + .await + .context("deleting RID spikes")?; + Ok(r.rows_affected() as usize) +} + +/// Delete the model rows of forecast_history issued at or after `from`, so +/// a catch-up recomputes them (after the readings they were computed from +/// changed). +pub async fn delete_model_forecasts_since(pool: &PgPool, from: NaiveDateTime) -> Result { + let r = sqlx::query("DELETE FROM forecast_history WHERE as_of >= $1 AND source = 'model'") + .bind(from) + .execute(pool) + .await + .context("deleting forecast_history rows")?; + Ok(r.rows_affected() as usize) +} + /// Day -> hours (0-23) that have at least one RID reading, for [start, end]. pub async fn recorded_hours_by_day( pool: &PgPool, diff --git a/rust/collector/src/db_tests.rs b/rust/collector/src/db_tests.rs index 26912ab..416dd97 100644 --- a/rust/collector/src/db_tests.rs +++ b/rust/collector/src/db_tests.rs @@ -191,6 +191,147 @@ async fn yesterdays_late_hours_are_recovered_after_midnight() { db.drop().await; } +/// A report source with fixed P.1 levels per hour of one report date. +struct ScriptedRid(Mutex>>); + +impl RidSource for ScriptedRid { + async fn fetch_report(&self, date: NaiveDate) -> Result> { + let script = self + .0 + .lock() + .unwrap() + .get(&date) + .cloned() + .unwrap_or_default(); + Ok(script + .into_iter() + .map(|(h, level)| Reading { + timestamp: date.and_time(NaiveTime::MIN) + Duration::hours(h), + station_num: 8, + water_level: level, + discharge: Some(45.0), + discharge_percent: None, + }) + .collect()) + } +} + +async fn p1_levels(pool: &PgPool) -> Vec<(NaiveDateTime, f64, Option)> { + sqlx::query_as( + "SELECT timestamp, water_level::float8, discharge::float8 FROM water_measurements + WHERE station_id = 8 ORDER BY timestamp", + ) + .fetch_all(pool) + .await + .unwrap() +} + +#[tokio::test] +async fn spikes_are_screened_and_refetch_repairs_the_store() { + let Some(db) = TestDb::new().await else { + return; + }; + let stations = Stations::bundled(); + let day = d(2026, 10, 24); + // what the store held before screening existed: the slip, level-only + db::save_rid( + &db.pool, + &[ + Reading { + timestamp: t(2026, 10, 24, 23, 0), + station_num: 8, + water_level: 1.04, + discharge: None, + discharge_percent: None, + }, + Reading { + timestamp: t(2026, 10, 25, 0, 0), + station_num: 8, + water_level: 10.6, + discharge: None, + discharge_percent: None, + }, + Reading { + timestamp: t(2026, 10, 25, 1, 0), + station_num: 8, + water_level: 1.07, + discharge: Some(45.0), + discharge_percent: None, + }, + ], + &stations, + ) + .await + .unwrap(); + + // RID still serves the slip: refetch deletes it and fills the discharge + let src = ScriptedRid(Mutex::new(HashMap::from([( + day, + vec![(23, 1.04), (24, 10.6)], + )]))); + let s = rid::refetch(&src, &db.pool, &stations, day, day) + .await + .unwrap(); + assert_eq!((s.saved, s.spikes, s.deleted), (1, 1, 1)); + assert_eq!(s.first_change, Some(t(2026, 10, 24, 23, 0))); + assert_eq!( + p1_levels(&db.pool).await, + vec![ + (t(2026, 10, 24, 23, 0), 1.04, Some(45.0)), + (t(2026, 10, 25, 1, 0), 1.07, Some(45.0)), + ] + ); + + // RID corrected it: the hour comes back with the right value + src.0 + .lock() + .unwrap() + .insert(day, vec![(23, 1.04), (24, 1.06)]); + let s = rid::refetch(&src, &db.pool, &stations, day, day) + .await + .unwrap(); + assert_eq!((s.saved, s.spikes, s.deleted, s.changed), (2, 0, 0, 1)); + assert_eq!(s.first_change, Some(t(2026, 10, 25, 0, 0))); + assert_eq!(p1_levels(&db.pool).await.len(), 3); + + // a live cycle: the newest hour jumps and waits for the next one + let today = d(2026, 10, 25); + src.0 + .lock() + .unwrap() + .insert(today, vec![(1, 1.07), (2, 9.9)]); + // (yesterday's report comes along: it still has holes) + let o = rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 25, 2, 10)).await; + assert_eq!((o.fetched, o.saved), (4, 3)); + assert_eq!( + p1_levels(&db.pool).await.last().unwrap().0, + t(2026, 10, 25, 1, 0) + ); + // 03:00 is back at 1.1: 02:00 was a spike and never stored + src.0 + .lock() + .unwrap() + .insert(today, vec![(1, 1.07), (2, 9.9), (3, 1.1)]); + let o = rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 25, 3, 10)).await; + assert_eq!((o.fetched, o.saved), (5, 4)); + let levels: Vec = p1_levels(&db.pool).await.iter().map(|r| r.1).collect(); + assert_eq!(levels, vec![1.04, 1.06, 1.07, 1.1]); + + // the forecasts issued from the changed hours can be redone + let mut f = forecast_row("P.1", 24, 0.9); + f.as_of = t(2026, 10, 25, 0, 0); + let mut g = forecast_row("P.1", 24, 0.0); + g.as_of = t(2026, 10, 24, 23, 0); + db::save_forecast_history(&db.pool, &[f, g]).await.unwrap(); + assert_eq!( + db::delete_model_forecasts_since(&db.pool, t(2026, 10, 25, 0, 0)) + .await + .unwrap(), + 1 + ); + db.drop().await; +} + #[tokio::test] async fn rid_upsert_revises_values_and_rounds_like_numeric() { let Some(db) = TestDb::new().await else { diff --git a/rust/collector/src/main.rs b/rust/collector/src/main.rs index 4fa07a9..89f4f56 100644 --- a/rust/collector/src/main.rs +++ b/rust/collector/src/main.rs @@ -62,6 +62,14 @@ enum Command { }, /// Re-fetch RID reports that have holes: a number of days back, or "all" FillGaps { days: String }, + /// Re-fetch the RID reports of FROM..=TO (YYYY-MM-DD; day D's report holds + /// D 01:00 to D+1 00:00) even without holes, through the spike screening: + /// corrects readings RID has revised and removes stored spikes + Refetch { + from: chrono::NaiveDate, + /// Last report date [default: FROM] + to: Option, + }, /// 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 @@ -87,6 +95,11 @@ enum Command { /// How many hours back to look #[arg(long, default_value_t = 336)] hours: i64, + /// First delete the model forecasts issued from this local time on + /// (e.g. 2026-10-25T00:00) so they are recomputed: after `refetch` + /// changed the readings they came from + #[arg(long)] + redo_from: Option, }, /// Show the readings the notifier judges and the ntfy transitions due now Notify { @@ -212,7 +225,43 @@ async fn run(cli: Cli) -> Result<()> { let rain = rain_file.as_deref().map(read_rain_file).transpose()?; collector.forecast(json, save, now, rain).await } - Command::CatchUpForecasts { hours } => collector.catch_up_forecasts(hours).await, + Command::CatchUpForecasts { hours, redo_from } => { + let redo_from = redo_from.as_deref().map(parse_local).transpose()?; + collector.catch_up_forecasts(hours, redo_from).await + } + Command::Refetch { from, to } => { + let to = to.unwrap_or(from); + if to < from { + bail!("refetch: TO ({to}) is before FROM ({from})"); + } + db::ensure_schema(&collector.pool).await?; + let s = rid::refetch( + &collector.rid, + &collector.pool, + &collector.stations(), + from, + to, + ) + .await?; + println!( + "{} readings saved ({} new or changed), {} spikes rejected, {} stored spikes deleted, {} held for want of the next reading", + s.saved, + s.changed - s.deleted, + s.spikes, + s.deleted, + s.held + ); + if let Some(t) = s.first_change { + let hours = (util::bangkok_now() - t).num_hours() + 2; + if hours <= 24 * 60 { + println!( + "forecasts issued since {t} came from other readings; recompute them with\n ping-collector catch-up-forecasts --hours {hours} --redo-from {}", + t.format("%Y-%m-%dT%H:%M") + ); + } + } + Ok(()) + } Command::Notify { dry_run } => collector.notify(dry_run).await, Command::CheckDb => { db::ensure_schema(&collector.pool).await?; diff --git a/rust/collector/src/rid.rs b/rust/collector/src/rid.rs index 8c29a53..0bb0372 100644 --- a/rust/collector/src/rid.rs +++ b/rust/collector/src/rid.rs @@ -9,6 +9,12 @@ //! every reading it gets (not only "the current hour"), and re-fetches //! yesterday's report while the store still has holes in it: hours 22-24 of a //! day are reachable only through that day's report. +//! +//! Before storing, readings are screened for single-hour spikes (see +//! [`judge`]): RID's report sometimes carries a slipped decimal point, a +//! flipped sign or a placeholder for one hour (P.1 10.6 m between 1.04 and +//! 1.07 on 2026-10-25 00:00), which would raise a false alarm and +//! distort the forecast features for three days. use crate::db; use crate::stations::Stations; @@ -17,7 +23,7 @@ use anyhow::{Context, Result}; use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime}; use serde_json::Value; use sqlx::PgPool; -use std::collections::{BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::future::Future; pub const N_STATIONS: u32 = 16; @@ -96,6 +102,235 @@ pub fn parse_report(report_date: NaiveDate, payload: &Value) -> Vec { out } +// ---- Screening ------------------------------------------------------------- + +/// Neighbours of a reading are looked for this many hours before and after it. +const SCREEN_WINDOW_H: i64 = 3; + +/// Gauges on flashy tributaries: genuine 1-h changes of 1.3-3.1 m are on +/// record (P.76 0.09 -> 3.18 m on 2026-05-29). Elsewhere the largest genuine +/// 1-h change since 2018 is 1.1 m (P.1: 0.56 m). +const FLASHY: &[&str] = &["P.76", "P.77", "P.81", "P.82", "P.84", "P.85", "P.87"]; + +/// A change from the previous reading larger than this (m) needs the next +/// reading to tell a spike from a real change. +pub fn jump_limit(code: &str) -> f64 { + if FLASHY.contains(&code) { 2.0 } else { 1.0 } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Verdict { + Keep, + /// The newest reading jumped more than twice the limit: wait for the next + /// one (an hour at most while RID is current) before storing it. + Hold, + /// Out and back within an hour or two: not stored. + Spike, +} + +/// `prev`: the last accepted level within [`SCREEN_WINDOW_H`] before the +/// reading, `next`: the following reading within the window, if any. +/// +/// A spike leaves the previous level by more than `limit` and the next +/// reading comes back, closer to the previous level than half of either +/// jump. A rising or falling limb (prev < level < next) or a step the next +/// reading confirms is kept. 2018-2026 history: 16 spikes (sign flips, +/// slipped decimal points, -5/-6/-9 m placeholders) and no genuine change +/// would have been rejected. +pub fn judge(limit: f64, prev: Option, level: f64, next: Option) -> Verdict { + let Some(p) = prev else { + return Verdict::Keep; + }; + let out = (level - p).abs(); + if out <= limit { + return Verdict::Keep; + } + let Some(n) = next else { + return if out > 2.0 * limit { + Verdict::Hold + } else { + Verdict::Keep + }; + }; + let back = (level - n).abs(); + if back > limit && (n - p).abs() < out.min(back) / 2.0 { + Verdict::Spike + } else { + Verdict::Keep + } +} + +/// A reading already in the database: (timestamp, report column, level, discharge). +pub type StoredReading = (NaiveDateTime, u32, f64, Option); + +#[derive(Debug, Default)] +pub struct Screened { + pub keep: Vec, + pub held: Vec, + pub spikes: Vec, + /// Stored rows at a spike's hour that are spikes themselves: delete. + pub delete: Vec<(NaiveDateTime, u32)>, +} + +/// Screen a batch against itself and the stored readings around it. A batch +/// reading replaces the stored one at the same hour; stored readings count +/// as accepted. `code` maps a report column to its station code. +pub fn screen( + readings: &[Reading], + stored: &[StoredReading], + code: impl Fn(u32) -> String, +) -> Screened { + #[derive(Clone, Copy)] + enum Src { + Batch(usize), + Stored, + } + let mut out = Screened::default(); + let mut by_station: BTreeMap> = BTreeMap::new(); + for (t, n, level, _) in stored { + by_station + .entry(*n) + .or_default() + .insert(*t, (*level, Src::Stored)); + } + let stored_at: HashMap<(NaiveDateTime, u32), f64> = + stored.iter().map(|(t, n, l, _)| ((*t, *n), *l)).collect(); + // the last copy of an hour wins, as in the upsert + for (i, r) in readings.iter().enumerate() { + by_station + .entry(r.station_num) + .or_default() + .insert(r.timestamp, (r.water_level, Src::Batch(i))); + } + let window = Duration::hours(SCREEN_WINDOW_H); + for (n, series) in &by_station { + let limit = jump_limit(&code(*n)); + let mut last: Option<(NaiveDateTime, f64)> = None; + let entries: Vec<_> = series.iter().map(|(t, v)| (*t, *v)).collect(); + for (k, &(t, (level, src))) in entries.iter().enumerate() { + let prev = last.filter(|(pt, _)| t - *pt <= window).map(|(_, l)| l); + let i = match src { + Src::Stored => { + last = Some((t, level)); + continue; + } + Src::Batch(i) => i, + }; + let next = entries + .get(k + 1) + .filter(|(nt, _)| *nt - t <= window) + .map(|(_, (l, _))| *l); + match judge(limit, prev, level, next) { + Verdict::Keep => { + out.keep.push(readings[i].clone()); + last = Some((t, level)); + } + Verdict::Hold => out.held.push(readings[i].clone()), + Verdict::Spike => { + out.spikes.push(readings[i].clone()); + if let Some(&old) = stored_at.get(&(t, *n)) { + if judge(limit, prev, old, next) == Verdict::Spike { + out.delete.push((t, *n)); + } else { + last = Some((t, old)); + } + } + } + } + } + } + out.keep.sort_by_key(|r| (r.timestamp, r.station_num)); + out +} + +#[derive(Debug, Default)] +pub struct StoreOutcome { + pub saved: usize, + pub held: usize, + pub spikes: usize, + pub deleted: usize, + /// Saved readings that were new or differ from the stored ones, plus deletions. + pub changed: usize, + /// The earliest hour whose stored data changed (forecasts from then on + /// were computed from other data). + pub first_change: Option, +} + +/// Screen `readings` against the stored ones around them, then save them and +/// delete stored spikes. +pub async fn store( + pool: &PgPool, + readings: &[Reading], + stations: &Stations, +) -> Result { + let mut out = StoreOutcome::default(); + let (Some(first), Some(last)) = ( + readings.iter().map(|r| r.timestamp).min(), + readings.iter().map(|r| r.timestamp).max(), + ) else { + return Ok(out); + }; + let window = Duration::hours(SCREEN_WINDOW_H); + let nums: Vec = readings + .iter() + .map(|r| r.station_num) + .collect::>() + .into_iter() + .collect(); + let stored = db::rid_readings_between(pool, &nums, first - window, last + window).await?; + let code = |n: u32| stations.get(n).code; + let s = screen(readings, &stored, code); + for r in &s.spikes { + tracing::warn!( + "RID {} {}: {} m is a one-hour spike, not stored", + r.timestamp, + code(r.station_num), + r.water_level + ); + } + for r in &s.held { + tracing::warn!( + "RID {} {}: {} m jumps from the previous reading; waiting for the next one", + r.timestamp, + code(r.station_num), + r.water_level + ); + } + let before: HashMap<(NaiveDateTime, u32), (f64, Option)> = stored + .iter() + .map(|(t, n, l, q)| ((*t, *n), (*l, *q))) + .collect(); + // NUMERIC(10,3) / (10,2): compare at the stored precision + let differs = |a: f64, b: f64, places: i32| { + let f = 10f64.powi(places); + (a * f).round() != (b * f).round() + }; + let mut changed: Vec = s + .keep + .iter() + .filter(|r| match before.get(&(r.timestamp, r.station_num)) { + None => true, + Some((l, q)) => { + differs(*l, r.water_level, 3) + || match (q, r.discharge) { + (Some(a), Some(b)) => differs(*a, b, 2), + (None, None) => false, + _ => true, + } + } + }) + .map(|r| r.timestamp) + .collect(); + changed.extend(s.delete.iter().map(|(t, _)| *t)); + out.saved = db::save_rid(pool, &s.keep, stations).await?; + out.deleted = db::delete_rid_readings(pool, &s.delete).await?; + out.held = s.held.len(); + out.spikes = s.spikes.len(); + out.changed = changed.len(); + out.first_change = changed.into_iter().min(); + Ok(out) +} + /// Anything that can produce a parsed RID report (HTTP in production, /// canned rows in tests). pub trait RidSource { @@ -248,12 +483,14 @@ pub async fn run_cycle( tracing::warn!("RID: no data fetched for {:?}", out.dates); return out; } - if let Err(e) = db::save_rid(pool, &readings, stations).await { - tracing::error!("RID save failed: {e:#}"); - out.error = Some(format!("save failed: {e:#}")); - return out; + match store(pool, &readings, stations).await { + Ok(s) => out.saved = s.saved, + Err(e) => { + tracing::error!("RID save failed: {e:#}"); + out.error = Some(format!("save failed: {e:#}")); + return out; + } } - out.saved = readings.len(); let latest = readings .iter() .map(|r| r.timestamp) @@ -308,8 +545,7 @@ pub async fn fill_gaps( } match src.fetch_report(*d).await { Ok(readings) if !readings.is_empty() => { - db::save_rid(pool, &readings, stations).await?; - filled += readings.len(); + filled += store(pool, &readings, stations).await?.saved; } Ok(_) => { empty += 1; @@ -332,6 +568,52 @@ pub async fn fill_gaps( Ok(filled) } +/// Re-fetch the reports of `from..=to` (report dates: day D's report holds +/// D 01:00 to D+1 00:00) whether or not they have holes, and store them +/// through the screening: corrects readings RID has revised since (and +/// fills the discharge of level-only hours), removes stored spikes. +pub async fn refetch( + src: &impl RidSource, + pool: &PgPool, + stations: &Stations, + from: NaiveDate, + to: NaiveDate, +) -> Result { + let mut total = StoreOutcome::default(); + let mut day = from; + while day <= to { + if day > from { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; // be polite + } + match src.fetch_report(day).await { + Ok(readings) => { + let s = store(pool, &readings, stations).await?; + tracing::info!( + "refetch {day}: {} fetched, {} saved ({} new or changed), {} spikes rejected, {} stored spikes deleted, {} held", + readings.len(), + s.saved, + s.changed - s.deleted, + s.spikes, + s.deleted, + s.held + ); + total.saved += s.saved; + total.held += s.held; + total.spikes += s.spikes; + total.deleted += s.deleted; + total.changed += s.changed; + total.first_change = match (total.first_change, s.first_change) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }; + } + Err(e) => tracing::warn!("refetch {day}: {e:#}"), + } + day += Duration::days(1); + } + Ok(total) +} + /// Next RID poll: `offset` minutes past the next hour when current, else in /// `retry` minutes (never later than that hourly poll). pub fn next_poll_time( @@ -407,6 +689,104 @@ mod tests { assert!(!by.contains_key(&5)); // negative discharge drops the reading } + #[test] + fn judge_spikes_steps_and_limbs() { + use Verdict::*; + // the P.1 slip of 2026-10-25 00:00, a sign flip, a placeholder + assert_eq!(judge(1.0, Some(1.04), 10.6, Some(1.07)), Spike); + assert_eq!(judge(1.0, Some(-1.18), 1.18, Some(-1.18)), Spike); + assert_eq!(judge(2.0, Some(-0.05), -5.0, Some(-0.05)), Spike); + // small changes, no neighbour, a confirmed step, rising limbs + assert_eq!(judge(1.0, Some(1.0), 1.9, Some(1.0)), Keep); + assert_eq!(judge(1.0, None, 10.6, Some(1.07)), Keep); + assert_eq!(judge(1.0, Some(1.0), 2.5, Some(2.6)), Keep); + assert_eq!(judge(2.0, Some(0.04), 1.61, Some(2.67)), Keep); + assert_eq!(judge(1.0, Some(0.82), 1.78, Some(2.82)), Keep); + // a peak that falls back less than halfway in the next hour is real + assert_eq!(judge(1.0, Some(1.0), 3.0, Some(1.8)), Keep); + // the newest reading: a jump over twice the limit waits + assert_eq!(judge(1.0, Some(1.04), 10.6, None), Hold); + assert_eq!(judge(1.0, Some(1.0), 2.5, None), Keep); + assert_eq!(judge(2.0, Some(0.09), 3.18, None), Keep); + assert_eq!(jump_limit("P.76"), 2.0); + assert_eq!(jump_limit("P.1"), 1.0); + } + + fn rd(h: u32, n: u32, level: f64) -> Reading { + Reading { + timestamp: t(2026, 10, 25, h, 0), + station_num: n, + water_level: level, + discharge: None, + discharge_percent: None, + } + } + + fn code(n: u32) -> String { + if n == 8 { + "P.1".into() + } else { + format!("P.{}", 19 + n) + } + } + + #[test] + fn screen_rejects_a_spike_and_deletes_its_stored_copy() { + let stored = vec![ + (t(2026, 10, 24, 23, 0), 8, 1.04, None), + (t(2026, 10, 25, 0, 0), 8, 10.6, None), + ]; + // the report re-fetched, RID still has the slip + let batch = vec![rd(0, 8, 10.6), rd(1, 8, 1.07), rd(2, 8, 1.09)]; + let s = screen(&batch, &stored, code); + assert_eq!(s.spikes, vec![rd(0, 8, 10.6)]); + assert_eq!(s.delete, vec![(t(2026, 10, 25, 0, 0), 8)]); + assert_eq!(s.keep, vec![rd(1, 8, 1.07), rd(2, 8, 1.09)]); + // RID corrected it: the new value replaces the stored spike + let s = screen(&[rd(0, 8, 1.06), rd(1, 8, 1.07)], &stored, code); + assert!(s.spikes.is_empty() && s.delete.is_empty()); + assert_eq!(s.keep, vec![rd(0, 8, 1.06), rd(1, 8, 1.07)]); + } + + #[test] + fn screen_keeps_a_good_stored_copy_and_judges_the_next_hour_by_it() { + let stored = vec![ + (t(2026, 10, 25, 0, 0), 8, 1.06, None), + (t(2026, 10, 25, 1, 0), 8, 1.07, None), + ]; + let s = screen(&[rd(1, 8, 10.7), rd(2, 8, 1.09)], &stored, code); + assert_eq!(s.spikes, vec![rd(1, 8, 10.7)]); + assert!(s.delete.is_empty()); + assert_eq!(s.keep, vec![rd(2, 8, 1.09)]); + } + + #[test] + fn screen_holds_the_newest_jump_until_the_next_reading() { + let stored = vec![(t(2026, 10, 25, 13, 0), 8, 1.04, None)]; + let s = screen(&[rd(14, 8, 10.6)], &stored, code); + assert_eq!(s.held, vec![rd(14, 8, 10.6)]); + assert!(s.keep.is_empty() && s.spikes.is_empty()); + // next poll: the report again, now with 15:00 + let s = screen(&[rd(14, 8, 10.6), rd(15, 8, 1.06)], &stored, code); + assert_eq!(s.spikes, vec![rd(14, 8, 10.6)]); + assert_eq!(s.keep, vec![rd(15, 8, 1.06)]); + // a real jump the next reading confirms is stored then + let s = screen(&[rd(14, 8, 3.5), rd(15, 8, 3.6)], &stored, code); + assert_eq!(s.keep, vec![rd(14, 8, 3.5), rd(15, 8, 3.6)]); + } + + #[test] + fn screen_window_and_stations_are_separate() { + // previous reading 4 h back: nothing to compare with + let stored = vec![(t(2026, 10, 25, 9, 0), 8, 1.0, None)]; + let s = screen(&[rd(13, 8, 5.0), rd(13, 1, 0.5)], &stored, code); + assert_eq!(s.keep.len(), 2); + // a jump at one gauge does not involve the others + let stored = vec![(t(2026, 10, 25, 12, 0), 1, 0.1, None)]; + let s = screen(&[rd(13, 8, 1.0), rd(13, 1, 0.12)], &stored, code); + assert_eq!(s.keep.len(), 2); + } + #[test] fn missing_rows_key_is_empty() { assert!(parse_report(d(2026, 7, 22), &json!({"unexpected": true})).is_empty()); diff --git a/rust/collector/src/service.rs b/rust/collector/src/service.rs index b0e7886..2f7162a 100644 --- a/rust/collector/src/service.rs +++ b/rust/collector/src/service.rs @@ -400,10 +400,22 @@ impl Collector { /// `ping-collector catch-up-forecasts`: `forecasting::catch_up` over /// `hours`, e.g. once after an upgrade for the nights before it. - pub async fn catch_up_forecasts(&self, hours: i64) -> Result<()> { + pub async fn catch_up_forecasts( + &self, + hours: i64, + redo_from: Option, + ) -> Result<()> { if !(1..=24 * 60).contains(&hours) { bail!("--hours must be between 1 and 1440 (Open-Meteo keeps 92 days of past rain)"); } + let now = bangkok_now(); + if let Some(from) = redo_from + && from < floor_hour(now) - Duration::hours(hours) + { + bail!( + "--redo-from {from} is more than --hours {hours} back: those hours would not be recomputed" + ); + } let mut store = ModelStore::new(&self.cfg.models_dir); let models = store.refresh()?; if models.is_empty() { @@ -413,11 +425,15 @@ impl Collector { ); } db::ensure_schema(&self.pool).await?; + if let Some(from) = redo_from { + let n = db::delete_model_forecasts_since(&self.pool, from).await?; + eprintln!("deleted {n} model forecasts issued since {from}; recomputing"); + } let c = forecasting::catch_up( &self.pool, &self.rain, &models, - bangkok_now(), + now, hours, &Default::default(), )