448 lines
15 KiB
Python
448 lines
15 KiB
Python
#!/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 = """<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<title>River Levels Dashboard</title>
|
|
<script src="https://cdn.jsdelivr.net/npm/chart.js@4"></script>
|
|
<style>
|
|
:root {
|
|
--bg: #0f172a; --card: #1e293b; --text: #e2e8f0; --muted: #94a3b8;
|
|
--accent: #38bdf8; --ok: #34d399; --warn: #fbbf24;
|
|
}
|
|
* { box-sizing: border-box; }
|
|
body {
|
|
margin: 0; padding: 24px; background: var(--bg); color: var(--text);
|
|
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
|
|
}
|
|
h1 { font-size: 1.4rem; margin-bottom: 4px; }
|
|
.meta { color: var(--muted); font-size: 0.85rem; margin-bottom: 24px; }
|
|
.grid {
|
|
display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr));
|
|
gap: 16px;
|
|
}
|
|
.card {
|
|
background: var(--card); border-radius: 10px; padding: 16px;
|
|
border: 1px solid #33415580;
|
|
}
|
|
.card h2 { font-size: 1rem; margin: 0 0 2px 0; }
|
|
.card .river { color: var(--accent); font-size: 0.8rem; margin-bottom: 8px; }
|
|
.value { font-size: 1.8rem; font-weight: 600; }
|
|
.unit { font-size: 0.9rem; color: var(--muted); font-weight: 400; }
|
|
.time { color: var(--muted); font-size: 0.75rem; margin-top: 2px; }
|
|
canvas { margin-top: 10px; max-height: 80px; }
|
|
.empty { color: var(--muted); padding: 40px; text-align: center; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>River Levels Dashboard</h1>
|
|
<div class="meta">Generated __GENERATED__ · Data: Environment Agency real-time flood-monitoring API (Open Government Licence)</div>
|
|
<div id="grid" class="grid"></div>
|
|
<div id="empty" class="empty" style="display:none">No station data yet — run the poller first.</div>
|
|
|
|
<script>
|
|
const stations = __DATA__;
|
|
const grid = document.getElementById('grid');
|
|
|
|
if (!stations.length) {
|
|
document.getElementById('empty').style.display = 'block';
|
|
}
|
|
|
|
stations.forEach((s, i) => {
|
|
const card = document.createElement('div');
|
|
card.className = 'card';
|
|
const t = new Date(s.latest_time).toLocaleString();
|
|
card.innerHTML = `
|
|
<h2>${s.label}</h2>
|
|
<div class="river">${s.river || ''}</div>
|
|
<div class="value">${s.latest_value != null ? s.latest_value.toFixed(2) : '—'}
|
|
<span class="unit">${s.unit || ''}</span></div>
|
|
<div class="time">${t}</div>
|
|
<canvas id="chart-${i}"></canvas>
|
|
`;
|
|
grid.appendChild(card);
|
|
|
|
const ctx = document.getElementById(`chart-${i}`).getContext('2d');
|
|
new Chart(ctx, {
|
|
type: 'line',
|
|
data: {
|
|
labels: s.history.map(h => h.t),
|
|
datasets: [{
|
|
data: s.history.map(h => h.v),
|
|
borderColor: '#38bdf8',
|
|
borderWidth: 2,
|
|
pointRadius: 0,
|
|
tension: 0.3,
|
|
fill: false,
|
|
}]
|
|
},
|
|
options: {
|
|
responsive: true,
|
|
plugins: { legend: { display: false } },
|
|
scales: { x: { display: false }, y: { display: false } },
|
|
elements: { line: { borderJoinStyle: 'round' } },
|
|
}
|
|
});
|
|
});
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 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()
|