diff --git a/rust/README.md b/rust/README.md index 4d29b53..7133d2b 100644 --- a/rust/README.md +++ b/rust/README.md @@ -25,7 +25,7 @@ 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 + # catch-up-forecasts command when stored levels 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 diff --git a/rust/collector/src/db_tests.rs b/rust/collector/src/db_tests.rs index 416dd97..57231e0 100644 --- a/rust/collector/src/db_tests.rs +++ b/rust/collector/src/db_tests.rs @@ -272,8 +272,8 @@ async fn spikes_are_screened_and_refetch_repairs_the_store() { 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!((s.saved, s.spikes, s.deleted, s.changed), (1, 1, 1, 2)); + assert_eq!(s.first_change, Some(t(2026, 10, 25, 0, 0))); assert_eq!( p1_levels(&db.pool).await, vec![ @@ -291,8 +291,17 @@ async fn spikes_are_screened_and_refetch_repairs_the_store() { .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!(s.first_change, None); // a new hour, nothing to redo assert_eq!(p1_levels(&db.pool).await.len(), 3); + // RID revises a level: the forecasts from that hour on need a redo + src.0 + .lock() + .unwrap() + .insert(day, vec![(23, 1.05), (24, 1.06)]); + let s = rid::refetch(&src, &db.pool, &stations, day, day) + .await + .unwrap(); + assert_eq!(s.first_change, Some(t(2026, 10, 24, 23, 0))); // a live cycle: the newest hour jumps and waits for the next one let today = d(2026, 10, 25); @@ -315,7 +324,7 @@ async fn spikes_are_screened_and_refetch_repairs_the_store() { 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]); + assert_eq!(levels, vec![1.05, 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); diff --git a/rust/collector/src/main.rs b/rust/collector/src/main.rs index 89f4f56..e5eed96 100644 --- a/rust/collector/src/main.rs +++ b/rust/collector/src/main.rs @@ -255,7 +255,7 @@ async fn run(cli: Cli) -> Result<()> { 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 {}", + "stored levels changed from {t} on: the forecasts issued since came from wrong levels; recompute them with\n ping-collector catch-up-forecasts --hours {hours} --redo-from {}", t.format("%Y-%m-%dT%H:%M") ); } diff --git a/rust/collector/src/rid.rs b/rust/collector/src/rid.rs index 0bb0372..c79b83b 100644 --- a/rust/collector/src/rid.rs +++ b/rust/collector/src/rid.rs @@ -251,8 +251,10 @@ pub struct StoreOutcome { 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). + /// The earliest hour whose stored level changed or was deleted: the + /// forecasts issued from then on came from a wrong level. (New hours and + /// added discharges do not count: the catch-up forecasts new hours, and + /// a forecast issued without the discharge is still what was issued.) pub first_change: Option, } @@ -322,12 +324,23 @@ pub async fn store( .map(|r| r.timestamp) .collect(); changed.extend(s.delete.iter().map(|(t, _)| *t)); + let level_changed = s + .keep + .iter() + .filter(|r| { + before + .get(&(r.timestamp, r.station_num)) + .is_some_and(|(l, _)| differs(*l, r.water_level, 3)) + }) + .map(|r| r.timestamp) + .chain(s.delete.iter().map(|(t, _)| *t)) + .min(); 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(); + out.first_change = level_changed; Ok(out) }