From fcaa9d5c7f2e7300b97b2450e2e99729b9b65347 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 23 Mar 2026 21:28:04 +0800 Subject: [PATCH] Store HKO intraday observations in runtime state --- src/data_collection/settlement_sources.py | 71 +++++++----------- src/database/runtime_state.py | 88 +++++++++++++++++++++++ 2 files changed, 115 insertions(+), 44 deletions(-) diff --git a/src/data_collection/settlement_sources.py b/src/data_collection/settlement_sources.py index e8eed266..d00e1129 100644 --- a/src/data_collection/settlement_sources.py +++ b/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() - os.makedirs(os.path.dirname(path), exist_ok=True) + mode = get_state_storage_mode() + 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] = {} - if os.path.exists(path): - try: - with open(path, "r", encoding="utf-8") as fh: - payload = json.load(fh) or {} - except Exception: - payload = {} - - existing_date = str(payload.get("date") or "").strip() - if existing_date != date_str: - payload = {"date": date_str, "points": []} - - indexed: Dict[str, Dict[str, Any]] = {} - for raw_point in payload.get("points") or []: - 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 + _official_intraday_repo.upsert_point( + source_code="hko", + station_code="HKO", + target_date=date_str, + observation_time=time_str, + value=round(float(current_temp), 1), + payload={"time": time_str, "temp": round(float(current_temp), 1)}, + ) + points = _official_intraday_repo.load_points( + source_code="hko", + station_code="HKO", + target_date=date_str, + ) + return self._sort_temp_points(points) def fetch_hko_settlement_current(self) -> Optional[Dict[str, Any]]: cache_key = "hko:hong_kong" diff --git a/src/database/runtime_state.py b/src/database/runtime_state.py index ba4b5db2..fba44700 100644 --- a/src/database/runtime_state.py +++ b/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