diff --git a/src/forecast_history.py b/src/forecast_history.py new file mode 100644 index 0000000..cdef15d --- /dev/null +++ b/src/forecast_history.py @@ -0,0 +1,172 @@ +"""Persistence for issued flood forecasts. + +Every background precompute stores what the deployed model predicted at that +moment — predicted 24/12/6 h peak, warning/danger probabilities, model +version. Keyed by (as_of, station, horizon), so hourly data yields one row +per station-horizon per hour regardless of how often the precompute runs. +This is the operational record that lets "predicted vs actual" be graphed +later without retraining historical models. +""" + +import datetime +import logging +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +class ForecastHistoryStore: + """SQL store (sqlite / postgresql / mysql), same pattern as HiiStore.""" + + 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"Forecast history 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, text + + self.engine = create_engine(self.connection_string, pool_pre_ping=True) + ddl = """ + CREATE TABLE IF NOT EXISTS forecast_history ( + as_of TIMESTAMP NOT NULL, + station_code VARCHAR(10) NOT NULL, + horizon_hours INTEGER NOT NULL, + predicted_max_level NUMERIC(8,3), + p_warning NUMERIC(7,5), + p_danger NUMERIC(7,5), + current_level NUMERIC(8,3), + model_version VARCHAR(64), + source VARCHAR(16), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (as_of, station_code, horizon_hours) + ) + """ + index = ( + "CREATE INDEX IF NOT EXISTS idx_forecast_history_station " + "ON forecast_history(station_code, as_of)" + ) + with self.engine.begin() as conn: + conn.execute(text(ddl)) + if self.db_type != "mysql": # MySQL lacks IF NOT EXISTS for indexes + conn.execute(text(index)) + return True + except Exception as error: + logger.error(f"ForecastHistoryStore failed to connect: {error}") + self.engine = None + return False + + def save_rows(self, rows: List[Dict]) -> int: + """Upsert forecast rows as returned by ml.predict (idempotent).""" + if not rows: + return 0 + if not self.engine and not self.connect(): + return 0 + from sqlalchemy import text + + cols = ( + "(as_of, station_code, horizon_hours, predicted_max_level, " + "p_warning, p_danger, current_level, model_version, source)" + ) + values = ( + "(:as_of, :station_code, :horizon_hours, :predicted_max_level, " + ":p_warning, :p_danger, :current_level, :model_version, :source)" + ) + update_cols = ( + "predicted_max_level", + "p_warning", + "p_danger", + "current_level", + "model_version", + "source", + ) + if self.db_type == "mysql": + updates = ", ".join(f"{c} = VALUES({c})" for c in update_cols) + sql = ( + f"INSERT INTO forecast_history {cols} VALUES {values} " + f"ON DUPLICATE KEY UPDATE {updates}" + ) + else: + updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in update_cols) + sql = ( + f"INSERT INTO forecast_history {cols} VALUES {values} " + f"ON CONFLICT (as_of, station_code, horizon_hours) " + f"DO UPDATE SET {updates}" + ) + + params = [] + for row in rows: + as_of = row.get("as_of") + if isinstance(as_of, str): + as_of = datetime.datetime.fromisoformat(as_of) + if as_of is None or row.get("station_code") is None: + continue + params.append( + { + "as_of": as_of, + "station_code": row["station_code"], + "horizon_hours": row.get("horizon_hours"), + "predicted_max_level": row.get("predicted_max_level"), + "p_warning": row.get("p_warning"), + "p_danger": row.get("p_danger"), + "current_level": row.get("current_level"), + "model_version": row.get("model_version"), + "source": row.get("source"), + } + ) + if not params: + return 0 + try: + with self.engine.begin() as conn: + conn.execute(text(sql), params) + return len(params) + except Exception as error: + logger.error(f"ForecastHistoryStore save failed: {error}") + return 0 + + def fetch( + self, + station_code: str, + start: Optional[datetime.datetime] = None, + end: Optional[datetime.datetime] = None, + horizon_hours: Optional[int] = None, + limit: int = 5000, + ) -> List[Dict]: + """Issued forecasts for one station, ascending by as_of.""" + if not self.engine and not self.connect(): + return [] + from sqlalchemy import text + + clauses = ["station_code = :code"] + params: Dict = {"code": station_code, "limit": limit} + if start is not None: + clauses.append("as_of >= :start") + params["start"] = start + if end is not None: + clauses.append("as_of <= :end") + params["end"] = end + if horizon_hours is not None: + clauses.append("horizon_hours = :horizon") + params["horizon"] = horizon_hours + sql = ( + "SELECT as_of, station_code, horizon_hours, predicted_max_level, " + "p_warning, p_danger, current_level, model_version, source " + f"FROM forecast_history WHERE {' AND '.join(clauses)} " + "ORDER BY as_of ASC, horizon_hours ASC LIMIT :limit" + ) + try: + with self.engine.connect() as conn: + rows = [dict(r._mapping) for r in conn.execute(text(sql), params)] + for row in rows: + for key, value in row.items(): + if hasattr(value, "is_finite"): # Decimal -> float + row[key] = float(value) + return rows + except Exception as error: + logger.error(f"ForecastHistoryStore fetch failed: {error}") + return [] diff --git a/src/static/dashboard.html b/src/static/dashboard.html index 3d5fc8e..bef0101 100644 --- a/src/static/dashboard.html +++ b/src/static/dashboard.html @@ -633,6 +633,34 @@ }; const sampled = rows.length > 2000 ? downsample(rows) : rows; + // Issued model predictions (recorded hourly by the server) — + // overlays what the model said would happen, when it said it. + let predSeries = null; + try { + const fr = await fetch(`/api/forecast/history/${encodeURIComponent(stationCode)}?${query}&horizon=24`); + const forecastRows = fr.ok ? await fr.json() : []; + if (forecastRows.length) { + const keyOf = (value) => { + const d = new Date(value); + return rows.length > 2000 + ? `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}` + : `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}-${d.getUTCHours()}`; + }; + const byKey = new Map(); + forecastRows.forEach((r) => { + if (r.predicted_max_level == null) return; + const key = keyOf(r.as_of); + if (!byKey.has(key)) byKey.set(key, []); + byKey.get(key).push(Number(r.predicted_max_level)); + }); + predSeries = sampled.map((row) => { + const vals = byKey.get(keyOf(row.timestamp)); + return vals ? vals.reduce((a, c) => a + c, 0) / vals.length : null; + }); + if (!predSeries.some((v) => v != null)) predSeries = null; + } + } catch (error) { /* overlay is optional */ } + // Safely clear existing chart instance if (state.historyChart) { state.historyChart.destroy(); state.historyChart = null; } // Clear any orphan Chart.js instance on the canvas @@ -645,7 +673,8 @@ labels: sampled.map((row) => new Date(row.timestamp).toLocaleString('en-TH', { timeZone: 'Asia/Bangkok', month: 'short', day: 'numeric', year: '2-digit', hour: '2-digit' })), datasets: [ { label: 'Discharge (m³/s)', data: sampled.map((row) => row.discharge), borderColor: '#087da5', backgroundColor: 'rgba(8,125,165,.12)', yAxisID: 'flow', pointRadius: 0, tension: .25 }, - { label: 'Water level (m)', data: sampled.map((row) => row.water_level), borderColor: '#d99018', yAxisID: 'level', pointRadius: 0, tension: .25 } + { label: 'Water level (m)', data: sampled.map((row) => row.water_level), borderColor: '#d99018', yAxisID: 'level', pointRadius: 0, tension: .25 }, + ...(predSeries ? [{ label: 'Model 24 h peak (as issued)', data: predSeries, borderColor: '#7c4fd0', borderDash: [6, 4], yAxisID: 'level', pointRadius: 0, tension: .25, spanGaps: true }] : []) ] }, options: { diff --git a/src/web_api.py b/src/web_api.py index 5975fab..801b29e 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -60,7 +60,9 @@ HISTORY_TTL = 300 # 5 minutes FORECAST_CACHE: Dict[str, tuple] = {} FORECAST_CACHE_LOCK = Lock() FORECAST_COMPUTE_LOCK = asyncio.Lock() # single-flight for expensive inference -FORECAST_TTL = 900 # 15 minutes +# The leader worker precomputes after every scrape cycle (hourly); the TTL just +# needs to outlive one cycle so user requests never trigger inference themselves. +FORECAST_TTL = int(os.getenv("FORECAST_TTL_SECONDS", "4500")) DB_STATS_CACHE: Dict[str, tuple] = {} DB_STATS_CACHE_LOCK = Lock() @@ -159,6 +161,20 @@ async def lifespan(app: FastAPI): db_config = Config.get_database_config() app_state["scraper"] = EnhancedWaterMonitorScraper(db_config) + # Forecast history store (SQL only): records what the model predicted + try: + if db_config["type"] in ("sqlite", "postgresql", "mysql"): + from .forecast_history import ForecastHistoryStore + + app_state["forecast_store"] = ForecastHistoryStore( + db_config["connection_string"], db_config["type"] + ) + else: + app_state["forecast_store"] = None + except Exception as e: + app_state["forecast_store"] = None + logger.error(f"Forecast history store init failed: {e}") + # Initialize HII/ThaiWater collector (rainfall + backup water level) try: from .hii_collector import create_collector_from_config @@ -248,6 +264,32 @@ app.add_middleware( ) +async def _precompute_forecasts(): + """Refresh the forecast cache and persist the issued forecasts (leader only).""" + try: + from .ml.predict import get_latest_forecasts + except ImportError: + return + try: + rows = await asyncio.to_thread(get_latest_forecasts) + except Exception as e: + logger.warning(f"Forecast precompute failed: {e}") + return + if not rows: + return + with FORECAST_CACHE_LOCK: + FORECAST_CACHE["all"] = (time.monotonic(), rows) + store = app_state.get("forecast_store") + if store: + try: + saved = await asyncio.to_thread(store.save_rows, rows) + logger.info( + f"Forecast precompute: cached {len(rows)} rows, persisted {saved}" + ) + except Exception as e: + logger.warning(f"Forecast history save failed: {e}") + + async def background_scraping_task(): """Background task for periodic data scraping""" while True: @@ -309,6 +351,12 @@ async def background_scraping_task(): except Exception as e: logger.error(f"HII collection failed: {e}") + # Precompute forecasts on fresh data: primes the response + # cache (user requests never pay for inference) and records + # what the model predicted for later predicted-vs-actual + # evaluation. + await _precompute_forecasts() + app_state["is_scraping"] = False # Calculate next run time @@ -366,6 +414,7 @@ def _send_umami_event(path: str, method: str, status: int, host: str, user_agent _CACHE_CONTROL_RULES = ( ("/static/", "public, max-age=3600"), ("/measurements/latest", "public, max-age=30"), + ("/api/forecast/", "public, max-age=300"), ("/api/hii/", "public, max-age=60"), ("/measurements/history", "public, max-age=300"), ("/forecast", "public, max-age=120"), @@ -1007,6 +1056,35 @@ async def _compute_forecasts(get_latest_forecasts): return data +@app.get("/api/forecast/history/{station_code}") +async def get_forecast_history( + station_code: str, + hours: int = Query(168, ge=1), + start: Optional[date] = Query(None, description="First day (overrides hours)"), + end: Optional[date] = Query(None, description="Last day, inclusive"), + horizon: Optional[int] = Query(None, ge=1, le=48), +): + """Issued model forecasts for a station — what the model predicted, when. + + Populated by the hourly precompute; enables predicted-vs-actual charts. + """ + increment_counter("api_requests", labels={"endpoint": "forecast_history"}) + store = app_state.get("forecast_store") + if not store: + return [] + end_dt = ( + datetime.combine(end, datetime.max.time()) if end else datetime.now() + ) + start_dt = ( + datetime.combine(start, datetime.min.time()) + if start + else end_dt - timedelta(hours=hours) + ) + return await asyncio.to_thread( + store.fetch, station_code, start_dt, end_dt, horizon + ) + + @app.get("/measurements/latest", response_model=List[MeasurementResponse]) async def get_latest_measurements(response: Response, limit: int = 100): """Get latest measurements from all stations""" diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index f9732f8..e0ad3ee 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -77,6 +77,8 @@ def test_dashboard_loads_station_history_chart(): html = DASHBOARD_PATH.read_text(encoding="utf-8") assert "Station history" in html + assert "/api/forecast/history/" in html + assert "Model 24 h peak (as issued)" in html assert "PostgreSQL" not in html assert "/measurements/history/" in html assert "history-chart" in html diff --git a/tests/test_forecast_history.py b/tests/test_forecast_history.py new file mode 100644 index 0000000..799b1fa --- /dev/null +++ b/tests/test_forecast_history.py @@ -0,0 +1,99 @@ +"""Tests for the issued-forecast archive (store + API endpoint).""" + +import asyncio +import datetime + +import pytest + +from src.forecast_history import ForecastHistoryStore + + +def _rows(as_of="2026-08-12T10:00:00"): + return [ + { + "as_of": as_of, + "station_code": "P.1", + "horizon_hours": h, + "predicted_max_level": 2.8 + h / 100, + "p_warning": 0.02, + "p_danger": 0.001, + "current_level": 2.76, + "model_version": "hgb-v1+test", + "trained_at": "2026-08-12T10:52:00", + "source": "model", + } + for h in (6, 12, 24) + ] + + +class TestForecastHistoryStore: + @pytest.fixture + def store(self, tmp_path): + store = ForecastHistoryStore(f"sqlite:///{tmp_path}/fh.db", "sqlite") + assert store.connect() + return store + + def test_rejects_non_sql(self): + with pytest.raises(ValueError): + ForecastHistoryStore("http://x", "victoriametrics") + + def test_roundtrip_and_upsert(self, store): + assert store.save_rows(_rows()) == 3 + # Same as_of again -> upsert, still 3 rows + assert store.save_rows(_rows()) == 3 + rows = store.fetch("P.1") + assert len(rows) == 3 + assert [r["horizon_hours"] for r in rows] == [6, 12, 24] + assert rows[2]["predicted_max_level"] == pytest.approx(3.04) + assert rows[0]["model_version"] == "hgb-v1+test" + + def test_fetch_filters(self, store): + store.save_rows(_rows("2026-08-12T10:00:00")) + store.save_rows(_rows("2026-08-12T11:00:00")) + only_24 = store.fetch("P.1", horizon_hours=24) + assert len(only_24) == 2 + assert all(r["horizon_hours"] == 24 for r in only_24) + windowed = store.fetch( + "P.1", + start=datetime.datetime(2026, 8, 12, 10, 30), + end=datetime.datetime(2026, 8, 12, 12, 0), + ) + assert len(windowed) == 3 # only the 11:00 issue + assert store.fetch("P.99") == [] + + def test_skips_malformed_rows(self, store): + rows = _rows() + [{"station_code": None, "as_of": None}] + assert store.save_rows(rows) == 3 + + +class TestForecastHistoryEndpoint: + def test_endpoint_returns_rows(self, tmp_path, monkeypatch): + from src import web_api + + store = ForecastHistoryStore(f"sqlite:///{tmp_path}/api-fh.db", "sqlite") + assert store.connect() + now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0) + store.save_rows(_rows(now.isoformat())) + monkeypatch.setitem(web_api.app_state, "forecast_store", store) + + rows = asyncio.run( + web_api.get_forecast_history( + "P.1", hours=48, start=None, end=None, horizon=24 + ) + ) + assert len(rows) == 1 + assert rows[0]["horizon_hours"] == 24 + assert rows[0]["station_code"] == "P.1" + + def test_endpoint_without_store(self, monkeypatch): + from src import web_api + + monkeypatch.setitem(web_api.app_state, "forecast_store", None) + assert ( + asyncio.run( + web_api.get_forecast_history( + "P.1", hours=168, start=None, end=None, horizon=None + ) + ) + == [] + )