"""Independent high-frequency observation collector for the web runtime.""" from __future__ import annotations import os import threading import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import Any, Callable, Iterable, List, Optional, Sequence, Tuple from loguru import logger from src.data_collection.amos_station_sources import AMOS_AIRPORT_CODES from src.data_collection.amsc_awos_sources import AMSC_AWOS_AIRPORTS from src.data_collection.city_registry import CITY_REGISTRY from src.data_collection.hko_obs_sources import HKO_STATIONS from src.database.db_manager import DBManager from src.database.runtime_state import ObservationCollectorStatusRepository from web.services.analysis_utils import parse_utc_datetime from web.services.canonical_engine import ( build_realtime_event_from_canonical, refresh_canonical_temperature_from_latest, ) from web.services.canonical_temperature import build_canonical_temperature from web.services.observation_freshness import build_observation_freshness from web.services.observation_source_adapters import ( ObservationRecord, collect_observation_source, ) from web.realtime_event_store_factory import create_realtime_event_store def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _env_int(name: str, default: int) -> int: raw = os.getenv(name) if raw is None: return default try: return int(raw) except Exception: return default def _normalized_cities(cities: Iterable[str]) -> Tuple[str, ...]: return tuple(sorted({str(city or "").strip().lower() for city in cities if str(city or "").strip()})) @dataclass(frozen=True) class ObservationSourceProfile: source: str cities: Tuple[str, ...] interval_sec: int class ObservationCollector: def __init__( self, *, weather: Any, profiles: Sequence[ObservationSourceProfile], cache_refresher: Optional[Callable[[str], Any]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, observation_store: Optional[Any] = None, realtime_event_store: Optional[Any] = None, realtime_broadcaster: Optional[Callable[[dict[str, Any]], Any]] = None, async_cache_refresh: Optional[bool] = None, cache_refresh_workers: Optional[int] = None, ) -> None: self.weather = weather self.profiles = list(profiles) self.cache_refresher = cache_refresher self.status_recorder = status_recorder self.observation_store = observation_store self.realtime_event_store = realtime_event_store self.realtime_broadcaster = realtime_broadcaster self._last_run_ts: dict[tuple[str, str], float] = {} self._lock = threading.Lock() self._cache_refresh_lock = threading.Lock() self._cache_refresh_inflight: set[str] = set() self._cache_refresh_async = ( _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_ASYNC", True) if async_cache_refresh is None else bool(async_cache_refresh) ) worker_count = max( 1, min( 4, int( cache_refresh_workers if cache_refresh_workers is not None else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 2) ), ), ) self._cache_refresh_executor: Optional[ThreadPoolExecutor] = ( ThreadPoolExecutor(max_workers=worker_count) if callable(cache_refresher) and self._cache_refresh_async else None ) def run_due_once(self, *, now_ts: Optional[float] = None) -> int: now = float(time.time() if now_ts is None else now_ts) due: List[tuple[ObservationSourceProfile, str, Optional[int]]] = [] with self._lock: for profile in self.profiles: interval = max(1, int(profile.interval_sec or 60)) for city in profile.cities: key = (profile.source, city) last_ts = float(self._last_run_ts.get(key) or 0.0) if now - last_ts >= interval: self._last_run_ts[key] = now due.append((profile, city, None)) due.extend(self._claim_due_refresh_requests(now)) completed = 0 for profile, city, request_id in due: started_wall = time.time() started_ts = now if now_ts is not None else started_wall ok = False error: Optional[str] = None try: ok = self._collect_city_source(profile.source, city) if ok: completed += 1 self._refresh_city_cache(city) else: error = "no_results" except Exception as exc: error = str(exc) or exc.__class__.__name__ self._store_raw_observation_status( source=profile.source, city=city, status=self._failure_status_from_exception(exc), error=error, ) logger.warning( "observation collector source failed source={} city={}: {}", profile.source, city, exc, ) finally: completed_ts = started_ts + max(0.0, time.time() - started_wall) self._record_source_status( profile=profile, city=city, due_ts=now, started_ts=started_ts, completed_ts=completed_ts, ok=ok, error=error, ) self._mark_refresh_request_done(request_id, ok=ok, error=error) return completed def _claim_due_refresh_requests(self, now: float) -> List[tuple[ObservationSourceProfile, str, Optional[int]]]: store = self.observation_store claimer = getattr(store, "claim_observation_refresh_requests", None) if not callable(claimer): return [] try: requests = claimer(limit=50, owner="observation_collector", now_ts=now) except Exception as exc: logger.debug("observation refresh request claim skipped: {}", exc) return [] due: List[tuple[ObservationSourceProfile, str, Optional[int]]] = [] for request in requests or []: if not isinstance(request, dict): continue city = str(request.get("city") or "").strip().lower() requested_source = str(request.get("source") or "").strip().lower() request_id = int(request.get("id") or 0) or None matched = False rate_limited = False for profile in self.profiles: profile_source = str(profile.source or "").strip().lower() if requested_source and requested_source != profile_source: continue if city not in set(profile.cities): continue key = (profile.source, city) interval = max(1, int(profile.interval_sec or 60)) with self._lock: last_ts = float(self._last_run_ts.get(key) or 0.0) if last_ts and now - last_ts < interval: rate_limited = True continue self._last_run_ts[key] = now due.append((profile, city, request_id)) matched = True if not matched: reason = "rate_limited" if rate_limited else "no_matching_profile" self._mark_refresh_request_done(request_id, ok=False, error=reason) return due def _mark_refresh_request_done( self, request_id: Optional[int], *, ok: bool, error: Optional[str], ) -> None: if not request_id: return store = self.observation_store marker = getattr(store, "mark_observation_refresh_request_done", None) if not callable(marker): return try: marker( request_id, status="done" if ok else "failed", error="" if ok else str(error or "collection_failed"), ) except Exception as exc: logger.debug("observation refresh request completion skipped id={}: {}", request_id, exc) def _record_source_status( self, *, profile: ObservationSourceProfile, city: str, due_ts: float, started_ts: float, completed_ts: float, ok: bool, error: Optional[str], ) -> None: if not self.status_recorder: return try: self.status_recorder.record_result( source=profile.source, city=city, interval_sec=profile.interval_sec, due_ts=due_ts, started_ts=started_ts, completed_ts=completed_ts, ok=ok, error=error, ) except Exception as exc: logger.warning( "observation collector status write failed source={} city={}: {}", profile.source, city, exc, ) def _collect_city_source(self, source: str, city: str) -> bool: normalized_source = str(source or "").strip().lower() normalized_city = str(city or "").strip().lower() if not normalized_source or not normalized_city: return False use_fahrenheit = bool(self.weather._uses_fahrenheit(normalized_city)) result = collect_observation_source( self.weather, normalized_source, normalized_city, use_fahrenheit=use_fahrenheit, ) if result.status == "unsupported": logger.debug("observation collector skipped unknown source={}", normalized_source) return False if result.status != "ok": self._store_raw_observation_status( source=result.source or normalized_source, city=result.city or normalized_city, status=result.status, error=result.error, ) return False wrote = self._store_raw_observations(result.records) if wrote <= 0: self._store_raw_observation_status( source=result.source or normalized_source, city=result.city or normalized_city, status="parse_error", error="source response had no usable temperature", ) return False return True @staticmethod def _failure_status_from_exception(exc: Exception) -> str: text = f"{exc.__class__.__name__} {exc}".lower() if "timeout" in text or "timed out" in text: return "timeout" if "401" in text or "403" in text or "auth" in text or "unauthor" in text: return "auth_error" return "error" def _store_canonical_temperature_from_observation( self, *, record: ObservationRecord, fetched_at: str, ) -> None: setter = getattr(self.observation_store, "set_canonical_temperature", None) if not callable(setter): return freshness = build_observation_freshness( source_code=record.source, source_label=record.source_label, observed_at=record.observed_at or None, observed_at_local=record.observed_at_local or None, ingested_at=fetched_at, now_utc=parse_utc_datetime(fetched_at), ) payload = { "name": record.city, "temp_symbol": "°F" if record.value_unit.startswith("f") else "°C", "updated_at": fetched_at, "current": { "temp": record.value, "source_code": record.source, "source_label": record.source_label, "settlement_source": record.source, "settlement_source_label": record.source_label, "station_code": record.station_code, "station_name": record.station_name, "observed_at": record.observed_at or None, "observed_at_local": record.observed_at_local or None, "obs_time": record.observed_at_local or record.observed_at, "freshness": freshness, "observation_status": "live", }, } canonical = build_canonical_temperature(record.city, payload, fetched_at=fetched_at) if not canonical: return try: setter(record.city, canonical) self._append_realtime_event(canonical) except Exception as exc: logger.debug( "canonical temperature write skipped source={} city={}: {}", record.source, record.city, exc, ) def _append_realtime_event(self, canonical: dict[str, Any]) -> None: store = self.realtime_event_store appender = getattr(store, "append_event", None) if not callable(appender): return event = build_realtime_event_from_canonical(canonical) if not event: return try: stored_event = appender(event) except Exception as exc: logger.debug("realtime event append skipped city={}: {}", canonical.get("city"), exc) return broadcaster = self.realtime_broadcaster if callable(broadcaster) and not bool(getattr(store, "uses_external_live_fanout", False)): try: broadcaster(stored_event) except Exception as exc: logger.debug("realtime event broadcast skipped city={}: {}", canonical.get("city"), exc) def _store_raw_observation_status( self, *, source: str, city: str, status: str, error: str = "", ) -> None: store = self.observation_store writer = getattr(store, "append_raw_observation", None) if not callable(writer): return fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) try: writer( source=source, city=city, fetched_at=fetched_at, status=status, payload={ "source": source, "city": city, "status": status, "error": str(error or "").strip(), }, ) except Exception as exc: logger.debug( "raw observation status write skipped source={} city={}: {}", source, city, exc, ) @staticmethod def _raw_payload_for_record(record: ObservationRecord) -> dict[str, Any]: payload = dict(record.payload) source = str(record.source or payload.get("source") or "").strip().lower() if source == "amsc_awos": if record.observed_at: payload["observation_time"] = record.observed_at if record.observed_at_local: payload["observation_time_local"] = record.observed_at_local return payload @staticmethod def _float_or_none(value: Any) -> Optional[float]: try: if value is None or value == "": return None return float(value) except (TypeError, ValueError): return None @staticmethod def _int_or_none(value: Any) -> Optional[int]: try: if value is None or value == "": return None return int(float(value)) except (TypeError, ValueError): return None def _store_amsc_runway_observations( self, record: ObservationRecord, payload: dict[str, Any], ) -> None: source = str(record.source or payload.get("source") or "").strip().lower() if source != "amsc_awos": return appender = getattr(self.observation_store, "append_runway_obs", None) if not callable(appender): return runway_obs = payload.get("runway_obs") if not isinstance(runway_obs, dict): return point_temperatures = runway_obs.get("point_temperatures") if not isinstance(point_temperatures, list) or not point_temperatures: return icao = str(payload.get("icao") or record.station_code or "").strip().upper() obs_time = str( payload.get("observation_time") or payload.get("observed_at") or record.observed_at or "" ).strip() if not icao or not obs_time: return for point in point_temperatures: if not isinstance(point, dict): continue runway = str(point.get("runway") or "").strip().upper() if not runway: continue try: appender( icao=icao, city=record.city, runway=runway, tdz_temp=self._float_or_none(point.get("tdz_temp")), mid_temp=self._float_or_none(point.get("mid_temp")), end_temp=self._float_or_none(point.get("end_temp")), target_runway_max=self._float_or_none(point.get("target_runway_max")), wind_dir=self._int_or_none(point.get("wind_dir")), wind_speed=self._float_or_none(point.get("wind_speed")), rvr=self._int_or_none(point.get("rvr")), mor=self._float_or_none(point.get("mor")), humidity=self._float_or_none(point.get("humidity")), otime_utc=obs_time, ) except Exception as exc: logger.debug( "AMSC runway observation write skipped city={} runway={}: {}", record.city, runway, exc, ) def _store_raw_observations(self, records: Sequence[ObservationRecord]) -> int: store = self.observation_store writer = getattr(store, "append_raw_observation", None) if not callable(writer): return len(records) fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) wrote = 0 written_records: list[ObservationRecord] = [] for record in records: try: payload = self._raw_payload_for_record(record) writer( source=record.source, city=record.city, value=record.value, observed_at=record.observed_at, fetched_at=fetched_at, station_code=record.station_code, station_name=record.station_name, runway=record.runway, value_unit=record.value_unit, status="ok", payload=payload, ) self._store_amsc_runway_observations(record, payload) written_records.append(record) wrote += 1 except Exception as exc: logger.debug( "raw observation store write skipped source={} city={}: {}", record.source, record.city, exc, ) if wrote: refreshed_cities: set[str] = set() for city in {record.city for record in written_records}: canonical = refresh_canonical_temperature_from_latest(self.observation_store, city) if canonical: refreshed_cities.add(city) self._append_realtime_event(canonical) for record in written_records: if record.city not in refreshed_cities: self._store_canonical_temperature_from_observation( record=record, fetched_at=fetched_at, ) logger.debug("raw observations stored count={}", wrote) return wrote def _refresh_city_cache(self, city: str) -> None: if not callable(self.cache_refresher): return normalized_city = str(city or "").strip().lower() if not normalized_city: return if self._cache_refresh_executor is None: self._refresh_city_cache_inline(normalized_city) return with self._cache_refresh_lock: if normalized_city in self._cache_refresh_inflight: return self._cache_refresh_inflight.add(normalized_city) try: self._cache_refresh_executor.submit(self._refresh_city_cache_task, normalized_city) except Exception as exc: with self._cache_refresh_lock: self._cache_refresh_inflight.discard(normalized_city) logger.warning( "observation collector cache refresh queue failed city={}: {}", normalized_city, exc, ) def _refresh_city_cache_task(self, city: str) -> None: try: self._refresh_city_cache_inline(city) finally: with self._cache_refresh_lock: self._cache_refresh_inflight.discard(city) def _refresh_city_cache_inline(self, city: str) -> None: try: self.cache_refresher(city) except Exception as exc: logger.warning("observation collector cache refresh failed city={}: {}", city, exc) def close(self) -> None: if self._cache_refresh_executor is not None: self._cache_refresh_executor.shutdown(wait=False) def build_observation_source_profiles() -> List[ObservationSourceProfile]: us_madis_cities = [ city for city, meta in CITY_REGISTRY.items() if str((meta or {}).get("icao") or "").strip().upper().startswith("K") ] turkish_mgm_cities = [ city for city in ("ankara", "istanbul") if city in CITY_REGISTRY ] metar_cities = [ city for city, meta in CITY_REGISTRY.items() if str((meta or {}).get("icao") or "").strip() ] return [ ObservationSourceProfile( source="amos", cities=_normalized_cities(AMOS_AIRPORT_CODES.keys()), interval_sec=max(30, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC", 60)), ), ObservationSourceProfile( source="amsc_awos", cities=_normalized_cities(AMSC_AWOS_AIRPORTS.keys()), interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC", 180)), ), ObservationSourceProfile( source="madis_hfmetar", cities=_normalized_cities(us_madis_cities), interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC", 300)), ), ObservationSourceProfile( source="cowin_obs", cities=("hong kong",), interval_sec=max(30, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC", 60)), ), ObservationSourceProfile( source="hko_obs", cities=_normalized_cities(HKO_STATIONS.keys()), interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC", 600)), ), ObservationSourceProfile( source="mgm", cities=_normalized_cities(turkish_mgm_cities), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_MGM_SEC", 300)), ), ObservationSourceProfile( source="jma_amedas", cities=("tokyo",), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_JMA_SEC", 600)), ), ObservationSourceProfile( source="singapore_mss", cities=("singapore",), interval_sec=max(30, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_MSS_SEC", 60)), ), ObservationSourceProfile( source="fmi", cities=("helsinki",), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_FMI_SEC", 600)), ), ObservationSourceProfile( source="knmi", cities=("amsterdam",), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_KNMI_SEC", 600)), ), ObservationSourceProfile( source="ims", cities=("tel aviv",), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_IMS_SEC", 600)), ), ObservationSourceProfile( source="aeroweb", cities=("paris",), interval_sec=max(600, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_AEROWEB_SEC", 900)), ), ObservationSourceProfile( source="cwa", cities=("taipei",), interval_sec=max(300, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CWA_SEC", 600)), ), ObservationSourceProfile( source="metar", cities=_normalized_cities(metar_cities), interval_sec=max(600, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_METAR_SEC", 1800)), ), ] _COLLECTOR_THREAD: Optional[threading.Thread] = None _COLLECTOR_LOCK = threading.Lock() def start_observation_collector_loop( *, weather: Any, cache_refresher: Optional[Callable[[str], Any]] = None, profiles: Optional[Sequence[ObservationSourceProfile]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, observation_store: Optional[Any] = None, realtime_event_store: Optional[Any] = None, realtime_broadcaster: Optional[Callable[[dict[str, Any]], Any]] = None, ) -> Optional[threading.Thread]: if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True): return None tick_sec = max(5, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC", 30)) initial_delay_sec = max( 0, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC", 5), ) selected_profiles = list(profiles or build_observation_source_profiles()) collector = ObservationCollector( weather=weather, profiles=selected_profiles, cache_refresher=cache_refresher, status_recorder=status_recorder or ObservationCollectorStatusRepository(), observation_store=observation_store or DBManager(), realtime_event_store=realtime_event_store or create_realtime_event_store(), realtime_broadcaster=realtime_broadcaster, ) global _COLLECTOR_THREAD with _COLLECTOR_LOCK: if _COLLECTOR_THREAD is not None and _COLLECTOR_THREAD.is_alive(): return _COLLECTOR_THREAD def _runner() -> None: logger.info( "observation collector started profiles={} tick_sec={}", len(selected_profiles), tick_sec, ) if initial_delay_sec: time.sleep(initial_delay_sec) while True: started = time.time() collector.run_due_once(now_ts=started) elapsed = time.time() - started time.sleep(max(1.0, tick_sec - elapsed)) _COLLECTOR_THREAD = threading.Thread( target=_runner, name="observation-collector", daemon=True, ) _COLLECTOR_THREAD.start() return _COLLECTOR_THREAD