diff --git a/src/data_collection/country_networks.py b/src/data_collection/country_networks.py index be1093a9..95105487 100644 --- a/src/data_collection/country_networks.py +++ b/src/data_collection/country_networks.py @@ -425,6 +425,35 @@ def _airport_primary_from_raw(city: str, raw: Dict[str, Any]) -> Dict[str, Any]: }, ) + cowin = raw.get("cowin_current") or {} + if cowin.get("temp") is not None: + return _normalize_station_row( + station_code=meta.get("icao") or str(cowin.get("icao") or "COWIN6087"), + station_label=meta.get("airport_name") or cowin.get("station_label") or meta.get("icao"), + temp=_safe_float(cowin["temp"]), + obs_time=str(cowin.get("obs_time") or metar.get("observation_time") or ""), + source_code="cowin_obs", + source_label="CoWIN 6087", + is_official=True, + is_airport_station=True, + is_settlement_anchor=False, + extra={ + "max_so_far": _safe_float(current.get("max_temp_so_far")), + "max_temp_time": current.get("max_temp_time"), + "obs_age_min": None, + "report_time": metar.get("report_time"), + "receipt_time": metar.get("receipt_time"), + "obs_time_epoch": metar.get("obs_time_epoch"), + "obs_time_utc_offset_seconds": 0, + "wind_speed_kt": _safe_float(current.get("wind_speed_kt")), + "wind_dir": _safe_float(current.get("wind_dir")), + "humidity": _safe_float(current.get("humidity")), + "visibility_mi": _safe_float(current.get("visibility_mi")), + "wx_desc": current.get("wx_desc"), + "raw_metar": current.get("raw_metar"), + }, + ) + aeroweb = raw.get("aeroweb") or {} aw_current = aeroweb.get("current") or {} if aw_current.get("temp") is not None: diff --git a/src/data_collection/cowin_sources.py b/src/data_collection/cowin_sources.py new file mode 100644 index 00000000..2317d602 --- /dev/null +++ b/src/data_collection/cowin_sources.py @@ -0,0 +1,142 @@ +"""CoWIN (Community Weather Information Network) 1-minute real-time data source. + +Fetches 1-minute temperature from HKU CoWIN API for Hong Kong. +Station 6087 (保良局陳守仁小學) provides true 1-minute observations. +No API key required. +""" + +from __future__ import annotations + +import os +import time +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional + +import requests +from loguru import logger + +from src.utils.metrics import record_source_call + +COWIN_BASE_URL = os.getenv("COWIN_BASE_URL", "").strip() or "https://cowin.hku.hk" +COWIN_SERIES_URL = f"{COWIN_BASE_URL}/API/data/CoWIN/series" +COWIN_STATION_ID = int(os.getenv("COWIN_HK_STATION_ID", "6087")) +COWIN_STATION_LABEL = os.getenv("COWIN_HK_STATION_LABEL", "").strip() or "保良局陳守仁小學 1min (CoWIN)" + + +class CowinSourceMixin: + + def _cowin_http_get(self, url: str) -> requests.Response: + getter = getattr(self, "_http_get", None) + if callable(getter): + resp = getter(url) + return resp + resp = self.session.get(url, timeout=self.timeout) + resp.raise_for_status() + return resp + + def fetch_cowin_obs_current( + self, + city: str, + use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + started = time.perf_counter() + city_key = str(city or "").strip().lower() + if city_key != "hong kong": + return None + + cache_key = f"cowin_obs:{city_key}:{use_fahrenheit}" + 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", "current", "cache_hit", + (time.perf_counter() - started) * 1000.0) + return cached["d"] + + try: + # Fetch last 10 minutes to get the latest reading + now = datetime.now(timezone.utc) + end_dt = now.strftime("%Y-%m-%dT%H:%M:%S") + start_dt = (now - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%S") + + params = { + "station_id": COWIN_STATION_ID, + "element_id": "temp", + "start_dt": start_dt, + "end_dt": end_dt, + } + resp = self._cowin_http_get(COWIN_SERIES_URL + "?" + "&".join( + f"{k}={v}" for k, v in params.items() + )) + payload = resp.json() if resp.content else {} + except Exception as exc: + logger.warning("CoWIN obs fetch failed city={} error={}", city_key, exc) + with self._cowin_obs_cache_lock: + stale = self._cowin_obs_cache.get(cache_key) + if stale: + record_source_call("cowin_obs", "current", "stale_cache", + (time.perf_counter() - started) * 1000.0) + return stale["d"] + record_source_call("cowin_obs", "current", "error", + (time.perf_counter() - started) * 1000.0) + return None + + minutely = payload.get("minutely") if isinstance(payload, dict) else None + if not minutely or not isinstance(minutely, list) or not minutely: + record_source_call("cowin_obs", "current", "no_data", + (time.perf_counter() - started) * 1000.0) + return None + + latest = minutely[-1] + try: + temp_c = float(latest["value1"]) + except (KeyError, ValueError, TypeError): + record_source_call("cowin_obs", "current", "no_temperature", + (time.perf_counter() - started) * 1000.0) + return None + + obs_time = str(latest.get("obstime") or "").strip() + + temp = round(temp_c * 9 / 5 + 32, 1) if use_fahrenheit else round(temp_c, 1) + + result = { + "source": "cowin_obs", + "timestamp": datetime.now(timezone.utc).isoformat(), + "station_code": str(COWIN_STATION_ID), + "station_name": COWIN_STATION_LABEL, + "icao": f"COWIN{COWIN_STATION_ID}", + "obs_time": obs_time or datetime.now(timezone.utc).isoformat(), + "current": { + "temp": temp, + }, + "temp_c": temp_c, + } + + with self._cowin_obs_cache_lock: + self._cowin_obs_cache[cache_key] = {"d": result, "t": now_ts} + record_source_call("cowin_obs", "current", "success", + (time.perf_counter() - started) * 1000.0) + return result + + def fetch_cowin_obs_official_nearby( + self, + city: str, + use_fahrenheit: bool = False, + ) -> list[Dict[str, Any]]: + current = self.fetch_cowin_obs_current(city, use_fahrenheit=use_fahrenheit) + if not current: + return [] + return [ + { + "name": COWIN_STATION_LABEL, + "station_label": COWIN_STATION_LABEL, + "lat": 22.3050, + "lon": 114.1670, + "temp": (current.get("current") or {}).get("temp"), + "icao": f"COWIN{COWIN_STATION_ID}", + "istNo": str(COWIN_STATION_ID), + "source": "cowin_obs", + "source_label": "CoWIN 6087", + "obs_time": current.get("obs_time"), + } + ] diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 35513e3b..3e960aa7 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -22,6 +22,7 @@ from src.data_collection.amsc_awos_sources import AmscAwosSourceMixin from src.data_collection.fmi_sources import FmiSourceMixin from src.data_collection.knmi_sources import KnmiSourceMixin from src.data_collection.hko_obs_sources import HkoObsSourceMixin +from src.data_collection.cowin_sources import CowinSourceMixin from src.data_collection.madis_sources import MadisSourceMixin from src.data_collection.singapore_mss_sources import SingaporeMssSourceMixin from src.data_collection.ims_sources import ImsSourceMixin @@ -30,7 +31,7 @@ from src.data_collection.aeroweb_sources import AerowebSourceMixin from src.database.db_manager import DBManager -class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, AmscAwosSourceMixin, FmiSourceMixin, KnmiSourceMixin, HkoObsSourceMixin, MadisSourceMixin, SingaporeMssSourceMixin, ImsSourceMixin, NcmSourceMixin, AerowebSourceMixin): +class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, AmscAwosSourceMixin, FmiSourceMixin, KnmiSourceMixin, HkoObsSourceMixin, CowinSourceMixin, MadisSourceMixin, SingaporeMssSourceMixin, ImsSourceMixin, NcmSourceMixin, AerowebSourceMixin): """ Multi-source weather data collector @@ -248,6 +249,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._madis_cache_ts: float = 0.0 self._madis_cache_lock = threading.Lock() self._hko_obs_cache_lock = threading.Lock() + self.cowin_obs_cache_ttl_sec = int( + os.getenv("COWIN_OBS_CACHE_TTL_SEC", "120") + ) + self._cowin_obs_cache: Dict[str, Dict] = {} + self._cowin_obs_cache_lock = threading.Lock() self.cwa_open_data_auth = ( os.getenv("CWA_OPEN_DATA_AUTH") or os.getenv("CWA_OPEN_DATA_API_KEY") @@ -322,6 +328,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour (self._fmi_cache, self._fmi_cache_lock, float(self.fmi_cache_ttl_sec * 2)), (self._knmi_cache, self._knmi_cache_lock, float(self.knmi_cache_ttl_sec * 2)), (self._hko_obs_cache, self._hko_obs_cache_lock, float(self.hko_obs_cache_ttl_sec * 2)), + (self._cowin_obs_cache, self._cowin_obs_cache_lock, float(self.cowin_obs_cache_ttl_sec * 5)), ]: stale = [ key @@ -825,6 +832,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._knmi_cache.pop(f"knmi:{normalized}:{use_fahrenheit}", None) with self._hko_obs_cache_lock: self._hko_obs_cache.pop(f"hko_obs:{normalized}:{use_fahrenheit}", None) + with self._cowin_obs_cache_lock: + self._cowin_obs_cache.pop(f"cowin_obs:{normalized}:{use_fahrenheit}", None) with self._settlement_cache_lock: city_meta = self.CITY_REGISTRY.get(normalized) or {} settlement_source = str(city_meta.get("settlement_source") or "").strip().lower() @@ -1116,6 +1125,31 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour except Exception: logger.exception("airport_obs_log append failed for hko_obs city={}", city_lower) + def _attach_cowin_official_nearby( + self, results: Dict, city_lower: str, use_fahrenheit: bool + ) -> None: + if city_lower != "hong kong": + return + rows = self.fetch_cowin_obs_official_nearby( + city_lower, use_fahrenheit=use_fahrenheit + ) + if not rows: + return + results["cowin_obs_nearby"] = rows + results["cowin_current"] = rows[0] + results["mgm_nearby"] = rows + results["nearby_source"] = "cowin_obs" + try: + row = rows[0] if rows else {} + DBManager().append_airport_obs( + icao=str(row.get("icao") or "COWIN6087"), + city=city_lower, + temp_c=row.get("temp"), + obs_time=str(row.get("obs_time") or datetime.now().isoformat()), + ) + except Exception: + logger.exception("airport_obs_log append failed for cowin_obs city={}", city_lower) + def _attach_cwa_settlement_nearby( self, results: Dict, city_lower: str, use_fahrenheit: bool ) -> None: @@ -1493,6 +1527,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_japan_official_nearby(results, city_lower, use_fahrenheit) self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit) self._attach_knmi_official_nearby(results, city_lower, use_fahrenheit) + self._attach_cowin_official_nearby(results, city_lower, use_fahrenheit) self._attach_hko_obs_official_nearby(results, city_lower, use_fahrenheit) self._attach_cwa_settlement_nearby(results, city_lower, use_fahrenheit) if city_lower == "warsaw": @@ -1544,6 +1579,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_japan_official_nearby(results, city_lower, use_fahrenheit) self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit) self._attach_knmi_official_nearby(results, city_lower, use_fahrenheit) + self._attach_cowin_official_nearby(results, city_lower, use_fahrenheit) self._attach_hko_obs_official_nearby(results, city_lower, use_fahrenheit) self._attach_cwa_settlement_nearby(results, city_lower, use_fahrenheit) if city_lower == "warsaw":