PolyWeather/web/observation_collector_service.py

"""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