diff --git a/.env.example b/.env.example index 75d842f..a6d7a16 100644 --- a/.env.example +++ b/.env.example @@ -9,3 +9,5 @@ DATABASE_PORT= DATABASE_USERNAME= DATABASE_PASSWORD= ALERT_ZONE=D1 +# Hysteresis deadband in metres around river thresholds (anti-flapping). Default 0.05. +RIVER_THRESHOLD_MARGIN=0.05 diff --git a/.gitignore b/.gitignore index 2ef9c53..9ba9087 100644 --- a/.gitignore +++ b/.gitignore @@ -11,4 +11,7 @@ dist/ .eslintcache # IDE -.vscode/settings.json \ No newline at end of file +.vscode/settings.json + +# Local agent / preview config +.claude/ \ No newline at end of file diff --git a/README.md b/README.md index a70f753..8149746 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,16 @@ The system interfaces with various official and unofficial sources to provide a - **Details:** Dynamically retrieves forecast maps for the following day via the Pretemp archive. - **Notification:** If conditions require it (presence of ongoing alerts), it sends the thunderstorm forecast image to the Telegram channel. +4. **River levels (Allerta Meteo hydrometric stations)** + - **Feature:** Monitoring of water levels at configured hydrometric stations. + - **Details:** Periodically polls the Allerta Meteo time-series endpoint for each registered station and compares the latest reading against operator-defined thresholds (`soglia1`, `soglia2`, `soglia3`). Station metadata and thresholds live in the `rivers` table; each new reading is appended to `river_levels` (de-duplicated by measurement time). A hysteresis deadband (`RIVER_THRESHOLD_MARGIN`, default 0.05 m) avoids alert flapping when a level hovers around a threshold. + - **Notification:** Sends a Telegram message only on crossing events — when a reading rises above or falls below a threshold relative to the previous check. + +5. **Flood prediction (empirical precursor model)** + - **Feature:** Warns that a downstream point of interest is likely to exceed a threshold, with an estimated lead time, based on what upstream gauges are doing now. + - **Details:** For each configured `river_link` (an upstream gauge → a downstream point), a daily calibration mines the accumulated `river_levels` history for past downstream exceedances and learns, from those events only, the upstream `precursor_level` and the typical `lead_time`. A link stays inactive until it has enough historical events. Online, when the upstream gauge reaches the learned level while rising, a single prediction is emitted and recorded in `link_predictions` for later scoring. No AI — event detection plus robust statistics. + - **Bootstrap:** Since the live API only retains ~2.4 days, historical data is backfilled from the free ARPAE-SIMC open archive (see below). Without a backfill the model simply stays dormant until enough live events accrue. + ## Scheduled Tasks (Crons) The application relies on scheduled tasks (crons) to automate the weather monitoring flow: @@ -31,6 +41,52 @@ The application relies on scheduled tasks (crons) to automate the weather monito - Verifies and sends the Pretemp map for the following day, provided there is an ongoing alert and the map hasn't been sent yet. - **Estofex Report Check** - Verifies and sends the Estofex map for the following day, following the same conditional logic based on ongoing alerts. +- **River Levels Check** + - Every 5 minutes, for each row in the `rivers` table, fetches the latest hydrometric reading and appends it to `river_levels`. Sends a Telegram message only when the reading crosses one of the configured thresholds since the previous check. +- **Flood Prediction Check** + - Every 5 minutes, evaluates each calibrated `river_link`: if the upstream gauge has reached its learned precursor level while rising, emits a single de-duplicated flood-arrival prediction. +- **Flood Calibration** + - Daily, re-learns every link's model (lead time + precursor level) from the accumulated `river_levels` history. Also runnable on demand after a backfill. + +## Database bootstrap + +Schema is applied manually (no migration tooling). The required tables are in [sql/rivers.sql](sql/rivers.sql): + +```bash +psql "$DATABASE_URL" -f sql/rivers.sql +``` + +## River stations CRUD + +Stations are managed via HTTP (port 3000): + +- `GET /rivers` — list registered stations +- `POST /rivers` — create; body `{ station_id, river_name, station_name, soglia1?, soglia2?, soglia3? }` +- `PATCH /rivers/:id` — update any of `river_name`, `station_name`, `soglia1`, `soglia2`, `soglia3` +- `DELETE /rivers/:id` — delete (cascades to `river_levels`) +- `POST /river-levels` — trigger an on-demand check (returns `{ checked, crossings, skipped, unchanged }`) +- `GET /rivers/nearest?lat=&lon=&limit=` — discovery: nearest stations to a point, with coordinates and the official soglie the Allerta sensor list reports (suggestions only) + +The `station_id` is the Allerta Meteo `idstazione`; threshold values are operator-defined (the `nearest` endpoint surfaces the portal's official soglie as suggestions, but they are not auto-applied). + +## Flood prediction (links, calibration, backfill) + +Both ends of a link must be registered as `rivers` so their readings accumulate in `river_levels`. + +- `GET /river-links` — list links and their learned models +- `POST /river-links` — create; body `{ upstream_river_id, downstream_river_id, target_threshold? }` (`target_threshold` 1–3, default 1) +- `DELETE /river-links/:id` — delete a link +- `POST /flood-calibration` — re-learn all link models from history (returns `{ calibrated, active, skipped }`) +- `POST /flood-prediction` — evaluate all links now (returns `{ evaluated, predictions, skipped }`) +- `POST /flood-backfill` — body `{ from: "YYYY-MM", to: "YYYY-MM" }`; backfills `river_levels` from the ARPAE-SIMC open archive (`https://dati-simc.arpae.it/opendata/osservati/meteo/storico/`) for every registered station. Returns `202` immediately and runs in the background (it streams ~20 MB/month). Run it once, then `POST /flood-calibration`. + +Example bootstrap for the Molinella stations (Idice S. Antonio + Reno Gandazzolo and their upstream gauges): + +```bash +# register stations, link them, then: +curl -X POST localhost:3000/flood-backfill -H 'content-type: application/json' -d '{"from":"2020-01","to":"2024-12"}' +curl -X POST localhost:3000/flood-calibration +``` ## Configuration and Installation diff --git a/index.html b/index.html new file mode 100644 index 0000000..3bda89d --- /dev/null +++ b/index.html @@ -0,0 +1,348 @@ + + + + + +Simulazione piena · Mida-Sync + + + +
+
+

🌊 Simulazione piena — preallarme a valle

+

Come il sistema stima quando una piena raggiunge la pianura, osservando l'idrometro a monte. Dimostrazione · Mida-Sync

+
+ +
+
+
+ +
+ + +
+
+
+ + +
+
+ + +
+
+ + +
+
+ + + +
+
☔Pioggia nel bacino
+
📈Preallarme a monte
+
🌊Onda in viaggio
+
🔴Soglia a valle
+
+ +
+
+
+
+
+ idrometro a monte + idrometro a valle + soglia + arrivo previsto +
+ +
+
Tempo simulato
0 h
+
Monte
– m
+
Valle
– m
+
Stato modello
In attesa
+
Arrivo previsto
–
+
Arrivo reale · errore
–
+
+
+ +
+

Come leggerlo

+

Quando l'idrometro a monte (blu) raggiunge il suo livello di preallarme, il modello lancia + l'avviso e stima fra quanto la piena toccherà la soglia a valle (riga rossa). L'onda viaggia lungo il + fiume: l'intervallo fra le due curve che attraversano la soglia è il tempo di anticipo. Alla fine puoi + confrontare arrivo previsto e arrivo reale: la differenza è l'errore tipico di una stima statistica + (le piene più forti viaggiano più veloci e arrivano un po' prima del previsto).

+

Prova a ridurre l'intensità: sotto una certa soglia il fiume non esonda e il + modello — correttamente — resta in silenzio.

