diff --git a/README.md b/README.md index b42b761..4ccbf8f 100644 --- a/README.md +++ b/README.md @@ -335,6 +335,7 @@ python src/demo_databases.py all # Test all databases ## 📚 Documentation ### Core Documentation +- **[Data Sources & API Catalog](docs/DATA_SOURCES.md)** - Every ingested and available data source (RID, ThaiWater/HII, dams, rainfall, forecasts) - **[Installation Guide](docs/DATABASE_DEPLOYMENT_GUIDE.md)** - Complete setup instructions - **[Scheduler Guide](docs/ENHANCED_SCHEDULER_GUIDE.md)** - 15-minute scheduling system - **[Geolocation Guide](docs/GEOLOCATION_GUIDE.md)** - Grafana geomap integration diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md new file mode 100644 index 0000000..1c18de4 --- /dev/null +++ b/docs/DATA_SOURCES.md @@ -0,0 +1,256 @@ +# Data Sources & External API Catalog + +Catalog of every data source available to the Ping River Monitor — what we ingest +today, what the ThaiWater/HII ecosystem exposes, and vetted external feeds for +future model inputs (rainfall, dam releases, forecasts). + +All "verified" claims below were empirically probed on **2026-08-11**. Endpoints +marked *(catalog)* were recovered from the thaiwater.net frontend JS bundle but +not exercised (auth required). + +--- + +## 1. Currently ingested — RID hydrology telemetry + +The only source persisted to the database and used by the ML pipeline. + +| | | +|---|---| +| Endpoint | `POST https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx` | +| Agency | Royal Irrigation Department (RID) | +| Auth | None | +| Cadence | Hourly (`hourlytime` 1.00–24.00; hour 24 = midnight next day) | +| Params | `DW[UtokID]=1`, `DW[BasinID]=6` (Ping), `DW[TimeCurrent]=`, `rows=100` | +| Variables | Water level (m, gauge datum), discharge (m³/s, `'***'` = malformed), discharge % of channel capacity | +| Stations | 16 P-series gauges (P.1 anchor at Nawarat Bridge; see `src/data/stations.json`) | +| Client | `src/water_scraper_v3.py` | + +Human-facing page: + +--- + +## 2. ThaiWater / HII ecosystem + +ThaiWater () is the National Hydroinformatics +Institute (HII) portal. It sits on **two distinct API layers** with very +different access rules. + +### 2.1 `api-v3.thaiwater.net` — open, no authentication ✅ + +Base: `https://api-v3.thaiwater.net/api/v1/thaiwater30/public/` + +Undocumented backend of the portal. No key, no session. No published rate +limits or terms of use — be a good citizen (hourly polls, cache, filter to +Ping basin `basin_code == 6`). + +#### `waterlevel_load` — national water-level snapshot (verified) + +``` +GET https://api-v3.thaiwater.net/api/v1/thaiwater30/public/waterlevel_load +``` + +- ~2.4 MB, 1,426 stations nationwide; **125 in Ping Basin, 61 in Chiang Mai + province, 34 with discharge**. +- Per station: `waterlevel_m`, `waterlevel_msl`, `waterlevel_msl_previous`, + `flow_rate`, `discharge`, `storage_percent`, `situation_level` (1–4 flood + severity), `diff_wl_bank`, bank/critical levels (`min_bank`, + `critical_level_msl`, `warning_level_m`, `critical_level_m`, `qmax`), + `river_name`, basin, geocode, agency, lat/long, `is_key_station`. +- Ping key stations present: P.1, P.67, P.75, P.81, P.92, P.17, P.87, P.4A, + P.7A, P.77, P.2A — plus RID mirrors (`ridhydro_P.1`, …), Tak-reach `TP.`/ + `TUP.` codes, and HII sensor clusters (`PIN001-011`, `CHM001-005`). +- **P.1 = internal station id `3226`** (lat 18.786961, long 99.005089). + +#### `waterlevel_graph` — hourly historical time series (verified) ⭐ + +``` +GET .../waterlevel_graph?station_type=tele_waterlevel&station_id=3226&start_date=2024-09-25&end_date=2024-10-08 +``` + +- Returns `{data: {graph_data: [{datetime, value, value_out, discharge}]}}`, + hourly. `value` is **water level in m MSL** (not gauge datum). +- **Respects arbitrary date ranges. Archive verified back to at least + 2019-08** (P.1 returned data for 2019-08-01). 14-day windows tested OK; + maximum window size not probed. +- Oct 2024 record flood fully present: peak 305.8 m MSL @ 2024-10-05 12:00, + discharge 656 m³/s. +- Datum conversion at P.1: 305.8 MSL peak = 5.30 m gauge ⇒ + **gauge ≈ MSL − 300.5 m** (verify per station before use; each station has + its own datum offset). +- Value: independent second historical source for cross-validating / gap-filling + the RID feed (RID grid is only ~56% filled). + +#### `rain_24h` — national rainfall snapshot (verified) + +``` +GET https://api-v3.thaiwater.net/api/v1/thaiwater30/public/rain_24h +``` + +- ~4.5 MB, 4,445 stations; **321 in Chiang Mai province**; agencies include + HII, DWR, RID. +- Per station: `rain_1h`, `rain_24h` (mm), `rainfall_datetime`, + `station.id` (small int — the graph key), `station.tele_station_oldcode` + (e.g. `CHM005`, `STN0410`, `ridtele_TUP.14`), `sub_basin_id`, lat/long, + basin, geocode, agency. +- **This is the missing rainfall input** for the flood model — near-real-time + hourly gauge rain across the upper Ping catchment. + +#### `rain_24h_graph` — trailing-window rainfall series (verified, limited) ⚠️ + +``` +GET .../rain_24h_graph?station_type=tele_rainfall&station_id=418&start_date=...&end_date=... +``` + +- `station_id` is the **small `station.id`** from `rain_24h` (e.g. 418 = + CHM005 "Chiang Mai 5", Mae Taeng), *not* the top-level record id. +- Returns hourly `{rainfall_datetime, rainfall_value}` — **but the date range + is IGNORED**: every request returns the same trailing ~36-hour window + (39 rows). Requests for 2020/2024 return identical data to today. +- Consequence: **no rainfall history via this API**. To build training data, + persist `rain_24h` from now on and backfill history from satellite QPE or an + HII data request (§4). + +#### Probed and NOT available on api-v3 (all HTTP 404) + +`dam_daily`, `dam`, `dam_json`, `big_dam`, `mainstream_dam`, `weather`, +`rain_graph`, `rainfall_graph`, `rain24hr_graph`. Dam data is v2-only (§2.2). + +### 2.2 `twa-api-public.thaiwater.net` — auth-gated (x-api-key / session) 🔒 + +The layer our existing `src/thaiwater.py` client uses +(`GET /v2/waterlevel` with `x-api-key: $THAIWATER_API_KEY`; wired to +`GET /sensors/thaiwater` in the web API, display-only, never persisted). +Without a key: HTTP 401/500. No public key-registration page was found — +obtain a sanctioned key from HII (). + +Full endpoint catalog *(catalog — recovered from frontend JS, not exercised)*: + +- **Water level / discharge**: `/v2/waterlevel`, `/v2/waterlevel/list`, + `/v2/waterlevel/{id}/detail`, `/v2/waterlevel/canal`, + `/v2/waterlevel/sea-waterlevel`, `/v2/waterlevel-discharge`, + `/v2/waterlevel-discharge/list`, `/v2/waterlevel-discharge/{id}/detail`, + `/v2/waterlevel-discharge/{id}/forecast-table`, + `/v2/waterlevel-discharge/forecast`, `/v2/waterlevel-discharge/forecast/list`, + `/v2/watergate`, `/v2/waterload-tide` +- **Dams** (incl. Bhumibol): `/v2/large-dam/daily-geo-json`, + `/v2/large-dam/daily/list`, `/v2/large-dam/hourly/list`, + `/v2/large-dam/daily/{id}/detail`, `/v2/large-dam/hourly/{id}/detail`, + `/v2/medium-dam/daily-geo-json`, `/v2/medium-dam/daily/list`, + `/v2/medium-dam/{id}/detail`, `/v2/summary/summary4dam`, + `/v2/summary/dam-summary`, `/v2/summary/dam-crisis` +- **Rainfall**: `/v2/rainfall/{type}`, `/v2/rainfall/{type}/list`, + `/v2/district-rain/actual-measure`, `/v2/district-rain/forecast`, + `/v2/district-rain/accumulate`, `/v2/summary/rainfall24h-ranking-province`, + `/v2/summary/rainfall-24hr-forecast`, `/v2/summary/rainfall-forecast`, + `/v2/summary/warning-rainfall-24h`, `/v2/summary/warning-rainfall-48h` +- **Weather / hazards**: `/v2/weather`, `/v2/storm`, `/v2/wave`, `/v2/pm25`, + `/v2/pm10`, `/v2/flood/flash-flood`, `/v2/flood/flash-flood-alert`, + `/v2/drought/alert`, `/v2/drought/risk-area/list`, + `/v2/summary/weather-summary`, `/v2/summary/temperature-forecast`, + `/v2/summary-area/rainfall` +- **Time-series / graph** (base `/data/platform/v1/public/`): + `tele_waterlevel/graph`, `flow/graph`, `latest_waterlevel/forecast/graph`, + `latest_watertide/forecast/graph`, `dam_pdaily_sum_by_date`, + `dam_pdaily_sum_by_region_graph`, `dam_rulecurve/graph`, + `medium_dam/graph_year`, `monthly_rainfall/stations`, + `monthly_rainfall/anomaly-stations`, `tele_watergate/graph`, + `salinity_forecast_cpy/graph`, `sea_waterlevel_forecast/graph`, + `latest_weather_area`, `latest_weather_area_daily` + +### 2.3 Other HII hosts + +| Host | What | Access | +|---|---|---| +| `https://standard.thaiwater.net` | **Official water-data standard** — canonical station/basin/province code registries, data-exchange formats, warning-level definitions (Thai) | Open, docs site | +| `https://api.hii.or.th/tiservice/v1/ws/{token}/isohyet/daily/latest/province/{code}` | Daily isohyet rainfall by province | Token in path | +| `https://live1.hii.or.th/product/latest/rain/one_map/data/*.tif` | Rainfall anomaly & 1–6-month forecast GeoTIFF rasters | Open | +| `https://data.hii.or.th` | HII open-data catalog — 36 datasets (rainfall telemetry, water level, weather, climate) | Open browsing | +| `https://tiwrm.hii.or.th` | Legacy reports | Open | + +Historical bulk telemetry: HII documents a request channel at +**nhcsoc@hii.or.th**. + +--- + +## 3. Dams & reservoirs + +> **Geography matters: Bhumibol Dam (Tak) is ~240 km DOWNSTREAM of P.1** and +> cannot influence Chiang Mai water levels. Do not use it as a P.1 feature. + +The predictive upstream reservoir is **Mae Ngat Somboon Chon** (Mae Ngat +tributary, joins the Ping above Chiang Mai; spilled 110 m³/s during the +Oct 2024 flood). Mae Kuang Udom Thara is the second upstream reservoir. + +| Source | What | Access | +|---|---|---| +| `https://lsim.rid.go.th/ForeCast?reservoirid=22` | Mae Ngat daily status/forecast (RID) | Open, scrape | +| `https://app.rid.go.th/reservoir/` | RID reservoir DB, per-reservoir daily detail with date-range URLs | Open, scrape (confirm URL pattern before hard-coding) | +| `https://water.egat.co.th` | EGAT dams (Bhumibol/Sirikit) hourly+daily inflow/outflow/level | Endpoint catalog not public; contact EGAT (0-2436-8186). Only relevant downstream of Bhumibol | +| ThaiWater `/v2/large-dam/*`, `dam_rulecurve/graph` | All large/medium dams incl. hourly | Requires HII API key (§2.2) | + +--- + +## 4. Rainfall & weather (external) + +### Near-real-time (usable in the live inference path) + +| Source | Cadence / latency | Access | Notes | +|---|---|---|---| +| HII `rain_24h` (§2.1) | Hourly, near-real-time | Open JSON | Primary rain-gauge feed; persist from now on | +| GSMaP NRT (JAXA) | Hourly, ~4 h latency, 0.1° | Free JAXA registration (FTP); or Google Earth Engine `JAXA/GPM_L3/GSMaP/v8/operational` (no registration) | Gauge-corrected `hourlyPrecipRateGC`; catchment-average rain where gauges are sparse | +| NASA IMERG **Early Run** | Half-hourly, ~4 h latency, 0.1° | Free Earthdata login | NASA-stack alternative to GSMaP | + +### Forecasts (the only way past the ~17 h physical lead-time cap) + +| Source | What | Access | +|---|---|---| +| **Open-Meteo** () | Hourly precip forecast ≤16 days, any lat/lon; **Historical Forecast API archive from 2021** (train on forecast-as-seen, leakage-free); Previous Runs API (fixed 1–7-day leads from Jan 2024); ERA5 back to 1940 | Free, no key, 10k calls/day, non-commercial w/ attribution | +| TMD NWP API (`https://data.tmd.go.th/nwpapi/v1/forecast/location/...`) | WRF 4.2 daily/hourly forecasts by place, processed ~06:00 daily | Free Bearer-token registration (`/nwpapi/doc/main/`) | +| GFS / ECMWF IFS open data | 0.25° global, 4×/day | Free (NOMADS / AWS / data.ecmwf.int); Open-Meteo already wraps both | + +### Training-only (too slow for live) + +| Source | Cadence | Latency | +|---|---|---| +| CHIRPS (`data.chc.ucsb.edu/products/CHIRPS-2.0/`) | Daily, 0.05° | ~2 days prelim / 3+ weeks final | +| IMERG Late / Final | Half-hourly | ~14 h / ~3.5 months | +| TMD observation API (`data.tmd.go.th/api/index1.php`) | 3-hourly / daily station obs, XML | Free uid+key registration | + +--- + +## 5. Historical / open-data portals + +| Portal | Content | +|---|---| +| `https://data.hii.or.th` | 36 HII datasets (rainfall telemetry the most viewed) | +| `https://data.go.th/dataset?organization=rid` | 4 RID datasets (API + ZIP) | +| `https://gdcatalog.go.th` | Nationwide daily rainfall-station catalogs | +| `https://hydro-1.net` | RID Upper-Northern Hydrology Center — hourly/daily tables, hydrology yearbooks (rating curves) for P-series stations; scrape/download | +| `https://water.rid.go.th/flood/flood/daily.pdf` | RID daily flood bulletin (PDF only) | + +--- + +## 6. Integration status & recommended order + +| Source | Status | Action | +|---|---|---| +| RID hourly gauges | ✅ Ingested (hourly → PostgreSQL) | — | +| 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) | +| 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) | +| HII API key (dams, forecasts) | ❌ | Contact HII for sanctioned access | + +**Collector configuration** (`src/hii_collector.py`): runs automatically every +scraping cycle (hourly cadence, even while the RID scraper is in 1-minute retry +mode) from both `--web-api` and continuous-monitoring modes; one-shot via +`python -m src.main --collect-hii`. Env vars: `ENABLE_HII_COLLECTION` +(default `true`), `HII_BASIN_CODE` (default `6` = Ping). Requires a SQL +`DB_TYPE` (sqlite/postgresql/mysql); tables are created automatically. + +**Caveats**: `api-v3` is an undocumented backend — no SLA, no ToS, may change +without notice. Poll hourly at most, cache aggressively, and pursue official +HII access for anything production-critical. diff --git a/docs/references/NOTABLE_DOCUMENTS.md b/docs/references/NOTABLE_DOCUMENTS.md index 1123da0..18b6249 100644 --- a/docs/references/NOTABLE_DOCUMENTS.md +++ b/docs/references/NOTABLE_DOCUMENTS.md @@ -34,6 +34,20 @@ This document contains important references and external resources related to th - **Usage**: Reference for individual station characteristics and historical data patterns - **Station**: P.76 - บ้านแม่อีไฮ (Banb Mae I Hai) +### **ThaiWater / HII (Hydro-Informatics Institute) Resources** + +#### **4. ThaiWater Portal** +- **URL**: https://twa.thaiwater.net +- **Description**: National water situation portal (rainfall, water level, dams, warnings) +- **Language**: Thai/English +- **Usage**: Backed by open and auth-gated APIs — full endpoint catalog in [DATA_SOURCES.md](../DATA_SOURCES.md) + +#### **5. ThaiWater Data Standard** +- **URL**: https://standard.thaiwater.net +- **Description**: Official water-data standard for exchange and warning — canonical station/basin/province code registries, data formats, warning-level definitions +- **Language**: Thai +- **Usage**: Reference for station metadata and warning-level semantics + ## 📊 **Data Sources and APIs** ### **Primary Data Source** diff --git a/src/config.py b/src/config.py index 24666ed..076a59e 100644 --- a/src/config.py +++ b/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")) diff --git a/src/hii_collector.py b/src/hii_collector.py new file mode 100644 index 0000000..60dd93d --- /dev/null +++ b/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) diff --git a/src/main.py b/src/main.py index 50dc2a5..62fa47e 100644 --- a/src/main.py +++ b/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() diff --git a/src/web_api.py b/src/web_api.py index aeb8e63..becbb16 100644 --- a/src/web_api.py +++ b/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 diff --git a/tests/test_hii_collector.py b/tests/test_hii_collector.py new file mode 100644 index 0000000..6ab38e4 --- /dev/null +++ b/tests/test_hii_collector.py @@ -0,0 +1,209 @@ +"""Tests for the HII/ThaiWater api-v3 collector (parsing + persistence).""" + +import datetime + +import pytest + +from src.hii_collector import ( + HiiStore, + parse_rain_records, + parse_waterlevel_records, + rid_code_from_oldcode, +) + + +def _rain_payload(): + return { + "result": "OK", + "data": [ + { + "id": 306091240, + "rain_1h": 0, + "rain_24h": "49.6", + "rainfall_datetime": "2026-08-11 13:00", + "agency": {"agency_shortname": {"en": "HII", "th": "สสน."}}, + "basin": {"basin_code": 6, "basin_name": {"en": "Ping Basin"}}, + "station": { + "id": 418, + "sub_basin_id": "0604", + "tele_station_lat": 19.12207, + "tele_station_long": 98.94447, + "tele_station_name": {"en": "Chiang Mai 5", "th": "แม่แตง"}, + "tele_station_oldcode": "CHM005", + }, + }, + # Wrong basin -> filtered out + { + "id": 2, + "rain_24h": 10, + "rainfall_datetime": "2026-08-11 13:00", + "basin": {"basin_code": 7}, + "station": {"id": 99}, + }, + # No timestamp -> skipped + { + "id": 3, + "rain_24h": 5, + "rainfall_datetime": None, + "basin": {"basin_code": 6}, + "station": {"id": 100}, + }, + ], + } + + +def _waterlevel_payload(): + return { + "waterlevel_data": { + "result": "OK", + "data": [ + { + "id": 1286124160, + "waterlevel_datetime": "2026-08-11 13:00", + "waterlevel_m": None, + "waterlevel_msl": "303.27", + "discharge": "335.00", + "flow_rate": None, + "storage_percent": "81.21", + "situation_level": 4, + "diff_wl_bank": "0.93", + "river_name": "แม่น้ำปิง", + "agency": {"agency_shortname": {"en": "RID"}}, + "basin": {"basin_code": 6}, + "station": { + "id": 3226, + "tele_station_lat": 18.786961, + "tele_station_long": 99.005089, + "tele_station_name": {"th": "สะพานนวรัฐ"}, + "tele_station_oldcode": "P.1", + "offset": 300.5, + "ground_level": 299.25, + "min_bank": 304.2, + "critical_level_msl": 304.2, + "critical_level_m": 3.7, + "qmax": 425, + "is_key_station": True, + }, + }, + { + "id": 2, + "waterlevel_datetime": "2026-08-11 13:00", + "waterlevel_msl": "200.0", + "basin": {"basin_code": 6}, + "station": {"id": 4000, "tele_station_oldcode": "ridhydro_P.67"}, + }, + # Wrong basin -> filtered out + { + "id": 3, + "waterlevel_datetime": "2026-08-11 13:00", + "basin": {"basin_code": 10}, + "station": {"id": 5000}, + }, + ], + } + } + + +class TestRidCodeNormalization: + def test_plain_code(self): + assert rid_code_from_oldcode("P.1") == "P.1" + + def test_ridhydro_prefix(self): + assert rid_code_from_oldcode("ridhydro_P.67") == "P.67" + + def test_letter_suffix(self): + assert rid_code_from_oldcode("ridhydro_P.4A") == "P.4A" + + def test_non_rid_codes(self): + assert rid_code_from_oldcode("CHM005") is None + assert rid_code_from_oldcode("ridtele_TUP.14") is None + assert rid_code_from_oldcode(None) is None + + +class TestParseRain: + def test_filters_and_parses(self): + records = parse_rain_records(_rain_payload()) + assert len(records) == 1 + r = records[0] + assert r["station_id"] == 418 + assert r["oldcode"] == "CHM005" + assert r["name_en"] == "Chiang Mai 5" + assert r["rain_1h"] == 0.0 + assert r["rain_24h"] == 49.6 + assert r["timestamp"] == datetime.datetime(2026, 8, 11, 13, 0) + assert r["agency"] == "HII" + + def test_empty_payload(self): + assert parse_rain_records({}) == [] + + +class TestParseWaterlevel: + def test_filters_and_parses(self): + records = parse_waterlevel_records(_waterlevel_payload()) + assert len(records) == 2 + p1 = records[0] + assert p1["station_id"] == 3226 + assert p1["rid_code"] == "P.1" + assert p1["wl_msl"] == 303.27 + assert p1["discharge"] == 335.0 + assert p1["flow_rate"] is None + assert p1["storage_percent"] == 81.21 + assert p1["situation_level"] == 4 + assert p1["offset_msl"] == 300.5 + assert p1["is_key_station"] is True + # MSL minus station offset recovers the familiar gauge level + assert p1["wl_msl"] - p1["offset_msl"] == pytest.approx(2.77) + assert records[1]["rid_code"] == "P.67" + + def test_empty_payload(self): + assert parse_waterlevel_records({}) == [] + + +class TestHiiStore: + @pytest.fixture + def store(self, tmp_path): + store = HiiStore(f"sqlite:///{tmp_path}/hii_test.db", "sqlite") + assert store.connect() + return store + + def test_rejects_non_sql_backend(self): + with pytest.raises(ValueError): + HiiStore("http://localhost:8428", "victoriametrics") + + def test_rain_roundtrip_and_upsert(self, store): + records = parse_rain_records(_rain_payload()) + assert store.save_rain(records) == 1 + # Same snapshot again -> upsert, still one row + assert store.save_rain(records) == 1 + + from sqlalchemy import text + + with store.engine.connect() as conn: + rows = conn.execute(text("SELECT COUNT(*) FROM hii_rainfall")).scalar() + stations = conn.execute( + text("SELECT oldcode FROM hii_rain_stations") + ).fetchall() + assert rows == 1 + assert stations == [("CHM005",)] + + def test_waterlevel_roundtrip(self, store): + records = parse_waterlevel_records(_waterlevel_payload()) + assert store.save_waterlevel(records) == 2 + + from sqlalchemy import text + + with store.engine.connect() as conn: + row = conn.execute( + text( + "SELECT s.rid_code, m.wl_msl, m.situation_level " + "FROM hii_waterlevel m JOIN hii_wl_stations s ON s.id = m.station_id " + "WHERE s.oldcode = 'P.1'" + ) + ).fetchone() + assert row is not None + assert row[0] == "P.1" + assert float(row[1]) == 303.27 + assert row[2] == 4 + + def test_save_empty(self, store): + assert store.save_rain([]) == 0