diff --git a/scripts/backfill_rain_db.py b/scripts/backfill_rain_db.py new file mode 100644 index 0000000..709310c --- /dev/null +++ b/scripts/backfill_rain_db.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Backfill the openmeteo_rain table with the full 2021+ catchment history. + +Usage: + uv run scripts/backfill_rain_db.py # DB from Config/.env + uv run scripts/backfill_rain_db.py --db-url postgresql://... +""" + +import argparse +import logging +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.config import Config +from src.ml.rain import backfill_db + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--db-url", default=None) + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + if args.db_url: + connection_string, db_type = args.db_url, args.db_url.split(":", 1)[0] + else: + cfg = Config.get_database_config() + if cfg["type"] not in ("sqlite", "postgresql", "mysql"): + print(f"requires a SQL database, got {cfg['type']}", file=sys.stderr) + return 1 + connection_string, db_type = cfg["connection_string"], cfg["type"] + + from sqlalchemy import create_engine + + engine = create_engine(connection_string, pool_pre_ping=True) + saved = backfill_db(engine, db_type) + print(f"backfilled {saved} hourly rows into openmeteo_rain") + return 0 if saved else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/ml/rain.py b/src/ml/rain.py index 9a77e8c..4f11811 100644 --- a/src/ml/rain.py +++ b/src/ml/rain.py @@ -162,6 +162,24 @@ def serving_series() -> Optional[pd.Series]: return None +def backfill_db(engine, db_type: str, chunk_rows: int = 5000) -> int: + """Push the full Open-Meteo history (2021+) into openmeteo_rain. + + Loads (or fetches) the archive cache and upserts in chunks; idempotent, + safe to re-run, and safe alongside the hourly live writer. + """ + history = load_history() + if history is None or history.empty: + logger.error("no rain history available to backfill") + return 0 + total = 0 + for start in range(0, len(history), chunk_rows): + part = history.iloc[start: start + chunk_rows] + total += save_to_db(part, engine, db_type) + logger.info(f"openmeteo_rain backfill: {total}/{len(history)} rows") + return total + + def save_to_db(df: pd.DataFrame, engine, db_type: str) -> int: """Upsert per-point + catchment-mean hourly rain into openmeteo_rain.