#!/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, timedelta, 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 fetch_historical_readings(
session: requests.Session, stations: list[dict], days: int
) -> list[Reading]:
since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")
params = {
"since": since,
"_limit": 3000,
}
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
measure_id = measure.get("@id", "")
if not measure_id:
continue
try:
resp = session.get(f"{measure_id}/readings", params=params, timeout=30)
resp.raise_for_status()
except requests.RequestException as exc:
print(f" Warning: could not fetch history for {label}: {exc}", file=sys.stderr)
continue
for item in resp.json().get("items", []):
readings.append(Reading(
station_id=station_id,
label=label,
river_name=river_name,
lat=lat,
lon=lon,
measure_id=measure_id,
unit=measure.get("unitName"),
qualifier=measure.get("qualifier"),
value=item.get("value"),
date_time=item.get("dateTime"),
))
print(f" {label}: {len([r for r in readings if r.station_id == station_id])} readings", flush=True)
return readings
def fetch_latest_reading(session: requests.Session, measure_id: str) -> Optional[dict]:
"""Fetch a measure's single most recent reading directly. Some stations
never populate the stations-endpoint's embedded `latestReading` field,
even though the measure's own /readings endpoint has current data."""
try:
resp = session.get(
f"{measure_id}/readings", params={"_sorted": "", "_limit": 1}, timeout=15
)
resp.raise_for_status()
except requests.RequestException:
return None
items = resp.json().get("items", [])
return items[0] if items else None
def extract_readings(session: requests.Session, 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
measure_id = measure.get("@id", "")
latest = measure.get("latestReading")
if not latest and measure_id:
latest = fetch_latest_reading(session, measure_id)
if not latest:
continue
readings.append(
Reading(
station_id=station_id,
label=label,
river_name=river_name,
lat=lat,
lon=lon,
measure_id=measure_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)
No station data yet — run the poller first.
"""
# --------------------------------------------------------------------------
# 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, backfill_days: int = 0, no_html: bool = False) -> None:
stations = fetch_level_stations(session, cfg["lat"], cfg["lon"], cfg["radius_km"])
conn = init_db(Path(cfg["db"]))
if backfill_days:
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{ts}] Backfilling {backfill_days} days of history for {len(stations)} stations...")
readings = fetch_historical_readings(session, stations, backfill_days)
else:
readings = extract_readings(session, stations)
new_rows = store_readings(conn, readings)
if not no_html:
build_dashboard(conn, Path(cfg["output"]))
conn.close()
if verbose:
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
suffix = "" if no_html else f" -> {cfg['output']}"
print(
f"[{ts}] {len(readings)} readings from {len(stations)} stations "
f"({new_rows} new){suffix}"
)
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)")
parser.add_argument("--backfill", action="store_true", help="Fetch full history (up to --backfill-days) for all stations, then exit")
parser.add_argument("--backfill-days", type=int, default=28, dest="backfill_days", help="How many days of history to backfill (default: 28, max: 28)")
parser.add_argument("--no-html", action="store_true", dest="no_html", help="Skip HTML dashboard generation (use when a web page reads the DB directly)")
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.backfill:
try:
poll_once(cfg, session, backfill_days=min(args.backfill_days, 28), no_html=args.no_html)
except requests.RequestException as exc:
print(f"Backfill failed: {exc}", file=sys.stderr)
sys.exit(1)
elif args.daemon:
print(f"Running as daemon, polling every {args.interval}s. Ctrl+C to stop.")
while True:
try:
poll_once(cfg, session, no_html=args.no_html)
except requests.RequestException as exc:
print(f"Poll failed: {exc}", file=sys.stderr)
time.sleep(args.interval)
else:
try:
poll_once(cfg, session, no_html=args.no_html)
except requests.RequestException as exc:
print(f"Poll failed: {exc}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()