From 9a0308d3801b4ce45da10df62c815adf6debe978 Mon Sep 17 00:00:00 2001
From: Simon Moore <simon.moore@theatari.com>
Date: Sun, 5 Jul 2026 00:25:46 +0100
Subject: [PATCH] Add --backfill flag to fetch historical readings from EA API

Pulls up to 28 days of 15-minute readings per station using the
/measures/{id}/readings?since= endpoint.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 river_monitor.py | 70 +++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 66 insertions(+), 4 deletions(-)

diff --git a/river_monitor.py b/river_monitor.py
index 9cf086e..01f971b 100644
--- a/river_monitor.py
+++ b/river_monitor.py
@@ -36,7 +36,7 @@ import sqlite3
 import sys
 import time
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Iterable, Optional
 
@@ -111,6 +111,55 @@ def fetch_level_stations(
     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 extract_readings(stations: Iterable[dict]) -> list[Reading]:
     readings: list[Reading] = []
     for station in stations:
@@ -388,10 +437,15 @@ def load_or_create_config(args: argparse.Namespace, session: requests.Session) -
 # Main
 # --------------------------------------------------------------------------
 
-def poll_once(cfg: dict, session: requests.Session, verbose: bool = True) -> None:
+def poll_once(cfg: dict, session: requests.Session, verbose: bool = True, backfill_days: int = 0) -> None:
     stations = fetch_level_stations(session, cfg["lat"], cfg["lon"], cfg["radius_km"])
-    readings = extract_readings(stations)
     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(stations)
     new_rows = store_readings(conn, readings)
     build_dashboard(conn, Path(cfg["output"]))
     conn.close()
@@ -414,6 +468,8 @@ def main() -> None:
     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)")
     args = parser.parse_args()
 
     session = requests.Session()
@@ -427,7 +483,13 @@ def main() -> None:
 
     print(f"Location: {cfg['lat']:.5f}, {cfg['lon']:.5f}  (radius {cfg['radius_km']} km)")
 
-    if args.daemon:
+    if args.backfill:
+        try:
+            poll_once(cfg, session, backfill_days=min(args.backfill_days, 28))
+        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: