Store HKO intraday observations in runtime state

src/data_collection/settlement_sources.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
 
 import csv
-import json
 import math
-import os
 import threading
 import time
 from datetime import datetime, timedelta, timezone
@@ -11,6 +9,16 @@ from typing import Any, Dict, List, Optional
 
 from loguru import logger
 
+from src.database.runtime_state import (
+    OfficialIntradayObservationRepository,
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+    get_state_storage_mode,
+)
+
+
+_official_intraday_repo = OfficialIntradayObservationRepository()
+
 
 class SettlementSourceMixin:
     IMGW_METEO_API_BASE = "https://meteo.imgw.pl/api/v1"
@@ -116,13 +124,6 @@ class SettlementSourceMixin:
                 return row
         return rows[0] if rows else None
 
-    def _get_runtime_data_dir(self) -> str:
-        configured = str(os.getenv("POLYWEATHER_RUNTIME_DATA_DIR") or "").strip()
-        if configured:
-            return configured
-        project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-        return os.path.join(project_root, "data")
-
     def _get_settlement_series_lock(self) -> threading.Lock:
         lock = getattr(self, "_settlement_series_lock", None)
         if lock is not None:
@@ -131,9 +132,6 @@ class SettlementSourceMixin:
         setattr(self, "_settlement_series_lock", lock)
         return lock
 
-    def _hko_today_obs_path(self) -> str:
-        return os.path.join(self._get_runtime_data_dir(), "hko_hong_kong_today_obs.json")
-
     @staticmethod
     def _sort_temp_points(points: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         def _key(item: Dict[str, Any]) -> tuple:
@@ -164,41 +162,26 @@ class SettlementSourceMixin:
         local_dt = obs_dt.astimezone(timezone(timedelta(hours=8)))
         date_str = local_dt.strftime("%Y-%m-%d")
         time_str = local_dt.strftime("%H:%M")
-        path = self._hko_today_obs_path()
+        mode = get_state_storage_mode()
-        os.makedirs(os.path.dirname(path), exist_ok=True)
+        if mode not in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+            return [{"time": time_str, "temp": round(float(current_temp), 1)}]
 
         lock = self._get_settlement_series_lock()
         with lock:
-            payload: Dict[str, Any] = {}
+            _official_intraday_repo.upsert_point(
-            if os.path.exists(path):
+                source_code="hko",
-                try:
+                station_code="HKO",
-                    with open(path, "r", encoding="utf-8") as fh:
+                target_date=date_str,
-                        payload = json.load(fh) or {}
+                observation_time=time_str,
-                except Exception:
+                value=round(float(current_temp), 1),
-                    payload = {}
+                payload={"time": time_str, "temp": round(float(current_temp), 1)},
-
+            )
-            existing_date = str(payload.get("date") or "").strip()
+            points = _official_intraday_repo.load_points(
-            if existing_date != date_str:
+                source_code="hko",
-                payload = {"date": date_str, "points": []}
+                station_code="HKO",
-
+                target_date=date_str,
-            indexed: Dict[str, Dict[str, Any]] = {}
+            )
-            for raw_point in payload.get("points") or []:
+            return self._sort_temp_points(points)
-                if not isinstance(raw_point, dict):
-                    continue
-                raw_time = str(raw_point.get("time") or "").strip()
-                raw_temp = self._safe_float(raw_point.get("temp"))
-                if not raw_time or raw_temp is None:
-                    continue
-                indexed[raw_time] = {"time": raw_time, "temp": round(raw_temp, 1)}
-
-            indexed[time_str] = {"time": time_str, "temp": round(float(current_temp), 1)}
-            points = self._sort_temp_points(list(indexed.values()))
-            payload = {"date": date_str, "station_code": "HKO", "points": points}
-
-            with open(path, "w", encoding="utf-8") as fh:
-                json.dump(payload, fh, ensure_ascii=False)
-
-            return points
 
     def fetch_hko_settlement_current(self) -> Optional[Dict[str, Any]]:
         cache_key = "hko:hong_kong"

src/database/runtime_state.py

@@ -132,6 +132,22 @@ class RuntimeStateDB:
             conn.execute(
                 "CREATE INDEX IF NOT EXISTS idx_open_meteo_cache_expires ON open_meteo_cache_store(source_kind, expires_at)"
             )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS official_intraday_observations_store (
+                    source_code TEXT NOT NULL,
+                    station_code TEXT NOT NULL,
+                    target_date TEXT NOT NULL,
+                    observation_time TEXT NOT NULL,
+                    value REAL NOT NULL,
+                    payload_json TEXT,
+                    PRIMARY KEY (source_code, station_code, observation_time)
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_official_intraday_obs_station_date ON official_intraday_observations_store(source_code, station_code, target_date, observation_time)"
+            )
             conn.commit()
 
 
@@ -466,6 +482,78 @@ class OpenMeteoCacheRepository:
             return 0.0
 
 
+class OfficialIntradayObservationRepository:
+    def __init__(self, db: Optional[RuntimeStateDB] = None):
+        self.db = db or RuntimeStateDB.instance()
+
+    def upsert_point(
+        self,
+        *,
+        source_code: str,
+        station_code: str,
+        target_date: str,
+        observation_time: str,
+        value: float,
+        payload: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        payload_json = json.dumps(payload, ensure_ascii=False) if payload is not None else None
+        with self.db.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO official_intraday_observations_store (
+                    source_code, station_code, target_date, observation_time, value, payload_json
+                ) VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(source_code, station_code, observation_time) DO UPDATE SET
+                    target_date = excluded.target_date,
+                    value = excluded.value,
+                    payload_json = excluded.payload_json
+                """,
+                (
+                    source_code,
+                    station_code,
+                    target_date,
+                    observation_time,
+                    float(value),
+                    payload_json,
+                ),
+            )
+            conn.commit()
+
+    def load_points(
+        self,
+        *,
+        source_code: str,
+        station_code: str,
+        target_date: str,
+    ) -> List[Dict[str, Any]]:
+        with self.db.connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT observation_time, value, payload_json
+                FROM official_intraday_observations_store
+                WHERE source_code = ? AND station_code = ? AND target_date = ?
+                ORDER BY observation_time
+                """,
+                (source_code, station_code, target_date),
+            ).fetchall()
+        out: List[Dict[str, Any]] = []
+        for row in rows:
+            point = {
+                "time": str(row["observation_time"] or "").strip(),
+                "temp": float(row["value"]),
+            }
+            if row["payload_json"]:
+                try:
+                    payload = json.loads(row["payload_json"])
+                except Exception:
+                    payload = None
+                if isinstance(payload, dict):
+                    point.update(payload)
+            if point["time"]:
+                out.append(point)
+        return out
+
+
 def _top_bucket(snapshot: Optional[List[Dict[str, Any]]]) -> Optional[int]:
     best_value = None
     best_prob = -1.0