feat: hourly HII/ThaiWater rainfall + backup water-level collection

Poll the open api-v3.thaiwater.net public endpoints (rain_24h,
waterlevel_load), filter to the Ping basin, and persist to new
hii_rain_stations/hii_rainfall and hii_wl_stations/hii_waterlevel tables
(auto-created; sqlite/postgresql/mysql). Water levels stay in their own
tables since HII reports m MSL from a different station set; rid_code
maps mirrors like ridhydro_P.1 to P.1 and offset_msl converts MSL to
gauge datum. Runs every scraping cycle in web-api and continuous modes
(hourly cadence even during 1-minute RID retry), one-shot via
--collect-hii; docs/DATA_SOURCES.md catalogs all probed endpoints.

src/config.py

@@ -78,6 +78,15 @@ class Config:
     # MySQL settings
     MYSQL_CONNECTION_STRING = os.getenv("MYSQL_CONNECTION_STRING")
 
+    # HII/ThaiWater open api-v3 collection (rainfall + backup water level)
+    # See docs/DATA_SOURCES.md. Requires a SQL DB_TYPE (sqlite/postgresql/mysql).
+    ENABLE_HII_COLLECTION = os.getenv("ENABLE_HII_COLLECTION", "true").lower() in (
+        "1",
+        "true",
+        "yes",
+    )
+    HII_BASIN_CODE = int(os.getenv("HII_BASIN_CODE", "6"))  # 6 = Ping Basin
+
     # Scheduler settings
     SCRAPING_INTERVAL_HOURS = int(os.getenv("SCRAPING_INTERVAL_HOURS", "1"))
 

src/hii_collector.py