+
+ +
+ Stima statistica a scopo dimostrativo — non è una previsione ufficiale. + Per allerte reali fare riferimento ad Allerta Meteo Emilia‑Romagna e alla Protezione Civile. +
+
+ + + + diff --git a/modello-predittivo-piene.pdf b/modello-predittivo-piene.pdf new file mode 100644 index 0000000..67e5fc4 Binary files /dev/null and b/modello-predittivo-piene.pdf differ diff --git a/sql/rivers.sql b/sql/rivers.sql new file mode 100644 index 0000000..add6a90 --- /dev/null +++ b/sql/rivers.sql @@ -0,0 +1,70 @@ +CREATE TABLE IF NOT EXISTS rivers ( + id SERIAL PRIMARY KEY, + station_id TEXT NOT NULL UNIQUE, + river_name TEXT NOT NULL, + station_name TEXT NOT NULL, + soglia1 NUMERIC, + soglia2 NUMERIC, + soglia3 NUMERIC, + created_on TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_on TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS river_levels ( + id SERIAL PRIMARY KEY, + river_id INTEGER NOT NULL REFERENCES rivers(id) ON DELETE CASCADE, + value NUMERIC NOT NULL, + measured_at TIMESTAMPTZ NOT NULL, + soglia1_above BOOLEAN, + soglia2_above BOOLEAN, + soglia3_above BOOLEAN, + created_on TIMESTAMPTZ NOT NULL DEFAULT now(), + -- One stored reading per (station, measurement time): the poller runs more often than the + -- sensor publishes, so this de-duplicates repeated reads of the same measurement. + UNIQUE (river_id, measured_at) +); + +-- For databases created before the UNIQUE clause above existed: +-- ALTER TABLE river_levels ADD CONSTRAINT river_levels_river_id_measured_at_key +-- UNIQUE (river_id, measured_at); + +CREATE INDEX IF NOT EXISTS river_levels_river_id_created_on_idx + ON river_levels (river_id, created_on DESC); + +CREATE INDEX IF NOT EXISTS river_levels_river_id_measured_at_idx + ON river_levels (river_id, measured_at ASC); + +-- Flood prediction: an upstream gauge whose readings historically preceded a threshold +-- exceedance at a downstream point of interest. Both ends are registered rivers so the poller +-- collects their readings into river_levels. Learned parameters are filled by calibration. +CREATE TABLE IF NOT EXISTS river_links ( + id SERIAL PRIMARY KEY, + upstream_river_id INTEGER NOT NULL REFERENCES rivers(id) ON DELETE CASCADE, + downstream_river_id INTEGER NOT NULL REFERENCES rivers(id) ON DELETE CASCADE, + -- Which downstream soglia (1/2/3) defines the exceedance event we predict. + target_threshold SMALLINT NOT NULL DEFAULT 1, + lead_time_minutes INTEGER, + precursor_level NUMERIC, + sample_size INTEGER NOT NULL DEFAULT 0, + model_json JSONB, + last_calibrated_on TIMESTAMPTZ, + created_on TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_on TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (upstream_river_id, downstream_river_id), + CHECK (upstream_river_id <> downstream_river_id) +); + +-- One row per prediction emitted, scored later against what actually happened downstream. +CREATE TABLE IF NOT EXISTS link_predictions ( + id SERIAL PRIMARY KEY, + link_id INTEGER NOT NULL REFERENCES river_links(id) ON DELETE CASCADE, + predicted_at TIMESTAMPTZ NOT NULL, + predicted_exceedance_at TIMESTAMPTZ NOT NULL, + upstream_value NUMERIC NOT NULL, + actual_exceedance_at TIMESTAMPTZ, + outcome TEXT NOT NULL DEFAULT 'pending', + created_on TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS link_predictions_link_id_predicted_at_idx + ON link_predictions (link_id, predicted_at DESC); diff --git a/src/config/config.ts b/src/config/config.ts index 0d447ee..6b629cf 100644 --- a/src/config/config.ts +++ b/src/config/config.ts @@ -12,6 +12,7 @@ export interface Config { telegram_token: string chat_id: string alert_zone: string + river_threshold_margin: number } export const getNodeEnv = (): NodeEnv => process.env.NODE_ENV as NodeEnv diff --git a/src/config/development.ts b/src/config/development.ts index 6c42f57..8b3a5d6 100644 --- a/src/config/development.ts +++ b/src/config/development.ts @@ -1,4 +1,5 @@ import { Config } from './config' +import { getRiverThresholdMargin } from './env' import dotenv from 'dotenv' dotenv.config() @@ -14,4 +15,5 @@ export const developmentConfig: Config = { telegram_token: process.env.TELEGRAM_2_TOKEN as string, chat_id: process.env.CHAT_ID as string, alert_zone: process.env.ALERT_ZONE || 'D1', + river_threshold_margin: getRiverThresholdMargin(), } diff --git a/src/config/env.ts b/src/config/env.ts new file mode 100644 index 0000000..9a6292f --- /dev/null +++ b/src/config/env.ts @@ -0,0 +1,22 @@ +// Environment-variable helpers shared by the per-environment config objects. Kept in its own +// module (importing nothing from ./config) so it does not create a runtime import cycle. + +const DEFAULT_RIVER_THRESHOLD_MARGIN = 0.05 + +/** + * Hysteresis deadband (in metres) applied around river thresholds to avoid alert flapping. + * Falls back to a small default when RIVER_THRESHOLD_MARGIN is unset or invalid. + */ +export const getRiverThresholdMargin = (): number => { + const raw = process.env.RIVER_THRESHOLD_MARGIN + + // Treat unset/blank as "use the default" — Number('') is 0, which would silently disable the + // deadband rather than fall back. + if (raw === undefined || raw.trim() === '') { + return DEFAULT_RIVER_THRESHOLD_MARGIN + } + + const margin = Number(raw) + + return Number.isFinite(margin) && margin >= 0 ? margin : DEFAULT_RIVER_THRESHOLD_MARGIN +} diff --git a/src/config/production.ts b/src/config/production.ts index 02cef6a..21cc30d 100644 --- a/src/config/production.ts +++ b/src/config/production.ts @@ -1,4 +1,5 @@ import { Config } from './config' +import { getRiverThresholdMargin } from './env' export const productionConfig: Config = { database: { @@ -11,4 +12,5 @@ export const productionConfig: Config = { telegram_token: process.env.TELEGRAM_TOKEN as string, chat_id: process.env.CHANNEL_ID as string, alert_zone: process.env.ALERT_ZONE || 'D1', + river_threshold_margin: getRiverThresholdMargin(), } diff --git a/src/models/link-prediction.ts b/src/models/link-prediction.ts new file mode 100644 index 0000000..61e8552 --- /dev/null +++ b/src/models/link-prediction.ts @@ -0,0 +1,36 @@ +import { database } from '..' + +const tableName = 'link_predictions' + +export type PredictionOutcome = 'pending' | 'hit' | 'miss' + +export interface LinkPrediction { + id: number + link_id: number + predicted_at: string + predicted_exceedance_at: string + upstream_value: number + actual_exceedance_at: string | null + outcome: PredictionOutcome + created_on: string +} + +export type CreatableLinkPrediction = Omit + +export const createLinkPrediction = async (prediction: CreatableLinkPrediction): Promise => { + return database.create(tableName, { + ...prediction, + actual_exceedance_at: null, + outcome: 'pending', + created_on: new Date().toISOString(), + }) +} + +export const getLatestLinkPrediction = async (linkId: number): Promise => { + const rows = await database.query( + `SELECT * FROM ${tableName} WHERE link_id = $1 ORDER BY predicted_at DESC, id DESC LIMIT 1`, + [linkId] + ) + + return rows[0] +} diff --git a/src/models/river-level.ts b/src/models/river-level.ts new file mode 100644 index 0000000..7d2c1a8 --- /dev/null +++ b/src/models/river-level.ts @@ -0,0 +1,104 @@ +import { database } from '..' + +const tableName = 'river_levels' + +export interface RiverLevel { + id: number + river_id: number + value: number + measured_at: string + soglia1_above: boolean | null + soglia2_above: boolean | null + soglia3_above: boolean | null + created_on: string +} + +export type CreatableRiverLevel = Omit + +export const getLatestRiverLevel = async (riverId: number): Promise => { + // Order by measured_at (not created_on): a historical backfill inserts old measurements with a + // current created_on, so created_on no longer tracks measurement recency. + const query = `SELECT * FROM ${tableName} WHERE river_id = $1 ORDER BY measured_at DESC, id DESC LIMIT 1` + + const rows = await database.query(query, [riverId]) + + return rows[0] +} + +export const createRiverLevel = async (level: CreatableRiverLevel): Promise => { + return database.create(tableName, { ...level, created_on: new Date().toISOString() }) +} + +/** + * Inserts a reading only if no row already exists for the same (river_id, measured_at). Returns + * the created row, or undefined when the reading was already stored. Relies on the + * UNIQUE (river_id, measured_at) constraint and keeps the table from accumulating duplicate rows + * when the sensor has not published a newer reading between polls. + */ +export const createRiverLevelIfNew = async (level: CreatableRiverLevel): Promise => { + return database.createOrIgnore( + tableName, + { ...level, created_on: new Date().toISOString() }, + 'river_id, measured_at' + ) +} + +export const getRiverLevelsSince = async (riverId: number, since: string): Promise => { + const query = `SELECT * FROM ${tableName} WHERE river_id = $1 AND measured_at >= $2 ORDER BY measured_at ASC` + + return database.query(query, [riverId, since]) +} + +export interface RiverReading { + value: number + measured_at: string +} + +/** + * Lightweight reading fetch for calibration: only the two columns the model needs, ordered in time. + * Keeps memory low on small hosts when scanning years of history (vs SELECT * full rows). + */ +export const getRiverReadingsSince = async (riverId: number, since: string): Promise => { + const query = `SELECT value, measured_at FROM ${tableName} WHERE river_id = $1 AND measured_at >= $2 ORDER BY measured_at ASC` + + return database.query(query, [riverId, since]) +} + +export type BackfillRiverLevel = Omit + +/** + * Inserts many readings in one statement, skipping any (river_id, measured_at) already present. + * Returns the number of rows actually inserted. Used by the historical backfill; callers must keep + * each batch under the Postgres parameter limit (7 params per row). + */ +export const bulkInsertRiverLevels = async (rows: BackfillRiverLevel[]): Promise => { + if (rows.length === 0) { + return 0 + } + + const now = new Date().toISOString() + const values: (string | number | boolean | null)[] = [] + const tuples = rows.map((row, index) => { + const base = index * 7 + values.push( + row.river_id, + row.value, + row.measured_at, + row.soglia1_above, + row.soglia2_above, + row.soglia3_above, + now + ) + return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5}, $${base + 6}, $${base + 7})` + }) + + const query = `INSERT INTO ${tableName} + (river_id, value, measured_at, soglia1_above, soglia2_above, soglia3_above, created_on) + VALUES ${tuples.join(', ')} + ON CONFLICT (river_id, measured_at) DO NOTHING + RETURNING id` + + const inserted = await database.query<{ id: number }>(query, values) + + return inserted.length +} diff --git a/src/models/river-link.ts b/src/models/river-link.ts new file mode 100644 index 0000000..e75d52d --- /dev/null +++ b/src/models/river-link.ts @@ -0,0 +1,76 @@ +import { database } from '..' + +const tableName = 'river_links' + +export interface RiverLink { + id: number + upstream_river_id: number + downstream_river_id: number + target_threshold: number + lead_time_minutes: number | null + precursor_level: number | null + sample_size: number + model_json: unknown | null + last_calibrated_on: string | null + created_on: string + updated_on: string +} + +export interface CreatableRiverLink { + upstream_river_id: number + downstream_river_id: number + target_threshold?: number +} + +export interface RiverLinkModelPatch { + lead_time_minutes: number | null + precursor_level: number | null + sample_size: number + model_json: unknown +} + +export const getRiverLinks = async (): Promise => { + return database.query(`SELECT * FROM ${tableName} ORDER BY id ASC`) +} + +export const getRiverLinkById = async (id: number): Promise => { + const rows = await database.query(`SELECT * FROM ${tableName} WHERE id = $1`, [id]) + + return rows[0] +} + +export const createRiverLink = async (link: CreatableRiverLink): Promise => { + const now = new Date().toISOString() + + return database.create(tableName, { + upstream_river_id: link.upstream_river_id, + downstream_river_id: link.downstream_river_id, + target_threshold: link.target_threshold ?? 1, + lead_time_minutes: null, + precursor_level: null, + sample_size: 0, + model_json: null, + last_calibrated_on: null, + created_on: now, + updated_on: now, + }) +} + +export const updateRiverLinkModel = async (id: number, patch: RiverLinkModelPatch): Promise => { + return database.edit( + tableName, + { + lead_time_minutes: patch.lead_time_minutes, + precursor_level: patch.precursor_level, + sample_size: patch.sample_size, + model_json: patch.model_json === undefined ? null : JSON.stringify(patch.model_json), + last_calibrated_on: new Date().toISOString(), + updated_on: new Date().toISOString(), + }, + id + ) +} + +export const deleteRiverLink = async (id: number): Promise => { + await database.delete(tableName, id) +} diff --git a/src/models/river.ts b/src/models/river.ts new file mode 100644 index 0000000..ee95d85 --- /dev/null +++ b/src/models/river.ts @@ -0,0 +1,54 @@ +import { database } from '..' + +const tableName = 'rivers' + +export interface River { + id: number + station_id: string + river_name: string + station_name: string + soglia1: number | null + soglia2: number | null + soglia3: number | null + created_on: string + updated_on: string +} + +export type CreatableRiver = Omit +export type RiverPatch = Partial> + +export const getRivers = async (): Promise => { + const query = `SELECT * FROM ${tableName} ORDER BY id ASC` + + return database.query(query) +} + +export const getRiverById = async (id: number): Promise => { + const query = `SELECT * FROM ${tableName} WHERE id = $1` + + const rows = await database.query(query, [id]) + + return rows[0] +} + +export const getRiverByStationId = async (stationId: string): Promise => { + const query = `SELECT * FROM ${tableName} WHERE station_id = $1` + + const rows = await database.query(query, [stationId]) + + return rows[0] +} + +export const createRiver = async (river: CreatableRiver): Promise => { + const now = new Date().toISOString() + + return database.create(tableName, { ...river, created_on: now, updated_on: now }) +} + +export const updateRiver = async (id: number, patch: RiverPatch): Promise => { + return database.edit(tableName, { ...patch, updated_on: new Date().toISOString() }, id) +} + +export const deleteRiver = async (id: number): Promise => { + await database.delete(tableName, id) +} diff --git a/src/routes/river-links.ts b/src/routes/river-links.ts new file mode 100644 index 0000000..3cc3541 --- /dev/null +++ b/src/routes/river-links.ts @@ -0,0 +1,139 @@ +import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify' +import { getRiverById } from '../models/river' +import { createRiverLink, deleteRiverLink, getRiverLinkById, getRiverLinks } from '../models/river-link' +import { runFloodCalibration } from '../tasks/flood-calibration' +import { runFloodPrediction } from '../tasks/flood-prediction' +import { runFloodBackfill } from '../tasks/flood-backfill' +import logger from '../logger' + +const createBodySchema = { + type: 'object', + required: ['upstream_river_id', 'downstream_river_id'], + additionalProperties: false, + properties: { + upstream_river_id: { type: 'integer', minimum: 1 }, + downstream_river_id: { type: 'integer', minimum: 1 }, + target_threshold: { type: 'integer', minimum: 1, maximum: 3 }, + }, +} + +const idParamSchema = { + type: 'object', + required: ['id'], + additionalProperties: false, + properties: { id: { type: 'integer', minimum: 1 } }, +} + +interface IdParams { + id: number +} + +interface CreateLinkBody { + upstream_river_id: number + downstream_river_id: number + target_threshold?: number +} + +const emptyBodySchema = { type: 'object', additionalProperties: false, maxProperties: 0 } + +const monthPattern = '^[0-9]{4}-(0[1-9]|1[0-2])$' +const backfillBodySchema = { + type: 'object', + required: ['from', 'to'], + additionalProperties: false, + properties: { + from: { type: 'string', pattern: monthPattern }, + to: { type: 'string', pattern: monthPattern }, + }, +} + +interface BackfillBody { + from: string + to: string +} + +export const registerRiverLinksRoutes = (fastify: FastifyInstance) => { + fastify.route({ + method: 'GET', + url: '/river-links', + handler: async (_request: FastifyRequest, reply: FastifyReply) => { + reply.status(200).send(await getRiverLinks()) + }, + }) + + fastify.route<{ Body: CreateLinkBody }>({ + method: 'POST', + url: '/river-links', + schema: { body: createBodySchema }, + handler: async (request, reply) => { + const { upstream_river_id, downstream_river_id, target_threshold } = request.body + + if (upstream_river_id === downstream_river_id) { + return reply.status(400).send({ error: 'upstream and downstream must differ' }) + } + + const [upstream, downstream] = await Promise.all([ + getRiverById(upstream_river_id), + getRiverById(downstream_river_id), + ]) + + if (!upstream || !downstream) { + return reply.status(404).send({ error: 'upstream or downstream river not found' }) + } + + const created = await createRiverLink({ upstream_river_id, downstream_river_id, target_threshold }) + reply.status(201).send(created) + }, + }) + + fastify.route<{ Params: IdParams }>({ + method: 'DELETE', + url: '/river-links/:id', + schema: { params: idParamSchema }, + handler: async (request, reply) => { + const link = await getRiverLinkById(request.params.id) + + if (!link) { + return reply.status(404).send({ error: 'River link not found' }) + } + + await deleteRiverLink(request.params.id) + reply.status(204).send(undefined) + }, + }) + + fastify.route({ + method: 'POST', + url: '/flood-calibration', + schema: { body: emptyBodySchema }, + handler: async (_request: FastifyRequest, reply: FastifyReply) => { + reply.status(200).send(await runFloodCalibration()) + }, + }) + + fastify.route({ + method: 'POST', + url: '/flood-prediction', + schema: { body: emptyBodySchema }, + handler: async (_request: FastifyRequest, reply: FastifyReply) => { + reply.status(200).send(await runFloodPrediction()) + }, + }) + + // Backfill runs for minutes (downloads/parses monthly archives), well beyond the request + // timeout, so it is started in the background and the caller gets an immediate acknowledgement. + fastify.route<{ Body: BackfillBody }>({ + method: 'POST', + url: '/flood-backfill', + schema: { body: backfillBodySchema }, + handler: async (request, reply) => { + const { from, to } = request.body + + void runFloodBackfill(from, to) + .then((summary) => logger.info({ summary }, 'Flood backfill finished')) + .catch((err) => logger.error({ err }, 'Flood backfill failed')) + + reply.status(202).send({ status: 'started', from, to }) + }, + }) +} diff --git a/src/routes/rivers.ts b/src/routes/rivers.ts new file mode 100644 index 0000000..a63ad0f --- /dev/null +++ b/src/routes/rivers.ts @@ -0,0 +1,184 @@ +import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify' +import { + createRiver, + deleteRiver, + getRiverById, + getRiverByStationId, + getRivers, + updateRiver, +} from '../models/river' +import { runRiverLevelCheck } from '../tasks/river-levels' +import { getSensorStations } from '../services/river-sensors' +import { findNearestStations, parseSensorStations } from '../utilities/river-stations' + +const nullableNumber = { type: ['number', 'null'] } + +const nearestQuerySchema = { + type: 'object', + required: ['lat', 'lon'], + additionalProperties: false, + properties: { + lat: { type: 'number', minimum: -90, maximum: 90 }, + lon: { type: 'number', minimum: -180, maximum: 180 }, + limit: { type: 'integer', minimum: 1, maximum: 50 }, + }, +} + +interface NearestQuery { + lat: number + lon: number + limit?: number +} + +const createBodySchema = { + type: 'object', + required: ['station_id', 'river_name', 'station_name'], + additionalProperties: false, + properties: { + station_id: { type: 'string', minLength: 1, maxLength: 64 }, + river_name: { type: 'string', minLength: 1, maxLength: 128 }, + station_name: { type: 'string', minLength: 1, maxLength: 128 }, + soglia1: nullableNumber, + soglia2: nullableNumber, + soglia3: nullableNumber, + }, +} + +const patchBodySchema = { + type: 'object', + additionalProperties: false, + minProperties: 1, + properties: { + river_name: { type: 'string', minLength: 1, maxLength: 128 }, + station_name: { type: 'string', minLength: 1, maxLength: 128 }, + soglia1: nullableNumber, + soglia2: nullableNumber, + soglia3: nullableNumber, + }, +} + +const idParamSchema = { + type: 'object', + required: ['id'], + additionalProperties: false, + properties: { + id: { type: 'integer', minimum: 1 }, + }, +} + +interface IdParams { + id: number +} + +interface CreateRiverBody { + station_id: string + river_name: string + station_name: string + soglia1?: number | null + soglia2?: number | null + soglia3?: number | null +} + +interface PatchRiverBody { + river_name?: string + station_name?: string + soglia1?: number | null + soglia2?: number | null + soglia3?: number | null +} + +export const registerRiversRoutes = (fastify: FastifyInstance) => { + fastify.route({ + method: 'GET', + url: '/rivers', + handler: async (_request: FastifyRequest, reply: FastifyReply) => { + const rivers = await getRivers() + reply.status(200).send(rivers) + }, + }) + + fastify.route<{ Body: CreateRiverBody }>({ + method: 'POST', + url: '/rivers', + schema: { body: createBodySchema }, + handler: async (request, reply) => { + const existing = await getRiverByStationId(request.body.station_id) + + if (existing) { + return reply.status(409).send({ error: 'Station already registered' }) + } + + const created = await createRiver({ + station_id: request.body.station_id, + river_name: request.body.river_name, + station_name: request.body.station_name, + soglia1: request.body.soglia1 ?? null, + soglia2: request.body.soglia2 ?? null, + soglia3: request.body.soglia3 ?? null, + }) + + reply.status(201).send(created) + }, + }) + + fastify.route<{ Params: IdParams; Body: PatchRiverBody }>({ + method: 'PATCH', + url: '/rivers/:id', + schema: { params: idParamSchema, body: patchBodySchema }, + handler: async (request, reply) => { + const river = await getRiverById(request.params.id) + + if (!river) { + return reply.status(404).send({ error: 'River not found' }) + } + + const updated = await updateRiver(request.params.id, request.body) + reply.status(200).send(updated) + }, + }) + + fastify.route<{ Params: IdParams }>({ + method: 'DELETE', + url: '/rivers/:id', + schema: { params: idParamSchema }, + handler: async (request, reply) => { + const river = await getRiverById(request.params.id) + + if (!river) { + return reply.status(404).send({ error: 'River not found' }) + } + + await deleteRiver(request.params.id) + reply.status(204).send(undefined) + }, + }) + + fastify.route<{ Querystring: NearestQuery }>({ + method: 'GET', + url: '/rivers/nearest', + schema: { querystring: nearestQuerySchema }, + handler: async (request, reply) => { + const stations = await getSensorStations() + const parsed = parseSensorStations(stations) + const nearest = findNearestStations( + parsed, + { lat: request.query.lat, lon: request.query.lon }, + request.query.limit ?? 5 + ) + + reply.status(200).send(nearest) + }, + }) + + fastify.route({ + method: 'POST', + url: '/river-levels', + schema: { + body: { type: 'object', additionalProperties: false, maxProperties: 0 }, + }, + handler: async (_request: FastifyRequest, reply: FastifyReply) => { + const summary = await runRiverLevelCheck() + reply.status(200).send(summary) + }, + }) +} diff --git a/src/scheduler.ts b/src/scheduler.ts index ca862e0..81b4cee 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -2,6 +2,9 @@ import { Cron } from 'croner' import { runMeteoAlertCheck } from './tasks/meteo-alerts' import { runPretempCheck } from './tasks/pretemp' import { runEstofexCheck } from './tasks/estofex' +import { runRiverLevelCheck } from './tasks/river-levels' +import { runFloodPrediction } from './tasks/flood-prediction' +import { runFloodCalibration } from './tasks/flood-calibration' import logger from './logger' const jobs: Cron[] = [] @@ -35,6 +38,10 @@ export const startScheduler = () => { schedule('meteo-alerts', '*/5 * * * *', runMeteoAlertCheck) schedule('pretemp', '*/5 * * * *', runPretempCheck) schedule('estofex', '*/5 * * * *', runEstofexCheck) + schedule('river-levels', '*/5 * * * *', runRiverLevelCheck) + schedule('flood-prediction', '*/5 * * * *', runFloodPrediction) + // Re-learn link models daily (cheap, reads accumulated history); also run on demand after a backfill. + schedule('flood-calibration', '15 3 * * *', runFloodCalibration) logger.info({ count: jobs.length }, 'Scheduler started') } diff --git a/src/server.ts b/src/server.ts index bc01335..20292e8 100644 --- a/src/server.ts +++ b/src/server.ts @@ -4,6 +4,8 @@ import i18next from 'i18next' import italian from './resources/locales/it.json' import { registerTestMessageRoutes } from './routes/test-message' import { registerForecastReportsRoutes } from './routes/forecast-reports' +import { registerRiversRoutes } from './routes/rivers' +import { registerRiverLinksRoutes } from './routes/river-links' import logger from './logger' const translations = { @@ -45,6 +47,8 @@ export const startServer = async (): Promise => { registerTestMessageRoutes(app) registerMeteoAlertsRoutes(app) registerForecastReportsRoutes(app) + registerRiversRoutes(app) + registerRiverLinksRoutes(app) await fastify.listen({ host: '127.0.0.1', diff --git a/src/services/arpae-archive.ts b/src/services/arpae-archive.ts new file mode 100644 index 0000000..62e5a81 --- /dev/null +++ b/src/services/arpae-archive.ts @@ -0,0 +1,129 @@ +import zlib from 'zlib' +import readline from 'readline' +import { http } from './http' +import logger from '../logger' + +// ARPAE-SIMC open-data historical archive: one gzipped JSONL file per month, each line a station's +// observation at one timestamp. Hydrometric level is the measurement block with timerange[0] === 254 +// carrying B13215 — the same signature as the live Allerta variable 254,0,0/1,-,-,-/B13215. +const ARCHIVE_BASE_URL = 'https://dati-simc.arpae.it/opendata/osservati/meteo/storico' + +const HYDRO_TIMERANGE = 254 +const LEVEL_VARIABLE = 'B13215' + +export interface ArchiveTarget { + riverId: number + soglia1: number | null + soglia2: number | null + soglia3: number | null +} + +export interface ParsedArchiveReading { + river_id: number + value: number + measured_at: string + soglia1_above: boolean | null + soglia2_above: boolean | null + soglia3_above: boolean | null +} + +interface ArchiveBlock { + timerange?: number[] + vars?: Record +} + +interface ArchiveLine { + lat?: number + lon?: number + date?: string + data?: ArchiveBlock[] +} + +// Stations are matched on the integer (lat, lon) coordinates, which the archive and the Allerta +// sensor list share (degrees × 1e5). +export const coordKey = (lat: number, lon: number): string => `${lat}:${lon}` + +const aboveOrNull = (value: number, threshold: number | null): boolean | null => + threshold == null ? null : value > threshold + +/** + * Parses one archive JSONL line into a river_levels reading, or null when the line is not one of the + * target stations, carries no hydrometric level, or is malformed. Pure — unit-testable without IO. + */ +export const parseArchiveLine = (line: string, index: Map): ParsedArchiveReading | null => { + let parsed: ArchiveLine + try { + parsed = JSON.parse(line) + } catch { + return null + } + + if (typeof parsed.lat !== 'number' || typeof parsed.lon !== 'number' || typeof parsed.date !== 'string') { + return null + } + + const target = index.get(coordKey(parsed.lat, parsed.lon)) + if (!target || !Array.isArray(parsed.data)) { + return null + } + + const block = parsed.data.find( + (b) => Array.isArray(b.timerange) && b.timerange[0] === HYDRO_TIMERANGE && b.vars?.[LEVEL_VARIABLE] != null + ) + const raw = block?.vars?.[LEVEL_VARIABLE]?.v + const value = typeof raw === 'number' ? raw : Number(raw) + + if (raw == null || !Number.isFinite(value)) { + return null + } + + return { + river_id: target.riverId, + value, + measured_at: new Date(parsed.date).toISOString(), + soglia1_above: aboveOrNull(value, target.soglia1), + soglia2_above: aboveOrNull(value, target.soglia2), + soglia3_above: aboveOrNull(value, target.soglia3), + } +} + +export const archiveUrlForMonth = (year: number, month: number): string => + `${ARCHIVE_BASE_URL}/${year}-${String(month).padStart(2, '0')}.json.gz` + +/** + * Streams one monthly archive file, returning the readings that match the target stations. Holds + * only the matched rows in memory (a few stations × ~2,976 readings/month), parsing line by line. + */ +export const fetchArchiveMonth = async ( + year: number, + month: number, + index: Map +): Promise => { + const url = archiveUrlForMonth(year, month) + const readings: ParsedArchiveReading[] = [] + + // Archive files are ~20 MB; override the default 15s client timeout so the download/stream has + // room to complete on a slow connection. + const response = await http.get(url, { responseType: 'stream', timeout: 180_000 }) + const lines = readline.createInterface({ + input: (response.data as NodeJS.ReadableStream).pipe(zlib.createGunzip()), + crlfDelay: Infinity, + }) + + try { + for await (const line of lines) { + if (line.length === 0) { + continue + } + const reading = parseArchiveLine(line, index) + if (reading) { + readings.push(reading) + } + } + } catch (error) { + logger.error({ err: error, url }, 'Failed while streaming ARPAE archive month') + throw error + } + + return readings +} diff --git a/src/services/river-sensors.ts b/src/services/river-sensors.ts new file mode 100644 index 0000000..462e1d4 --- /dev/null +++ b/src/services/river-sensors.ts @@ -0,0 +1,79 @@ +import { http } from './http' +import logger from '../logger' + +export interface RiverSensorTimePoint { + t: number + v: number +} + +export interface RawSensorStation { + idstazione: string + nomestaz: string + lon: string + lat: string + value: number | null + soglia1: number | null + soglia2: number | null + soglia3: number | null +} + +const VARIABILE = '254,0,0/1,-,-,-/B13215' +const TIME_SERIES_BASE_URL = 'https://allertameteo.regione.emilia-romagna.it/o/api/allerta/get-time-series/' +const SENSOR_VALUES_BASE_URL = + 'https://allertameteo.regione.emilia-romagna.it/o/api/allerta/get-sensor-values-no-time' + +export const getRiverSensorTimeSeries = async (stationId: string): Promise => { + const url = `${TIME_SERIES_BASE_URL}?stazione=${encodeURIComponent(stationId)}&variabile=${VARIABILE}` + + try { + const response = await http.get(url).then((res) => res.data) + + if (!Array.isArray(response)) { + return [] + } + + return response.filter( + (point): point is RiverSensorTimePoint => + point != null && typeof point.t === 'number' && typeof point.v === 'number' + ) + } catch (error) { + logger.error({ err: error, stationId }, 'Failed to retrieve river sensor time series') + throw error + } +} + +export const getLatestRiverSensorValue = async (stationId: string): Promise => { + const series = await getRiverSensorTimeSeries(stationId) + + if (series.length === 0) { + return undefined + } + + return series.reduce((latest, point) => (point.t > latest.t ? point : latest), series[0]) +} + +/** + * Fetches the full station list (one call, all hydrometric stations) used for discovery and + * nearest-station lookup. Note: the per-station `value` is unreliable (often null) for this + * variable, so this endpoint is NOT a substitute for the time-series readings — it is metadata + * (name, coordinates, official thresholds) only. The leading `{ time }` element is dropped. + */ +export const getSensorStations = async (): Promise => { + const url = `${SENSOR_VALUES_BASE_URL}?variabile=${VARIABILE}` + + try { + const response = await http.get(url).then((res) => res.data) + + if (!Array.isArray(response)) { + return [] + } + + return response.filter( + (entry): entry is RawSensorStation => + entry != null && typeof (entry as RawSensorStation).idstazione === 'string' + ) + } catch (error) { + logger.error({ err: error }, 'Failed to retrieve river sensor station list') + throw error + } +} diff --git a/src/tasks/flood-backfill.ts b/src/tasks/flood-backfill.ts new file mode 100644 index 0000000..f365fbf --- /dev/null +++ b/src/tasks/flood-backfill.ts @@ -0,0 +1,118 @@ +import { getRivers } from '../models/river' +import { bulkInsertRiverLevels } from '../models/river-level' +import { getSensorStations } from '../services/river-sensors' +import { ArchiveTarget, coordKey, fetchArchiveMonth } from '../services/arpae-archive' +import logger from '../logger' + +const log = logger.child({ task: 'flood-backfill' }) + +const INSERT_BATCH_SIZE = 5000 + +export interface FloodBackfillSummary { + months: number + fetched: number + inserted: number + targets: number +} + +interface YearMonth { + year: number + month: number +} + +const toNumberOrNull = (value: number | null): number | null => { + if (value == null) { + return null + } + const n = Number(value) + return Number.isFinite(n) ? n : null +} + +export const expandMonthRange = (from: string, to: string): YearMonth[] => { + const [fy, fm] = from.split('-').map(Number) + const [ty, tm] = to.split('-').map(Number) + + if (![fy, fm, ty, tm].every(Number.isFinite) || fm < 1 || fm > 12 || tm < 1 || tm > 12) { + throw new Error(`Invalid month range: ${from}..${to}`) + } + + const months: YearMonth[] = [] + let year = fy + let month = fm + while (year < ty || (year === ty && month <= tm)) { + months.push({ year, month }) + month += 1 + if (month > 12) { + month = 1 + year += 1 + } + } + + return months +} + +/** + * Builds a coordinate -> river index from the registered rivers and the live Allerta sensor list. + * Both the archive and the sensor list encode coordinates as the same degrees×1e5 integers, so the + * raw integer strings are used as the join key. + */ +const buildCoordIndex = async (): Promise> => { + const [rivers, stations] = await Promise.all([getRivers(), getSensorStations()]) + const stationCoords = new Map(stations.map((s) => [s.idstazione, { lat: Number(s.lat), lon: Number(s.lon) }])) + + const index = new Map() + for (const river of rivers) { + const coords = stationCoords.get(river.station_id) + if (!coords || !Number.isFinite(coords.lat) || !Number.isFinite(coords.lon)) { + log.warn({ riverId: river.id, stationId: river.station_id }, 'No archive coordinates for station; skipping') + continue + } + + index.set(coordKey(coords.lat, coords.lon), { + riverId: river.id, + soglia1: toNumberOrNull(river.soglia1), + soglia2: toNumberOrNull(river.soglia2), + soglia3: toNumberOrNull(river.soglia3), + }) + } + + return index +} + +const insertInBatches = async (readings: Parameters[0]): Promise => { + let inserted = 0 + for (let i = 0; i < readings.length; i += INSERT_BATCH_SIZE) { + inserted += await bulkInsertRiverLevels(readings.slice(i, i + INSERT_BATCH_SIZE)) + } + return inserted +} + +/** + * Backfills river_levels from the ARPAE historical archive for every registered station over the + * inclusive month range (e.g. "2020-01".."2024-12"). Idempotent: existing (river_id, measured_at) + * rows are skipped, so it is safe to re-run. Run on demand, then trigger calibration afterwards. + */ +export const runFloodBackfill = async (from: string, to: string): Promise => { + const months = expandMonthRange(from, to) + const index = await buildCoordIndex() + const summary: FloodBackfillSummary = { months: months.length, fetched: 0, inserted: 0, targets: index.size } + + if (index.size === 0) { + log.warn('No target stations resolved; nothing to backfill') + return summary + } + + for (const { year, month } of months) { + try { + const readings = await fetchArchiveMonth(year, month, index) + const inserted = await insertInBatches(readings) + summary.fetched += readings.length + summary.inserted += inserted + log.info({ year, month, fetched: readings.length, inserted }, 'Backfilled archive month') + } catch (err) { + log.error({ err, year, month }, 'Failed to backfill archive month; continuing') + } + } + + return summary +} diff --git a/src/tasks/flood-calibration.ts b/src/tasks/flood-calibration.ts new file mode 100644 index 0000000..a577c57 --- /dev/null +++ b/src/tasks/flood-calibration.ts @@ -0,0 +1,87 @@ +import { getRiverById } from '../models/river' +import { getRiverReadingsSince } from '../models/river-level' +import { getRiverLinks, updateRiverLinkModel } from '../models/river-link' +import { calibrateLink, isLinkActive, toReadings } from '../utilities/flood-prediction' +import logger from '../logger' + +const log = logger.child({ task: 'flood-calibration' }) + +// Bound how far back calibration reads, so memory stays modest on small hosts and the model +// reflects recent channel behaviour (~6 years still spans many flood seasons). +const CALIBRATION_LOOKBACK_DAYS = 6 * 365 +const DAY_MS = 24 * 60 * 60 * 1000 + +export interface FloodCalibrationSummary { + calibrated: number + active: number + skipped: number +} + +const thresholdFor = ( + river: { soglia1: number | null; soglia2: number | null; soglia3: number | null }, + target: number +): number | null => { + const raw = target === 1 ? river.soglia1 : target === 2 ? river.soglia2 : target === 3 ? river.soglia3 : null + + return raw == null ? null : Number(raw) +} + +/** + * Re-learns every link's model from the readings accumulated in river_levels. Safe to run + * repeatedly; links without enough historical events simply stay inactive (sample_size below the + * minimum). Intended to run on a slow schedule (e.g. daily) and on demand after a backfill. + */ +export const runFloodCalibration = async (): Promise => { + const links = await getRiverLinks() + const summary: FloodCalibrationSummary = { calibrated: 0, active: 0, skipped: 0 } + const since = new Date(Date.now() - CALIBRATION_LOOKBACK_DAYS * DAY_MS).toISOString() + + for (const link of links) { + try { + const downstream = await getRiverById(link.downstream_river_id) + const upstream = await getRiverById(link.upstream_river_id) + + if (!downstream || !upstream) { + summary.skipped += 1 + continue + } + + const threshold = thresholdFor(downstream, link.target_threshold) + + if (threshold == null) { + log.warn({ linkId: link.id, downstreamId: downstream.id }, 'Downstream soglia not set; skipping link') + summary.skipped += 1 + continue + } + + const [downstreamRows, upstreamRows] = await Promise.all([ + getRiverReadingsSince(link.downstream_river_id, since), + getRiverReadingsSince(link.upstream_river_id, since), + ]) + + const model = calibrateLink(toReadings(upstreamRows), toReadings(downstreamRows), threshold) + + await updateRiverLinkModel(link.id, { + lead_time_minutes: Number.isFinite(model.leadTimeMinutes) ? Math.round(model.leadTimeMinutes) : null, + precursor_level: Number.isFinite(model.precursorLevel) ? model.precursorLevel : null, + sample_size: model.sampleSize, + model_json: model, + }) + + summary.calibrated += 1 + if (isLinkActive(model)) { + summary.active += 1 + } + + log.info( + { linkId: link.id, sampleSize: model.sampleSize, active: isLinkActive(model) }, + 'Calibrated flood link' + ) + } catch (err) { + log.error({ err, linkId: link.id }, 'Failed to calibrate flood link; continuing') + summary.skipped += 1 + } + } + + return summary +} diff --git a/src/tasks/flood-prediction.ts b/src/tasks/flood-prediction.ts new file mode 100644 index 0000000..5c5b082 --- /dev/null +++ b/src/tasks/flood-prediction.ts @@ -0,0 +1,98 @@ +import { getRiverById, River } from '../models/river' +import { getRiverLevelsSince } from '../models/river-level' +import { getRiverLinks } from '../models/river-link' +import { createLinkPrediction, getLatestLinkPrediction } from '../models/link-prediction' +import { LinkModel, predict, toReadings } from '../utilities/flood-prediction' +import { sendFloodPredictionMessage } from '../utilities/telegram' +import logger from '../logger' + +const log = logger.child({ task: 'flood-prediction' }) + +const RECENT_WINDOW_MINUTES = 6 * 60 +const MIN_DEDUP_MINUTES = 180 +const MINUTE_MS = 60_000 + +export interface FloodPredictionSummary { + evaluated: number + predictions: number + skipped: number +} + +const modelFromLink = (link: { + lead_time_minutes: number | null + precursor_level: number | null + sample_size: number +}): LinkModel => ({ + leadTimeMinutes: link.lead_time_minutes ?? NaN, + precursorLevel: link.precursor_level == null ? NaN : Number(link.precursor_level), + sampleSize: link.sample_size, + leadSpreadMinutes: NaN, +}) + +/** + * Online step (runs on the fast cron): for each calibrated link, checks whether the upstream gauge + * has reached the learned precursor level while rising and, if so, emits a single flood-arrival + * prediction. De-duplicates so a sustained high upstream level does not re-fire every poll. + */ +export const runFloodPrediction = async (): Promise => { + const links = await getRiverLinks() + const summary: FloodPredictionSummary = { evaluated: 0, predictions: 0, skipped: 0 } + const nowMs = Date.now() + const since = new Date(nowMs - RECENT_WINDOW_MINUTES * MINUTE_MS).toISOString() + + for (const link of links) { + summary.evaluated += 1 + + // Isolate per-link failures (a DB read for one link must not abort the whole run). + try { + const model = modelFromLink(link) + const upstreamRows = await getRiverLevelsSince(link.upstream_river_id, since) + const prediction = predict(toReadings(upstreamRows), model, nowMs) + + if (!prediction) { + continue + } + + // Suppress repeats while the upstream stays high: one warning per lead-time-sized window. + const dedupMinutes = Math.max(model.leadTimeMinutes, MIN_DEDUP_MINUTES) + const previous = await getLatestLinkPrediction(link.id) + if (previous && nowMs - new Date(previous.predicted_at).getTime() < dedupMinutes * MINUTE_MS) { + summary.skipped += 1 + continue + } + + const [downstream, upstream] = await Promise.all([ + getRiverById(link.downstream_river_id), + getRiverById(link.upstream_river_id), + ]) + + if (!downstream || !upstream) { + summary.skipped += 1 + continue + } + + // Send first, persist on success: a transient Telegram failure then retries on the next + // poll instead of being permanently suppressed by the de-dup check above. + await sendFloodPredictionMessage(downstream as River, upstream as River, { + upstreamValue: prediction.upstreamValue, + leadTimeMinutes: prediction.leadTimeMinutes, + predictedExceedanceAt: prediction.predictedExceedanceAt, + targetThreshold: link.target_threshold, + }) + + await createLinkPrediction({ + link_id: link.id, + predicted_at: new Date(nowMs).toISOString(), + predicted_exceedance_at: new Date(prediction.predictedExceedanceAt).toISOString(), + upstream_value: prediction.upstreamValue, + }) + + summary.predictions += 1 + } catch (err) { + log.error({ err, linkId: link.id }, 'Failed to evaluate flood link; continuing') + summary.skipped += 1 + } + } + + return summary +} diff --git a/src/tasks/river-levels.ts b/src/tasks/river-levels.ts new file mode 100644 index 0000000..2c68b6d --- /dev/null +++ b/src/tasks/river-levels.ts @@ -0,0 +1,100 @@ +import { getRivers } from '../models/river' +import { createRiverLevelIfNew, getLatestRiverLevel } from '../models/river-level' +import { getLatestRiverSensorValue } from '../services/river-sensors' +import { detectThresholdCrossings, getActiveThresholds, ThresholdBooleans } from '../utilities/river-sensors' +import { sendRiverLevelCrossingMessage } from '../utilities/telegram' +import { config } from '../config/config' +import logger from '../logger' + +const log = logger.child({ task: 'river-levels' }) + +export interface RiverLevelCheckSummary { + checked: number + crossings: number + skipped: number + unchanged: number +} + +export const runRiverLevelCheck = async (): Promise => { + const rivers = await getRivers() + const summary: RiverLevelCheckSummary = { checked: 0, crossings: 0, skipped: 0, unchanged: 0 } + + if (rivers.length === 0) { + return summary + } + + for (const river of rivers) { + // Isolate per-river failures (a sensor fetch or DB error for one station must not abort the + // whole run); the scheduler would otherwise only retry on the next tick. + try { + const latest = await getLatestRiverSensorValue(river.station_id) + + if (!latest) { + summary.skipped += 1 + continue + } + + const previous = await getLatestRiverLevel(river.id) + + // The reading window is polled more frequently than the sensor publishes, so skip when + // the latest reading is not newer than the one already stored — avoids duplicate rows + // and redundant crossing evaluation. + if (previous && latest.t <= new Date(previous.measured_at).getTime()) { + summary.unchanged += 1 + continue + } + + const currentValue = Number(latest.v) + const thresholds = getActiveThresholds(river) + const previousState: ThresholdBooleans | undefined = previous + ? { + soglia1_above: previous.soglia1_above, + soglia2_above: previous.soglia2_above, + soglia3_above: previous.soglia3_above, + } + : undefined + + const { crossings, nextState } = detectThresholdCrossings( + currentValue, + thresholds, + previousState, + config.river_threshold_margin + ) + + const inserted = await createRiverLevelIfNew({ + river_id: river.id, + value: currentValue, + measured_at: new Date(latest.t).toISOString(), + soglia1_above: nextState.soglia1_above, + soglia2_above: nextState.soglia2_above, + soglia3_above: nextState.soglia3_above, + }) + + // A concurrent run may have already persisted this reading; only alert for the run that + // actually stored it, so a crossing is never announced twice. + if (!inserted) { + summary.unchanged += 1 + continue + } + + for (const crossing of crossings) { + try { + await sendRiverLevelCrossingMessage(river, crossing, currentValue) + summary.crossings += 1 + } catch (err) { + log.error( + { err, riverId: river.id, stationId: river.station_id, threshold: crossing.threshold.key }, + 'Failed to send river level crossing Telegram message' + ) + } + } + + summary.checked += 1 + } catch (err) { + log.error({ err, riverId: river.id, stationId: river.station_id }, 'Failed to process river; continuing') + summary.skipped += 1 + } + } + + return summary +} diff --git a/src/utilities/flood-prediction.ts b/src/utilities/flood-prediction.ts new file mode 100644 index 0000000..13222b4 --- /dev/null +++ b/src/utilities/flood-prediction.ts @@ -0,0 +1,234 @@ +/** + * Empirical "precursor-of-exceedance" flood model. + * + * For a downstream point of interest we learn, purely from history, the upstream-gauge level + * (`precursorLevel`) and the time offset (`leadTimeMinutes`) that preceded a threshold exceedance + * downstream. Online, when the upstream gauge reaches that learned level while rising, we warn + * that the downstream point is likely to exceed its threshold roughly one lead time later. + * + * No AI/regression — just event detection plus robust statistics over past events. The model is + * deliberately conservative on `precursorLevel` (a low percentile of pre-event upstream peaks) so + * it tends to catch events at the cost of some false positives. The lead time is measured from the + * upstream peak to the downstream onset, so the online prediction (triggered while the upstream is + * still rising) is slightly early-biased; the prediction-scoring feedback loop corrects this drift. + */ + +export interface Reading { + measuredAt: number // epoch ms + value: number +} + +export interface ExceedanceEvent { + onsetAt: number + peakAt: number + peakValue: number +} + +export interface PrecursorObservation { + leadMinutes: number + upstreamPeak: number +} + +export interface LinkModel { + leadTimeMinutes: number + precursorLevel: number + sampleSize: number + leadSpreadMinutes: number +} + +export interface Prediction { + predictedExceedanceAt: number // epoch ms + leadTimeMinutes: number + upstreamValue: number +} + +export interface FloodModelOptions { + maxLookbackMinutes: number + minLeadMinutes: number + minSamples: number + precursorPercentile: number + minSeparationMinutes: number +} + +export const DEFAULT_FLOOD_OPTIONS: FloodModelOptions = { + maxLookbackMinutes: 48 * 60, + minLeadMinutes: 15, + minSamples: 3, + precursorPercentile: 0.25, + // Treat brief dips back under the threshold as part of the same flood, so one flood is one + // event (not many micro-events when the level hugs the threshold). + minSeparationMinutes: 6 * 60, +} + +const MINUTE_MS = 60_000 + +const sortByTime = (readings: Reading[]): Reading[] => [...readings].sort((a, b) => a.measuredAt - b.measuredAt) + +export const median = (values: number[]): number => percentile(values, 0.5) + +export const percentile = (values: number[], p: number): number => { + if (values.length === 0) { + return NaN + } + + const sorted = [...values].sort((a, b) => a - b) + const rank = Math.min(Math.max(p, 0), 1) * (sorted.length - 1) + const low = Math.floor(rank) + const high = Math.ceil(rank) + + if (low === high) { + return sorted[low] + } + + return sorted[low] + (sorted[high] - sorted[low]) * (rank - low) +} + +/** + * Splits a level series into maximal runs strictly above `threshold`. Each run yields one event + * with its onset (first sample above), peak time and peak value. + */ +export const detectExceedanceEvents = ( + series: Reading[], + threshold: number, + minSeparationMinutes = 0 +): ExceedanceEvent[] => { + const sorted = sortByTime(series) + const separationMs = minSeparationMinutes * MINUTE_MS + const events: ExceedanceEvent[] = [] + let current: ExceedanceEvent | null = null + let belowSince: number | null = null + + for (const point of sorted) { + if (point.value > threshold) { + belowSince = null + if (!current) { + current = { onsetAt: point.measuredAt, peakAt: point.measuredAt, peakValue: point.value } + } else if (point.value > current.peakValue) { + current.peakValue = point.value + current.peakAt = point.measuredAt + } + } else if (current) { + // Only close the event once the level has stayed below the threshold for at least the + // separation window; shorter dips are absorbed into the same flood. + if (belowSince === null) { + belowSince = point.measuredAt + } + if (point.measuredAt - belowSince >= separationMs) { + events.push(current) + current = null + belowSince = null + } + } + } + + if (current) { + events.push(current) + } + + return events +} + +/** + * Finds the upstream peak in the lookback window before a downstream event and the lead time from + * that peak to the event onset. Returns null when there is no usable upstream data in the window + * or the lead is below the minimum. + */ +export const findPrecursor = ( + upstream: Reading[], + event: ExceedanceEvent, + options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS +): PrecursorObservation | null => { + const windowStart = event.onsetAt - options.maxLookbackMinutes * MINUTE_MS + const candidates = upstream.filter((p) => p.measuredAt >= windowStart && p.measuredAt <= event.onsetAt) + + if (candidates.length === 0) { + return null + } + + const peak = candidates.reduce((best, p) => (p.value > best.value ? p : best), candidates[0]) + const leadMinutes = (event.onsetAt - peak.measuredAt) / MINUTE_MS + + if (leadMinutes < options.minLeadMinutes) { + return null + } + + return { leadMinutes, upstreamPeak: peak.value } +} + +/** + * Learns a link model from upstream readings and the downstream level series. `sampleSize` is the + * number of usable historical events; the model is only meaningful once it reaches `minSamples` + * (see {@link isLinkActive}). + */ +export const calibrateLink = ( + upstream: Reading[], + downstream: Reading[], + threshold: number, + options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS +): LinkModel => { + const events = detectExceedanceEvents(downstream, threshold, options.minSeparationMinutes) + const observations = events + .map((event) => findPrecursor(upstream, event, options)) + .filter((obs): obs is PrecursorObservation => obs !== null) + + const leads = observations.map((o) => o.leadMinutes) + const peaks = observations.map((o) => o.upstreamPeak) + + return { + sampleSize: observations.length, + leadTimeMinutes: observations.length > 0 ? median(leads) : NaN, + precursorLevel: observations.length > 0 ? percentile(peaks, options.precursorPercentile) : NaN, + leadSpreadMinutes: + observations.length > 0 ? percentile(leads, 0.75) - percentile(leads, 0.25) : NaN, + } +} + +export const isLinkActive = (model: LinkModel, options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS): boolean => + model.sampleSize >= options.minSamples && + Number.isFinite(model.leadTimeMinutes) && + Number.isFinite(model.precursorLevel) + +const isRising = (readings: Reading[]): boolean => { + if (readings.length < 2) { + // A single reading cannot establish an upward trend (e.g. first poll after a cold start or + // sensor outage), so do not fire a prediction from it. + return false + } + + return readings[readings.length - 1].value > readings[readings.length - 2].value +} + +/** + * Converts stored river_levels rows into the time/value Reading shape used by this module. Accepts + * the value as string or number because node-postgres returns NUMERIC columns as strings. + */ +export const toReadings = (rows: { value: number | string; measured_at: string }[]): Reading[] => + rows.map((row) => ({ value: Number(row.value), measuredAt: new Date(row.measured_at).getTime() })) + +/** + * Emits a prediction when the most recent upstream reading has reached the learned precursor level + * while rising. Returns null otherwise (model inactive, below precursor, or receding). + */ +export const predict = ( + upstreamReadings: Reading[], + model: LinkModel, + nowMs: number, + options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS +): Prediction | null => { + if (!isLinkActive(model, options) || upstreamReadings.length === 0) { + return null + } + + const sorted = sortByTime(upstreamReadings) + const latest = sorted[sorted.length - 1] + + if (latest.value < model.precursorLevel || !isRising(sorted)) { + return null + } + + return { + predictedExceedanceAt: nowMs + model.leadTimeMinutes * MINUTE_MS, + leadTimeMinutes: model.leadTimeMinutes, + upstreamValue: latest.value, + } +} diff --git a/src/utilities/river-sensors.ts b/src/utilities/river-sensors.ts new file mode 100644 index 0000000..5e1f76f --- /dev/null +++ b/src/utilities/river-sensors.ts @@ -0,0 +1,88 @@ +import { River } from '../models/river' + +export type ThresholdKey = 'soglia1' | 'soglia2' | 'soglia3' + +export interface ActiveThreshold { + key: ThresholdKey + value: number +} + +export interface ThresholdCrossing { + threshold: ActiveThreshold + direction: 'above' | 'below' +} + +export interface ThresholdBooleans { + soglia1_above: boolean | null + soglia2_above: boolean | null + soglia3_above: boolean | null +} + +export interface CrossingDetectionResult { + crossings: ThresholdCrossing[] + nextState: ThresholdBooleans +} + +const THRESHOLD_KEYS: ThresholdKey[] = ['soglia1', 'soglia2', 'soglia3'] + +export const getActiveThresholds = (river: River): ActiveThreshold[] => { + return THRESHOLD_KEYS.reduce((acc, key) => { + const raw = river[key] + const value = raw == null ? null : Number(raw) + + if (value != null && Number.isFinite(value)) { + acc.push({ key, value }) + } + + return acc + }, []) +} + +/** + * Detects threshold crossings with an optional hysteresis `margin` (deadband) to avoid alert + * flapping when a reading hovers around a threshold. With the default margin of 0 this reduces + * exactly to a strict `value > threshold` comparison. + * + * Schmitt-trigger semantics once a previous side is known: + * - while above, the level must drop to `threshold - margin` before it counts as below; + * - while below, the level must exceed `threshold + margin` before it counts as above. + * The first observation (no previous side) is seeded with a plain comparison and emits no crossing. + */ +export const detectThresholdCrossings = ( + currentValue: number, + thresholds: ActiveThreshold[], + previousState: ThresholdBooleans | undefined, + margin = 0 +): CrossingDetectionResult => { + const nextState: ThresholdBooleans = { + soglia1_above: null, + soglia2_above: null, + soglia3_above: null, + } + const crossings: ThresholdCrossing[] = [] + + for (const threshold of thresholds) { + const aboveKey = `${threshold.key}_above` as keyof ThresholdBooleans + const previousAbove = previousState?.[aboveKey] ?? null + + let currentAbove: boolean + if (previousAbove === null) { + currentAbove = currentValue > threshold.value + } else if (previousAbove) { + currentAbove = currentValue > threshold.value - margin + } else { + currentAbove = currentValue > threshold.value + margin + } + + nextState[aboveKey] = currentAbove + + if (previousState && previousAbove !== null && previousAbove !== currentAbove) { + crossings.push({ + threshold, + direction: currentAbove ? 'above' : 'below', + }) + } + } + + return { crossings, nextState } +} diff --git a/src/utilities/river-stations.ts b/src/utilities/river-stations.ts new file mode 100644 index 0000000..732f7fd --- /dev/null +++ b/src/utilities/river-stations.ts @@ -0,0 +1,91 @@ +import { RawSensorStation } from '../services/river-sensors' + +export interface SensorStation { + stationId: string + name: string + lat: number + lon: number + soglia1: number | null + soglia2: number | null + soglia3: number | null +} + +export interface Coordinates { + lat: number + lon: number +} + +export interface NearestStation extends SensorStation { + distanceKm: number +} + +// The Allerta Meteo sensor list encodes coordinates as integer strings scaled by 1e5 +// (e.g. "4457480" -> 44.57480, "1170609" -> 11.70609). +const COORDINATE_SCALE = 100000 + +const parseCoordinate = (raw: string): number | null => { + const value = Number(raw) + + return Number.isFinite(value) ? value / COORDINATE_SCALE : null +} + +// A soglia of 0 means "no official threshold" for this variable (e.g. tide/coastal stations), +// so it is normalised to null rather than treated as a real 0 m threshold. +const parseThreshold = (raw: number | null): number | null => { + if (raw == null) { + return null + } + + const value = Number(raw) + + return Number.isFinite(value) && value !== 0 ? value : null +} + +export const parseSensorStations = (raw: RawSensorStation[]): SensorStation[] => { + return raw.reduce((acc, entry) => { + const lat = parseCoordinate(entry.lat) + const lon = parseCoordinate(entry.lon) + + if (lat == null || lon == null) { + return acc + } + + acc.push({ + stationId: entry.idstazione, + name: entry.nomestaz, + lat, + lon, + soglia1: parseThreshold(entry.soglia1), + soglia2: parseThreshold(entry.soglia2), + soglia3: parseThreshold(entry.soglia3), + }) + + return acc + }, []) +} + +const EARTH_RADIUS_KM = 6371 + +const toRadians = (degrees: number): number => (degrees * Math.PI) / 180 + +export const haversineKm = (a: Coordinates, b: Coordinates): number => { + const dLat = toRadians(b.lat - a.lat) + const dLon = toRadians(b.lon - a.lon) + const lat1 = toRadians(a.lat) + const lat2 = toRadians(b.lat) + + const h = Math.sin(dLat / 2) ** 2 + Math.cos(lat1) * Math.cos(lat2) * Math.sin(dLon / 2) ** 2 + + return 2 * EARTH_RADIUS_KM * Math.asin(Math.sqrt(h)) +} + +export const findNearestStations = ( + stations: SensorStation[], + point: Coordinates, + limit = 5 +): NearestStation[] => { + return stations + .map((station) => ({ ...station, distanceKm: haversineKm(station, point) })) + .sort((a, b) => a.distanceKm - b.distanceKm) + .slice(0, Math.max(0, limit)) +} diff --git a/src/utilities/telegram.ts b/src/utilities/telegram.ts index 61a01a9..acec08b 100644 --- a/src/utilities/telegram.ts +++ b/src/utilities/telegram.ts @@ -2,10 +2,18 @@ import { sendTelegramMessage } from '../services/telegram' import { InlineKeyboardButton } from 'telegraf/typings/core/types/typegram' import { translateKey } from './common' import { ParsedMeteoAlert } from './meteo-alerts' +import { ThresholdCrossing } from './river-sensors' +import { River } from '../models/river' import { config } from '../config/config' const separator = '--------------------------------' +const thresholdLabel: Record = { + soglia1: 'soglia 1', + soglia2: 'soglia 2', + soglia3: 'soglia 3', +} + export const sendNewTomorrowAlertMessage = async (alert: ParsedMeteoAlert) => { const criticDataMessage = Object.keys(alert.criticZoneData) .map((key) => { @@ -41,3 +49,62 @@ ${separator} }, }) } + +export const sendRiverLevelCrossingMessage = async ( + river: River, + crossing: ThresholdCrossing, + currentValue: number +) => { + const label = thresholdLabel[crossing.threshold.key] + const directionLine = + crossing.direction === 'above' ? `⬆️ Superata ${label}` : `⬇️ Rientrata sotto ${label}` + + const textMessage = `🌊 Livello del ${river.river_name} — ${river.station_name} +${directionLine} +Livello attuale: ${currentValue} m +Soglia: ${crossing.threshold.value} m` + + // No parse_mode: the body has no markup and river/station names are operator-supplied, so HTML + // parsing would only risk breaking on characters like & or <. + await sendTelegramMessage(config.chat_id, textMessage) +} + +export interface FloodPredictionPayload { + upstreamValue: number + leadTimeMinutes: number + predictedExceedanceAt: number + targetThreshold: number +} + +export const formatLeadTime = (minutes: number): string => { + const total = Math.max(0, Math.round(minutes)) + const hours = Math.floor(total / 60) + const mins = total % 60 + + if (hours === 0) { + return `${mins} min` + } + + return mins === 0 ? `${hours}h` : `${hours}h ${mins}min` +} + +export const sendFloodPredictionMessage = async ( + downstream: River, + upstream: River, + payload: FloodPredictionPayload +) => { + const eta = new Date(payload.predictedExceedanceAt).toLocaleString('it-IT', { + timeZone: 'Europe/Rome', + weekday: 'short', + hour: '2-digit', + minute: '2-digit', + }) + + const textMessage = `🌊⚠️ Possibile piena in arrivo +Monte: ${upstream.river_name} — ${upstream.station_name}: ${payload.upstreamValue} m +Storicamente questo livello a monte ha preceduto il superamento della soglia ${payload.targetThreshold} a ${downstream.river_name} — ${downstream.station_name}. +Arrivo stimato: tra ~${formatLeadTime(payload.leadTimeMinutes)} (≈ ${eta}) +ℹ️ Stima statistica, non una previsione ufficiale: verificare sempre le fonti ufficiali (allertameteo.regione.emilia-romagna.it, Protezione Civile).` + + await sendTelegramMessage(config.chat_id, textMessage) +} diff --git a/tests/services/arpae-archive.ts b/tests/services/arpae-archive.ts new file mode 100644 index 0000000..e8510d6 --- /dev/null +++ b/tests/services/arpae-archive.ts @@ -0,0 +1,89 @@ +import { expect } from 'chai' +import { describe, it } from 'mocha' +import { ArchiveTarget, archiveUrlForMonth, coordKey, parseArchiveLine } from '../../src/services/arpae-archive' + +const S_ANTONIO_LAT = 4457480 +const S_ANTONIO_LON = 1170609 + +const index = new Map([ + [coordKey(S_ANTONIO_LAT, S_ANTONIO_LON), { riverId: 1, soglia1: 10.5, soglia2: 12.2, soglia3: 13.7 }], +]) + +// Mirrors the verified archive shape: a metadata block then measurement blocks; the hydrometric +// level is the timerange[0] === 254 block carrying B13215. +const buildLine = (overrides: Record = {}): string => + JSON.stringify({ + network: 'simnbo', + ident: null, + lat: S_ANTONIO_LAT, + lon: S_ANTONIO_LON, + date: '2023-05-03T12:00:00Z', + data: [ + { vars: { B01019: { v: 'S. Antonio' } } }, + { timerange: [1, 0, 900], level: [1, null, null, null], vars: { B13215: { v: 99 } } }, + { timerange: [254, 0, 0], level: [1, null, null, null], vars: { B13215: { v: 14.39 } } }, + ], + ...overrides, + }) + +describe('tests/services/arpae-archive', () => { + describe('archiveUrlForMonth', () => { + it('zero-pads the month', () => { + expect(archiveUrlForMonth(2023, 5)).to.equal( + 'https://dati-simc.arpae.it/opendata/osservati/meteo/storico/2023-05.json.gz' + ) + }) + }) + + describe('parseArchiveLine', () => { + it('extracts the 254/B13215 reading and computes soglia flags', () => { + const reading = parseArchiveLine(buildLine(), index) + + expect(reading).to.deep.equal({ + river_id: 1, + value: 14.39, + measured_at: '2023-05-03T12:00:00.000Z', + soglia1_above: true, + soglia2_above: true, + soglia3_above: true, + }) + }) + + it('ignores non-hydrometric timeranges and uses the 254 block', () => { + // The [1,0,900] block (value 99) must not be picked. + const reading = parseArchiveLine(buildLine(), index) + expect(reading?.value).to.equal(14.39) + }) + + it('returns null for a station not in the index', () => { + expect(parseArchiveLine(buildLine({ lat: 1, lon: 2 }), index)).to.equal(null) + }) + + it('returns null when there is no hydrometric block', () => { + const line = buildLine({ data: [{ vars: { B01019: { v: 'S. Antonio' } } }] }) + expect(parseArchiveLine(line, index)).to.equal(null) + }) + + it('returns null when the level value is null', () => { + const line = buildLine({ + data: [{ timerange: [254, 0, 0], level: [1, null, null, null], vars: { B13215: { v: null } } }], + }) + expect(parseArchiveLine(line, index)).to.equal(null) + }) + + it('returns null on malformed JSON', () => { + expect(parseArchiveLine('{not json', index)).to.equal(null) + }) + + it('leaves a soglia flag null when that threshold is unset', () => { + const sparseIndex = new Map([ + [coordKey(S_ANTONIO_LAT, S_ANTONIO_LON), { riverId: 2, soglia1: 10.5, soglia2: null, soglia3: null }], + ]) + const reading = parseArchiveLine(buildLine(), sparseIndex) + + expect(reading?.soglia1_above).to.equal(true) + expect(reading?.soglia2_above).to.equal(null) + expect(reading?.soglia3_above).to.equal(null) + }) + }) +}) diff --git a/tests/services/river-sensors.ts b/tests/services/river-sensors.ts new file mode 100644 index 0000000..7e3b021 --- /dev/null +++ b/tests/services/river-sensors.ts @@ -0,0 +1,114 @@ +import { expect } from 'chai' +import { afterEach, beforeEach, describe, it } from 'mocha' +import sinon, { SinonStub } from 'sinon' +import { http } from '../../src/services/http' +import { getLatestRiverSensorValue, getRiverSensorTimeSeries } from '../../src/services/river-sensors' + +describe('tests/services/river-sensors', () => { + const expectedUrl = + 'https://allertameteo.regione.emilia-romagna.it/o/api/allerta/get-time-series/?stazione=3130&variabile=254,0,0/1,-,-,-/B13215' + + let axiosGetStub: SinonStub + + beforeEach(() => { + axiosGetStub = sinon.stub(http, 'get') + }) + + afterEach(() => { + axiosGetStub.restore() + }) + + describe('getRiverSensorTimeSeries', () => { + it('builds the correct URL and returns the full array', async () => { + const series = [ + { t: 1, v: 1.5 }, + { t: 2, v: 1.6 }, + ] + + axiosGetStub.resolves({ data: series }) + + const result = await getRiverSensorTimeSeries('3130') + + expect(result).to.deep.equal(series) + expect(axiosGetStub.calledOnceWithExactly(expectedUrl)).to.equal(true) + }) + + it('returns empty array when response is not an array', async () => { + axiosGetStub.resolves({ data: null }) + + const result = await getRiverSensorTimeSeries('3130') + + expect(result).to.deep.equal([]) + }) + + it('filters out malformed points', async () => { + axiosGetStub.resolves({ + data: [ + { t: 1, v: 1.5 }, + { t: 'bad', v: 1.6 }, + null, + { t: 3, v: 'bad' }, + { t: 4, v: 2.0 }, + ], + }) + + const result = await getRiverSensorTimeSeries('3130') + + expect(result).to.deep.equal([ + { t: 1, v: 1.5 }, + { t: 4, v: 2.0 }, + ]) + }) + + it('URL-encodes the station id', async () => { + axiosGetStub.resolves({ data: [] }) + + await getRiverSensorTimeSeries('abc/123') + + expect( + axiosGetStub.calledOnceWithExactly( + 'https://allertameteo.regione.emilia-romagna.it/o/api/allerta/get-time-series/?stazione=abc%2F123&variabile=254,0,0/1,-,-,-/B13215' + ) + ).to.equal(true) + }) + + it('rethrows and logs on HTTP failure', async () => { + axiosGetStub.rejects(new Error('boom')) + + let caught: unknown + + try { + await getRiverSensorTimeSeries('3130') + } catch (err) { + caught = err + } + + expect(caught).to.be.instanceOf(Error) + expect((caught as Error).message).to.equal('boom') + }) + }) + + describe('getLatestRiverSensorValue', () => { + it('returns undefined when the series is empty', async () => { + axiosGetStub.resolves({ data: [] }) + + const result = await getLatestRiverSensorValue('3130') + + expect(result).to.equal(undefined) + }) + + it('returns the point with the greatest timestamp', async () => { + axiosGetStub.resolves({ + data: [ + { t: 10, v: 1.1 }, + { t: 30, v: 1.3 }, + { t: 20, v: 1.2 }, + ], + }) + + const result = await getLatestRiverSensorValue('3130') + + expect(result).to.deep.equal({ t: 30, v: 1.3 }) + }) + }) +}) diff --git a/tests/tasks/flood-backfill.ts b/tests/tasks/flood-backfill.ts new file mode 100644 index 0000000..5081317 --- /dev/null +++ b/tests/tasks/flood-backfill.ts @@ -0,0 +1,24 @@ +import { expect } from 'chai' +import { describe, it } from 'mocha' +import { expandMonthRange } from '../../src/tasks/flood-backfill' + +describe('tests/tasks/flood-backfill', () => { + describe('expandMonthRange', () => { + it('expands a single month', () => { + expect(expandMonthRange('2023-05', '2023-05')).to.deep.equal([{ year: 2023, month: 5 }]) + }) + + it('expands across a year boundary inclusively', () => { + expect(expandMonthRange('2023-11', '2024-02')).to.deep.equal([ + { year: 2023, month: 11 }, + { year: 2023, month: 12 }, + { year: 2024, month: 1 }, + { year: 2024, month: 2 }, + ]) + }) + + it('throws on an invalid month', () => { + expect(() => expandMonthRange('2023-13', '2024-01')).to.throw() + }) + }) +}) diff --git a/tests/tasks/flood-calibration.ts b/tests/tasks/flood-calibration.ts new file mode 100644 index 0000000..f381d6e --- /dev/null +++ b/tests/tasks/flood-calibration.ts @@ -0,0 +1,145 @@ +import { expect } from 'chai' +import { afterEach, beforeEach, describe, it } from 'mocha' +import sinon, { SinonStub } from 'sinon' +import * as riverModel from '../../src/models/river' +import * as riverLevelModel from '../../src/models/river-level' +import * as riverLinkModel from '../../src/models/river-link' +import { River } from '../../src/models/river' +import { RiverLink } from '../../src/models/river-link' +import { runFloodCalibration } from '../../src/tasks/flood-calibration' + +const T0 = new Date('2026-01-01T00:00:00.000Z').getTime() +const MIN = 60_000 +const DAY = 24 * 60 * MIN + +const iso = (ms: number) => new Date(ms).toISOString() + +const river = (id: number, soglia1: number | null): River => ({ + id, + station_id: String(id), + river_name: 'Idice', + station_name: `st-${id}`, + soglia1, + soglia2: null, + soglia3: null, + created_on: iso(T0), + updated_on: iso(T0), +}) + +const link = (overrides: Partial = {}): RiverLink => ({ + id: 1, + upstream_river_id: 10, + downstream_river_id: 20, + target_threshold: 1, + lead_time_minutes: null, + precursor_level: null, + sample_size: 0, + model_json: null, + last_calibrated_on: null, + created_on: iso(T0), + updated_on: iso(T0), + ...overrides, +}) + +// Three downstream exceedances (>10) 3 days apart, each preceded by an upstream peak. +const buildPair = (dayOffset: number, leadMin: number, upstreamPeak: number) => { + const onset = T0 + dayOffset * DAY + return { + downstream: [ + { value: 9, measured_at: iso(onset - 30 * MIN) }, + { value: 11, measured_at: iso(onset) }, + { value: 12, measured_at: iso(onset + 30 * MIN) }, + { value: 8, measured_at: iso(onset + 60 * MIN) }, + ], + upstream: [ + { value: upstreamPeak - 2, measured_at: iso(onset - (leadMin + 60) * MIN) }, + { value: upstreamPeak, measured_at: iso(onset - leadMin * MIN) }, + { value: upstreamPeak - 1, measured_at: iso(onset - (leadMin - 30) * MIN) }, + ], + } +} + +const e = [buildPair(2, 120, 5), buildPair(5, 150, 6), buildPair(8, 135, 5.5)] +const downstreamRows = e.flatMap((p) => p.downstream) +const upstreamRows = e.flatMap((p) => p.upstream) + +describe('tests/tasks/flood-calibration', () => { + let getLinks: SinonStub + let getRiver: SinonStub + let getLevels: SinonStub + let update: SinonStub + + beforeEach(() => { + getLinks = sinon.stub(riverLinkModel, 'getRiverLinks') + getRiver = sinon.stub(riverModel, 'getRiverById') + getLevels = sinon.stub(riverLevelModel, 'getRiverReadingsSince') + update = sinon.stub(riverLinkModel, 'updateRiverLinkModel').resolves({} as never) + }) + + afterEach(() => sinon.restore()) + + it('learns and persists an active model from historical events', async () => { + getLinks.resolves([link()]) + getRiver.withArgs(20).resolves(river(20, 10)) + getRiver.withArgs(10).resolves(river(10, 5)) + getLevels.withArgs(20).resolves(downstreamRows) + getLevels.withArgs(10).resolves(upstreamRows) + + const summary = await runFloodCalibration() + + expect(update.calledOnce).to.equal(true) + const patch = update.firstCall.args[1] + expect(patch.sample_size).to.equal(3) + expect(patch.lead_time_minutes).to.equal(135) + expect(patch.precursor_level).to.be.closeTo(5.25, 1e-9) + expect(summary).to.deep.equal({ calibrated: 1, active: 1, skipped: 0 }) + }) + + it('skips a link whose downstream soglia is not set', async () => { + getLinks.resolves([link()]) + getRiver.withArgs(20).resolves(river(20, null)) + getRiver.withArgs(10).resolves(river(10, 5)) + + const summary = await runFloodCalibration() + + expect(update.called).to.equal(false) + expect(summary).to.deep.equal({ calibrated: 0, active: 0, skipped: 1 }) + }) + + it('skips a link with a missing river', async () => { + getLinks.resolves([link()]) + getRiver.withArgs(20).resolves(undefined) + getRiver.withArgs(10).resolves(river(10, 5)) + + const summary = await runFloodCalibration() + + expect(summary.skipped).to.equal(1) + expect(update.called).to.equal(false) + }) + + it('isolates a per-link failure and continues', async () => { + getLinks.resolves([link({ id: 1 }), link({ id: 2 })]) + getRiver.withArgs(20).resolves(river(20, 10)) + getRiver.withArgs(10).resolves(river(10, 5)) + getLevels.withArgs(20).rejects(new Error('db down')) + + const summary = await runFloodCalibration() + + expect(summary.skipped).to.equal(2) + expect(summary.calibrated).to.equal(0) + }) + + it('persists an inactive model when there are too few events', async () => { + getLinks.resolves([link()]) + getRiver.withArgs(20).resolves(river(20, 10)) + getRiver.withArgs(10).resolves(river(10, 5)) + getLevels.withArgs(20).resolves(e[0].downstream) // single event only + getLevels.withArgs(10).resolves(e[0].upstream) + + const summary = await runFloodCalibration() + + const patch = update.firstCall.args[1] + expect(patch.sample_size).to.equal(1) + expect(summary).to.deep.equal({ calibrated: 1, active: 0, skipped: 0 }) + }) +}) diff --git a/tests/tasks/flood-prediction.ts b/tests/tasks/flood-prediction.ts new file mode 100644 index 0000000..1c847ba --- /dev/null +++ b/tests/tasks/flood-prediction.ts @@ -0,0 +1,140 @@ +import { expect } from 'chai' +import { afterEach, beforeEach, describe, it } from 'mocha' +import sinon, { SinonStub } from 'sinon' +import * as riverModel from '../../src/models/river' +import * as riverLevelModel from '../../src/models/river-level' +import * as riverLinkModel from '../../src/models/river-link' +import * as linkPredictionModel from '../../src/models/link-prediction' +import * as telegram from '../../src/utilities/telegram' +import { River } from '../../src/models/river' +import { RiverLink } from '../../src/models/river-link' +import { runFloodPrediction } from '../../src/tasks/flood-prediction' + +const makeLink = (overrides: Partial = {}): RiverLink => ({ + id: 1, + upstream_river_id: 10, + downstream_river_id: 20, + target_threshold: 1, + lead_time_minutes: 120, + precursor_level: 5, + sample_size: 3, + model_json: null, + last_calibrated_on: null, + created_on: new Date().toISOString(), + updated_on: new Date().toISOString(), + ...overrides, +}) + +const river = (id: number, name: string): River => ({ + id, + station_id: String(id), + river_name: name, + station_name: name, + soglia1: 10, + soglia2: 12, + soglia3: 14, + created_on: new Date().toISOString(), + updated_on: new Date().toISOString(), +}) + +// Two readings, rising, ending at `value`, within the recent window. +const risingTo = (value: number) => [ + { value: value - 0.3, measured_at: new Date(Date.now() - 30 * 60_000).toISOString() }, + { value, measured_at: new Date(Date.now() - 15 * 60_000).toISOString() }, +] + +describe('tests/tasks/flood-prediction', () => { + let getLinks: SinonStub + let getLevels: SinonStub + let getLatestPred: SinonStub + let createPred: SinonStub + let getRiver: SinonStub + let send: SinonStub + + beforeEach(() => { + getLinks = sinon.stub(riverLinkModel, 'getRiverLinks') + getLevels = sinon.stub(riverLevelModel, 'getRiverLevelsSince') + getLatestPred = sinon.stub(linkPredictionModel, 'getLatestLinkPrediction').resolves(undefined) + createPred = sinon.stub(linkPredictionModel, 'createLinkPrediction').resolves({} as never) + getRiver = sinon.stub(riverModel, 'getRiverById') + send = sinon.stub(telegram, 'sendFloodPredictionMessage').resolves() + getRiver.withArgs(20).resolves(river(20, 'Idice')) + getRiver.withArgs(10).resolves(river(10, 'Idice monte')) + }) + + afterEach(() => sinon.restore()) + + it('emits and persists a prediction when an active link sees a rising upstream above precursor', async () => { + getLinks.resolves([makeLink()]) + getLevels.resolves(risingTo(5.4)) + + const summary = await runFloodPrediction() + + expect(send.calledOnce).to.equal(true) + expect(createPred.calledOnce).to.equal(true) + expect(summary).to.deep.equal({ evaluated: 1, predictions: 1, skipped: 0 }) + }) + + it('does not predict when the upstream is below the precursor level', async () => { + getLinks.resolves([makeLink()]) + getLevels.resolves(risingTo(4.0)) + + const summary = await runFloodPrediction() + + expect(send.called).to.equal(false) + expect(summary.predictions).to.equal(0) + }) + + it('does not predict for an inactive (under-calibrated) link', async () => { + getLinks.resolves([makeLink({ sample_size: 1 })]) + getLevels.resolves(risingTo(5.4)) + + const summary = await runFloodPrediction() + + expect(send.called).to.equal(false) + expect(summary).to.deep.equal({ evaluated: 1, predictions: 0, skipped: 0 }) + }) + + it('suppresses a repeat prediction inside the dedup window', async () => { + getLinks.resolves([makeLink()]) + getLevels.resolves(risingTo(5.4)) + getLatestPred.resolves({ + id: 9, + link_id: 1, + predicted_at: new Date(Date.now() - 10 * 60_000).toISOString(), + predicted_exceedance_at: new Date().toISOString(), + upstream_value: 5.4, + actual_exceedance_at: null, + outcome: 'pending', + created_on: new Date().toISOString(), + }) + + const summary = await runFloodPrediction() + + expect(send.called).to.equal(false) + expect(summary).to.deep.equal({ evaluated: 1, predictions: 0, skipped: 1 }) + }) + + it('isolates a per-link failure and keeps evaluating the rest', async () => { + getLinks.resolves([makeLink({ id: 1, upstream_river_id: 10 }), makeLink({ id: 2, upstream_river_id: 11 })]) + getLevels.withArgs(10).rejects(new Error('db down')) + getLevels.withArgs(11).resolves(risingTo(4.0)) + + const summary = await runFloodPrediction() + + expect(summary.evaluated).to.equal(2) + expect(summary.skipped).to.equal(1) + }) + + it('does not persist a prediction when the Telegram send fails (so it retries next poll)', async () => { + getLinks.resolves([makeLink()]) + getLevels.resolves(risingTo(5.4)) + send.rejects(new Error('telegram down')) + + const summary = await runFloodPrediction() + + expect(createPred.called).to.equal(false) + expect(summary.predictions).to.equal(0) + expect(summary.skipped).to.equal(1) + }) +}) diff --git a/tests/tasks/river-levels.ts b/tests/tasks/river-levels.ts new file mode 100644 index 0000000..b5dcce9 --- /dev/null +++ b/tests/tasks/river-levels.ts @@ -0,0 +1,133 @@ +import { expect } from 'chai' +import { afterEach, beforeEach, describe, it } from 'mocha' +import sinon, { SinonStub } from 'sinon' +import * as riverModel from '../../src/models/river' +import * as riverLevelModel from '../../src/models/river-level' +import * as sensorsService from '../../src/services/river-sensors' +import * as telegram from '../../src/utilities/telegram' +import { River } from '../../src/models/river' +import { RiverLevel } from '../../src/models/river-level' +import { runRiverLevelCheck } from '../../src/tasks/river-levels' + +const makeRiver = (overrides: Partial = {}): River => ({ + id: 1, + station_id: '3130', + river_name: 'Idice', + station_name: 'S. Antonio', + soglia1: 10.5, + soglia2: 12.2, + soglia3: 13.7, + created_on: new Date().toISOString(), + updated_on: new Date().toISOString(), + ...overrides, +}) + +const makeLevel = (overrides: Partial = {}): RiverLevel => ({ + id: 1, + river_id: 1, + value: 9, + measured_at: '2026-06-20T00:00:00.000Z', + soglia1_above: false, + soglia2_above: false, + soglia3_above: false, + created_on: new Date().toISOString(), + ...overrides, +}) + +// A timestamp newer than the default makeLevel.measured_at above. +const NEWER_T = new Date('2026-06-21T00:00:00.000Z').getTime() + +describe('tests/tasks/river-levels', () => { + let getRivers: SinonStub + let getLatest: SinonStub + let getLatestLevel: SinonStub + let createIfNew: SinonStub + let sendMessage: SinonStub + + beforeEach(() => { + getRivers = sinon.stub(riverModel, 'getRivers') + getLatest = sinon.stub(sensorsService, 'getLatestRiverSensorValue') + getLatestLevel = sinon.stub(riverLevelModel, 'getLatestRiverLevel') + createIfNew = sinon.stub(riverLevelModel, 'createRiverLevelIfNew').resolves(makeLevel()) + sendMessage = sinon.stub(telegram, 'sendRiverLevelCrossingMessage').resolves() + }) + + afterEach(() => { + sinon.restore() + }) + + it('seeds the first observation without sending a message', async () => { + getRivers.resolves([makeRiver()]) + getLatestLevel.resolves(undefined) + getLatest.resolves({ t: NEWER_T, v: 11 }) + + const summary = await runRiverLevelCheck() + + expect(createIfNew.calledOnce).to.equal(true) + expect(sendMessage.called).to.equal(false) + expect(summary).to.deep.equal({ checked: 1, crossings: 0, skipped: 0, unchanged: 0 }) + }) + + it('sends one message when a reading crosses above a threshold', async () => { + getRivers.resolves([makeRiver()]) + getLatestLevel.resolves(makeLevel({ soglia1_above: false })) + getLatest.resolves({ t: NEWER_T, v: 11 }) // 11 > soglia1 (10.5) + default margin (0.05) + + const summary = await runRiverLevelCheck() + + expect(sendMessage.calledOnce).to.equal(true) + const [, crossing] = sendMessage.firstCall.args + expect(crossing).to.deep.include({ direction: 'above' }) + expect(summary.crossings).to.equal(1) + expect(summary.checked).to.equal(1) + }) + + it('skips when the latest reading is not newer than the stored one', async () => { + getRivers.resolves([makeRiver()]) + getLatestLevel.resolves(makeLevel({ measured_at: '2026-06-22T00:00:00.000Z' })) + getLatest.resolves({ t: new Date('2026-06-21T00:00:00.000Z').getTime(), v: 11 }) + + const summary = await runRiverLevelCheck() + + expect(createIfNew.called).to.equal(false) + expect(sendMessage.called).to.equal(false) + expect(summary.unchanged).to.equal(1) + }) + + it('counts a sensor fetch failure as skipped and keeps processing other rivers', async () => { + getRivers.resolves([makeRiver({ id: 1, station_id: 'A' }), makeRiver({ id: 2, station_id: 'B' })]) + getLatestLevel.resolves(undefined) + getLatest.withArgs('A').rejects(new Error('boom')) + getLatest.withArgs('B').resolves({ t: NEWER_T, v: 9 }) + + const summary = await runRiverLevelCheck() + + expect(summary.skipped).to.equal(1) + expect(summary.checked).to.equal(1) + }) + + it('does not abort the loop when sending a crossing message fails', async () => { + getRivers.resolves([makeRiver()]) + getLatestLevel.resolves(makeLevel({ soglia1_above: false })) + getLatest.resolves({ t: NEWER_T, v: 11 }) + sendMessage.rejects(new Error('telegram down')) + + const summary = await runRiverLevelCheck() + + expect(summary.crossings).to.equal(0) + expect(summary.checked).to.equal(1) + }) + + it('does not send when the reading was already persisted by a concurrent run', async () => { + getRivers.resolves([makeRiver()]) + getLatestLevel.resolves(makeLevel({ soglia1_above: false })) + getLatest.resolves({ t: NEWER_T, v: 11 }) + createIfNew.resolves(undefined) + + const summary = await runRiverLevelCheck() + + expect(sendMessage.called).to.equal(false) + expect(summary.unchanged).to.equal(1) + expect(summary.checked).to.equal(0) + }) +}) diff --git a/tests/utilities/flood-prediction.ts b/tests/utilities/flood-prediction.ts new file mode 100644 index 0000000..fe9e6f4 --- /dev/null +++ b/tests/utilities/flood-prediction.ts @@ -0,0 +1,172 @@ +import { expect } from 'chai' +import { describe, it } from 'mocha' +import { + calibrateLink, + detectExceedanceEvents, + findPrecursor, + isLinkActive, + LinkModel, + median, + percentile, + predict, + Reading, +} from '../../src/utilities/flood-prediction' + +const T0 = new Date('2026-01-01T00:00:00.000Z').getTime() +const MIN = 60_000 +const DAY = 24 * 60 * MIN + +const at = (offsetMin: number, value: number): Reading => ({ measuredAt: T0 + offsetMin * MIN, value }) + +describe('tests/utilities/flood-prediction', () => { + describe('percentile / median', () => { + it('interpolates percentiles and computes the median', () => { + expect(median([1, 2, 3])).to.equal(2) + expect(median([1, 2, 3, 4])).to.equal(2.5) + expect(percentile([5, 5.5, 6], 0.25)).to.be.closeTo(5.25, 1e-9) + expect(percentile([], 0.5)).to.be.NaN + }) + }) + + describe('detectExceedanceEvents', () => { + it('returns no events when the series never exceeds the threshold', () => { + expect(detectExceedanceEvents([at(0, 5), at(15, 6), at(30, 7)], 10)).to.deep.equal([]) + }) + + it('captures a single run with its onset and peak', () => { + const events = detectExceedanceEvents([at(0, 9), at(15, 11), at(30, 13), at(45, 12), at(60, 8)], 10) + + expect(events).to.have.length(1) + expect(events[0].onsetAt).to.equal(T0 + 15 * MIN) + expect(events[0].peakAt).to.equal(T0 + 30 * MIN) + expect(events[0].peakValue).to.equal(13) + }) + + it('separates two runs split by a drop below the threshold', () => { + const events = detectExceedanceEvents([at(0, 11), at(15, 9), at(30, 12), at(45, 8)], 10) + + expect(events.map((e) => e.onsetAt)).to.deep.equal([T0, T0 + 30 * MIN]) + }) + + it('closes an event that is still above the threshold at the end of the series', () => { + const events = detectExceedanceEvents([at(0, 9), at(15, 11), at(30, 12)], 10) + + expect(events).to.have.length(1) + expect(events[0].peakValue).to.equal(12) + }) + + it('merges runs separated by a dip shorter than the separation window', () => { + // dip below at +30 then back above at +45 (15-min gap) with a 60-min separation -> one flood + const events = detectExceedanceEvents([at(0, 11), at(30, 9), at(45, 12)], 10, 60) + + expect(events).to.have.length(1) + expect(events[0].onsetAt).to.equal(T0) + expect(events[0].peakValue).to.equal(12) + }) + + it('keeps runs separated by a dip longer than the separation window', () => { + const events = detectExceedanceEvents([at(0, 11), at(30, 9), at(95, 9), at(110, 12)], 10, 60) + + expect(events).to.have.length(2) + }) + }) + + describe('findPrecursor', () => { + const event = { onsetAt: T0 + 600 * MIN, peakAt: T0 + 610 * MIN, peakValue: 12 } + + it('finds the upstream peak in the window and the lead to onset', () => { + const upstream = [at(420, 4), at(480, 6), at(540, 5)] // peak 6 at +480, onset at +600 -> 120 min lead + const result = findPrecursor(upstream, event) + + expect(result).to.not.equal(null) + expect(result?.upstreamPeak).to.equal(6) + expect(result?.leadMinutes).to.equal(120) + }) + + it('returns null when there is no upstream data in the lookback window', () => { + expect( + findPrecursor([at(0, 4)], event, { + maxLookbackMinutes: 60, + minLeadMinutes: 15, + minSamples: 3, + precursorPercentile: 0.25, + minSeparationMinutes: 0, + }) + ).to.equal(null) + }) + + it('returns null when the lead is below the minimum', () => { + const upstream = [at(595, 6)] // only 5 min before onset + expect(findPrecursor(upstream, event)).to.equal(null) + }) + }) + + describe('calibrateLink', () => { + // Three downstream exceedance events spaced 3 days apart (> 48h lookback) so each upstream + // peak only falls inside its own event's window. + const buildPair = (dayOffset: number, leadMin: number, upstreamPeak: number) => { + const onset = dayOffset * DAY + const onsetMin = onset / MIN + return { + downstream: [at(onsetMin - 30, 9), at(onsetMin, 11), at(onsetMin + 30, 12), at(onsetMin + 60, 8)], + upstream: [ + at(onsetMin - leadMin - 60, upstreamPeak - 2), + at(onsetMin - leadMin, upstreamPeak), + at(onsetMin - leadMin + 30, upstreamPeak - 1), + ], + } + } + + const e1 = buildPair(2, 120, 5) + const e2 = buildPair(5, 150, 6) + const e3 = buildPair(8, 135, 5.5) + const downstream = [...e1.downstream, ...e2.downstream, ...e3.downstream] + const upstream = [...e1.upstream, ...e2.upstream, ...e3.upstream] + + it('learns lead time and precursor level from historical events', () => { + const model = calibrateLink(upstream, downstream, 10) + + expect(model.sampleSize).to.equal(3) + expect(model.leadTimeMinutes).to.equal(135) // median of [120,150,135] + expect(model.precursorLevel).to.be.closeTo(5.25, 1e-9) // 25th pct of [5,5.5,6] + expect(isLinkActive(model)).to.equal(true) + }) + + it('stays inactive when there are fewer than minSamples events', () => { + const model = calibrateLink(e1.upstream, e1.downstream, 10) + + expect(model.sampleSize).to.equal(1) + expect(isLinkActive(model)).to.equal(false) + }) + }) + + describe('predict', () => { + const model: LinkModel = { leadTimeMinutes: 135, precursorLevel: 5.25, sampleSize: 3, leadSpreadMinutes: 30 } + const now = T0 + 1000 * MIN + + it('predicts an arrival when the upstream reaches the precursor level while rising', () => { + const prediction = predict([at(960, 5.0), at(990, 5.3)], model, now) + + expect(prediction).to.not.equal(null) + expect(prediction?.predictedExceedanceAt).to.equal(now + 135 * MIN) + expect(prediction?.upstreamValue).to.equal(5.3) + }) + + it('returns null when the upstream is below the precursor level', () => { + expect(predict([at(960, 4.9), at(990, 5.0)], model, now)).to.equal(null) + }) + + it('returns null when the upstream is receding even if above the precursor level', () => { + expect(predict([at(960, 5.6), at(990, 5.3)], model, now)).to.equal(null) + }) + + it('returns null when the model is inactive', () => { + const inactive: LinkModel = { ...model, sampleSize: 1 } + expect(predict([at(960, 5.0), at(990, 5.3)], inactive, now)).to.equal(null) + }) + + it('returns null on a single reading (cannot confirm a rising trend)', () => { + expect(predict([at(990, 5.5)], model, now)).to.equal(null) + }) + }) +}) diff --git a/tests/utilities/river-sensors.ts b/tests/utilities/river-sensors.ts new file mode 100644 index 0000000..21c5f23 --- /dev/null +++ b/tests/utilities/river-sensors.ts @@ -0,0 +1,229 @@ +import { expect } from 'chai' +import { describe, it } from 'mocha' +import { River } from '../../src/models/river' +import { detectThresholdCrossings, getActiveThresholds, ThresholdBooleans } from '../../src/utilities/river-sensors' + +const makeRiver = (overrides: Partial = {}): River => ({ + id: 1, + station_id: '3130', + river_name: 'Reno', + station_name: 'Casalecchio Chiusa', + soglia1: 1.5, + soglia2: 2.2, + soglia3: 3.0, + created_on: new Date().toISOString(), + updated_on: new Date().toISOString(), + ...overrides, +}) + +describe('tests/utilities/river-sensors', () => { + describe('getActiveThresholds', () => { + it('returns all three thresholds when all are set', () => { + const thresholds = getActiveThresholds(makeRiver()) + + expect(thresholds).to.deep.equal([ + { key: 'soglia1', value: 1.5 }, + { key: 'soglia2', value: 2.2 }, + { key: 'soglia3', value: 3.0 }, + ]) + }) + + it('drops null thresholds', () => { + const thresholds = getActiveThresholds(makeRiver({ soglia2: null })) + + expect(thresholds).to.deep.equal([ + { key: 'soglia1', value: 1.5 }, + { key: 'soglia3', value: 3.0 }, + ]) + }) + + it('coerces numeric strings (pg returns NUMERIC as string)', () => { + const river = makeRiver({ + soglia1: '1.5' as unknown as number, + soglia2: null, + soglia3: '3.0' as unknown as number, + }) + + expect(getActiveThresholds(river)).to.deep.equal([ + { key: 'soglia1', value: 1.5 }, + { key: 'soglia3', value: 3.0 }, + ]) + }) + + it('returns empty array when no thresholds are set', () => { + expect(getActiveThresholds(makeRiver({ soglia1: null, soglia2: null, soglia3: null }))).to.deep.equal([]) + }) + }) + + describe('detectThresholdCrossings', () => { + const thresholds = [ + { key: 'soglia1' as const, value: 1.5 }, + { key: 'soglia2' as const, value: 2.2 }, + { key: 'soglia3' as const, value: 3.0 }, + ] + + it('seeds state without emitting crossings on first observation', () => { + const result = detectThresholdCrossings(2.5, thresholds, undefined) + + expect(result.crossings).to.deep.equal([]) + expect(result.nextState).to.deep.equal({ + soglia1_above: true, + soglia2_above: true, + soglia3_above: false, + }) + }) + + it('emits no crossing when the side is unchanged', () => { + const previous: ThresholdBooleans = { + soglia1_above: true, + soglia2_above: false, + soglia3_above: false, + } + + const result = detectThresholdCrossings(2.0, thresholds, previous) + + expect(result.crossings).to.deep.equal([]) + expect(result.nextState).to.deep.equal(previous) + }) + + it('emits an above crossing when the level rises past a threshold', () => { + const previous: ThresholdBooleans = { + soglia1_above: false, + soglia2_above: false, + soglia3_above: false, + } + + const result = detectThresholdCrossings(1.8, thresholds, previous) + + expect(result.crossings).to.deep.equal([ + { threshold: { key: 'soglia1', value: 1.5 }, direction: 'above' }, + ]) + expect(result.nextState).to.deep.equal({ + soglia1_above: true, + soglia2_above: false, + soglia3_above: false, + }) + }) + + it('emits a below crossing when the level drops under a threshold', () => { + const previous: ThresholdBooleans = { + soglia1_above: true, + soglia2_above: false, + soglia3_above: false, + } + + const result = detectThresholdCrossings(1.2, thresholds, previous) + + expect(result.crossings).to.deep.equal([ + { threshold: { key: 'soglia1', value: 1.5 }, direction: 'below' }, + ]) + }) + + it('emits one crossing per threshold when multiple thresholds flip at once', () => { + const previous: ThresholdBooleans = { + soglia1_above: false, + soglia2_above: false, + soglia3_above: false, + } + + const result = detectThresholdCrossings(3.5, thresholds, previous) + + expect(result.crossings).to.deep.equal([ + { threshold: { key: 'soglia1', value: 1.5 }, direction: 'above' }, + { threshold: { key: 'soglia2', value: 2.2 }, direction: 'above' }, + { threshold: { key: 'soglia3', value: 3.0 }, direction: 'above' }, + ]) + expect(result.nextState).to.deep.equal({ + soglia1_above: true, + soglia2_above: true, + soglia3_above: true, + }) + }) + + it('treats a null previous boolean as no-prior-reference for that threshold', () => { + const previous: ThresholdBooleans = { + soglia1_above: null, + soglia2_above: false, + soglia3_above: null, + } + + const result = detectThresholdCrossings(2.5, thresholds, previous) + + expect(result.crossings).to.deep.equal([ + { threshold: { key: 'soglia2', value: 2.2 }, direction: 'above' }, + ]) + expect(result.nextState).to.deep.equal({ + soglia1_above: true, + soglia2_above: true, + soglia3_above: false, + }) + }) + + it('only considers provided active thresholds', () => { + const previous: ThresholdBooleans = { + soglia1_above: false, + soglia2_above: false, + soglia3_above: false, + } + + const result = detectThresholdCrossings(2.5, [{ key: 'soglia1', value: 1.5 }], previous) + + expect(result.crossings).to.deep.equal([ + { threshold: { key: 'soglia1', value: 1.5 }, direction: 'above' }, + ]) + expect(result.nextState).to.deep.equal({ + soglia1_above: true, + soglia2_above: null, + soglia3_above: null, + }) + }) + }) + + describe('detectThresholdCrossings with hysteresis margin', () => { + const single = [{ key: 'soglia1' as const, value: 1.5 }] + const margin = 0.1 + + it('holds "below" while inside the upper deadband', () => { + const previous: ThresholdBooleans = { soglia1_above: false, soglia2_above: null, soglia3_above: null } + + const result = detectThresholdCrossings(1.55, single, previous, margin) + + expect(result.crossings).to.deep.equal([]) + expect(result.nextState.soglia1_above).to.equal(false) + }) + + it('emits "above" only once the level clears threshold + margin', () => { + const previous: ThresholdBooleans = { soglia1_above: false, soglia2_above: null, soglia3_above: null } + + const result = detectThresholdCrossings(1.65, single, previous, margin) + + expect(result.crossings).to.deep.equal([{ threshold: { key: 'soglia1', value: 1.5 }, direction: 'above' }]) + expect(result.nextState.soglia1_above).to.equal(true) + }) + + it('holds "above" while inside the lower deadband', () => { + const previous: ThresholdBooleans = { soglia1_above: true, soglia2_above: null, soglia3_above: null } + + const result = detectThresholdCrossings(1.45, single, previous, margin) + + expect(result.crossings).to.deep.equal([]) + expect(result.nextState.soglia1_above).to.equal(true) + }) + + it('emits "below" only once the level drops under threshold - margin', () => { + const previous: ThresholdBooleans = { soglia1_above: true, soglia2_above: null, soglia3_above: null } + + const result = detectThresholdCrossings(1.35, single, previous, margin) + + expect(result.crossings).to.deep.equal([{ threshold: { key: 'soglia1', value: 1.5 }, direction: 'below' }]) + expect(result.nextState.soglia1_above).to.equal(false) + }) + + it('seeds the first observation with a plain comparison, ignoring the margin', () => { + const result = detectThresholdCrossings(1.55, single, undefined, margin) + + expect(result.crossings).to.deep.equal([]) + expect(result.nextState.soglia1_above).to.equal(true) + }) + }) +}) diff --git a/tests/utilities/river-stations.ts b/tests/utilities/river-stations.ts new file mode 100644 index 0000000..2690483 --- /dev/null +++ b/tests/utilities/river-stations.ts @@ -0,0 +1,78 @@ +import { expect } from 'chai' +import { describe, it } from 'mocha' +import { RawSensorStation } from '../../src/services/river-sensors' +import { findNearestStations, haversineKm, parseSensorStations } from '../../src/utilities/river-stations' + +const rawStation = (overrides: Partial = {}): RawSensorStation => ({ + idstazione: '3130', + nomestaz: 'S. Antonio', + lat: '4457480', + lon: '1170609', + value: null, + soglia1: 10.5, + soglia2: 12.2, + soglia3: 13.7, + ...overrides, +}) + +describe('tests/utilities/river-stations', () => { + describe('parseSensorStations', () => { + it('scales the integer-string coordinates by 1e5', () => { + const [station] = parseSensorStations([rawStation()]) + + expect(station.lat).to.be.closeTo(44.5748, 1e-9) + expect(station.lon).to.be.closeTo(11.70609, 1e-9) + expect(station.stationId).to.equal('3130') + expect(station.name).to.equal('S. Antonio') + }) + + it('normalises a 0 soglia to null (no official threshold)', () => { + const [station] = parseSensorStations([rawStation({ soglia1: 0, soglia2: 8.5, soglia3: 0 })]) + + expect(station.soglia1).to.equal(null) + expect(station.soglia2).to.equal(8.5) + expect(station.soglia3).to.equal(null) + }) + + it('drops entries with unparseable coordinates', () => { + const result = parseSensorStations([rawStation({ lat: 'n/a' }), rawStation({ idstazione: '3149' })]) + + expect(result).to.have.length(1) + expect(result[0].stationId).to.equal('3149') + }) + }) + + describe('haversineKm', () => { + it('returns 0 for the same point', () => { + expect(haversineKm({ lat: 44.6, lon: 11.6 }, { lat: 44.6, lon: 11.6 })).to.equal(0) + }) + + it('returns ~111 km for one degree of latitude', () => { + expect(haversineKm({ lat: 0, lon: 0 }, { lat: 1, lon: 0 })).to.be.closeTo(111.19, 0.5) + }) + + it('matches the known S. Antonio -> Molinella distance', () => { + expect(haversineKm({ lat: 44.5748, lon: 11.70609 }, { lat: 44.617, lon: 11.667 })).to.be.closeTo(5.6, 0.5) + }) + }) + + describe('findNearestStations', () => { + const stations = parseSensorStations([ + rawStation({ idstazione: 'far', lat: '4482117', lon: '1099474' }), + rawStation({ idstazione: 'near', lat: '4457480', lon: '1170609' }), + rawStation({ idstazione: 'mid', lat: '4467307', lon: '1162513' }), + ]) + const molinella = { lat: 44.617, lon: 11.667 } + + it('sorts by ascending distance and attaches distanceKm', () => { + const result = findNearestStations(stations, molinella) + + expect(result.map((s) => s.stationId)).to.deep.equal(['near', 'mid', 'far']) + expect(result[0].distanceKm).to.be.lessThan(result[1].distanceKm) + }) + + it('respects the limit', () => { + expect(findNearestStations(stations, molinella, 1).map((s) => s.stationId)).to.deep.equal(['near']) + }) + }) +}) diff --git a/tests/utilities/telegram.ts b/tests/utilities/telegram.ts new file mode 100644 index 0000000..2b562ae --- /dev/null +++ b/tests/utilities/telegram.ts @@ -0,0 +1,63 @@ +import { expect } from 'chai' +import { afterEach, beforeEach, describe, it } from 'mocha' +import sinon, { SinonStub } from 'sinon' +import * as telegramService from '../../src/services/telegram' +import { sendRiverLevelCrossingMessage } from '../../src/utilities/telegram' +import { River } from '../../src/models/river' +import { ThresholdCrossing } from '../../src/utilities/river-sensors' + +const makeRiver = (overrides: Partial = {}): River => ({ + id: 1, + station_id: '3130', + river_name: 'Idice', + station_name: 'S. Antonio', + soglia1: 10.5, + soglia2: 12.2, + soglia3: 13.7, + created_on: new Date().toISOString(), + updated_on: new Date().toISOString(), + ...overrides, +}) + +describe('tests/utilities/telegram', () => { + let sendStub: SinonStub + + beforeEach(() => { + sendStub = sinon.stub(telegramService, 'sendTelegramMessage').resolves() + }) + + afterEach(() => { + sendStub.restore() + }) + + describe('sendRiverLevelCrossingMessage', () => { + it('sends an "above" message and passes no parse_mode (HTML-unsafe names stay literal)', async () => { + const crossing: ThresholdCrossing = { + threshold: { key: 'soglia2', value: 12.2 }, + direction: 'above', + } + + await sendRiverLevelCrossingMessage(makeRiver({ river_name: 'Reno & Idice' }), crossing, 12.5) + + expect(sendStub.calledOnce).to.equal(true) + const [, text, extra] = sendStub.firstCall.args + expect(extra).to.equal(undefined) + expect(text).to.contain('Reno & Idice') + expect(text).to.contain('Superata soglia 2') + expect(text).to.contain('Livello attuale: 12.5 m') + expect(text).to.contain('Soglia: 12.2 m') + }) + + it('sends a "below" message when the level falls back under a threshold', async () => { + const crossing: ThresholdCrossing = { + threshold: { key: 'soglia1', value: 10.5 }, + direction: 'below', + } + + await sendRiverLevelCrossingMessage(makeRiver(), crossing, 10.1) + + const [, text] = sendStub.firstCall.args + expect(text).to.contain('Rientrata sotto soglia 1') + }) + }) +})