From 64569636a532b22566443e005a262ff869700212 Mon Sep 17 00:00:00 2001 From: Nico Montanari Date: Tue, 21 Apr 2026 11:29:43 +0200 Subject: [PATCH 1/5] feat: add river levels monitoring and management features --- README.md | 27 +++++ sql/rivers.sql | 25 +++++ src/models/river-level.ts | 28 +++++ src/models/river.ts | 54 +++++++++ src/routes/rivers.ts | 148 +++++++++++++++++++++++++ src/scheduler.ts | 2 + src/server.ts | 2 + src/services/river-sensors.ts | 40 +++++++ src/tasks/river-levels.ts | 73 +++++++++++++ src/utilities/river-sensors.ts | 69 ++++++++++++ src/utilities/telegram.ts | 25 +++++ tests/services/river-sensors.ts | 114 +++++++++++++++++++ tests/utilities/river-sensors.ts | 181 +++++++++++++++++++++++++++++++ 13 files changed, 788 insertions(+) create mode 100644 sql/rivers.sql create mode 100644 src/models/river-level.ts create mode 100644 src/models/river.ts create mode 100644 src/routes/rivers.ts create mode 100644 src/services/river-sensors.ts create mode 100644 src/tasks/river-levels.ts create mode 100644 src/utilities/river-sensors.ts create mode 100644 tests/services/river-sensors.ts create mode 100644 tests/utilities/river-sensors.ts diff --git a/README.md b/README.md index a70f753..36421c6 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,11 @@ 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 check is appended to `river_levels`. + - **Notification:** Sends a Telegram message only on crossing events — when a reading rises above or falls below a threshold relative to the previous check. + ## Scheduled Tasks (Crons) The application relies on scheduled tasks (crons) to automate the weather monitoring flow: @@ -31,6 +36,28 @@ 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. + +## 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 }`) + +The `station_id` is the Allerta Meteo `idstazione`; threshold values must be looked up manually from the [Allerta Meteo portal](https://allertameteo.regione.emilia-romagna.it/) and stored alongside the station. ## Configuration and Installation diff --git a/sql/rivers.sql b/sql/rivers.sql new file mode 100644 index 0000000..2e8b0c4 --- /dev/null +++ b/sql/rivers.sql @@ -0,0 +1,25 @@ +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() +); + +CREATE INDEX IF NOT EXISTS river_levels_river_id_created_on_idx + ON river_levels (river_id, created_on DESC); diff --git a/src/models/river-level.ts b/src/models/river-level.ts new file mode 100644 index 0000000..56db2ef --- /dev/null +++ b/src/models/river-level.ts @@ -0,0 +1,28 @@ +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 => { + const query = `SELECT * FROM ${tableName} WHERE river_id = $1 ORDER BY created_on 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() }) +} 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/rivers.ts b/src/routes/rivers.ts new file mode 100644 index 0000000..2800b60 --- /dev/null +++ b/src/routes/rivers.ts @@ -0,0 +1,148 @@ +import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify' +import { + createRiver, + deleteRiver, + getRiverById, + getRiverByStationId, + getRivers, + updateRiver, +} from '../models/river' +import { runRiverLevelCheck } from '../tasks/river-levels' + +const nullableNumber = { type: ['number', 'null'] } + +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({ + 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..1787376 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -2,6 +2,7 @@ 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 logger from './logger' const jobs: Cron[] = [] @@ -35,6 +36,7 @@ export const startScheduler = () => { schedule('meteo-alerts', '*/5 * * * *', runMeteoAlertCheck) schedule('pretemp', '*/5 * * * *', runPretempCheck) schedule('estofex', '*/5 * * * *', runEstofexCheck) + schedule('river-levels', '*/5 * * * *', runRiverLevelCheck) logger.info({ count: jobs.length }, 'Scheduler started') } diff --git a/src/server.ts b/src/server.ts index bc01335..c4e1894 100644 --- a/src/server.ts +++ b/src/server.ts @@ -4,6 +4,7 @@ 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 logger from './logger' const translations = { @@ -45,6 +46,7 @@ export const startServer = async (): Promise => { registerTestMessageRoutes(app) registerMeteoAlertsRoutes(app) registerForecastReportsRoutes(app) + registerRiversRoutes(app) await fastify.listen({ host: '127.0.0.1', diff --git a/src/services/river-sensors.ts b/src/services/river-sensors.ts new file mode 100644 index 0000000..369064e --- /dev/null +++ b/src/services/river-sensors.ts @@ -0,0 +1,40 @@ +import { http } from './http' +import logger from '../logger' + +export interface RiverSensorTimePoint { + t: number + v: number +} + +const VARIABILE = '254,0,0/1,-,-,-/B13215' +const TIME_SERIES_BASE_URL = 'https://allertameteo.regione.emilia-romagna.it/o/api/allerta/get-time-series/' + +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]) +} diff --git a/src/tasks/river-levels.ts b/src/tasks/river-levels.ts new file mode 100644 index 0000000..44db897 --- /dev/null +++ b/src/tasks/river-levels.ts @@ -0,0 +1,73 @@ +import { getRivers } from '../models/river' +import { createRiverLevel, 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 logger from '../logger' + +const log = logger.child({ task: 'river-levels' }) + +export interface RiverLevelCheckSummary { + checked: number + crossings: number + skipped: number +} + +export const runRiverLevelCheck = async (): Promise => { + const rivers = await getRivers() + const summary: RiverLevelCheckSummary = { checked: 0, crossings: 0, skipped: 0 } + + if (rivers.length === 0) { + return summary + } + + for (const river of rivers) { + const latest = await getLatestRiverSensorValue(river.station_id).catch((err) => { + log.error({ err, stationId: river.station_id, riverId: river.id }, 'Failed to fetch latest river sensor value') + return undefined + }) + + if (!latest) { + summary.skipped += 1 + continue + } + + const currentValue = Number(latest.v) + const thresholds = getActiveThresholds(river) + const previous = await getLatestRiverLevel(river.id) + 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) + + await createRiverLevel({ + 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, + }) + + 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 + } + + return summary +} diff --git a/src/utilities/river-sensors.ts b/src/utilities/river-sensors.ts new file mode 100644 index 0000000..227f890 --- /dev/null +++ b/src/utilities/river-sensors.ts @@ -0,0 +1,69 @@ +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 + }, []) +} + +export const detectThresholdCrossings = ( + currentValue: number, + thresholds: ActiveThreshold[], + previousState: ThresholdBooleans | undefined +): CrossingDetectionResult => { + const nextState: ThresholdBooleans = { + soglia1_above: null, + soglia2_above: null, + soglia3_above: null, + } + const crossings: ThresholdCrossing[] = [] + + for (const threshold of thresholds) { + const currentAbove = currentValue > threshold.value + const aboveKey = `${threshold.key}_above` as keyof ThresholdBooleans + nextState[aboveKey] = currentAbove + + const previousAbove = previousState?.[aboveKey] ?? null + + if (previousState && previousAbove !== null && previousAbove !== currentAbove) { + crossings.push({ + threshold, + direction: currentAbove ? 'above' : 'below', + }) + } + } + + return { crossings, nextState } +} diff --git a/src/utilities/telegram.ts b/src/utilities/telegram.ts index 61a01a9..2dcbd54 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,20 @@ ${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` + + await sendTelegramMessage(config.chat_id, textMessage, { parse_mode: 'HTML' }) +} 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/utilities/river-sensors.ts b/tests/utilities/river-sensors.ts new file mode 100644 index 0000000..356a81f --- /dev/null +++ b/tests/utilities/river-sensors.ts @@ -0,0 +1,181 @@ +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, + }) + }) + }) +}) From d13e47dad2f984cab2fbb176229e5433022f7b16 Mon Sep 17 00:00:00 2001 From: Nico Montanari Date: Mon, 22 Jun 2026 17:56:58 +0200 Subject: [PATCH 2/5] feat: implement flood calibration and prediction tasks - Add flood calibration task to re-learn link models from historical river levels. - Introduce flood prediction task to evaluate calibrated links and send alerts based on upstream readings. - Create utility functions for flood prediction, including model calibration and exceedance event detection. - Implement river station utilities for parsing sensor data and calculating distances. - Add tests for flood calibration, prediction, river levels, and utility functions to ensure functionality and reliability. --- .env.example | 2 + README.md | 35 ++++- sql/rivers.sql | 47 +++++- src/config/config.ts | 1 + src/config/development.ts | 2 + src/config/env.ts | 22 +++ src/config/production.ts | 2 + src/models/link-prediction.ts | 36 +++++ src/models/river-level.ts | 63 +++++++- src/models/river-link.ts | 76 ++++++++++ src/routes/river-links.ts | 139 ++++++++++++++++++ src/routes/rivers.ts | 36 +++++ src/scheduler.ts | 5 + src/server.ts | 2 + src/services/arpae-archive.ts | 129 +++++++++++++++++ src/services/river-sensors.ts | 39 +++++ src/tasks/flood-backfill.ts | 118 +++++++++++++++ src/tasks/flood-calibration.ts | 83 +++++++++++ src/tasks/flood-prediction.ts | 98 +++++++++++++ src/tasks/river-levels.ts | 107 ++++++++------ src/utilities/flood-prediction.ts | 215 ++++++++++++++++++++++++++++ src/utilities/river-sensors.ts | 27 +++- src/utilities/river-stations.ts | 91 ++++++++++++ src/utilities/telegram.ts | 43 +++++- tests/services/arpae-archive.ts | 89 ++++++++++++ tests/tasks/flood-backfill.ts | 24 ++++ tests/tasks/river-levels.ts | 133 +++++++++++++++++ tests/utilities/flood-prediction.ts | 149 +++++++++++++++++++ tests/utilities/river-sensors.ts | 48 +++++++ tests/utilities/river-stations.ts | 78 ++++++++++ tests/utilities/telegram.ts | 63 ++++++++ 31 files changed, 1952 insertions(+), 50 deletions(-) create mode 100644 src/config/env.ts create mode 100644 src/models/link-prediction.ts create mode 100644 src/models/river-link.ts create mode 100644 src/routes/river-links.ts create mode 100644 src/services/arpae-archive.ts create mode 100644 src/tasks/flood-backfill.ts create mode 100644 src/tasks/flood-calibration.ts create mode 100644 src/tasks/flood-prediction.ts create mode 100644 src/utilities/flood-prediction.ts create mode 100644 src/utilities/river-stations.ts create mode 100644 tests/services/arpae-archive.ts create mode 100644 tests/tasks/flood-backfill.ts create mode 100644 tests/tasks/river-levels.ts create mode 100644 tests/utilities/flood-prediction.ts create mode 100644 tests/utilities/river-stations.ts create mode 100644 tests/utilities/telegram.ts 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/README.md b/README.md index 36421c6..8149746 100644 --- a/README.md +++ b/README.md @@ -23,9 +23,14 @@ The system interfaces with various official and unofficial sources to provide a 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 check is appended to `river_levels`. + - **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: @@ -38,6 +43,10 @@ The application relies on scheduled tasks (crons) to automate the weather monito - 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 @@ -55,9 +64,29 @@ Stations are managed via HTTP (port 3000): - `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 }`) +- `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`. -The `station_id` is the Allerta Meteo `idstazione`; threshold values must be looked up manually from the [Allerta Meteo portal](https://allertameteo.regione.emilia-romagna.it/) and stored alongside the station. +- `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/sql/rivers.sql b/sql/rivers.sql index 2e8b0c4..add6a90 100644 --- a/sql/rivers.sql +++ b/sql/rivers.sql @@ -18,8 +18,53 @@ CREATE TABLE IF NOT EXISTS river_levels ( soglia1_above BOOLEAN, soglia2_above BOOLEAN, soglia3_above BOOLEAN, - created_on TIMESTAMPTZ NOT NULL DEFAULT now() + 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 index 56db2ef..1b4a449 100644 --- a/src/models/river-level.ts +++ b/src/models/river-level.ts @@ -16,7 +16,9 @@ export interface RiverLevel { export type CreatableRiverLevel = Omit export const getLatestRiverLevel = async (riverId: number): Promise => { - const query = `SELECT * FROM ${tableName} WHERE river_id = $1 ORDER BY created_on DESC, id DESC LIMIT 1` + // 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]) @@ -26,3 +28,62 @@ export const getLatestRiverLevel = async (riverId: number): 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 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/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 index 2800b60..a63ad0f 100644 --- a/src/routes/rivers.ts +++ b/src/routes/rivers.ts @@ -8,9 +8,28 @@ import { 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'], @@ -134,6 +153,23 @@ export const registerRiversRoutes = (fastify: FastifyInstance) => { }, }) + 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', diff --git a/src/scheduler.ts b/src/scheduler.ts index 1787376..81b4cee 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -3,6 +3,8 @@ 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[] = [] @@ -37,6 +39,9 @@ export const startScheduler = () => { 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 c4e1894..20292e8 100644 --- a/src/server.ts +++ b/src/server.ts @@ -5,6 +5,7 @@ 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 = { @@ -47,6 +48,7 @@ export const startServer = async (): Promise => { 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 index 369064e..462e1d4 100644 --- a/src/services/river-sensors.ts +++ b/src/services/river-sensors.ts @@ -6,8 +6,21 @@ export interface RiverSensorTimePoint { 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}` @@ -38,3 +51,29 @@ export const getLatestRiverSensorValue = async (stationId: string): Promise (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..5001ae9 --- /dev/null +++ b/src/tasks/flood-calibration.ts @@ -0,0 +1,83 @@ +import { getRiverById } from '../models/river' +import { getRiverLevelsSince } 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' }) + +const EPOCH_START = '1970-01-01T00:00:00.000Z' + +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 } + + 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([ + getRiverLevelsSince(link.downstream_river_id, EPOCH_START), + getRiverLevelsSince(link.upstream_river_id, EPOCH_START), + ]) + + 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 index 44db897..2c68b6d 100644 --- a/src/tasks/river-levels.ts +++ b/src/tasks/river-levels.ts @@ -1,8 +1,9 @@ import { getRivers } from '../models/river' -import { createRiverLevel, getLatestRiverLevel } from '../models/river-level' +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' }) @@ -11,62 +12,88 @@ 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 } + const summary: RiverLevelCheckSummary = { checked: 0, crossings: 0, skipped: 0, unchanged: 0 } if (rivers.length === 0) { return summary } for (const river of rivers) { - const latest = await getLatestRiverSensorValue(river.station_id).catch((err) => { - log.error({ err, stationId: river.station_id, riverId: river.id }, 'Failed to fetch latest river sensor value') - return undefined - }) + // 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 - } + 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 previous = await getLatestRiverLevel(river.id) - const previousState: ThresholdBooleans | undefined = previous - ? { - soglia1_above: previous.soglia1_above, - soglia2_above: previous.soglia2_above, - soglia3_above: previous.soglia3_above, - } - : undefined + 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) + const { crossings, nextState } = detectThresholdCrossings( + currentValue, + thresholds, + previousState, + config.river_threshold_margin + ) - await createRiverLevel({ - 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, - }) + 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, + }) - 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' - ) + // 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 + 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..f217025 --- /dev/null +++ b/src/utilities/flood-prediction.ts @@ -0,0 +1,215 @@ +/** + * 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 +} + +export const DEFAULT_FLOOD_OPTIONS: FloodModelOptions = { + maxLookbackMinutes: 48 * 60, + minLeadMinutes: 15, + minSamples: 3, + precursorPercentile: 0.25, +} + +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): ExceedanceEvent[] => { + const sorted = sortByTime(series) + const events: ExceedanceEvent[] = [] + let current: ExceedanceEvent | null = null + + for (const point of sorted) { + if (point.value > threshold) { + 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) { + events.push(current) + current = 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) + 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 index 227f890..5e1f76f 100644 --- a/src/utilities/river-sensors.ts +++ b/src/utilities/river-sensors.ts @@ -38,10 +38,21 @@ export const getActiveThresholds = (river: River): ActiveThreshold[] => { }, []) } +/** + * 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 + previousState: ThresholdBooleans | undefined, + margin = 0 ): CrossingDetectionResult => { const nextState: ThresholdBooleans = { soglia1_above: null, @@ -51,12 +62,20 @@ export const detectThresholdCrossings = ( const crossings: ThresholdCrossing[] = [] for (const threshold of thresholds) { - const currentAbove = currentValue > threshold.value const aboveKey = `${threshold.key}_above` as keyof ThresholdBooleans - nextState[aboveKey] = currentAbove - 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, 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 2dcbd54..bea4700 100644 --- a/src/utilities/telegram.ts +++ b/src/utilities/telegram.ts @@ -64,5 +64,46 @@ ${directionLine} Livello attuale: ${currentValue} m Soglia: ${crossing.threshold.value} m` - await sendTelegramMessage(config.chat_id, textMessage, { parse_mode: 'HTML' }) + // 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})` + + 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/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/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..b1ad4b0 --- /dev/null +++ b/tests/utilities/flood-prediction.ts @@ -0,0 +1,149 @@ +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) + }) + }) + + 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 })).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 index 356a81f..21c5f23 100644 --- a/tests/utilities/river-sensors.ts +++ b/tests/utilities/river-sensors.ts @@ -178,4 +178,52 @@ describe('tests/utilities/river-sensors', () => { }) }) }) + + 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') + }) + }) +}) From da4079304d3a364bb56e3a7ada17be51469c3327 Mon Sep 17 00:00:00 2001 From: Nico Montanari Date: Wed, 24 Jun 2026 12:15:51 +0200 Subject: [PATCH 3/5] feat: enhance flood calibration and prediction with new river readings and tests --- src/models/river-level.ts | 15 +++ src/tasks/flood-calibration.ts | 12 ++- src/utilities/flood-prediction.ts | 27 +++++- src/utilities/telegram.ts | 3 +- tests/tasks/flood-calibration.ts | 145 ++++++++++++++++++++++++++++ tests/tasks/flood-prediction.ts | 140 +++++++++++++++++++++++++++ tests/utilities/flood-prediction.ts | 25 ++++- 7 files changed, 357 insertions(+), 10 deletions(-) create mode 100644 tests/tasks/flood-calibration.ts create mode 100644 tests/tasks/flood-prediction.ts diff --git a/src/models/river-level.ts b/src/models/river-level.ts index 1b4a449..7d2c1a8 100644 --- a/src/models/river-level.ts +++ b/src/models/river-level.ts @@ -49,6 +49,21 @@ export const getRiverLevelsSince = async (riverId: number, since: string): Promi 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 /** diff --git a/src/tasks/flood-calibration.ts b/src/tasks/flood-calibration.ts index 5001ae9..a577c57 100644 --- a/src/tasks/flood-calibration.ts +++ b/src/tasks/flood-calibration.ts @@ -1,12 +1,15 @@ import { getRiverById } from '../models/river' -import { getRiverLevelsSince } from '../models/river-level' +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' }) -const EPOCH_START = '1970-01-01T00:00:00.000Z' +// 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 @@ -31,6 +34,7 @@ const thresholdFor = ( 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 { @@ -51,8 +55,8 @@ export const runFloodCalibration = async (): Promise => } const [downstreamRows, upstreamRows] = await Promise.all([ - getRiverLevelsSince(link.downstream_river_id, EPOCH_START), - getRiverLevelsSince(link.upstream_river_id, EPOCH_START), + getRiverReadingsSince(link.downstream_river_id, since), + getRiverReadingsSince(link.upstream_river_id, since), ]) const model = calibrateLink(toReadings(upstreamRows), toReadings(downstreamRows), threshold) diff --git a/src/utilities/flood-prediction.ts b/src/utilities/flood-prediction.ts index f217025..13222b4 100644 --- a/src/utilities/flood-prediction.ts +++ b/src/utilities/flood-prediction.ts @@ -47,6 +47,7 @@ export interface FloodModelOptions { minLeadMinutes: number minSamples: number precursorPercentile: number + minSeparationMinutes: number } export const DEFAULT_FLOOD_OPTIONS: FloodModelOptions = { @@ -54,6 +55,9 @@ export const DEFAULT_FLOOD_OPTIONS: FloodModelOptions = { 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 @@ -83,13 +87,20 @@ export const percentile = (values: number[], p: number): number => { * 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): ExceedanceEvent[] => { +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) { @@ -97,8 +108,16 @@ export const detectExceedanceEvents = (series: Reading[], threshold: number): Ex current.peakAt = point.measuredAt } } else if (current) { - events.push(current) - current = null + // 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 + } } } @@ -147,7 +166,7 @@ export const calibrateLink = ( threshold: number, options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS ): LinkModel => { - const events = detectExceedanceEvents(downstream, threshold) + const events = detectExceedanceEvents(downstream, threshold, options.minSeparationMinutes) const observations = events .map((event) => findPrecursor(upstream, event, options)) .filter((obs): obs is PrecursorObservation => obs !== null) diff --git a/src/utilities/telegram.ts b/src/utilities/telegram.ts index bea4700..acec08b 100644 --- a/src/utilities/telegram.ts +++ b/src/utilities/telegram.ts @@ -103,7 +103,8 @@ export const sendFloodPredictionMessage = async ( 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})` +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/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/utilities/flood-prediction.ts b/tests/utilities/flood-prediction.ts index b1ad4b0..fe9e6f4 100644 --- a/tests/utilities/flood-prediction.ts +++ b/tests/utilities/flood-prediction.ts @@ -54,6 +54,21 @@ describe('tests/utilities/flood-prediction', () => { 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', () => { @@ -69,7 +84,15 @@ describe('tests/utilities/flood-prediction', () => { }) 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 })).to.equal(null) + 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', () => { From 8484c087b88c7cca49ad57202de164bffc9dfb31 Mon Sep 17 00:00:00 2001 From: Nico Montanari Date: Wed, 24 Jun 2026 13:05:46 +0200 Subject: [PATCH 4/5] docs: interactive flood-model demo (GitHub Pages) + model explainer PDF - index.html: self-contained, dependency-free simulation of an upstream->downstream flood with animated map, hydrograph chart, step indicator and predicted-vs-real arrival; real Idice/Reno parameters; carries a 'not an official forecast' disclaimer. - modello-predittivo-piene.pdf: plain-language explainer of the empirical model, the 4 phases, the monitored stations, the backtest results and the usage limits. --- index.html | 348 +++++++++++++++++++++++++++++++++++ modello-predittivo-piene.pdf | Bin 0 -> 8679 bytes 2 files changed, 348 insertions(+) create mode 100644 index.html create mode 100644 modello-predittivo-piene.pdf 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 0000000000000000000000000000000000000000..67e5fc483a1dfbe303865c3f57abb784d882e6b7 GIT binary patch literal 8679 zcmdUV+5V!+vgUm~g#!u>fS@2E&I}GXfdk4U%BT#E=j?SZ`USf0`o5pF`PZ(scm1k% z_gUxia3M*UL}W&M5oAVWVa4(=hBUqv{`dd-pZ_N#IrrrG7OiteaAN0ay&XG~_Q=@D z~!!or@a=w0Jm~SYLW$^S-%Ah=+ssCuzGmRguMuzy&A~MY% zEiyxWwZ3(`-<;U~V)Vt~PliM5rf*q5$<7bgD^os%$w{?Jt8FES&4Vm3;Zn8M(d!G4%{?%07fO#YTp(_Xa< z(IODMNu-u-U5C zci9wHXDglBpZX#GqJF+W{5#8kfEHe!!{G5r5%^E6{=dLW{zZ|kpG5xCBtLxZzne+0 zFW~&Cz%O&n6Fb$`_T?;+IBk59Qo#K{&)=PSWWB90@qRKa_H!^AeBOTY%s?7s#Yy}} zlzd)VsgyxSPse%(Nz8s>24ZQvhLbfCZ(w-Z!Yef#tku9j`~984Cw|?XOW+2M{hh_k z`QzV{r~S%6X}@%&{~4 zt?>1;JABvbpSm9f_2h({S|z}RO3G$QdXi9dv8BeC#mKatJgc7LG(?qJ~h5O94z-xW1Z6tFAIF8 z&M3>C-GCa)@+k*cn%Gg<=6%v>LZ(;U+I<~%U2WI(U)$w1e-L{EjpBK7=B;kvGb#7# z)3~Gi&&Qb34vB{p?&IkTuR5Y#AbTTL_9O5C_EcQ~%yX}G(Kmg6!2w--FV?9|=QK=6hJB1Wp4Ri0>szviru3%n!CIH!HQ#U923p%2Q3xC>=cY|oPA7IfpcJh3(A?u*HA*!jrEY3q=HeebeciN{^@f{*Z9DcGFiJ1)+$ z`3YNq=xLQFZiA~=Zx6Laui9-t4)Or^)*UuGMhycy;94uXh{cu;^XLpj(C30IoIt2T zVG^~_p8f4{`)Sx+^@Rn6pxYk%3#a@PG_&5gwsosmtwB3_d|C`THJ26cY7r}jeHix! zCN$cP$4B((UuMfs=rh!1>;Sk~9VZ8kbk^u5bWgNAb9UL*1YDUq`!;2bIQ1+Z^ijKdM|&^dQS>!JOcHW{URSQ#sdIHWC=tTUBXwu_rD}iL+oi^C}?5)d^xG?KI^U1Vwi_D5~l3R`CsI+_g zqc)$I`pd0u)$?(A4Cqm_q};Z-WBnNAPI!AXHlvMNebVOv3U$gkpt1#)DqtIYZkfGj zf!$$VuSU&7?vY0i@7vzZ9q3}W)xok*IVuQ~BP~}MybR`wcy=At7Xb5C@5yKDdogfD zcJTB@<`VU~DB{j*+rL0y6d|)l1*&<6s~pwn4za=w>%C7p#Z3)t&MAmnIE4be0n`6C z50S1Xxg_2-9^2D&>Pa&a4#mU%x?B2gae?H~q}!RUxtPjv_$VJGH9PVzee}|>>SJMZ z6Gn$)o2blshtxZTYJ??4|m& z0apsf?Nh477c{P1eZ3GDE+u;fk6AQ7uyGn7(F@^d7tx2V;jr`wkIs6LpSioK*_ysu zLglT6WtVU-V);E~IUn$=P$&c)7~9B)3ByKCk$-R&72OMt>^!zj2W@+00vH01++Bl+ zyi~rwvXV6GV5^M|ih45K)z|$#;5b8|p_DoMkmc2r+v$Uh(7@|lE*X!qT@F{}ZLj<2 z-K#rLanWLJz@bi|5*-IRdKdFagv(2PCO}I0>>;#TfGRsnUCuYxVSseI`nn4ghW1f- zz$-D!RhjwvQ4Ob6zt)uX>1K+Uj(IDaQ*Az`IrNP^28S%r?3MeK_GcBzE4q!C@x7XO zr(hW_qSY&*L{WV7x*?1T!Dj;vgC}bGBO5a@=;Lr=iPP)Y%4Ak245%IKLwDqLjCySW z_(kW2%>ZiIe*5E>#=pEqcUL&=PB&PoLj%rltBv6E39D#h1iYZx_(rCF zWxt(fQN^3R4LB9fmy*=cVLF@=ry|jTXx-S}^{fJ_IW_9R8{Hi5+t}$wp3c{m0#=!J z)bu))mrg4q;WYC{tk^$XddpiV)yvb`2-K;Gi8T7_%P>9&XYSm{V4aYIXHhU2jysin1IN^F|&cu&`y&2W_k^Au}MD>L~yD3QBf>XV+$ z&(3iT3z$kZbh?KSAmujD>d8?cv;fc0srk5+FbX-^jE-<|?Fersc~v8(?$_$qVKWW4jgZ~E3Nce7gdqdm&)a~TD>o%3>pTaN zsc<*M7*DgCB=DLG2);K_CaXQeaPd)!-pDvri@D_9EHuXEoSqvsu+VC5sn0XmAx0;6 zRB*vOD?7NdIzC2ZGpP7J-Yh<(R}~o34N#StmU@t0Tk_FZZY!S)dIyUe^eI)a_#rQJbAuQe_aXRGBDGiZn(I8CJ3NMOM|pqcoJ9S==y+%LcEyj01~+< zF@>yXamM^jeo-n=k;U9EYqxXh)yM3%?Hy)r?YMKc+q{Yf)u5pau&ptJ96&Xu~Vh_T;_=t7w)qaKA?hCI4HEgHW$#+RR4EjX`QN)dUaEi$N ze(_6D7kqB!j!bC^Q?i-ImW9-H%k=d6MM59h{@N@q9Zmq^_w~bMUyjtLXODCbd$KlK z?80HZs?x(oU1g_9`fOYy%kd$}rMtNK=A0L-e^{PYGsK7C)6M#3@ZZqA}Y%pn%-u=b(4jcG4`izXdr z{q)UsLVeC)cI*i*aeY;gyFW9;U4EdB=7KgB@ApArS+MHZ%zq+{2>=H|sh1n+m$pB8 ziCc*tF6nq~wQO0qVT4JjX!$`3QUXG5JND@7#ZPXi%-{`0IBmDX+P>%KO1ofubAY|& z+>Q?r6%iYI@vTaKQS0AMOSl7{jgT)pS|=ZCv)W`P?AN1k10gKaSX>&FQKyuX(yv@0 zI=)o(NC<}q+{6!6TrTG%Td1^~;JH4#2y>&&;9C)av(5baKu6blf7@OL)zW=)4;nmo z)x!%81>~vGnvuDAsW8vQ{Xkmy{CcJM#RFHFs{LA9=$Mbfd>!@n-St6v*RQ#T5iQJP z$!(?!hE2Jo=gomTsyIn(4QS6Tbcgi#FscUkt(gCm6O#9ahw0(WPKO`xb$WA)Ui>PuHk?1Qj@Zw+x&Pu87AITy zFI!lyMN+{8%vVS*S?=s2g1-boST!kO)s4BK5>$yXHMP&e$gg+Y<@mCXMI8t^!^pPu zUADAq>Xp?;+_EoNS=L%cp`bkV`!ox&d+l&Gje7ObY;_l~@wnZN%eL-*y2k}P5=i4# z)vEGfJlYvOqdIjlQ@te+zYA8^@Vu&{mU!KQ{;<###-<=Q>YLe^Jl5{yz1*PNG~hs0 zwD;i)(UR);StWZ9Ps|gj<@qt`7hD0^uf#?&L&E2)UN*9{xSIrrg zcdfb%=o5~cSTq~bdU-=heDzZ)9tTqf_mh<#Rx8H>nO_fKVvw!6H?rY^#XLRXkOudv zL^0TXeJ0`H%+PPP!BbJXW|rTEIL!}Zw?hoW2}BVHtZ~zM6&cvfoZ!#cpo`6;;X2tQ zUO(r&PY1WUoFP0=b%A`gHJKDX-C11Np4=t59f|^R>TRNi`gB*t*iuE8Y+XS0HEgx{ zSuO8s&s(Lc4xWqi{4uTf#>>H3I9#&ZuzWo|d-%*}5N;*s4JpF=<>!d8+5lczrAo>> zI@Db5oAZ}J&|Yag$}u?4K*ML}O%`c8WQC?~OU|{mXeP=vXhvJY;Q4-{_uDAg(9%q> z7hbiNV0NfAK5w`3^IfsXWf7$yKHV0dlTAfifCC!mM|Fvx$sVt`VSN}}rTECSxvn_j z=UgLhd8_(G9S;Lw_+-4x495&)d#A=#SI}?Zvgwbk-Uh)O@ixJ%ZEh-@9_#gKF$2~+ zv%M&zwvA`?N8>%cnN6p!`!e1HmG*AXxG%jAeOenwvl@A-q?B^qB`tQfOTTBGcEyUa zzYs@gqn!IU5#w5QhMgudZ?9kH`yxvaK;DS&0J{%s1*UN+AlQ_KmnCK=OFGwri}q-L zz4?10>D=v{m2`GqMd5VH*E~TBCv1wb zd`U{#y`-*D@;37zMddT!|C}` zNMZXC;pW$Df5@-b=q%p$#_#@Xiae8<_aG?>>_XT0Q)= z=h(RKs$QE#@3-CISX5@&WsPgL7F$>rszc81E%0e)1_~y54ZLYAjd8i2+s!sJrEJ0` zv3TEZ2{$e~1k@Ecf>%XmMJ`X-08Gpzhj_}^Dk5gt9+Clw>I}qT{nNqM2h>J=bEb42 zYG|!6*mu%B%&;Ed`{&ZUFYPwF(^`+gcJ)~ql71kMwqQ~7lSj{eY2_5L(R&Dw=g;mWs24dZH;nH?Uiu?$ztd3>l69b|iM zY~1Nd`sCVL`XO$y*$0de@;(3~Qb!|*c&VTp+N~(Wv`tQTC*p$d-7nIape%2@UxOZE zzzj~dd1S5#Tf0$U0)x`OR%ybEP~F@8%4ux6dp%yRC1k7_%Wfp@r>L)x=_-5db&7!8 zpHpJzc;ON$40$09(d*IeXf}IyqY;xuV>}e!>uQ~Z?B46KIyNfFj4LG?U6`v6zR((v}WGV|~iz3Mf>*st@R%uZc=_vGO9gT`*&UqZPxvo1`HtdvZFppOD|#-EU-Q+aP#_j;>hs}{Q( z&H4?ni(K@59iBh;5-}VLL8Xz~N~`dMN-Xl|73(H2m-FKK8s=A~&%UFlR%?yIW_C%i z=m{mVX@WD+*g=!PPnG2uER08w3atwTAYr2QT2)s1t1YSPHyL&LlGV6d(Ms+v=Nd2! z71h)aGzQ@faw=44Y1*CVQ ziB6uF6?B{50sYWZ)iiJ2BG+if*D7-()p7IzK@b;?@+>#pkNVKUcSY?@u8Q+wJGe{zP3?XzlOLe! z)kjQ+TM-s23qp~ImJn7=jZz>0FI)|u4=rqtYdg01=xql(e!T~8U51ygie1BxbQv~= z5J@iBxm8Az^(`lvd1Fne9#QIijEO}Y$N75Hh;@S{DaivF-n+rW2|fj$MkkW4*dvgU zs{UBKp+tkq-V45slHGl7d{%Iy2R=w&)kXxhlt*c{Ha`Ut=!h~)JLbJX?+l%G&TdC~ zSW)Kb{o-G&Z~yjm+;6oEMtrON@~g zt^P0GZ8X!e-Y@-#Mp}OTg_c6?HyV|W?ALxR;x~R6M*P+pbq`ZFZdsG4=LYA#j7D6B;hEz%_#ns01q`QEy|m`}%V~>m=p=TN%=bd4^ewG?Cu!k49dLSr0+oaRaJ~%uy_jY_-``D6 OVPtx)qrffsZ~qrB@ybyE literal 0 HcmV?d00001 From ba48d3fe45ebadd897784b572f09a821e28fc4aa Mon Sep 17 00:00:00 2001 From: Nico Montanari Date: Wed, 24 Jun 2026 13:12:28 +0200 Subject: [PATCH 5/5] chore: gitignore local .claude preview/agent config --- .gitignore | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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