@@ -0,0 +1,456 @@
+"""Collector for HII/ThaiWater open api-v3 feeds (rainfall + water level).
+
+Polls the unauthenticated api-v3.thaiwater.net public endpoints, filters to the
+Ping basin, and persists to dedicated tables alongside the RID data:
+
+- hii_rain_stations / hii_rainfall        (rain_1h / rain_24h gauge telemetry)
+- hii_wl_stations   / hii_waterlevel      (independent water-level source, m MSL)
+
+Water levels are kept in a separate table (not a column on water_measurements)
+because HII reports in m MSL from a different station universe; the per-station
+``offset`` column (gauge zero in m MSL) converts to gauge datum when needed.
+
+See docs/DATA_SOURCES.md for the endpoint catalog and quirks.
+"""
+
+import datetime
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+HII_API_BASE = "https://api-v3.thaiwater.net/api/v1/thaiwater30/public"
+PING_BASIN_CODE = 6
+
+# Matches 'P.1', 'ridhydro_P.67', 'ridtele_TUP.14' -> canonical RID code suffix
+_RID_CODE_RE = re.compile(r"(?:^|_)(P\.\d+[A-Z]?)$")
+
+
+def _to_float(value: Any) -> Optional[float]:
+    """API numerics arrive as strings ('335.00'), numbers, or None."""
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _parse_datetime(value: Any) -> Optional[datetime.datetime]:
+    """Timestamps are Thai local time, e.g. '2026-08-11 13:00'."""
+    if not value:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"):
+        try:
+            return datetime.datetime.strptime(value, fmt)
+        except ValueError:
+            continue
+    return None
+
+
+def _name(station: Dict, lang: str) -> Optional[str]:
+    name = station.get("tele_station_name")
+    if isinstance(name, dict):
+        return name.get(lang)
+    return name if lang == "th" else None
+
+
+def rid_code_from_oldcode(oldcode: Optional[str]) -> Optional[str]:
+    """Normalize a ThaiWater oldcode to the RID P-code it mirrors, if any."""
+    if not oldcode:
+        return None
+    match = _RID_CODE_RE.search(oldcode)
+    return match.group(1) if match else None
+
+
+def parse_rain_records(
+    payload: Dict, basin_code: int = PING_BASIN_CODE
+) -> List[Dict]:
+    """Extract per-station rainfall rows from a rain_24h payload."""
+    records = []
+    for row in payload.get("data") or []:
+        basin = row.get("basin") or {}
+        if basin.get("basin_code") != basin_code:
+            continue
+        station = row.get("station") or {}
+        station_id = station.get("id")
+        timestamp = _parse_datetime(row.get("rainfall_datetime"))
+        if station_id is None or timestamp is None:
+            continue
+        records.append(
+            {
+                "station_id": station_id,
+                "oldcode": station.get("tele_station_oldcode"),
+                "name_th": _name(station, "th"),
+                "name_en": _name(station, "en"),
+                "latitude": _to_float(station.get("tele_station_lat")),
+                "longitude": _to_float(station.get("tele_station_long")),
+                "sub_basin_id": str(station.get("sub_basin_id") or "") or None,
+                "agency": ((row.get("agency") or {}).get("agency_shortname") or {}).get(
+                    "en"
+                ),
+                "timestamp": timestamp,
+                "rain_1h": _to_float(row.get("rain_1h")),
+                "rain_24h": _to_float(row.get("rain_24h")),
+            }
+        )
+    return records
+
+
+def parse_waterlevel_records(
+    payload: Dict, basin_code: int = PING_BASIN_CODE
+) -> List[Dict]:
+    """Extract per-station water-level rows from a waterlevel_load payload."""
+    data = (payload.get("waterlevel_data") or {}).get("data") or []
+    records = []
+    for row in data:
+        basin = row.get("basin") or {}
+        if basin.get("basin_code") != basin_code:
+            continue
+        station = row.get("station") or {}
+        station_id = station.get("id")
+        timestamp = _parse_datetime(row.get("waterlevel_datetime"))
+        if station_id is None or timestamp is None:
+            continue
+        oldcode = station.get("tele_station_oldcode")
+        records.append(
+            {
+                "station_id": station_id,
+                "oldcode": oldcode,
+                "rid_code": rid_code_from_oldcode(oldcode),
+                "name_th": _name(station, "th"),
+                "name_en": _name(station, "en"),
+                "latitude": _to_float(station.get("tele_station_lat")),
+                "longitude": _to_float(station.get("tele_station_long")),
+                "agency": ((row.get("agency") or {}).get("agency_shortname") or {}).get(
+                    "en"
+                ),
+                "river_name": row.get("river_name"),
+                "offset_msl": _to_float(station.get("offset")),
+                "ground_level_msl": _to_float(station.get("ground_level")),
+                "min_bank_msl": _to_float(station.get("min_bank")),
+                "critical_level_msl": _to_float(station.get("critical_level_msl")),
+                "critical_level_m": _to_float(station.get("critical_level_m")),
+                "qmax": _to_float(station.get("qmax")),
+                "is_key_station": bool(station.get("is_key_station")),
+                "timestamp": timestamp,
+                "wl_msl": _to_float(row.get("waterlevel_msl")),
+                "wl_m": _to_float(row.get("waterlevel_m")),
+                "discharge": _to_float(row.get("discharge")),
+                "flow_rate": _to_float(row.get("flow_rate")),
+                "storage_percent": _to_float(row.get("storage_percent")),
+                "situation_level": row.get("situation_level"),
+                "diff_wl_bank": _to_float(row.get("diff_wl_bank")),
+            }
+        )
+    return records
+
+
+class HiiClient:
+    """HTTP client for the open api-v3 public endpoints."""
+
+    def __init__(
+        self,
+        base_url: str = HII_API_BASE,
+        session: Optional[requests.Session] = None,
+        timeout: int = 90,
+    ):
+        self.base_url = base_url.rstrip("/")
+        self.session = session or requests.Session()
+        self.timeout = timeout
+
+    def _get(self, endpoint: str) -> Dict:
+        response = self.session.get(
+            f"{self.base_url}/{endpoint}", 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)
+
+    def fetch_waterlevel(self, basin_code: int = PING_BASIN_CODE) -> List[Dict]:
+        return parse_waterlevel_records(self._get("waterlevel_load"), basin_code)
+
+
+class HiiStore:
+    """SQL persistence for HII feeds (sqlite / postgresql / mysql).
+
+    Reuses the app's main relational database (same connection string as
+    the RID tables) but writes to its own hii_* tables.
+    """
+
+    def __init__(self, connection_string: str, db_type: str):
+        self.db_type = db_type.lower()
+        if self.db_type not in ("sqlite", "postgresql", "mysql"):
+            raise ValueError(
+                f"HII collection requires a SQL database, got '{db_type}'"
+            )
+        self.connection_string = connection_string
+        self.engine = None
+
+    def connect(self) -> bool:
+        try:
+            from sqlalchemy import create_engine
+
+            self.engine = create_engine(self.connection_string, pool_pre_ping=True)
+            self._create_tables()
+            return True
+        except Exception as e:
+            logger.error(f"HiiStore failed to connect: {e}")
+            self.engine = None
+            return False
+
+    def _pk(self) -> str:
+        if self.db_type == "sqlite":
+            return "INTEGER PRIMARY KEY AUTOINCREMENT"
+        if self.db_type == "postgresql":
+            return "BIGSERIAL PRIMARY KEY"
+        return "BIGINT AUTO_INCREMENT PRIMARY KEY"
+
+    def _create_tables(self):
+        from sqlalchemy import text
+
+        bool_type = "BOOLEAN" if self.db_type != "mysql" else "TINYINT(1)"
+        ddl = [
+            """
+            CREATE TABLE IF NOT EXISTS hii_rain_stations (
+                id INTEGER PRIMARY KEY,
+                oldcode VARCHAR(60),
+                name_th VARCHAR(255),
+                name_en VARCHAR(255),
+                latitude NUMERIC(10,6),
+                longitude NUMERIC(10,6),
+                sub_basin_id VARCHAR(10),
+                agency VARCHAR(40),
+                updated_at TIMESTAMP
+            )
+            """,
+            f"""
+            CREATE TABLE IF NOT EXISTS hii_rainfall (
+                id {self._pk()},
+                station_id INTEGER NOT NULL,
+                timestamp TIMESTAMP NOT NULL,
+                rain_1h NUMERIC(7,2),
+                rain_24h NUMERIC(8,2),
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT uq_hii_rainfall UNIQUE (station_id, timestamp)
+            )
+            """,
+            f"""
+            CREATE TABLE IF NOT EXISTS hii_wl_stations (
+                id INTEGER PRIMARY KEY,
+                oldcode VARCHAR(60),
+                rid_code VARCHAR(10),
+                name_th VARCHAR(255),
+                name_en VARCHAR(255),
+                latitude NUMERIC(10,6),
+                longitude NUMERIC(10,6),
+                agency VARCHAR(40),
+                river_name VARCHAR(255),
+                offset_msl NUMERIC(8,3),
+                ground_level_msl NUMERIC(8,3),
+                min_bank_msl NUMERIC(8,3),
+                critical_level_msl NUMERIC(8,3),
+                critical_level_m NUMERIC(8,3),
+                qmax NUMERIC(10,2),
+                is_key_station {bool_type},
+                updated_at TIMESTAMP
+            )
+            """,
+            f"""
+            CREATE TABLE IF NOT EXISTS hii_waterlevel (
+                id {self._pk()},
+                station_id INTEGER NOT NULL,
+                timestamp TIMESTAMP NOT NULL,
+                wl_msl NUMERIC(8,3),
+                wl_m NUMERIC(8,3),
+                discharge NUMERIC(10,2),
+                flow_rate NUMERIC(10,2),
+                storage_percent NUMERIC(6,2),
+                situation_level INTEGER,
+                diff_wl_bank NUMERIC(8,3),
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT uq_hii_waterlevel UNIQUE (station_id, timestamp)
+            )
+            """,
+            "CREATE INDEX IF NOT EXISTS idx_hii_rainfall_ts ON hii_rainfall(timestamp)",
+            "CREATE INDEX IF NOT EXISTS idx_hii_waterlevel_ts ON hii_waterlevel(timestamp)",
+        ]
+        # MySQL (<8.0.13 semantics) lacks CREATE INDEX IF NOT EXISTS; the unique
+        # constraints already cover the hot (station_id, timestamp) lookups there.
+        if self.db_type == "mysql":
+            ddl = ddl[:4]
+        with self.engine.begin() as conn:
+            for statement in ddl:
+                conn.execute(text(statement))
+
+    def _upsert(self, table: str, key_cols: List[str], value_cols: List[str]) -> str:
+        cols = key_cols + value_cols
+        col_list = ", ".join(cols)
+        params = ", ".join(f":{c}" for c in 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)
+            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)
+        return (
+            f"INSERT INTO {table} ({col_list}) VALUES ({params}) "
+            f"ON DUPLICATE KEY UPDATE {updates}"
+        )
+
+    def save_rain(self, records: List[Dict]) -> int:
+        return self._save(
+            records,
+            station_table="hii_rain_stations",
+            station_cols=[
+                "oldcode",
+                "name_th",
+                "name_en",
+                "latitude",
+                "longitude",
+                "sub_basin_id",
+                "agency",
+            ],
+            measurement_table="hii_rainfall",
+            measurement_cols=["rain_1h", "rain_24h"],
+        )
+
+    def save_waterlevel(self, records: List[Dict]) -> int:
+        return self._save(
+            records,
+            station_table="hii_wl_stations",
+            station_cols=[
+                "oldcode",
+                "rid_code",
+                "name_th",
+                "name_en",
+                "latitude",
+                "longitude",
+                "agency",
+                "river_name",
+                "offset_msl",
+                "ground_level_msl",
+                "min_bank_msl",
+                "critical_level_msl",
+                "critical_level_m",
+                "qmax",
+                "is_key_station",
+            ],
+            measurement_table="hii_waterlevel",
+            measurement_cols=[
+                "wl_msl",
+                "wl_m",
+                "discharge",
+                "flow_rate",
+                "storage_percent",
+                "situation_level",
+                "diff_wl_bank",
+            ],
+        )
+
+    def _save(
+        self,
+        records: List[Dict],
+        station_table: str,
+        station_cols: List[str],
+        measurement_table: str,
+        measurement_cols: List[str],
+    ) -> int:
+        if not records:
+            return 0
+        if not self.engine and not self.connect():
+            return 0
+        from sqlalchemy import text
+
+        now = datetime.datetime.now()
+        station_sql = self._upsert(
+            station_table, ["id"], station_cols + ["updated_at"]
+        )
+        measurement_sql = self._upsert(
+            measurement_table, ["station_id", "timestamp"], measurement_cols
+        )
+        # Dedupe stations (one row per station per snapshot anyway) and build
+        # parameter dicts limited to each statement's columns.
+        stations = {}
+        measurements = []
+        for record in records:
+            sid = record["station_id"]
+            station_row = {c: record.get(c) for c in station_cols}
+            station_row.update({"id": sid, "updated_at": now})
+            stations[sid] = station_row
+            measurement_row = {c: record.get(c) for c in measurement_cols}
+            measurement_row.update(
+                {"station_id": sid, "timestamp": record["timestamp"]}
+            )
+            measurements.append(measurement_row)
+
+        try:
+            with self.engine.begin() as conn:
+                conn.execute(text(station_sql), list(stations.values()))
+                conn.execute(text(measurement_sql), measurements)
+            return len(measurements)
+        except Exception as e:
+            logger.error(f"HiiStore save to {measurement_table} failed: {e}")
+            return 0
+
+
+class HiiCollector:
+    """Fetch + persist one snapshot of both HII feeds."""
+
+    def __init__(
+        self,
+        db_config: Dict,
+        basin_code: int = PING_BASIN_CODE,
+        client: Optional[HiiClient] = None,
+    ):
+        self.client = client or HiiClient()
+        self.basin_code = basin_code
+        self.store = HiiStore(
+            connection_string=db_config["connection_string"],
+            db_type=db_config["type"],
+        )
+
+    def run_cycle(self) -> Dict[str, int]:
+        """Collect both feeds; each is independent and failure-isolated."""
+        counts = {"rainfall": 0, "waterlevel": 0}
+        try:
+            counts["rainfall"] = self.store.save_rain(
+                self.client.fetch_rain(self.basin_code)
+            )
+        except Exception as e:
+            logger.error(f"HII rainfall collection failed: {e}")
+        try:
+            counts["waterlevel"] = self.store.save_waterlevel(
+                self.client.fetch_waterlevel(self.basin_code)
+            )
+        except Exception as e:
+            logger.error(f"HII waterlevel collection failed: {e}")
+        logger.info(
+            f"HII collection: {counts['rainfall']} rainfall, "
+            f"{counts['waterlevel']} waterlevel rows saved"
+        )
+        return counts
+
+
+def create_collector_from_config() -> Optional[HiiCollector]:
+    """Build a collector from app Config; None when disabled or non-SQL DB."""
+    from .config import Config
+
+    if not Config.ENABLE_HII_COLLECTION:
+        return None
+    db_config = Config.get_database_config()
+    if db_config["type"] not in ("sqlite", "postgresql", "mysql"):
+        logger.warning(
+            f"HII collection skipped: DB_TYPE '{db_config['type']}' is not SQL"
+        )
+        return None
+    return HiiCollector(db_config, basin_code=Config.HII_BASIN_CODE)

