fix: restore Hong Kong CoWIN reference curve

This commit is contained in:
2569718930@qq.com
2026-05-28 08:14:27 +08:00
parent 1a7d2a487f
commit 725727d763
6 changed files with 282 additions and 32 deletions
+22 -14
View File
@@ -1214,20 +1214,28 @@ def build_country_network_snapshot(city: str, raw: Dict[str, Any]) -> Dict[str,
} }
airport_primary_today_obs = ((raw.get("metar") or {}).get("today_obs") or []) airport_primary_today_obs = ((raw.get("metar") or {}).get("today_obs") or [])
if airport_primary.get("source_code") == "cowin_obs": if airport_primary.get("source_code") == "cowin_obs":
try: live_points = raw.get("cowin_today_obs") or []
from src.database.runtime_state import OfficialIntradayObservationRepository if live_points:
repo = OfficialIntradayObservationRepository() airport_primary_today_obs = [
local_now = datetime.now(timezone.utc) + timedelta(seconds=city_offset or 0) {"time": p["time"], "temp": p["temp"]}
local_date_str = local_now.strftime("%Y-%m-%d") for p in live_points
points = repo.load_points( if isinstance(p, dict) and p.get("time") and p.get("temp") is not None
source_code="cowin_obs", ]
station_code=airport_primary.get("station_code") or "6087", if not airport_primary_today_obs:
target_date=local_date_str, try:
) from src.database.runtime_state import OfficialIntradayObservationRepository
if points: repo = OfficialIntradayObservationRepository()
airport_primary_today_obs = [{"time": p["time"], "temp": p["temp"]} for p in points] local_now = datetime.now(timezone.utc) + timedelta(seconds=city_offset or 0)
except Exception: local_date_str = local_now.strftime("%Y-%m-%d")
pass points = repo.load_points(
source_code="cowin_obs",
station_code=airport_primary.get("station_code") or "6087",
target_date=local_date_str,
)
if points:
airport_primary_today_obs = [{"time": p["time"], "temp": p["temp"]} for p in points]
except Exception:
pass
return { return {
"provider_code": provider.provider_code, "provider_code": provider.provider_code,
+122 -15
View File
@@ -11,6 +11,7 @@ import os
import time import time
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from urllib.parse import urlencode
import requests import requests
from loguru import logger from loguru import logger
@@ -21,6 +22,16 @@ COWIN_BASE_URL = os.getenv("COWIN_BASE_URL", "").strip() or "https://cowin.hku.h
COWIN_SERIES_URL = f"{COWIN_BASE_URL}/API/data/CoWIN/series" COWIN_SERIES_URL = f"{COWIN_BASE_URL}/API/data/CoWIN/series"
COWIN_STATION_ID = int(os.getenv("COWIN_HK_STATION_ID", "6087")) COWIN_STATION_ID = int(os.getenv("COWIN_HK_STATION_ID", "6087"))
COWIN_STATION_LABEL = os.getenv("COWIN_HK_STATION_LABEL", "").strip() or "保良局陳守仁小學 1min (CoWIN)" COWIN_STATION_LABEL = os.getenv("COWIN_HK_STATION_LABEL", "").strip() or "保良局陳守仁小學 1min (CoWIN)"
COWIN_HK_UTC_OFFSET_SECONDS = 8 * 60 * 60
def _cowin_tls_fallback_enabled() -> bool:
raw = os.getenv("COWIN_ALLOW_INSECURE_TLS_FALLBACK", "1").strip().lower()
return raw not in {"0", "false", "no", "off"}
def _is_tls_certificate_error(exc: Exception) -> bool:
return "CERTIFICATE_VERIFY_FAILED" in str(exc).upper()
def _cowin_obs_time_to_iso(value: Any) -> Optional[str]: def _cowin_obs_time_to_iso(value: Any) -> Optional[str]:
@@ -32,22 +43,52 @@ def _cowin_obs_time_to_iso(value: Any) -> Optional[str]:
except (ValueError, TypeError): except (ValueError, TypeError):
return None return None
if dt.tzinfo is None: if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc) dt = dt.replace(tzinfo=timezone(timedelta(seconds=COWIN_HK_UTC_OFFSET_SECONDS)))
else: return dt.isoformat()
dt = dt.astimezone(timezone.utc)
return dt.isoformat().replace("+00:00", "Z")
class CowinSourceMixin: class CowinSourceMixin:
def _cowin_http_get(self, url: str) -> requests.Response: def _cowin_http_get(self, url: str) -> requests.Response:
getter = getattr(self, "_http_get", None) getter = getattr(self, "_http_get", None)
if callable(getter): try:
resp = getter(url) if callable(getter):
return getter(url)
resp = self.session.get(url, timeout=self.timeout)
resp.raise_for_status()
return resp return resp
resp = self.session.get(url, timeout=self.timeout) except Exception as exc:
resp.raise_for_status() if not _cowin_tls_fallback_enabled() or not _is_tls_certificate_error(exc):
return resp raise
logger.warning(
"CoWIN TLS verification failed; retrying unverified HTTPS for public station data: {}",
exc,
)
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
resp = requests.get(
url,
timeout=self.timeout,
verify=False,
headers={"User-Agent": getattr(self, "user_agent", "PolyWeather/1.0")},
)
resp.raise_for_status()
return resp
def _cowin_series_payload(self, params: Dict[str, Any]) -> Dict[str, Any]:
resp = self._cowin_http_get(COWIN_SERIES_URL + "?" + urlencode(params))
return resp.json() if resp.content else {}
@staticmethod
def _cowin_local_day_bounds(now_utc: Optional[datetime] = None) -> tuple[str, datetime, datetime]:
utc_now = now_utc or datetime.now(timezone.utc)
if utc_now.tzinfo is None:
utc_now = utc_now.replace(tzinfo=timezone.utc)
else:
utc_now = utc_now.astimezone(timezone.utc)
hk_tz = timezone(timedelta(seconds=COWIN_HK_UTC_OFFSET_SECONDS))
local_now = utc_now.astimezone(hk_tz)
local_start = local_now.replace(hour=0, minute=0, second=0, microsecond=0)
return local_now.strftime("%Y-%m-%d"), local_start, local_now
def fetch_cowin_obs_current( def fetch_cowin_obs_current(
self, self,
@@ -69,8 +110,9 @@ class CowinSourceMixin:
return cached["d"] return cached["d"]
try: try:
# Fetch last 10 minutes to get the latest reading # CoWIN series API expects Hong Kong local timestamps.
now = datetime.now(timezone.utc) hk_tz = timezone(timedelta(seconds=COWIN_HK_UTC_OFFSET_SECONDS))
now = datetime.now(timezone.utc).astimezone(hk_tz)
end_dt = now.strftime("%Y-%m-%dT%H:%M:%S") end_dt = now.strftime("%Y-%m-%dT%H:%M:%S")
start_dt = (now - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%S") start_dt = (now - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%S")
@@ -80,10 +122,7 @@ class CowinSourceMixin:
"start_dt": start_dt, "start_dt": start_dt,
"end_dt": end_dt, "end_dt": end_dt,
} }
resp = self._cowin_http_get(COWIN_SERIES_URL + "?" + "&".join( payload = self._cowin_series_payload(params)
f"{k}={v}" for k, v in params.items()
))
payload = resp.json() if resp.content else {}
except Exception as exc: except Exception as exc:
logger.warning("CoWIN obs fetch failed city={} error={}", city_key, exc) logger.warning("CoWIN obs fetch failed city={} error={}", city_key, exc)
with self._cowin_obs_cache_lock: with self._cowin_obs_cache_lock:
@@ -133,6 +172,74 @@ class CowinSourceMixin:
(time.perf_counter() - started) * 1000.0) (time.perf_counter() - started) * 1000.0)
return result return result
def fetch_cowin_obs_today_series(
self,
city: str,
use_fahrenheit: bool = False,
now_utc: Optional[datetime] = None,
) -> list[Dict[str, Any]]:
started = time.perf_counter()
city_key = str(city or "").strip().lower()
if city_key != "hong kong":
return []
local_date_str, start_local, end_local = self._cowin_local_day_bounds(now_utc)
cache_key = f"cowin_obs_today:{city_key}:{use_fahrenheit}:{local_date_str}"
now_ts = time.time()
with self._cowin_obs_cache_lock:
cached = self._cowin_obs_cache.get(cache_key)
if cached and now_ts - cached["t"] < self.cowin_obs_cache_ttl_sec:
record_source_call("cowin_obs", "today_series", "cache_hit",
(time.perf_counter() - started) * 1000.0)
return cached["d"]
try:
payload = self._cowin_series_payload(
{
"station_id": COWIN_STATION_ID,
"element_id": "temp",
"start_dt": start_local.strftime("%Y-%m-%dT%H:%M:%S"),
"end_dt": end_local.strftime("%Y-%m-%dT%H:%M:%S"),
}
)
except Exception as exc:
logger.warning("CoWIN today series fetch failed city={} error={}", city_key, exc)
record_source_call("cowin_obs", "today_series", "error",
(time.perf_counter() - started) * 1000.0)
return []
minutely = payload.get("minutely") if isinstance(payload, dict) else None
if not isinstance(minutely, list) or not minutely:
record_source_call("cowin_obs", "today_series", "no_data",
(time.perf_counter() - started) * 1000.0)
return []
hk_tz = timezone(timedelta(seconds=COWIN_HK_UTC_OFFSET_SECONDS))
points_by_time: Dict[str, Dict[str, Any]] = {}
for row in minutely:
if not isinstance(row, dict):
continue
obs_iso = _cowin_obs_time_to_iso(row.get("obstime"))
if not obs_iso:
continue
try:
obs_dt = datetime.fromisoformat(obs_iso.replace("Z", "+00:00")).astimezone(hk_tz)
temp_c = float(row["value1"])
except (KeyError, ValueError, TypeError):
continue
if obs_dt.strftime("%Y-%m-%d") != local_date_str:
continue
temp = round(temp_c * 9 / 5 + 32, 1) if use_fahrenheit else round(temp_c, 1)
time_label = obs_dt.strftime("%H:%M")
points_by_time[time_label] = {"time": time_label, "temp": temp}
points = sorted(points_by_time.values(), key=lambda item: item["time"])
with self._cowin_obs_cache_lock:
self._cowin_obs_cache[cache_key] = {"d": points, "t": now_ts}
record_source_call("cowin_obs", "today_series", "success",
(time.perf_counter() - started) * 1000.0)
return points
def fetch_cowin_obs_official_nearby( def fetch_cowin_obs_official_nearby(
self, self,
city: str, city: str,
+5
View File
@@ -1266,6 +1266,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
return return
results["cowin_obs_nearby"] = rows results["cowin_obs_nearby"] = rows
results["cowin_current"] = rows[0] results["cowin_current"] = rows[0]
today_obs = self.fetch_cowin_obs_today_series(
city_lower, use_fahrenheit=use_fahrenheit
)
if today_obs:
results["cowin_today_obs"] = today_obs
self._emit_temperature_patch_if_changed( self._emit_temperature_patch_if_changed(
city_lower, city_lower,
rows[0].get("temp"), rows[0].get("temp"),
+43
View File
@@ -358,6 +358,49 @@ def test_hong_kong_cowin_primary_uses_station_6087_history(monkeypatch):
] ]
def test_hong_kong_cowin_primary_prefers_live_intraday_series(monkeypatch):
from src.database import runtime_state
class FakeOfficialIntradayObservationRepository:
def load_points(self, *, source_code, station_code, target_date):
return [{"time": "09:00", "temp": 30.0}]
monkeypatch.setattr(
runtime_state,
"OfficialIntradayObservationRepository",
FakeOfficialIntradayObservationRepository,
)
snapshot = build_country_network_snapshot(
"hong kong",
{
"cowin_current": {
"temp": 31.3,
"obs_time": "2026-05-27T02:41:00Z",
"istNo": "6087",
"icao": "COWIN6087",
"station_label": "保良局陳守仁小學 1min (CoWIN)",
},
"cowin_today_obs": [
{"time": "10:40", "temp": 31.1},
{"time": "10:41", "temp": 31.3},
],
"settlement_current": {
"station_code": "HKO",
"station_name": "HK Observatory",
"observation_time": "2026-05-27T10:40:00+08:00",
"current": {"temp": 31.2},
},
},
)
assert snapshot["airport_primary_current"]["source_code"] == "cowin_obs"
assert snapshot["airport_primary_today_obs"] == [
{"time": "10:40", "temp": 31.1},
{"time": "10:41", "temp": 31.3},
]
def test_moscow_provider_uses_realtime_metar_cluster_not_station_archive_rows(): def test_moscow_provider_uses_realtime_metar_cluster_not_station_archive_rows():
raw = { raw = {
"metar": { "metar": {
+87
View File
@@ -0,0 +1,87 @@
import threading
from datetime import datetime, timezone
import httpx
from src.data_collection import cowin_sources
from src.data_collection.cowin_sources import CowinSourceMixin
class _FakeResponse:
content = b"{}"
def __init__(self, payload):
self._payload = payload
self.text = "{}"
def json(self):
return self._payload
def raise_for_status(self):
return None
class _FakeCowinCollector(CowinSourceMixin):
timeout = 2
cowin_obs_cache_ttl_sec = 0
_cowin_obs_cache = {}
_cowin_obs_cache_lock = threading.Lock()
user_agent = "test"
def _http_get(self, url):
raise httpx.ConnectError("[SSL: CERTIFICATE_VERIFY_FAILED] unable to get local issuer certificate")
def test_cowin_current_retries_without_tls_verification_when_chain_is_incomplete(monkeypatch):
calls = []
def fake_get(url, **kwargs):
calls.append({"url": url, **kwargs})
return _FakeResponse(
{
"station": 6087,
"minutely": [
{"obstime": "2026-05-28T07:59:00", "value1": 30.1},
{"obstime": "2026-05-28T08:00:00", "value1": 30.0},
],
}
)
monkeypatch.setattr(cowin_sources.requests, "get", fake_get)
current = _FakeCowinCollector().fetch_cowin_obs_current("hong kong")
assert current is not None
assert current["station_code"] == "6087"
assert current["current"]["temp"] == 30.0
assert current["obs_time"] == "2026-05-28T08:00:00+08:00"
assert calls
assert calls[0]["verify"] is False
def test_cowin_today_series_returns_hong_kong_local_intraday_points(monkeypatch):
def fake_get(url, **kwargs):
return _FakeResponse(
{
"station": 6087,
"minutely": [
{"obstime": "2026-05-27T23:59:00", "value1": 29.8},
{"obstime": "2026-05-28T00:00:00", "value1": 29.9},
{"obstime": "2026-05-28T07:58:00", "value1": 30.1},
{"obstime": "2026-05-28T08:00:00", "value1": 30.0},
],
}
)
monkeypatch.setattr(cowin_sources.requests, "get", fake_get)
points = _FakeCowinCollector().fetch_cowin_obs_today_series(
"hong kong",
now_utc=datetime(2026, 5, 28, 0, 5, tzinfo=timezone.utc),
)
assert points == [
{"time": "00:00", "temp": 29.9},
{"time": "07:58", "temp": 30.1},
{"time": "08:00", "temp": 30.0},
]
+3 -3
View File
@@ -21,9 +21,9 @@ def test_aeroweb_obs_time_is_utc_aware():
) )
def test_cowin_obs_time_uses_utc_window_when_timezone_is_missing(): def test_cowin_obs_time_keeps_hong_kong_timezone_when_timezone_is_missing():
assert _cowin_obs_time_to_iso("2026-05-27T01:15:00") == "2026-05-27T01:15:00Z" assert _cowin_obs_time_to_iso("2026-05-27T09:15:00") == "2026-05-27T09:15:00+08:00"
assert _cowin_obs_time_to_iso("2026-05-27T09:15:00+08:00") == "2026-05-27T01:15:00Z" assert _cowin_obs_time_to_iso("2026-05-27T09:15:00+08:00") == "2026-05-27T09:15:00+08:00"
def test_hko_one_minute_obs_time_keeps_hong_kong_timezone(): def test_hko_one_minute_obs_time_keeps_hong_kong_timezone():