From 382daa7d86c132f5d60fee0af23b3ebc447b9d75 Mon Sep 17 00:00:00 2001 From: grabowski Date: Thu, 13 Aug 2026 23:10:36 +0700 Subject: [PATCH] perf: backfill one dam per request via the api/dam range endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET app.rid.go.th/reservoir/api/dam?dam_id&date_start&date_end returns a single dam's whole date range in one response — Mae Ngat's 2009-today archive is ~4 chunked requests instead of the ~2,900 one-day POSTs the all-dams path needs. Field names differ from api/dams and are mapped in parse_dam_range_records, verified equal on spot-checked dates; the range endpoint also carries DMD_ULevel, the reservoir level in m MSL that api/dams stopped publishing after ~2013. scripts/backfill_rid_reservoir.py defaults to the fast per-dam path (--all-dams keeps the full-fleet crawl, --refresh rewrites stored days). Already-stored dates are still skipped, junk values are still bounded, and the consecutive-failure abort still applies. --- scripts/backfill_rid_reservoir.py | 79 ++++++- src/rid_reservoir.py | 278 ++++++++++++++++++++-- tests/test_rid_reservoir.py | 377 ++++++++++++++++++++++++++++++ 3 files changed, 705 insertions(+), 29 deletions(-) diff --git a/scripts/backfill_rid_reservoir.py b/scripts/backfill_rid_reservoir.py index a088871..17f3f30 100644 --- a/scripts/backfill_rid_reservoir.py +++ b/scripts/backfill_rid_reservoir.py @@ -1,14 +1,20 @@ #!/usr/bin/env python3 """Backfill rid_reservoir_daily with RID large-dam history (Mae Ngat et al.). -One request per day against app.rid.go.th/reservoir/api/dams (archive reaches -back to at least 2009). Days already stored are skipped, so reruns only fetch -what is missing — safe alongside the hourly live collector, and a rerun -repairs holes left by transient failures. +Two paths, both idempotent and both skipping what is already stored, so a +rerun repairs holes left by transient failures and is safe alongside the +hourly live collector: + + --dam-id (default: Mae Ngat) one dam, whole range, via api/dam — a handful + of requests for the entire 2009-today archive + --all-dams all ~35 dams, one request per calendar day via + api/dams — thousands of requests, ~25 minutes Usage: - uv run scripts/backfill_rid_reservoir.py # missing days since 2018-08-01 - uv run scripts/backfill_rid_reservoir.py --start 2015-01-01 + uv run scripts/backfill_rid_reservoir.py # Mae Ngat since 2018-08-01 + uv run scripts/backfill_rid_reservoir.py --start 2009-01-01 # full archive + uv run scripts/backfill_rid_reservoir.py --refresh # rewrite stored days too + uv run scripts/backfill_rid_reservoir.py --all-dams --start 2015-01-01 uv run scripts/backfill_rid_reservoir.py --db-url postgresql://... """ @@ -21,7 +27,12 @@ import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from src.config import Config -from src.rid_reservoir import RidReservoirStore, backfill +from src.rid_reservoir import ( + MAE_NGAT_DAM_ID, + RidReservoirStore, + backfill, + backfill_dam, +) DEFAULT_START = datetime.date(2018, 8, 1) # start of the water_measurements grid @@ -34,6 +45,27 @@ def main(argv=None) -> int: parser.add_argument("--end", type=datetime.date.fromisoformat, default=None) parser.add_argument("--db-url", default=None) parser.add_argument("--throttle", type=float, default=0.4) + parser.add_argument( + "--dam-id", + default=MAE_NGAT_DAM_ID, + help="dam to backfill via the fast range endpoint (default Mae Ngat)", + ) + parser.add_argument( + "--all-dams", + action="store_true", + help="every dam, one request per calendar day (slow full-fleet path)", + ) + parser.add_argument( + "--chunk-days", + type=int, + default=1830, + help="days per range request; the endpoint imposes no limit of its own", + ) + parser.add_argument( + "--refresh", + action="store_true", + help="re-fetch days already stored (adds level_msl to api/dams rows)", + ) args = parser.parse_args(argv) logging.basicConfig( @@ -59,10 +91,35 @@ def main(argv=None) -> int: return 1 end = args.end or datetime.date.today() span_days = (end - args.start).days + 1 - missing = span_days - len(store.present_dates(args.start, end)) - saved = backfill(store, args.start, end, throttle_seconds=args.throttle) - print(f"backfilled {saved} dam-day rows ({missing} days were missing)") - return 0 if saved or missing == 0 else 1 + dam_id = None if args.all_dams else args.dam_id + missing = span_days - len(store.present_dates(args.start, end, dam_id=dam_id)) + if args.all_dams: + saved = backfill(store, args.start, end, throttle_seconds=args.throttle) + print(f"backfilled {saved} dam-day rows ({missing} days were missing)") + return 0 if saved or missing == 0 else 1 + + stats = {} + saved = backfill_dam( + store, + dam_id=args.dam_id, + start=args.start, + end=end, + chunk_days=args.chunk_days, + throttle_seconds=max(args.throttle, 1.0), + skip_present=not args.refresh, + stats=stats, + ) + still_missing = span_days - len( + store.present_dates(args.start, end, dam_id=args.dam_id) + ) + print( + f"backfilled {saved} dam-day rows for dam {args.dam_id} " + f"(requests: {stats.get('requests', 0)}, {missing} days were missing, " + f"{still_missing} never published by the source)" + ) + # A rerun saves nothing once the archive is complete — only a real + # transport/database failure is an error here. + return 1 if stats.get("aborted") or stats.get("failures") else 0 if __name__ == "__main__": diff --git a/src/rid_reservoir.py b/src/rid_reservoir.py index 4f85446..54b2417 100644 --- a/src/rid_reservoir.py +++ b/src/rid_reservoir.py @@ -7,6 +7,13 @@ dam in Thailand — storage, inflow and outflow in MCM — with archive depth back to at least 2009. GET returns 404 ("Unknown method."); the POST body may be empty but must carry a Content-Length. +A sibling endpoint transposes that axis: ``GET .../api/dam`` with +``dam_id``/``date_start``/``date_end`` returns ONE dam over a whole date +range. It is GET-only (POST answers 404 "Unknown method.") and served the +full 2009-01-01..today archive — 6,362 rows, 4.7 MB — in a single ~6 s +response, so backfilling one dam costs one request instead of one per +calendar day. Field names differ from api/dams; see parse_dam_range_records. + The reservoir that matters for P.1 flood forecasting is Mae Ngat Somboon Chon (DAM_ID 200103), the only large dam upstream of Chiang Mai: during the Oct 2024 record flood it reached 113% of usable capacity with inflow spikes @@ -26,7 +33,9 @@ import requests logger = logging.getLogger(__name__) RID_DAMS_URL = "https://app.rid.go.th/reservoir/api/dams" +RID_DAM_RANGE_URL = "https://app.rid.go.th/reservoir/api/dam" MAE_NGAT_DAM_ID = "200103" +RID_ARCHIVE_START = datetime.date(2009, 1, 1) # earliest date api/dam serves # Per-column NUMERIC capacity; source junk beyond these becomes NULL instead # of overflowing the insert and discarding the whole daily batch. @@ -91,6 +100,62 @@ def parse_dam_records(payload: Dict) -> List[Dict]: return records +def parse_dam_range_records(payload: Dict) -> List[Dict]: + """Flatten one api/dam single-dam range payload into the same rows as + parse_dam_records, so both endpoints feed one store. + + api/dam names its columns differently and suffixes each measurement + ``_curr`` / ``_prev``; ``_prev`` is the SAME calendar date one year + earlier (confirmed against api/dams' DMD_Date_prev) and is dropped — + those days are rows of their own. Mapping, verified equal to api/dams + on 2019-01-05, 2024-09-24, 2024-10-05 and 2026-08-08: + + DMD_QUse_curr -> storage_mcm (identical) + PERCENT_DMD_QUse_curr -> storage_pct (2 dp; api/dams rounds to + whole percent, 112.62 vs 113) + DMD_Inflow_curr -> inflow_mcm (identical) + DMD_Outflow_curr -> outflow_mcm (identical) + DMD_ULevel_curr -> level_msl (only source of the level: + api/dams' DMD_Q is ' - ' for + all 35 dams, and api/dam's + DMD_Q_curr is a constant 0.00) + + Dam metadata (name, region, capacities) and coordinates are carried too, + so the range path never has to blank rid_dams. + """ + coords = payload.get("dam_coordinates") or {} + payload_dam_id = payload.get("dam_id") + records = [] + for row in payload.get("dam_data") or []: + dam_id = row.get("DAM_ID") or payload_dam_id + try: + date = datetime.date.fromisoformat(row.get("DATE_curr") or "") + except ValueError: + continue + if not dam_id: + continue + records.append( + { + "dam_id": dam_id, + "region": row.get("DAM_Region") or payload.get("dam_region"), + "name_th": row.get("DAM_Name") or payload.get("dam_name"), + "latitude": _to_float(coords.get("lat")), + "longitude": _to_float(coords.get("lng")), + "capacity_max_mcm": _to_float(row.get("DAM_QMax")), + "capacity_normal_mcm": _to_float(row.get("DAM_QStore")), + "date": date, + "storage_mcm": _to_float(row.get("DMD_QUse_curr")), + "storage_pct": _to_float(row.get("PERCENT_DMD_QUse_curr")), + "inflow_mcm": _to_float(row.get("DMD_Inflow_curr")), + "outflow_mcm": _to_float(row.get("DMD_Outflow_curr")), + # An MSL elevation of exactly 0 is "not published", not a + # reading — these dams sit between 45 and 400 m. + "level_msl": _to_float(row.get("DMD_ULevel_curr")) or None, + } + ) + return records + + class RidReservoirClient: """HTTP client for the RID reservoir daily-status API.""" @@ -98,9 +163,11 @@ class RidReservoirClient: self, url: str = RID_DAMS_URL, session: Optional[requests.Session] = None, - timeout: int = 60, + timeout: int = 120, + range_url: str = RID_DAM_RANGE_URL, ): self.url = url + self.range_url = range_url self.session = session or requests.Session() self.timeout = timeout @@ -116,6 +183,36 @@ class RidReservoirClient: response.raise_for_status() return parse_dam_records(response.json()) + def fetch_dam_range( + self, dam_id: str, start: datetime.date, end: datetime.date + ) -> List[Dict]: + """Every published day in [start, end] for one dam, in one request. + + GET only — api/dam answers POST with 404 "Unknown method.", the exact + opposite of api/dams. An unrecognised dam_id still returns HTTP 200 + but with a PHP notice page instead of JSON, so a decode failure is + reported as a bad request rather than a transport error. + """ + response = self.session.get( + self.range_url, + params={ + "dam_id": dam_id, + "date_start": start.isoformat(), + "date_end": end.isoformat(), + "percent": "", + }, + timeout=self.timeout, + ) + response.raise_for_status() + try: + payload = response.json() + except ValueError as e: + raise ValueError( + f"api/dam returned non-JSON for dam_id '{dam_id}' " + f"({start}..{end}) — unknown dam_id?" + ) from e + return parse_dam_range_records(payload) + class RidReservoirStore: """SQL persistence for dam metadata + daily measurements. @@ -205,20 +302,53 @@ class RidReservoirStore: except Exception as e: logger.warning(f"rid_reservoir_daily migration skipped: {e}") - def _upsert(self, table: str, key_cols: List[str], value_cols: List[str]) -> str: + def _upsert( + self, + table: str, + key_cols: List[str], + value_cols: List[str], + preserve_cols: "tuple[str, ...]" = (), + ) -> str: + """Build an upsert; columns in `preserve_cols` keep their stored value + when the incoming one is NULL. + + Dam metadata needs that: any payload that omits a name or coordinate + would otherwise blank a good rid_dams row on every later write. + """ cols = key_cols + value_cols col_list = ", ".join(cols) params = ", ".join(f":{c}" for c in cols) + conflict = ", ".join(key_cols) if self.db_type == "sqlite": - return f"INSERT OR REPLACE INTO {table} ({col_list}) VALUES ({params})" - if self.db_type == "postgresql": - updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in value_cols) - conflict = ", ".join(key_cols) + if not preserve_cols: + return f"INSERT OR REPLACE INTO {table} ({col_list}) VALUES ({params})" + updates = ", ".join( + f"{c} = COALESCE(excluded.{c}, {table}.{c})" + if c in preserve_cols + else f"{c} = excluded.{c}" + for c in value_cols + ) return ( f"INSERT INTO {table} ({col_list}) VALUES ({params}) " f"ON CONFLICT ({conflict}) DO UPDATE SET {updates}" ) - updates = ", ".join(f"{c} = VALUES({c})" for c in value_cols) + if self.db_type == "postgresql": + updates = ", ".join( + f"{c} = COALESCE(EXCLUDED.{c}, {table}.{c})" + if c in preserve_cols + else f"{c} = EXCLUDED.{c}" + for c in value_cols + ) + return ( + f"INSERT INTO {table} ({col_list}) VALUES ({params}) " + f"ON CONFLICT ({conflict}) DO UPDATE SET {updates}" + ) + updates = ", ".join( + f"{c} = COALESCE(VALUES({c}), {c})" + if c in preserve_cols + else f"{c} = VALUES({c})" + for c in value_cols + ) return ( f"INSERT INTO {table} ({col_list}) VALUES ({params}) " f"ON DUPLICATE KEY UPDATE {updates}" @@ -247,9 +377,20 @@ class RidReservoirStore: "outflow_mcm", "level_msl", ] - dam_sql = self._upsert("rid_dams", ["dam_id"], dam_cols + ["updated_at"]) + dam_sql = self._upsert( + "rid_dams", + ["dam_id"], + dam_cols + ["updated_at"], + preserve_cols=tuple(dam_cols), # never blank metadata we already have + ) measure_sql = self._upsert( - "rid_reservoir_daily", ["dam_id", "date"], measure_cols + "rid_reservoir_daily", + ["dam_id", "date"], + measure_cols, + # Only api/dam carries a level (api/dams' DMD_Q is ' - ' for every + # dam), so the hourly collector would blank the backfilled level + # of today and yesterday on every cycle. + preserve_cols=("level_msl",), ) now = datetime.datetime.now() dams = {} @@ -275,21 +416,31 @@ class RidReservoirStore: return 0 def present_dates( - self, start: datetime.date, end: datetime.date + self, + start: datetime.date, + end: datetime.date, + dam_id: Optional[str] = None, ) -> "set[datetime.date]": - """Dates in [start, end] that already have rows, for backfill skipping.""" + """Dates in [start, end] that already have rows, for backfill skipping. + + `dam_id` narrows the answer to one dam: the per-day fleet backfill can + treat any stored date as done, but a per-dam backfill must not skip a + date merely because some other dam published it. + """ if not self.engine and not self.connect(): return set() from sqlalchemy import text + sql = ( + "SELECT DISTINCT date FROM rid_reservoir_daily " + "WHERE date >= :start AND date <= :end" + ) + params = {"start": start, "end": end} + if dam_id is not None: + sql += " AND dam_id = :dam_id" + params["dam_id"] = dam_id with self.engine.begin() as conn: - values = conn.execute( - text( - "SELECT DISTINCT date FROM rid_reservoir_daily " - "WHERE date >= :start AND date <= :end" - ), - {"start": start, "end": end}, - ).fetchall() + values = conn.execute(text(sql), params).fetchall() dates = set() for (value,) in values: if isinstance(value, str): # sqlite returns ISO strings @@ -374,6 +525,97 @@ def backfill( return total +def backfill_dam( + store: RidReservoirStore, + dam_id: str = MAE_NGAT_DAM_ID, + start: datetime.date = RID_ARCHIVE_START, + end: Optional[datetime.date] = None, + client: Optional[RidReservoirClient] = None, + chunk_days: int = 1830, + throttle_seconds: float = 1.0, + skip_present: bool = True, + stats: Optional[Dict] = None, +) -> int: + """Backfill ONE dam over [start, end] using the range endpoint. + + Costs one request per chunk instead of one per calendar day: Mae Ngat's + whole 2009-today archive is ~4 requests here versus ~6,400 with + `backfill`. No server-side range limit was observed (2009-01-01..today + answered in full), so `chunk_days` exists only to bound the response size + and the time a single request can hang, not to satisfy the API. + + Chunks whose dates are already stored are skipped without a request, and + returned rows are filtered to the missing dates, so a rerun repairs holes + rather than rewriting the archive. `skip_present=False` re-fetches + everything, which is how rows first written by the api/dams collector + gain a level_msl and two-decimal storage_pct. + + Returns rows saved. Some dates stay missing however often this runs — + the source simply never published them (72 days of Mae Ngat's archive, + absent from api/dams too) — so a 0-row rerun is normal and callers must + not read it as failure; pass `stats` to get the request/failure counts + that actually distinguish an outage. + """ + client = client or RidReservoirClient() + end = end or datetime.date.today() + counters = {"requests": 0, "failures": 0, "aborted": False} + if stats is not None: + stats.update(counters) + counters = stats + if not store.engine and not store.connect(): + logger.error("backfill_dam aborted: database connection failed") + counters["aborted"] = True + return 0 + span_days = (end - start).days + 1 + if span_days <= 0: + return 0 + present = store.present_dates(start, end, dam_id=dam_id) if skip_present else set() + logger.info( + f"backfill_dam {dam_id}: {span_days - len(present)} of {span_days} " + f"days missing in [{start}, {end}]" + ) + total = 0 + failures = 0 + chunk_start = start + while chunk_start <= end: + chunk_end = min(chunk_start + datetime.timedelta(days=chunk_days - 1), end) + wanted = { + chunk_start + datetime.timedelta(days=i) + for i in range((chunk_end - chunk_start).days + 1) + } - present + if not wanted: + chunk_start = chunk_end + datetime.timedelta(days=1) + continue + try: + counters["requests"] += 1 + records = client.fetch_dam_range(dam_id, chunk_start, chunk_end) + records = [r for r in records if r["date"] in wanted] + saved = store.save(records) + if records and not saved: + raise RuntimeError("database save persisted 0 rows") + total += saved + failures = 0 + logger.info( + f"backfill_dam {dam_id} [{chunk_start}, {chunk_end}]: " + f"{saved} rows ({total} total)" + ) + except Exception as e: + failures += 1 + counters["failures"] += 1 + logger.warning( + f"backfill_dam {dam_id} [{chunk_start}, {chunk_end}] failed " + f"({failures} in a row): {e}" + ) + if failures >= 5: + logger.error("5 consecutive failures — aborting backfill_dam") + counters["aborted"] = True + break + chunk_start = chunk_end + datetime.timedelta(days=1) + if chunk_start <= end: + time.sleep(throttle_seconds) + return total + + def create_collector_from_config() -> Optional[RidReservoirCollector]: """Build a collector from app Config; None when disabled or non-SQL DB.""" from .config import Config diff --git a/tests/test_rid_reservoir.py b/tests/test_rid_reservoir.py index fdcf69d..76898fa 100644 --- a/tests/test_rid_reservoir.py +++ b/tests/test_rid_reservoir.py @@ -9,6 +9,8 @@ from src.rid_reservoir import ( RidReservoirCollector, RidReservoirStore, backfill, + backfill_dam, + parse_dam_range_records, parse_dam_records, ) @@ -56,6 +58,47 @@ def _dams_payload(date="2026-08-13"): } +def _range_row(date, **overrides): + """One api/dam row, shaped like the real 2024-09-24 Mae Ngat response.""" + row = { + "DAM_ID": "200103", + "DATE_curr": date, + "DAM_Name": "เขื่อนแม่งัดสมบูรณ์ชล", + "DAM_Region": "เหนือ", + "DAM_QMax": "323.00", + "DAM_QStore": "265.00", + "DAM_QUsage": "253.00", + "DUL_Useless": "12.00", + "DMD_ULevel_curr": "395.91", + "DMD_Q_curr": "0.00", + "DMD_Inflow_curr": "19.06", + "DMD_Outflow_curr": "0.13", + "VAL_DMD_Q_curr": "242.87", + "DMD_QUse_curr": "254.87", + "PERCENT_DMD_QUse_curr": "96.18", + # Same calendar date one year earlier -> must be ignored, not stored + "DMD_ULevel_prev": "391.52", + "DMD_QUse_prev": "195.64", + "PERCENT_DMD_QUse_prev": "73.83", + "DMD_Inflow_prev": "1.64", + "DMD_Outflow_prev": "3.79", + } + row.update(overrides) + return row + + +def _range_payload(dates=("2024-09-24",), rows=None): + return { + "dam_id": "200103", + "dam_name": "เขื่อนแม่งัดสมบูรณ์ชล", + "dam_region": "เหนือ", + "dam_coordinates": {"lat": 19.16138, "lng": 99.04011}, + "dam_data": rows if rows is not None else [_range_row(d) for d in dates], + "max": {"max_DMD_QUse_curr": "254.87"}, + "min": {"min_DMD_QUse_curr": "254.87"}, + } + + class FakeClient: def __init__(self, payload=None, fail_dates=()): self.payload = payload or _dams_payload() @@ -71,6 +114,28 @@ class FakeClient: return parse_dam_records(self.payload) +class FakeRangeClient: + """Serves every day in the requested window, minus `gaps` (the real + endpoint omits scattered days rather than returning empty rows).""" + + def __init__(self, gaps=(), fail_chunks=()): + self.gaps = {datetime.date.fromisoformat(d) for d in gaps} + self.fail_chunks = set(fail_chunks) # (start, end) tuples that raise + self.calls = [] + + def fetch_dam_range(self, dam_id, start, end): + self.calls.append((dam_id, start, end)) + if (start, end) in self.fail_chunks: + raise ConnectionError("boom") + days = [ + start + datetime.timedelta(days=i) for i in range((end - start).days + 1) + ] + return parse_dam_range_records( + _range_payload(rows=[_range_row(d.isoformat()) for d in days + if d not in self.gaps]) + ) + + class TestParsing: def test_parse_dam_records(self): records = parse_dam_records(_dams_payload()) @@ -88,6 +153,46 @@ class TestParsing: def test_parse_empty_payload(self): assert parse_dam_records({}) == [] + def test_parse_dam_range_records(self): + records = parse_dam_range_records(_range_payload(("2024-09-24",))) + assert len(records) == 1 # the _prev columns are last year, not a row + row = records[0] + assert row["dam_id"] == MAE_NGAT_DAM_ID + assert row["date"] == datetime.date(2024, 9, 24) + assert row["region"] == "เหนือ" + assert row["latitude"] == 19.16138 + assert row["longitude"] == 99.04011 + assert row["capacity_max_mcm"] == 323.0 + assert row["capacity_normal_mcm"] == 265.0 + # Agrees with api/dams for this date, at higher percent precision + assert row["storage_mcm"] == 254.87 + assert row["storage_pct"] == 96.18 + assert row["inflow_mcm"] == 19.06 + assert row["outflow_mcm"] == 0.13 + assert row["level_msl"] == 395.91 # DMD_ULevel, absent from api/dams + + def test_parse_dam_range_records_produces_same_keys_as_dams(self): + assert set(parse_dam_range_records(_range_payload())[0]) == set( + parse_dam_records(_dams_payload())[0] + ) + + def test_parse_dam_range_zero_level_is_missing_not_a_reading(self): + payload = _range_payload(rows=[_range_row("2024-09-24", DMD_ULevel_curr="0.00")]) + assert parse_dam_range_records(payload)[0]["level_msl"] is None + + def test_parse_dam_range_skips_broken_rows(self): + payload = _range_payload( + rows=[ + _range_row("n/a"), # unparseable date + _range_row("2024-09-24"), + ] + ) + assert len(parse_dam_range_records(payload)) == 1 + + def test_parse_dam_range_empty_payload(self): + assert parse_dam_range_records({}) == [] + assert parse_dam_range_records(_range_payload(rows=[])) == [] + class TestStore: @pytest.fixture @@ -160,6 +265,55 @@ class TestStore: # Outside the window -> excluded assert store.present_dates(lo, datetime.date(2026, 8, 12)) == set() + def test_daily_collector_does_not_blank_a_backfilled_level(self, store): + date = "2026-08-13" + assert store.save(parse_dam_range_records(_range_payload((date,)))) == 1 + # api/dams has no level column at all; its rewrite must not clear one + dams_row = [r for r in parse_dam_records(_dams_payload(date)) + if r["dam_id"] == MAE_NGAT_DAM_ID] + assert dams_row[0]["level_msl"] is None + store.save(dams_row) + from sqlalchemy import text + + with store.engine.begin() as conn: + row = conn.execute( + text( + "SELECT level_msl, storage_mcm FROM rid_reservoir_daily " + "WHERE dam_id = :d" + ), + {"d": MAE_NGAT_DAM_ID}, + ).fetchone() + assert float(row[0]) == 395.91 # kept + assert float(row[1]) == 222.01 # published columns still overwritten + + def test_present_dates_per_dam(self, store): + lo, hi = datetime.date(2026, 8, 1), datetime.date(2026, 8, 31) + store.save(parse_dam_records(_dams_payload())) + day = datetime.date(2026, 8, 13) + assert store.present_dates(lo, hi, dam_id=MAE_NGAT_DAM_ID) == {day} + # Another dam having the date must not mark this one done + assert store.present_dates(lo, hi, dam_id="999999") == set() + + def test_null_metadata_does_not_wipe_known_dam_details(self, store): + store.save(parse_dam_records(_dams_payload())) + blank = parse_dam_records(_dams_payload("2026-08-14")) + for record in blank: # a payload that omits metadata + record.update({"name_th": None, "latitude": None, "region": None}) + assert store.save(blank) == 2 + from sqlalchemy import text + + with store.engine.begin() as conn: + row = conn.execute( + text( + "SELECT name_th, latitude, region FROM rid_dams " + "WHERE dam_id = :d" + ), + {"d": MAE_NGAT_DAM_ID}, + ).fetchone() + assert row[0] == "เขื่อนแม่งัดสมบูรณ์ชล" + assert float(row[1]) == 19.16138 + assert row[2] == "เหนือ" + class TestCollectorAndBackfill: def test_run_cycle_today_and_yesterday(self, tmp_path): @@ -245,3 +399,226 @@ class TestCollectorAndBackfill: ) assert saved == 4 # first 2 days succeeded, then 5 failures -> abort assert len(client.calls) == 7 + + +class TestBackfillDam: + @pytest.fixture + def store(self, tmp_path): + store = RidReservoirStore(f"sqlite:///{tmp_path}/range.db", "sqlite") + assert store.connect() + return store + + def test_whole_range_in_one_request(self, store): + client = FakeRangeClient() + saved = backfill_dam( + store, + start=datetime.date(2024, 9, 24), + end=datetime.date(2024, 10, 6), + client=client, + throttle_seconds=0, + ) + assert saved == 13 + assert client.calls == [ + (MAE_NGAT_DAM_ID, datetime.date(2024, 9, 24), datetime.date(2024, 10, 6)) + ] + + def test_chunks_long_ranges(self, store): + client = FakeRangeClient() + saved = backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 10), + client=client, + chunk_days=4, + throttle_seconds=0, + ) + assert saved == 10 + assert client.calls == [ + (MAE_NGAT_DAM_ID, datetime.date(2024, 1, 1), datetime.date(2024, 1, 4)), + (MAE_NGAT_DAM_ID, datetime.date(2024, 1, 5), datetime.date(2024, 1, 8)), + (MAE_NGAT_DAM_ID, datetime.date(2024, 1, 9), datetime.date(2024, 1, 10)), + ] + + def test_source_gaps_are_tolerated(self, store): + client = FakeRangeClient(gaps=("2024-01-03",)) + saved = backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 5), + client=client, + throttle_seconds=0, + ) + assert saved == 4 # the day the source never published stays absent + assert store.present_dates( + datetime.date(2024, 1, 1), datetime.date(2024, 1, 5) + ) == { + datetime.date(2024, 1, d) for d in (1, 2, 4, 5) + } + + def test_stored_days_are_skipped_and_whole_chunks_cost_no_request(self, store): + client = FakeRangeClient() + backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 4), + client=client, + throttle_seconds=0, + ) + # Rerun over a wider window: the stored chunk is not re-requested and + # only the missing days are written + saved = backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 8), + client=client, + chunk_days=4, + throttle_seconds=0, + ) + assert saved == 4 + assert client.calls[1:] == [ + (MAE_NGAT_DAM_ID, datetime.date(2024, 1, 5), datetime.date(2024, 1, 8)) + ] + + def test_partly_stored_chunk_saves_only_missing_days(self, store): + client = FakeRangeClient() + backfill_dam( + store, + start=datetime.date(2024, 1, 3), + end=datetime.date(2024, 1, 3), + client=client, + throttle_seconds=0, + ) + saved = backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 5), + client=client, + throttle_seconds=0, + ) + assert saved == 4 # day 3 already present, requested but not rewritten + + def test_refresh_rewrites_stored_days(self, store): + client = FakeRangeClient() + window = dict( + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 3), + client=client, + throttle_seconds=0, + ) + assert backfill_dam(store, **window) == 3 + assert backfill_dam(store, skip_present=False, **window) == 3 + assert len(client.calls) == 2 + + def test_junk_values_are_nulled_not_fatal(self, store): + client = FakeRangeClient() + client.fetch_dam_range = lambda dam_id, start, end: parse_dam_range_records( + _range_payload( + rows=[ + _range_row( + "2019-01-05", + DMD_QUse_curr="343292.00", + PERCENT_DMD_QUse_curr="87798.47", + DMD_Outflow_curr="1e12", # beyond NUMERIC(10,2) -> NULL + ) + ] + ) + ) + assert ( + backfill_dam( + store, + start=datetime.date(2019, 1, 5), + end=datetime.date(2019, 1, 5), + client=client, + throttle_seconds=0, + ) + == 1 + ) + from sqlalchemy import text + + with store.engine.begin() as conn: + row = conn.execute( + text( + "SELECT storage_pct, outflow_mcm FROM rid_reservoir_daily " + "WHERE date = '2019-01-05'" + ) + ).fetchone() + assert float(row[0]) == 87798.47 + assert row[1] is None + + def test_stats_separate_an_empty_rerun_from_an_outage(self, store): + window = dict( + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 3), + throttle_seconds=0, + ) + backfill_dam(store, client=FakeRangeClient(), **window) + # Everything already stored: no request, no failure, still a success + stats = {} + assert backfill_dam(store, client=FakeRangeClient(), stats=stats, **window) == 0 + assert stats == {"requests": 0, "failures": 0, "aborted": False} + # A source gap keeps requesting, but still reports no failure + gapped = FakeRangeClient(gaps=("2024-01-05",)) + stats = {} + assert ( + backfill_dam( + store, + client=gapped, + stats=stats, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 1, 5), + throttle_seconds=0, + ) + == 1 + ) + assert stats["requests"] == 1 and not stats["failures"] + + def test_stats_record_failures(self, store): + chunk = (datetime.date(2024, 1, 1), datetime.date(2024, 1, 3)) + client = FakeRangeClient(fail_chunks=[chunk]) + stats = {} + assert ( + backfill_dam( + store, + client=client, + stats=stats, + start=chunk[0], + end=chunk[1], + throttle_seconds=0, + ) + == 0 + ) + assert stats["failures"] == 1 and not stats["aborted"] + + def test_aborts_after_consecutive_failures(self, store): + start = datetime.date(2024, 1, 1) + chunks = [ + (start + datetime.timedelta(days=i), start + datetime.timedelta(days=i)) + for i in range(30) + ] + client = FakeRangeClient(fail_chunks=chunks[1:]) + stats = {} + saved = backfill_dam( + store, + start=start, + end=start + datetime.timedelta(days=29), + client=client, + chunk_days=1, + throttle_seconds=0, + stats=stats, + ) + assert saved == 1 # first chunk succeeded, then 5 failures -> abort + assert len(client.calls) == 6 + assert stats["aborted"] and stats["failures"] == 5 + + def test_aborts_when_db_saves_nothing(self, store, monkeypatch): + monkeypatch.setattr(store, "save", lambda records: 0) # broken DB + client = FakeRangeClient() + backfill_dam( + store, + start=datetime.date(2024, 1, 1), + end=datetime.date(2024, 3, 1), + client=client, + chunk_days=1, + throttle_seconds=0, + ) + assert len(client.calls) == 5 # aborted, not one request per chunk