From d72496f404a3d90fc92618950e578c2702648d10 Mon Sep 17 00:00:00 2001 From: grabowski Date: Tue, 11 Aug 2026 15:11:08 +0700 Subject: [PATCH] feat: backfill hii_waterlevel from the HII waterlevel_graph archive scripts/backfill_hii_waterlevel.py walks the api-v3 waterlevel_graph endpoint (hourly wl_msl + discharge, archive back to ~2019) in full-year windows per station and upserts into hii_waterlevel. Defaults to the RID-mirror and key stations; --stations/--all/--start/--end/--chunk-days override. History upserts touch only wl_msl and discharge so colliding live-snapshot rows keep storage_percent/situation_level. Idempotent and safe to re-run. --- docs/DATA_SOURCES.md | 2 +- scripts/backfill_hii_waterlevel.py | 18 +++ src/hii_backfill.py | 222 +++++++++++++++++++++++++++++ src/hii_collector.py | 43 +++++- tests/test_hii_collector.py | 103 +++++++++++++ 5 files changed, 383 insertions(+), 5 deletions(-) create mode 100644 scripts/backfill_hii_waterlevel.py create mode 100644 src/hii_backfill.py diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md index 1c18de4..36588e1 100644 --- a/docs/DATA_SOURCES.md +++ b/docs/DATA_SOURCES.md @@ -238,7 +238,7 @@ Oct 2024 flood). Mae Kuang Udom Thara is the second upstream reservoir. | ThaiWater `/v2/waterlevel` | 🟡 Display-only (`src/thaiwater.py`, needs `THAIWATER_API_KEY`, never persisted) | Optionally persist | | HII `rain_24h` | ✅ Ingested hourly via `src/hii_collector.py` → `hii_rain_stations` + `hii_rainfall` (Ping-filtered; ~300 stations) | — | | HII `waterlevel_load` | ✅ Ingested hourly via `src/hii_collector.py` → `hii_wl_stations` + `hii_waterlevel` (125 Ping stations, m MSL; `rid_code` column maps mirrors like `ridhydro_P.1` → `P.1`, `offset_msl` converts MSL → gauge datum) | — | -| HII `waterlevel_graph` | ❌ Not ingested | Backfill script to cross-validate / gap-fill RID history (≥2019) | +| HII `waterlevel_graph` | ✅ Backfill via `scripts/backfill_hii_waterlevel.py` → `hii_waterlevel` (hourly MSL + discharge, archive ≥2019; full-year windows per request; upserts never overwrite live-snapshot columns) | Run once on the box: `python scripts/backfill_hii_waterlevel.py` (defaults: 2019-01-01 → today, RID-mirror + key stations; `--stations P.1,P.67`, `--all` for every Ping station) | | Mae Ngat reservoir (lsim.rid.go.th) | ❌ Not ingested | Add daily storage/release scrape | | Satellite QPE (GSMaP/IMERG) | ❌ | Backfill training rainfall (GEE) | | Open-Meteo forecasts | ❌ | Add forecast features (live + 2021 archive for training) | diff --git a/scripts/backfill_hii_waterlevel.py b/scripts/backfill_hii_waterlevel.py new file mode 100644 index 0000000..b7cc854 --- /dev/null +++ b/scripts/backfill_hii_waterlevel.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""CLI entry point for backfilling hii_waterlevel from the HII archive. + +Usage: + python scripts/backfill_hii_waterlevel.py # key stations, 2019..today + python scripts/backfill_hii_waterlevel.py --stations P.1,P.67 + python scripts/backfill_hii_waterlevel.py --start 2024-09-01 --end 2024-11-01 --all +""" + +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.hii_backfill import main + +if __name__ == "__main__": + sys.exit(0 if main() else 1) diff --git a/src/hii_backfill.py b/src/hii_backfill.py new file mode 100644 index 0000000..a86f167 --- /dev/null +++ b/src/hii_backfill.py @@ -0,0 +1,222 @@ +"""Backfill historical water levels from the HII waterlevel_graph endpoint. + +The api-v3 waterlevel_graph archive reaches back to ~2019 with hourly +wl_msl + discharge. This module walks a date range in chunks per station and +upserts into hii_waterlevel (idempotent; safe to re-run and to overlap with +the live snapshot collector). Station metadata comes from a live +waterlevel_load fetch, so hii_wl_stations is populated/refreshed as a side +effect. + +Usage: python scripts/backfill_hii_waterlevel.py --start 2019-01-01 +""" + +import argparse +import datetime +import logging +import time +from typing import Dict, List, Optional + +from .hii_collector import ( + HiiClient, + HiiStore, + PING_BASIN_CODE, + _parse_datetime, + _to_float, +) + +logger = logging.getLogger(__name__) + +DEFAULT_START = datetime.date(2019, 1, 1) +DEFAULT_CHUNK_DAYS = 365 # full-year windows verified working (8,760 rows, ~700KB) +DEFAULT_SLEEP_SECONDS = 1.0 + + +def parse_graph_rows(payload: Dict) -> List[Dict]: + """Extract history rows from a waterlevel_graph payload (skips empty rows).""" + rows = (payload.get("data") or {}).get("graph_data") or [] + records = [] + for row in rows: + timestamp = _parse_datetime(row.get("datetime")) + wl_msl = _to_float(row.get("value")) + discharge = _to_float(row.get("discharge")) + if timestamp is None or (wl_msl is None and discharge is None): + continue + records.append( + {"timestamp": timestamp, "wl_msl": wl_msl, "discharge": discharge} + ) + return records + + +def fetch_waterlevel_history( + client: HiiClient, + station_id: int, + start_date: datetime.date, + end_date: datetime.date, +) -> List[Dict]: + """Hourly wl_msl + discharge history (archive reaches back to ~2019).""" + payload = client.get( + "waterlevel_graph", + params={ + "station_type": "tele_waterlevel", + "station_id": station_id, + "start_date": start_date.isoformat(), + "end_date": end_date.isoformat(), + }, + ) + return parse_graph_rows(payload) + + +def chunk_date_range( + start: datetime.date, end: datetime.date, chunk_days: int +) -> List[tuple]: + """Split [start, end] into inclusive (start, end) windows.""" + chunks = [] + cursor = start + while cursor <= end: + chunk_end = min(cursor + datetime.timedelta(days=chunk_days - 1), end) + chunks.append((cursor, chunk_end)) + cursor = chunk_end + datetime.timedelta(days=1) + return chunks + + +def select_stations( + station_records: List[Dict], + codes: Optional[List[str]] = None, + all_stations: bool = False, +) -> List[Dict]: + """Pick stations to backfill from parsed waterlevel_load records. + + Default: stations that mirror a RID gauge (rid_code) or are flagged + is_key_station — the ones relevant to the flood model. Explicit codes + match rid_code or oldcode; --all takes every station in the basin. + """ + if all_stations: + return station_records + if codes: + wanted = {c.strip().upper() for c in codes if c.strip()} + return [ + r + for r in station_records + if (r.get("rid_code") or "").upper() in wanted + or (r.get("oldcode") or "").upper() in wanted + ] + return [r for r in station_records if r.get("rid_code") or r.get("is_key_station")] + + +def backfill( + store: HiiStore, + client: Optional[HiiClient] = None, + start: datetime.date = DEFAULT_START, + end: Optional[datetime.date] = None, + codes: Optional[List[str]] = None, + all_stations: bool = False, + chunk_days: int = DEFAULT_CHUNK_DAYS, + sleep_seconds: float = DEFAULT_SLEEP_SECONDS, + basin_code: int = PING_BASIN_CODE, +) -> Dict[str, int]: + """Run the backfill; returns {'stations': n, 'rows': n, 'errors': n}.""" + client = client or HiiClient() + end = end or datetime.date.today() + + logger.info("Fetching station catalog from waterlevel_load...") + station_records = client.fetch_waterlevel(basin_code) + # Refresh station metadata (and today's snapshot) while we have it + store.save_waterlevel(station_records) + + stations = select_stations(station_records, codes=codes, all_stations=all_stations) + if not stations: + logger.error("No stations matched the selection") + return {"stations": 0, "rows": 0, "errors": 0} + + chunks = chunk_date_range(start, end, chunk_days) + logger.info( + f"Backfilling {len(stations)} stations x {len(chunks)} windows " + f"({start} .. {end}, {chunk_days}-day chunks)" + ) + + totals = {"stations": len(stations), "rows": 0, "errors": 0} + for station in stations: + sid = station["station_id"] + label = station.get("rid_code") or station.get("oldcode") or str(sid) + station_rows = 0 + for chunk_start, chunk_end in chunks: + try: + rows = fetch_waterlevel_history(client, sid, chunk_start, chunk_end) + station_rows += store.save_waterlevel_history(sid, rows) + except Exception as e: + totals["errors"] += 1 + logger.warning( + f"{label}: {chunk_start}..{chunk_end} failed: {e}" + ) + time.sleep(sleep_seconds) + totals["rows"] += station_rows + logger.info(f"{label} (id {sid}): {station_rows} rows saved") + + logger.info( + f"Backfill complete: {totals['rows']} rows across " + f"{totals['stations']} stations, {totals['errors']} failed windows" + ) + return totals + + +def main(argv: Optional[List[str]] = None) -> bool: + parser = argparse.ArgumentParser( + description="Backfill hii_waterlevel from the HII waterlevel_graph archive" + ) + parser.add_argument( + "--start", + type=datetime.date.fromisoformat, + default=DEFAULT_START, + help=f"First date to fetch (default {DEFAULT_START})", + ) + parser.add_argument( + "--end", + type=datetime.date.fromisoformat, + default=None, + help="Last date to fetch (default today)", + ) + parser.add_argument( + "--stations", + help="Comma-separated codes (rid_code or oldcode, e.g. P.1,P.67,CHM004). " + "Default: all RID-mirror and key stations", + ) + parser.add_argument( + "--all", + action="store_true", + help="Backfill every Ping-basin station (125+; slow)", + ) + parser.add_argument( + "--chunk-days", type=int, default=DEFAULT_CHUNK_DAYS, help="Window size" + ) + parser.add_argument( + "--sleep", + type=float, + default=DEFAULT_SLEEP_SECONDS, + help="Pause between requests in seconds", + ) + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" + ) + + from .config import Config + + db_config = Config.get_database_config() + if db_config["type"] not in ("sqlite", "postgresql", "mysql"): + logger.error(f"Backfill requires a SQL DB_TYPE, got '{db_config['type']}'") + return False + store = HiiStore(db_config["connection_string"], db_config["type"]) + if not store.connect(): + return False + + totals = backfill( + store, + start=args.start, + end=args.end, + codes=args.stations.split(",") if args.stations else None, + all_stations=args.all, + chunk_days=args.chunk_days, + sleep_seconds=args.sleep, + ) + return totals["rows"] > 0 and totals["errors"] == 0 diff --git a/src/hii_collector.py b/src/hii_collector.py index a097cda..01aab1b 100644 --- a/src/hii_collector.py +++ b/src/hii_collector.py @@ -162,18 +162,18 @@ class HiiClient: self.session = session or requests.Session() self.timeout = timeout - def _get(self, endpoint: str) -> Dict: + def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict: response = self.session.get( - f"{self.base_url}/{endpoint}", timeout=self.timeout + f"{self.base_url}/{endpoint}", params=params, timeout=self.timeout ) response.raise_for_status() return response.json() def fetch_rain(self, basin_code: int = PING_BASIN_CODE) -> List[Dict]: - return parse_rain_records(self._get("rain_24h"), basin_code) + return parse_rain_records(self.get("rain_24h"), basin_code) def fetch_waterlevel(self, basin_code: int = PING_BASIN_CODE) -> List[Dict]: - return parse_waterlevel_records(self._get("waterlevel_load"), basin_code) + return parse_waterlevel_records(self.get("waterlevel_load"), basin_code) class HiiStore: @@ -350,6 +350,41 @@ class HiiStore: ], ) + def save_waterlevel_history(self, station_id: int, rows: List[Dict]) -> int: + """Upsert backfilled history rows, touching only wl_msl and discharge. + + Live-snapshot rows for the same (station, hour) keep their extra + columns (storage_percent, situation_level, ...) untouched. + """ + if not rows: + return 0 + if not self.engine and not self.connect(): + return 0 + from sqlalchemy import text + + cols = "(station_id, timestamp, wl_msl, discharge)" + values = "(:station_id, :timestamp, :wl_msl, :discharge)" + if self.db_type == "mysql": + sql = ( + f"INSERT INTO hii_waterlevel {cols} VALUES {values} " + "ON DUPLICATE KEY UPDATE wl_msl = VALUES(wl_msl), " + "discharge = VALUES(discharge)" + ) + else: # sqlite (>=3.24) and postgresql share upsert syntax + sql = ( + f"INSERT INTO hii_waterlevel {cols} VALUES {values} " + "ON CONFLICT (station_id, timestamp) DO UPDATE SET " + "wl_msl = EXCLUDED.wl_msl, discharge = EXCLUDED.discharge" + ) + params = [{**row, "station_id": station_id} for row in rows] + try: + with self.engine.begin() as conn: + conn.execute(text(sql), params) + return len(params) + except Exception as e: + logger.error(f"HiiStore history save failed: {e}") + return 0 + def _save( self, records: List[Dict], diff --git a/tests/test_hii_collector.py b/tests/test_hii_collector.py index 6ab38e4..c90f9c8 100644 --- a/tests/test_hii_collector.py +++ b/tests/test_hii_collector.py @@ -4,6 +4,7 @@ import datetime import pytest +from src.hii_backfill import chunk_date_range, parse_graph_rows, select_stations from src.hii_collector import ( HiiStore, parse_rain_records, @@ -207,3 +208,105 @@ class TestHiiStore: def test_save_empty(self, store): assert store.save_rain([]) == 0 + + def test_history_upsert_preserves_snapshot_columns(self, store): + # A live snapshot row exists with extra columns populated + store.save_waterlevel(parse_waterlevel_records(_waterlevel_payload())) + # Backfill collides on the same (station, timestamp) with new values + rows = [ + { + "timestamp": datetime.datetime(2026, 8, 11, 13, 0), + "wl_msl": 303.30, + "discharge": 340.0, + }, + { + "timestamp": datetime.datetime(2019, 8, 1, 1, 0), + "wl_msl": 301.71, + "discharge": 11.7, + }, + ] + assert store.save_waterlevel_history(3226, rows) == 2 + + from sqlalchemy import text + + with store.engine.connect() as conn: + collided = conn.execute( + text( + "SELECT wl_msl, discharge, storage_percent, situation_level " + "FROM hii_waterlevel WHERE station_id = 3226 " + "AND timestamp = '2026-08-11 13:00:00'" + ) + ).fetchone() + historical = conn.execute( + text( + "SELECT wl_msl FROM hii_waterlevel WHERE station_id = 3226 " + "AND timestamp = '2019-08-01 01:00:00'" + ) + ).fetchone() + # wl_msl/discharge updated, snapshot-only columns untouched + assert float(collided[0]) == 303.30 + assert float(collided[1]) == 340.0 + assert float(collided[2]) == 81.21 + assert collided[3] == 4 + assert float(historical[0]) == 301.71 + + +class TestParseGraphRows: + def test_parses_and_skips_empty(self): + payload = { + "data": { + "graph_data": [ + {"datetime": "2024-10-05 12:00", "value": 305.8, "discharge": 656}, + {"datetime": "2024-10-05 13:00", "value": None, "discharge": None}, + {"datetime": None, "value": 300.0, "discharge": 1}, + ] + } + } + rows = parse_graph_rows(payload) + assert rows == [ + { + "timestamp": datetime.datetime(2024, 10, 5, 12, 0), + "wl_msl": 305.8, + "discharge": 656.0, + } + ] + + def test_empty_payload(self): + assert parse_graph_rows({}) == [] + + +class TestBackfillHelpers: + def test_chunk_date_range(self): + chunks = chunk_date_range( + datetime.date(2024, 1, 1), datetime.date(2024, 3, 1), 31 + ) + assert chunks[0] == (datetime.date(2024, 1, 1), datetime.date(2024, 1, 31)) + assert chunks[-1][1] == datetime.date(2024, 3, 1) + # Contiguous, no overlap + for (_, prev_end), (next_start, _) in zip(chunks, chunks[1:]): + assert next_start == prev_end + datetime.timedelta(days=1) + + def test_chunk_single_day(self): + d = datetime.date(2024, 1, 1) + assert chunk_date_range(d, d, 365) == [(d, d)] + + def test_select_default_keeps_rid_and_key_stations(self): + records = [ + {"station_id": 1, "rid_code": "P.1", "is_key_station": True}, + {"station_id": 2, "rid_code": None, "is_key_station": False}, + {"station_id": 3, "rid_code": None, "is_key_station": True}, + ] + assert [r["station_id"] for r in select_stations(records)] == [1, 3] + + def test_select_by_code_matches_rid_code_and_oldcode(self): + records = [ + {"station_id": 1, "rid_code": "P.1", "oldcode": "ridhydro_P.1"}, + {"station_id": 2, "rid_code": None, "oldcode": "CHM004"}, + {"station_id": 3, "rid_code": "P.67", "oldcode": "P.67"}, + ] + selected = select_stations(records, codes=["p.1", "chm004"]) + assert [r["station_id"] for r in selected] == [1, 2] + + def test_select_all(self): + records = [{"station_id": 1}, {"station_id": 2}] + assert select_stations(records, all_stations=True) == records