src/main.py

@@ -86,6 +86,19 @@ def run_continuous_monitoring():
 
         alerting = WaterLevelAlertSystem()
 
+        # Initialize HII/ThaiWater collector (rainfall + backup water level)
+        hii_collector = None
+        try:
+            from .hii_collector import create_collector_from_config
+
+            hii_collector = create_collector_from_config()
+            if hii_collector:
+                logger.info(
+                    "HII collection enabled (Ping-basin rainfall + water level)"
+                )
+        except Exception as e:
+            logger.error(f"HII collector initialization failed: {e}")
+
         # Setup signal handlers
         setup_signal_handlers(scraper)
 
@@ -108,6 +121,14 @@ def run_continuous_monitoring():
         retry_mode = not initial_success
         last_successful_fetch = None if not initial_success else datetime.now()
 
+        last_hii_run = None
+        if hii_collector:
+            try:
+                hii_collector.run_cycle()
+                last_hii_run = datetime.now()
+            except Exception as e:
+                logger.error(f"HII collection failed: {e}")
+
         if retry_mode:
             logger.warning("No data fetched in initial run - entering retry mode")
             next_run = datetime.now() + timedelta(minutes=1)
@@ -126,6 +147,18 @@ def run_continuous_monitoring():
                 logger.info("Running scheduled data collection...")
                 success = scraper.run_scraping_cycle()
 
+                # HII feeds update hourly; keep collecting on that cadence even
+                # when the RID scraper is in 1-minute retry mode.
+                if hii_collector and (
+                    last_hii_run is None
+                    or current_time - last_hii_run >= timedelta(minutes=55)
+                ):
+                    try:
+                        hii_collector.run_cycle()
+                        last_hii_run = current_time
+                    except Exception as e:
+                        logger.error(f"HII collection failed: {e}")
+
                 if success:
                     last_successful_fetch = current_time
 
@@ -180,6 +213,27 @@ def run_continuous_monitoring():
     return True
 
 
+def run_hii_collection():
+    """Run a single HII/ThaiWater collection cycle (rainfall + water level)"""
+    try:
+        Config.validate_config()
+        from .hii_collector import create_collector_from_config
+
+        collector = create_collector_from_config()
+        if not collector:
+            logger.error(
+                "HII collection unavailable (disabled via ENABLE_HII_COLLECTION "
+                "or DB_TYPE is not a SQL database)"
+            )
+            return False
+
+        counts = collector.run_cycle()
+        return counts["rainfall"] > 0 or counts["waterlevel"] > 0
+    except Exception as e:
+        logger.error(f"HII collection failed: {e}")
+        return False
+
+
 def run_gap_filling(days_back: Optional[int]):
     """Run gap filling for missing data (days_back=None scans the whole range)"""
     if days_back is None:
