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>
This commit is contained in:
2026-07-05 00:25:46 +01:00
parent afe742dcce
commit 9a0308d380

View File

@@ -36,7 +36,7 @@ import sqlite3
import sys import sys
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path from pathlib import Path
from typing import Iterable, Optional from typing import Iterable, Optional
@@ -111,6 +111,55 @@ def fetch_level_stations(
return resp.json().get("items", []) 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]: def extract_readings(stations: Iterable[dict]) -> list[Reading]:
readings: list[Reading] = [] readings: list[Reading] = []
for station in stations: for station in stations:
@@ -388,10 +437,15 @@ def load_or_create_config(args: argparse.Namespace, session: requests.Session) -
# Main # 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"]) stations = fetch_level_stations(session, cfg["lat"], cfg["lon"], cfg["radius_km"])
readings = extract_readings(stations)
conn = init_db(Path(cfg["db"])) 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) new_rows = store_readings(conn, readings)
build_dashboard(conn, Path(cfg["output"])) build_dashboard(conn, Path(cfg["output"]))
conn.close() 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("--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("--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("--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() args = parser.parse_args()
session = requests.Session() session = requests.Session()
@@ -427,7 +483,13 @@ def main() -> None:
print(f"Location: {cfg['lat']:.5f}, {cfg['lon']:.5f} (radius {cfg['radius_km']} km)") 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.") print(f"Running as daemon, polling every {args.interval}s. Ctrl+C to stop.")
while True: while True:
try: try: