From afe742dccef7c9d52e6f8049eb5d451bad1f7754 Mon Sep 17 00:00:00 2001 From: Simon Moore Date: Sun, 5 Jul 2026 00:09:00 +0100 Subject: [PATCH] Initial commit: river level monitoring service Co-Authored-By: Claude Sonnet 4.6 --- README.md | 108 ++++++++++ river-monitor.service | 12 ++ river-monitor.timer | 10 + river_monitor.py | 447 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 577 insertions(+) create mode 100644 README.md create mode 100644 river-monitor.service create mode 100644 river-monitor.timer create mode 100644 river_monitor.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..f68f39c --- /dev/null +++ b/README.md @@ -0,0 +1,108 @@ +# River Levels Monitor + +Polls the UK Environment Agency's [flood-monitoring API](https://environment.data.gov.uk/flood-monitoring/doc/reference) +for river level stations within a radius you choose, archives every reading +to a local SQLite database, and generates a self-contained HTML dashboard. + +Data on the EA network itself typically updates every 15 minutes, so +polling more often than that gains you nothing. + +## 1. Install + +```bash +mkdir -p ~/river-monitor +cp river_monitor.py ~/river-monitor/ +pip install --user requests # only dependency +``` + +## 2. First run + +Give it a UK postcode (or `--lat`/`--lon`) and a search radius in km. This +location is saved to `~/river-data/config.json` so you don't need to repeat +it on later runs. + +```bash +python3 ~/river-monitor/river_monitor.py --postcode "WS15 3RZ" --radius 15 +``` + +This creates: + +- `~/river-data/river_levels.db` — the SQLite archive (`stations` and + `readings` tables) +- `~/river-data/dashboard.html` — open this in a browser to see current + levels and a small trend chart per station +- `~/river-data/config.json` — remembered location/paths for future runs + +Subsequent runs need no arguments: + +```bash +python3 ~/river-monitor/river_monitor.py +``` + +## 3. Polling periodically + +Two options — pick whichever fits your setup: + +### Option A: systemd timer (recommended on Kubuntu) + +```bash +mkdir -p ~/.config/systemd/user +cp river-monitor.service river-monitor.timer ~/.config/systemd/user/ +systemctl --user daemon-reload +systemctl --user enable --now river-monitor.timer + +# Check it's scheduled: +systemctl --user list-timers river-monitor.timer + +# Check logs: +journalctl --user -u river-monitor.service -f +``` + +### Option B: cron + +```bash +crontab -e +``` + +Add: + +``` +*/15 * * * * /usr/bin/python3 /home/YOURUSER/river-monitor/river_monitor.py >> /home/YOURUSER/river-data/poll.log 2>&1 +``` + +### Option C: built-in loop (no scheduler needed) + +```bash +python3 ~/river-monitor/river_monitor.py --daemon --interval 900 +``` + +Runs in the foreground (or under `tmux`/`screen`), polling every 900 +seconds until stopped. + +## Data model + +**stations**: `station_id`, `label`, `river_name`, `lat`, `lon` + +**readings**: `station_id`, `measure_id`, `unit`, `qualifier`, `value`, +`date_time` (from the EA), `fetched_at` (when you polled it). Unique on +`(measure_id, date_time)` so re-polling before a new reading is published +never creates duplicates — safe to poll as often as you like. + +Query it directly any time, e.g. levels for the last 24h: + +```bash +sqlite3 ~/river-data/river_levels.db \ + "SELECT label, date_time, value, unit FROM readings + JOIN stations USING(station_id) + WHERE date_time > datetime('now','-1 day') + ORDER BY date_time DESC;" +``` + +## Notes + +- Only stations that measure river **level** (as opposed to rainfall or + flow-only gauges) are included. Edit `parameter=level` in + `fetch_level_stations()` if you also want rainfall/tide stations. +- No API key is required — this is the EA's open data API. +- Attribution (per the Open Government Licence): *"This uses Environment + Agency flood and river level data from the real-time data API (Beta)."* diff --git a/river-monitor.service b/river-monitor.service new file mode 100644 index 0000000..d95fcd4 --- /dev/null +++ b/river-monitor.service @@ -0,0 +1,12 @@ +[Unit] +Description=Poll UK Environment Agency river levels and rebuild dashboard +Wants=network-online.target +After=network-online.target + +[Service] +Type=oneshot +# Adjust the path below to wherever you copy river_monitor.py +ExecStart=/usr/bin/python3 %h/river-monitor/river_monitor.py + +[Install] +WantedBy=default.target diff --git a/river-monitor.timer b/river-monitor.timer new file mode 100644 index 0000000..3edbf4b --- /dev/null +++ b/river-monitor.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Run river-monitor every 15 minutes (matches EA data refresh rate) + +[Timer] +OnBootSec=2min +OnUnitActiveSec=15min +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/river_monitor.py b/river_monitor.py new file mode 100644 index 0000000..9cf086e --- /dev/null +++ b/river_monitor.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python3 +""" +river_monitor.py + +Polls the UK Environment Agency's real-time flood-monitoring API +(https://environment.data.gov.uk/flood-monitoring/doc/reference) for river +level stations within a given radius, archives every reading to a local +SQLite database, and regenerates a self-contained HTML dashboard. + +Typical usage +------------- +First run, resolving location from a postcode: + + ./river_monitor.py --postcode WS15 3RZ --radius 15 + +Subsequent runs (e.g. from cron/systemd) reuse the saved location: + + ./river_monitor.py + +Run continuously instead of via an external scheduler: + + ./river_monitor.py --daemon --interval 900 + +Data is polled from the EA network roughly every 15 minutes, so there is +little value polling much more often than that. + +Attribution: This uses Environment Agency flood and river level data from +the real-time data API (Beta), under the Open Government Licence. +""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +import sys +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional + +import requests + +BASE_URL = "https://environment.data.gov.uk/flood-monitoring" +POSTCODES_API = "https://api.postcodes.io/postcodes/" + +DEFAULT_DIR = Path.home() / "river-data" +DEFAULT_DB = DEFAULT_DIR / "river_levels.db" +DEFAULT_HTML = DEFAULT_DIR / "dashboard.html" +DEFAULT_CONFIG = DEFAULT_DIR / "config.json" + +USER_AGENT = "river-monitor/1.0 (personal homelab script)" + + +# -------------------------------------------------------------------------- +# Data model +# -------------------------------------------------------------------------- + +@dataclass +class Reading: + station_id: str + label: str + river_name: str + lat: Optional[float] + lon: Optional[float] + measure_id: str + unit: Optional[str] + qualifier: Optional[str] + value: Optional[float] + date_time: Optional[str] + + +# -------------------------------------------------------------------------- +# Location helpers +# -------------------------------------------------------------------------- + +def resolve_postcode(postcode: str, session: requests.Session) -> tuple[float, float]: + """Resolve a UK postcode to (lat, long) using the free postcodes.io API.""" + clean = postcode.strip().replace(" ", "") + resp = session.get(POSTCODES_API + clean, timeout=15) + resp.raise_for_status() + data = resp.json() + if data.get("status") != 200 or "result" not in data: + raise ValueError(f"Could not resolve postcode {postcode!r}: {data}") + result = data["result"] + return result["latitude"], result["longitude"] + + +# -------------------------------------------------------------------------- +# EA flood-monitoring API +# -------------------------------------------------------------------------- + +def fetch_level_stations( + session: requests.Session, lat: float, lon: float, dist_km: float +) -> list[dict]: + """ + Return all stations within dist_km of (lat, lon) that measure river level. + The EA API embeds each measure's latest reading directly in this + response, so a single call gives us metadata *and* current values. + """ + params = { + "lat": lat, + "long": lon, + "dist": dist_km, + "parameter": "level", + "_limit": 2000, + } + resp = session.get(f"{BASE_URL}/id/stations", params=params, timeout=30) + resp.raise_for_status() + return resp.json().get("items", []) + + +def extract_readings(stations: Iterable[dict]) -> list[Reading]: + readings: list[Reading] = [] + for station in stations: + station_id = station.get("stationReference") or station.get("@id", "").rsplit("/", 1)[-1] + label = station.get("label", "Unknown") + if isinstance(label, list): + label = label[0] + river_name = station.get("riverName", "") + lat = station.get("lat") + lon = station.get("long") + + measures = station.get("measures", []) + if isinstance(measures, dict): + measures = [measures] + + for measure in measures: + if measure.get("parameter") != "level": + continue + latest = measure.get("latestReading") + if not latest: + continue + readings.append( + Reading( + station_id=station_id, + label=label, + river_name=river_name, + lat=lat, + lon=lon, + measure_id=measure.get("@id", ""), + unit=measure.get("unitName"), + qualifier=measure.get("qualifier"), + value=latest.get("value"), + date_time=latest.get("dateTime"), + ) + ) + return readings + + +# -------------------------------------------------------------------------- +# Storage +# -------------------------------------------------------------------------- + +def init_db(db_path: Path) -> sqlite3.Connection: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(db_path) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS stations ( + station_id TEXT PRIMARY KEY, + label TEXT, + river_name TEXT, + lat REAL, + lon REAL + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS readings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + station_id TEXT NOT NULL, + measure_id TEXT NOT NULL, + unit TEXT, + qualifier TEXT, + value REAL, + date_time TEXT NOT NULL, + fetched_at TEXT NOT NULL, + UNIQUE(measure_id, date_time) + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_readings_station_time ON readings(station_id, date_time)" + ) + conn.commit() + return conn + + +def store_readings(conn: sqlite3.Connection, readings: list[Reading]) -> int: + fetched_at = datetime.now(timezone.utc).isoformat() + new_rows = 0 + for r in readings: + conn.execute( + """ + INSERT INTO stations (station_id, label, river_name, lat, lon) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(station_id) DO UPDATE SET + label=excluded.label, river_name=excluded.river_name, + lat=excluded.lat, lon=excluded.lon + """, + (r.station_id, r.label, r.river_name, r.lat, r.lon), + ) + cur = conn.execute( + """ + INSERT OR IGNORE INTO readings + (station_id, measure_id, unit, qualifier, value, date_time, fetched_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + (r.station_id, r.measure_id, r.unit, r.qualifier, r.value, r.date_time, fetched_at), + ) + new_rows += cur.rowcount + conn.commit() + return new_rows + + +# -------------------------------------------------------------------------- +# Dashboard generation +# -------------------------------------------------------------------------- + +def build_dashboard(conn: sqlite3.Connection, output_path: Path, history_points: int = 50) -> None: + stations = conn.execute( + "SELECT station_id, label, river_name, lat, lon FROM stations ORDER BY label" + ).fetchall() + + station_payload = [] + for station_id, label, river_name, lat, lon in stations: + rows = conn.execute( + """ + SELECT date_time, value, unit FROM readings + WHERE station_id = ? + ORDER BY date_time DESC LIMIT ? + """, + (station_id, history_points), + ).fetchall() + if not rows: + continue + rows = list(reversed(rows)) # chronological order for charting + latest_time, latest_value, unit = rows[-1] + station_payload.append( + { + "id": station_id, + "label": label, + "river": river_name or "", + "lat": lat, + "lon": lon, + "unit": unit, + "latest_value": latest_value, + "latest_time": latest_time, + "history": [{"t": t, "v": v} for t, v, _ in rows], + } + ) + + generated_at = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") + data_json = json.dumps(station_payload) + + html = HTML_TEMPLATE.replace("__DATA__", data_json).replace("__GENERATED__", generated_at) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(html, encoding="utf-8") + + +HTML_TEMPLATE = """ + + + +River Levels Dashboard + + + + +

River Levels Dashboard

+
Generated __GENERATED__ · Data: Environment Agency real-time flood-monitoring API (Open Government Licence)
+
+ + + + + +""" + + +# -------------------------------------------------------------------------- +# Config persistence (so you don't have to pass --lat/--lon every time) +# -------------------------------------------------------------------------- + +def load_or_create_config(args: argparse.Namespace, session: requests.Session) -> dict: + config_path = Path(args.config) + + if args.lat is not None and args.lon is not None: + lat, lon = args.lat, args.lon + elif args.postcode: + lat, lon = resolve_postcode(args.postcode, session) + elif config_path.exists(): + saved = json.loads(config_path.read_text()) + return saved + else: + raise SystemExit( + "No location known yet. Provide --postcode 'XXX YYY' or --lat/--lon " + "on the first run; it will be remembered for future runs." + ) + + cfg = { + "lat": lat, + "lon": lon, + "radius_km": args.radius, + "db": str(args.db), + "output": str(args.output), + } + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text(json.dumps(cfg, indent=2)) + return cfg + + +# -------------------------------------------------------------------------- +# Main +# -------------------------------------------------------------------------- + +def poll_once(cfg: dict, session: requests.Session, verbose: bool = True) -> None: + stations = fetch_level_stations(session, cfg["lat"], cfg["lon"], cfg["radius_km"]) + readings = extract_readings(stations) + conn = init_db(Path(cfg["db"])) + new_rows = store_readings(conn, readings) + build_dashboard(conn, Path(cfg["output"])) + conn.close() + if verbose: + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print( + f"[{ts}] {len(readings)} readings from {len(stations)} stations " + f"({new_rows} new) -> {cfg['output']}" + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Poll and archive UK river levels.") + parser.add_argument("--postcode", help="UK postcode to centre the search on (first run only)") + parser.add_argument("--lat", type=float, help="Latitude to centre the search on") + parser.add_argument("--lon", type=float, help="Longitude to centre the search on") + parser.add_argument("--radius", type=float, default=15.0, help="Search radius in km (default: 15)") + parser.add_argument("--db", type=Path, default=DEFAULT_DB, help="Path to SQLite archive") + parser.add_argument("--output", type=Path, default=DEFAULT_HTML, help="Path to write dashboard HTML") + parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG, help="Path to saved location config") + parser.add_argument("--daemon", action="store_true", help="Run continuously instead of once") + parser.add_argument("--interval", type=int, default=900, help="Seconds between polls in --daemon mode (default: 900)") + args = parser.parse_args() + + session = requests.Session() + session.headers.update({"User-Agent": USER_AGENT}) + + try: + cfg = load_or_create_config(args, session) + except (requests.RequestException, ValueError) as exc: + print(f"Error resolving location: {exc}", file=sys.stderr) + sys.exit(1) + + print(f"Location: {cfg['lat']:.5f}, {cfg['lon']:.5f} (radius {cfg['radius_km']} km)") + + if args.daemon: + print(f"Running as daemon, polling every {args.interval}s. Ctrl+C to stop.") + while True: + try: + poll_once(cfg, session) + except requests.RequestException as exc: + print(f"Poll failed: {exc}", file=sys.stderr) + time.sleep(args.interval) + else: + try: + poll_once(cfg, session) + except requests.RequestException as exc: + print(f"Poll failed: {exc}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main()