@@ -504,6 +558,12 @@ Examples:
         "--alert-test", action="store_true", help="Send test alert message to Matrix"
     )
 
+    parser.add_argument(
+        "--collect-hii",
+        action="store_true",
+        help="Run one HII/ThaiWater collection cycle (rainfall + water level)",
+    )
+
     parser.add_argument(
         "--log-level",
         choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
@@ -558,6 +618,8 @@ Examples:
             success = run_alert_check()
         elif args.alert_test:
             success = run_alert_test()
+        elif args.collect_hii:
+            success = run_hii_collection()
         else:
             success = run_continuous_monitoring()
 

src/web_api.py

@@ -118,6 +118,17 @@ async def lifespan(app: FastAPI):
     db_config = Config.get_database_config()
     app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)
 
+    # Initialize HII/ThaiWater collector (rainfall + backup water level)
+    try:
+        from .hii_collector import create_collector_from_config
+
+        app_state["hii_collector"] = create_collector_from_config()
+        if app_state["hii_collector"]:
+            logger.info("HII collection enabled (Ping-basin rainfall + water level)")
+    except Exception as e:
+        app_state["hii_collector"] = None
+        logger.error(f"HII collector initialization failed: {e}")
+
     # Initialize health checks
     health_manager = HealthCheckManager()
     health_manager.add_check(DatabaseHealthCheck(app_state["scraper"].db_adapter))
@@ -220,6 +231,22 @@ async def background_scraping_task():
                         increment_counter("scraping_cycles_failed")
                         logger.error(f"Background scraping cycle failed: {e}")
 
+                    # HII rainfall/water-level snapshot (independent of RID cycle)
+                    hii_collector = app_state.get("hii_collector")
+                    if hii_collector:
+                        try:
+                            hii_counts = await asyncio.get_event_loop().run_in_executor(
+                                None, hii_collector.run_cycle
+                            )
+                            set_gauge(
+                                "hii_rainfall_rows_saved", hii_counts["rainfall"]
+                            )
+                            set_gauge(
+                                "hii_waterlevel_rows_saved", hii_counts["waterlevel"]
+                            )
+                        except Exception as e:
+                            logger.error(f"HII collection failed: {e}")
+
                 app_state["is_scraping"] = False
 
             # Calculate next run time