diff --git a/web/analysis_service.py b/web/analysis_service.py index a3b48d9f..9999c954 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -35,6 +35,10 @@ from web.services.city_payloads import ( build_city_market_scan_payload as _city_payload_market_scan, build_city_summary_payload as _city_payload_summary, ) +from web.services.observation_freshness import ( + build_observation_freshness as _build_observation_freshness, + observation_age_min as _observation_age_min, +) from web.services.analysis_signals import ( _build_deviation_monitor, _build_taf_signal, @@ -189,183 +193,6 @@ def _metar_is_current_local_day( return local_dt.strftime("%Y-%m-%d") == local_date -_OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = { - "amos": { - "label": "AMOS", - "native_update_interval_sec": 60, - "fresh_window_sec": 180, - "expected_grace_sec": 180, - "stale_after_sec": 900, - }, - "amsc_awos": { - "label": "AMSC AWOS", - "native_update_interval_sec": 60, - "fresh_window_sec": 180, - "expected_grace_sec": 180, - "stale_after_sec": 900, - }, - "jma": { - "label": "JMA", - "native_update_interval_sec": 600, - "fresh_window_sec": 900, - "expected_grace_sec": 600, - "stale_after_sec": 2700, - }, - "fmi": { - "label": "FMI", - "native_update_interval_sec": 600, - "fresh_window_sec": 900, - "expected_grace_sec": 600, - "stale_after_sec": 2700, - }, - "knmi": { - "label": "KNMI", - "native_update_interval_sec": 600, - "fresh_window_sec": 900, - "expected_grace_sec": 600, - "stale_after_sec": 2700, - }, - "hko": { - "label": "HKO", - "native_update_interval_sec": 600, - "fresh_window_sec": 900, - "expected_grace_sec": 600, - "stale_after_sec": 2700, - }, - "cwa": { - "label": "CWA", - "native_update_interval_sec": 600, - "fresh_window_sec": 900, - "expected_grace_sec": 600, - "stale_after_sec": 2700, - }, - "mgm": { - "label": "MGM", - "native_update_interval_sec": 900, - "fresh_window_sec": 900, - "expected_grace_sec": 900, - "stale_after_sec": 3600, - }, - "metar": { - "label": "METAR", - "native_update_interval_sec": 900, - "fresh_window_sec": 600, - "expected_grace_sec": 900, - "stale_after_sec": 3600, - }, - "noaa": { - "label": "NOAA", - "native_update_interval_sec": 900, - "fresh_window_sec": 600, - "expected_grace_sec": 900, - "stale_after_sec": 3600, - }, - "wunderground": { - "label": "METAR", - "native_update_interval_sec": 900, - "fresh_window_sec": 600, - "expected_grace_sec": 900, - "stale_after_sec": 3600, - }, -} - - -def _canonical_observation_source_code(value: Any) -> str: - raw = str(value or "").strip().lower() - if not raw: - return "metar" - if "amos" in raw: - return "amos" - if "jma" in raw: - return "jma" - if "fmi" in raw: - return "fmi" - if "knmi" in raw: - return "knmi" - if "hko" in raw: - return "hko" - if "cwa" in raw: - return "cwa" - if "mgm" in raw: - return "mgm" - if "noaa" in raw: - return "noaa" - if "wunderground" in raw or raw == "wu": - return "wunderground" - return raw - - -def _observation_age_min(value: Any, now_utc: Optional[datetime] = None) -> Optional[int]: - obs_dt = _parse_utc_datetime(value) - if obs_dt is None: - return None - now = now_utc or datetime.now(timezone.utc) - return max(0, int((now - obs_dt).total_seconds() / 60)) - - -def _optional_str(value: Any) -> Optional[str]: - raw = str(value or "").strip() - return raw or None - - -def _build_observation_freshness( - *, - source_code: Any, - source_label: Any = None, - observed_at: Any = None, - observed_at_local: Any = None, - ingested_at: Any = None, - age_min: Optional[int] = None, - now_utc: Optional[datetime] = None, -) -> Dict[str, Any]: - code = _canonical_observation_source_code(source_code or source_label) - profile = _OBSERVATION_SOURCE_PROFILES.get(code) or _OBSERVATION_SOURCE_PROFILES["metar"] - now = now_utc or datetime.now(timezone.utc) - obs_dt = _parse_utc_datetime(observed_at) - age_sec = None - if age_min is not None: - try: - age_sec = max(0, int(age_min) * 60) - except Exception: - age_sec = None - if age_sec is None and obs_dt is not None: - age_sec = max(0, int((now - obs_dt).total_seconds())) - - if age_sec is None: - status = "unknown" - reason = "observation_time_missing" - elif age_sec <= int(profile["fresh_window_sec"]): - status = "fresh" - reason = "within_native_fresh_window" - elif age_sec <= int(profile["native_update_interval_sec"]) + int(profile["expected_grace_sec"]): - status = "expected_wait" - reason = "within_source_expected_cadence" - elif age_sec <= int(profile["stale_after_sec"]): - status = "delayed" - reason = "past_expected_cadence" - else: - status = "stale" - reason = "past_stale_threshold" - - expected_next = ( - obs_dt + timedelta(seconds=int(profile["native_update_interval_sec"])) - if obs_dt is not None - else None - ) - return { - "source_code": code, - "source_label": str(source_label or profile["label"]), - "observed_at": obs_dt.isoformat() if obs_dt is not None else _optional_str(observed_at), - "observed_at_local": _optional_str(observed_at_local), - "ingested_at": _optional_str(ingested_at), - "native_update_interval_sec": int(profile["native_update_interval_sec"]), - "expected_next_update_at": expected_next.isoformat() if expected_next is not None else None, - "freshness_status": status, - "freshness_reason": reason, - "age_sec": age_sec, - } - - def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) -> None: now = datetime.now(timezone.utc).isoformat() with _ANALYSIS_CACHE_STATS_LOCK: diff --git a/web/services/observation_freshness.py b/web/services/observation_freshness.py new file mode 100644 index 00000000..f3fd0173 --- /dev/null +++ b/web/services/observation_freshness.py @@ -0,0 +1,199 @@ +"""Observation source freshness profiles and helpers. + +Extracted from analysis_service.py to keep the god module leaner. +""" + +from __future__ import annotations + +from datetime import datetime, timezone, timedelta +from typing import Any, Dict, Optional + + +_OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = { + "amos": { + "label": "AMOS", + "native_update_interval_sec": 60, + "fresh_window_sec": 180, + "expected_grace_sec": 180, + "stale_after_sec": 900, + }, + "amsc_awos": { + "label": "AMSC AWOS", + "native_update_interval_sec": 60, + "fresh_window_sec": 180, + "expected_grace_sec": 180, + "stale_after_sec": 900, + }, + "jma": { + "label": "JMA", + "native_update_interval_sec": 600, + "fresh_window_sec": 900, + "expected_grace_sec": 600, + "stale_after_sec": 2700, + }, + "fmi": { + "label": "FMI", + "native_update_interval_sec": 600, + "fresh_window_sec": 900, + "expected_grace_sec": 600, + "stale_after_sec": 2700, + }, + "knmi": { + "label": "KNMI", + "native_update_interval_sec": 600, + "fresh_window_sec": 900, + "expected_grace_sec": 600, + "stale_after_sec": 2700, + }, + "hko": { + "label": "HKO", + "native_update_interval_sec": 600, + "fresh_window_sec": 900, + "expected_grace_sec": 600, + "stale_after_sec": 2700, + }, + "cwa": { + "label": "CWA", + "native_update_interval_sec": 600, + "fresh_window_sec": 900, + "expected_grace_sec": 600, + "stale_after_sec": 2700, + }, + "mgm": { + "label": "MGM", + "native_update_interval_sec": 900, + "fresh_window_sec": 900, + "expected_grace_sec": 900, + "stale_after_sec": 3600, + }, + "metar": { + "label": "METAR", + "native_update_interval_sec": 900, + "fresh_window_sec": 600, + "expected_grace_sec": 900, + "stale_after_sec": 3600, + }, + "noaa": { + "label": "NOAA", + "native_update_interval_sec": 900, + "fresh_window_sec": 600, + "expected_grace_sec": 900, + "stale_after_sec": 3600, + }, + "wunderground": { + "label": "METAR", + "native_update_interval_sec": 900, + "fresh_window_sec": 600, + "expected_grace_sec": 900, + "stale_after_sec": 3600, + }, +} + + +def parse_utc_datetime(value: Any) -> Optional[datetime]: + raw = str(value or "").strip() + if not raw or "T" not in raw: + return None + try: + dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except Exception: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + + +def observation_age_min(value: Any, now_utc: Optional[datetime] = None) -> Optional[int]: + obs_dt = parse_utc_datetime(value) + if obs_dt is None: + return None + now = now_utc or datetime.now(timezone.utc) + return max(0, int((now - obs_dt).total_seconds() / 60)) + + +def canonical_observation_source_code(value: Any) -> str: + raw = str(value or "").strip().lower() + if not raw: + return "metar" + if "amos" in raw: + return "amos" + if "jma" in raw: + return "jma" + if "fmi" in raw: + return "fmi" + if "knmi" in raw: + return "knmi" + if "hko" in raw: + return "hko" + if "cwa" in raw: + return "cwa" + if "mgm" in raw: + return "mgm" + if "noaa" in raw: + return "noaa" + if "wunderground" in raw or raw == "wu": + return "wunderground" + return raw + + +def optional_str(value: Any) -> Optional[str]: + raw = str(value or "").strip() + return raw or None + + +def build_observation_freshness( + *, + source_code: Any, + source_label: Any = None, + observed_at: Any = None, + observed_at_local: Any = None, + ingested_at: Any = None, + age_min: Optional[int] = None, + now_utc: Optional[datetime] = None, +) -> Dict[str, Any]: + code = canonical_observation_source_code(source_code or source_label) + profile = _OBSERVATION_SOURCE_PROFILES.get(code) or _OBSERVATION_SOURCE_PROFILES["metar"] + now = now_utc or datetime.now(timezone.utc) + obs_dt = parse_utc_datetime(observed_at) + age_sec = None + if age_min is not None: + try: + age_sec = max(0, int(age_min) * 60) + except Exception: + age_sec = None + if age_sec is None and obs_dt is not None: + age_sec = max(0, int((now - obs_dt).total_seconds())) + + if age_sec is None: + status = "unknown" + reason = "observation_time_missing" + elif age_sec <= int(profile["fresh_window_sec"]): + status = "fresh" + reason = "within_native_fresh_window" + elif age_sec <= int(profile["native_update_interval_sec"]) + int(profile["expected_grace_sec"]): + status = "expected_wait" + reason = "within_source_expected_cadence" + elif age_sec <= int(profile["stale_after_sec"]): + status = "delayed" + reason = "past_expected_cadence" + else: + status = "stale" + reason = "past_stale_threshold" + + expected_next = ( + obs_dt + timedelta(seconds=int(profile["native_update_interval_sec"])) + if obs_dt is not None + else None + ) + return { + "source_code": code, + "source_label": str(source_label or profile["label"]), + "observed_at": obs_dt.isoformat() if obs_dt is not None else optional_str(observed_at), + "observed_at_local": optional_str(observed_at_local), + "ingested_at": optional_str(ingested_at), + "native_update_interval_sec": int(profile["native_update_interval_sec"]), + "expected_next_update_at": expected_next.isoformat() if expected_next is not None else None, + "freshness_status": status, + "freshness_reason": reason, + "age_sec": age_sec, + }