diff --git a/tests/test_canonical_engine.py b/tests/test_canonical_engine.py index 8efa16f1..defa5c41 100644 --- a/tests/test_canonical_engine.py +++ b/tests/test_canonical_engine.py @@ -83,6 +83,47 @@ def test_canonical_engine_ignores_failed_latest_rows(): assert canonical["value"] == 24.0 +def test_canonical_engine_prefers_mgm_over_metar_for_turkish_city(): + from web.services.canonical_engine import build_canonical_temperature_from_observations + + canonical = build_canonical_temperature_from_observations( + "ankara", + [ + { + "source": "metar", + "city": "ankara", + "station_code": "LTAC", + "station_name": "Ankara Esenboga", + "value": 14.0, + "value_unit": "c", + "observed_at": "2026-06-15T14:20:00+00:00", + "fetched_at": "2026-06-15T14:21:00+00:00", + "status": "ok", + "payload": {"source_label": "METAR"}, + "updated_at_ts": 10.0, + }, + { + "source": "mgm", + "city": "ankara", + "station_code": "17128", + "station_name": "Esenboga Airport", + "value": 19.0, + "value_unit": "c", + "observed_at": "2026-06-15T14:20:00+00:00", + "fetched_at": "2026-06-15T14:21:00+00:00", + "status": "ok", + "payload": {"source_label": "MGM"}, + "updated_at_ts": 9.0, + }, + ], + ) + + assert canonical is not None + assert canonical["source"] == "mgm" + assert canonical["source_label"] == "MGM" + assert canonical["value"] == 19.0 + + def test_canonical_engine_builds_realtime_event_from_canonical(): from web.services.canonical_engine import build_realtime_event_from_canonical diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 3c0aff5d..a32a377c 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -87,12 +87,15 @@ def test_observation_collector_profiles_match_source_cadence(): assert profiles["madis_hfmetar"].interval_sec == 300 assert profiles["cowin_obs"].interval_sec == 60 assert profiles["hko_obs"].interval_sec == 600 + assert profiles["mgm"].interval_sec == 300 assert "qingdao" in profiles["amsc_awos"].cities assert {"seoul", "busan"}.issubset(set(profiles["amos"].cities)) assert "new york" in profiles["madis_hfmetar"].cities assert "hong kong" in profiles["cowin_obs"].cities assert {"hong kong", "shenzhen"}.issubset(set(profiles["hko_obs"].cities)) + assert {"ankara", "istanbul"}.issubset(set(profiles["mgm"].cities)) assert SOURCE_CADENCE_SECONDS["amsc_awos"] == 180 + assert SOURCE_CADENCE_SECONDS["mgm"] == 300 def test_observation_collector_run_due_once_collects_without_panel_cache_refresh(): diff --git a/tests/test_observation_source_adapters.py b/tests/test_observation_source_adapters.py index 494a672a..f53676a2 100644 --- a/tests/test_observation_source_adapters.py +++ b/tests/test_observation_source_adapters.py @@ -81,6 +81,52 @@ def test_source_adapter_flattens_nearby_source_lists(): assert [record.value for record in result.records] == [28.1, 27.6] +def test_source_adapter_collects_mgm_with_keyword_flags_and_station_metadata(): + from web.services.observation_source_adapters import collect_observation_source + + calls = [] + + class FakeWeather: + def _attach_turkish_mgm_data( + self, + results, + city, + *, + include_mgm=True, + include_nearby=True, + ): + calls.append((city, include_mgm, include_nearby)) + if not include_mgm: + return + results["mgm"] = { + "source": "mgm", + "source_label": "MGM", + "station_code": "17128", + "station_name": "Esenboga Airport", + "obs_time": "2026-06-15T14:20:00Z", + "current": {"temp": 19.0}, + } + + result = collect_observation_source( + FakeWeather(), + "mgm", + "ankara", + use_fahrenheit=False, + ) + + assert calls == [("ankara", True, True)] + assert result.status == "ok" + assert len(result.records) == 1 + + record = result.records[0] + assert record.source == "mgm" + assert record.source_label == "MGM" + assert record.value == 19.0 + assert record.observed_at == "2026-06-15T14:20:00Z" + assert record.station_code == "17128" + assert record.station_name == "Esenboga Airport" + + def test_source_adapter_reports_parse_error_for_unusable_source_rows(): from web.services.observation_source_adapters import collect_observation_source diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 3ad46b85..09cb5c68 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1829,6 +1829,77 @@ def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch assert refresh_calls == 0 +def test_stale_ankara_chart_data_overlays_latest_mgm_canonical(monkeypatch): + import asyncio + + class FakeCache: + def get_city_cache(self, kind, city): + assert kind == "full" + assert city == "ankara" + return { + "payload": { + "name": "ankara", + "local_date": "2026-06-14", + "local_time": "13:12", + "current": { + "temp": 16.0, + "source_code": "metar", + "settlement_source": "metar", + "settlement_source_label": "METAR", + "observed_at": "2026-06-14T09:50:00+00:00", + }, + "airport_primary": { + "temp": 16.0, + "source_code": "metar", + "source_label": "METAR", + "obs_time": "2026-06-14T09:50:00+00:00", + }, + "hourly": {"times": ["2026-06-14T09:00:00Z"], "temps": [16.0]}, + "deb": {"prediction": 23.0}, + }, + } + + def get_canonical_temperature(self, city): + assert city == "ankara" + return { + "payload": { + "city": "ankara", + "value": 19.0, + "temp_symbol": "°C", + "source": "mgm", + "source_label": "MGM", + "source_role": "settlement_official", + "station_code": "17128", + "station_name": "Esenboga Airport", + "observed_at": "2026-06-15T14:20:00+00:00", + "observed_at_local": "17:20", + "freshness_sec": 60, + "freshness_status": "fresh", + "fetched_at": "2026-06-15T14:21:00+00:00", + "confidence": 0.9, + "deb_prediction": 23.0, + } + } + + def enqueue_observation_refresh_request(self, **_kwargs): + return True + + monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache()) + monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: False) + monkeypatch.setattr(city_api, "_start_city_full_stale_refresh", lambda city: None) + monkeypatch.setattr(city_api.legacy_routes, "_overlay_latest_wunderground_current", lambda city, payload: payload) + + payload = asyncio.run(city_api._get_city_chart_data("ankara", force_refresh=False)) + + assert payload["current"]["temp"] == 19.0 + assert payload["current"]["source_code"] == "mgm" + assert payload["current"]["settlement_source_label"] == "MGM" + assert payload["airport_primary"]["temp"] == 19.0 + assert payload["airport_primary"]["source_code"] == "mgm" + assert payload["airport_primary"]["station_code"] == "17128" + assert payload["deb"]["prediction"] == 23.0 + + def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(monkeypatch): import asyncio diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 587c08ef..36095b0f 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -586,6 +586,11 @@ def build_observation_source_profiles() -> List[ObservationSourceProfile]: 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 + ] return [ ObservationSourceProfile( source="amos", @@ -612,6 +617,11 @@ def build_observation_source_profiles() -> List[ObservationSourceProfile]: 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)), + ), ] diff --git a/web/services/canonical_engine.py b/web/services/canonical_engine.py index 4b36b1e8..7598fe24 100644 --- a/web/services/canonical_engine.py +++ b/web/services/canonical_engine.py @@ -14,6 +14,7 @@ from web.realtime_patch_schema import normalize_observation_patch _SETTLEMENT_SOURCE_ADAPTERS = { "hko": {"hko_obs", "cowin_obs"}, "cwa": {"cwa"}, + "mgm": {"mgm"}, "noaa": {"madis_hfmetar", "metar", "noaa"}, "wunderground": {"amsc_awos", "amos", "madis_hfmetar", "metar", "wunderground"}, } @@ -23,6 +24,7 @@ _SOURCE_WEIGHTS = { "amos": 700, "hko_obs": 680, "cowin_obs": 660, + "mgm": 640, "madis_hfmetar": 500, "metar": 460, } @@ -35,6 +37,11 @@ _FRESHNESS_WEIGHTS = { "stale": 5, } +_TURKISH_MGM_STATION_CODES = { + "ankara": "17128", + "istanbul": "17058", +} + def _normalized_city(city: Any) -> str: return str(city or "").strip().lower() @@ -116,8 +123,13 @@ def _candidate_canonical(city: str, row: dict[str, Any]) -> Optional[dict[str, A def _score(city: str, row: dict[str, Any], canonical: dict[str, Any]) -> tuple[int, float]: meta = CITY_REGISTRY.get(city) or {} station_code = _station_code(row.get("station_code")) - settlement_station_code = _station_code(meta.get("settlement_station_code") or meta.get("icao")) - settlement_source = _normalized_source(meta.get("settlement_source")) + is_turkish_mgm_city = city in _TURKISH_MGM_STATION_CODES + settlement_station_code = _station_code( + _TURKISH_MGM_STATION_CODES.get(city) + if is_turkish_mgm_city + else (meta.get("settlement_station_code") or meta.get("icao")) + ) + settlement_source = "mgm" if is_turkish_mgm_city else _normalized_source(meta.get("settlement_source")) expected_sources = _SETTLEMENT_SOURCE_ADAPTERS.get(settlement_source, {settlement_source}) station_name = str(row.get("station_name") or "").strip().lower() candidates = { diff --git a/web/services/canonical_temperature.py b/web/services/canonical_temperature.py index 03dc4473..d6e518f8 100644 --- a/web/services/canonical_temperature.py +++ b/web/services/canonical_temperature.py @@ -218,6 +218,8 @@ def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> O "max_so_far": _first_float(canonical.get("max_so_far"), value), "source_code": source, "source_label": source_label, + "station_code": canonical.get("station_code"), + "station_name": canonical.get("station_name"), "obs_time": observed_at or observed_at_local, "freshness": freshness, } diff --git a/web/services/city_api.py b/web/services/city_api.py index 47ba71d5..a104ce34 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -16,7 +16,10 @@ from loguru import logger import web.routes as legacy_routes from web.analysis_service import _runway_history_temp_for_city from web.services.canonical_temperature import build_city_weather_from_canonical -from web.services.latest_observation_overlay import overlay_latest_amsc_observation +from web.services.latest_observation_overlay import ( + overlay_latest_amsc_observation, + parse_observation_epoch, +) from web.services.request_timing import ServerTimingRecorder _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None @@ -297,11 +300,12 @@ def _start_city_cache_stale_refresh( async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: + latest_payload = await _overlay_cached_canonical_observation(city, payload) latest_payload = await run_in_threadpool( overlay_latest_amsc_observation, legacy_routes._CACHE_DB, city, - payload, + latest_payload, ) return await run_in_threadpool( legacy_routes._overlay_latest_wunderground_current, @@ -310,6 +314,70 @@ async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Di ) +def _observation_block_epoch(block: Any) -> Optional[int]: + if not isinstance(block, dict): + return None + freshness = block.get("freshness") if isinstance(block.get("freshness"), dict) else {} + values = ( + block.get("observed_at"), + block.get("observation_time"), + block.get("obs_time"), + freshness.get("observed_at"), + ) + epochs = [epoch for epoch in (parse_observation_epoch(value) for value in values) if epoch is not None] + return max(epochs) if epochs else None + + +def _payload_observation_epoch(payload: Dict[str, Any]) -> Optional[int]: + if not isinstance(payload, dict): + return None + blocks = ( + payload.get("current"), + payload.get("airport_primary"), + payload.get("airport_current"), + ) + epochs = [epoch for epoch in (_observation_block_epoch(block) for block in blocks) if epoch is not None] + return max(epochs) if epochs else None + + +def _merge_latest_observation_payload( + payload: Dict[str, Any], + latest_payload: Dict[str, Any], +) -> Dict[str, Any]: + if not isinstance(payload, dict) or not isinstance(latest_payload, dict): + return payload + latest_epoch = _payload_observation_epoch(latest_payload) + if latest_epoch is None: + return payload + current_epoch = _payload_observation_epoch(payload) + if current_epoch is not None and current_epoch >= latest_epoch: + return payload + + next_payload = deepcopy(payload) + for key in ("current", "airport_primary", "airport_current"): + latest_block = latest_payload.get(key) + if not isinstance(latest_block, dict) or not latest_block: + continue + base_block = next_payload.get(key) if isinstance(next_payload.get(key), dict) else {} + next_payload[key] = {**base_block, **latest_block} + if isinstance(latest_payload.get("canonical_temperature"), dict): + next_payload["canonical_temperature"] = latest_payload["canonical_temperature"] + if latest_payload.get("updated_at"): + next_payload["updated_at"] = latest_payload.get("updated_at") + if latest_payload.get("temp_symbol"): + next_payload["temp_symbol"] = latest_payload.get("temp_symbol") + return next_payload + + +async def _overlay_cached_canonical_observation(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: + if not isinstance(payload, dict) or not payload: + return payload + canonical_payload = await _get_canonical_city_payload(city, detail_depth=str(payload.get("detail_depth") or "full")) + if not canonical_payload: + return payload + return _merge_latest_observation_payload(payload, canonical_payload) + + def _overlay_cached_runway_history_from_db(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: if not isinstance(payload, dict) or not payload: return payload @@ -405,6 +473,7 @@ async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, An async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, Any]: if force_refresh: payload = await _get_city_full_data(city, force_refresh=True) + payload = await _overlay_cached_canonical_observation(city, payload) payload = await _run_optional_city_chart_overlay( city=city, overlay_name="runway_history", @@ -433,6 +502,7 @@ async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, A if payload: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): _start_city_full_stale_refresh(city) + payload = await _overlay_cached_canonical_observation(city, payload) payload = await _run_optional_city_chart_overlay( city=city, overlay_name="runway_history", diff --git a/web/services/observation_source_adapters.py b/web/services/observation_source_adapters.py index 73baa513..d4d7472c 100644 --- a/web/services/observation_source_adapters.py +++ b/web/services/observation_source_adapters.py @@ -36,6 +36,12 @@ _ATTACH_METHODS: dict[str, str] = { "madis_hfmetar": "_attach_madis_hfmetar_data", "hko_obs": "_attach_hko_obs_official_nearby", "cowin_obs": "_attach_cowin_official_nearby", + "mgm": "_attach_turkish_mgm_data", +} + +_TURKISH_MGM_STATION_CODES = { + "ankara": "17128", + "istanbul": "17058", } @@ -115,6 +121,31 @@ def _record_from_row( ) +def _enrich_mgm_results(results: dict[str, Any], city: str) -> None: + mgm = results.get("mgm") + if not isinstance(mgm, dict): + return + current = mgm.get("current") if isinstance(mgm.get("current"), dict) else {} + station_code = str( + mgm.get("station_code") + or mgm.get("istNo") + or _TURKISH_MGM_STATION_CODES.get(city) + or "" + ).strip() + station_name = str( + mgm.get("station_name") + or current.get("station_name") + or current.get("station_label") + or "" + ).strip() + mgm.setdefault("source", "mgm") + mgm.setdefault("source_label", "MGM") + if station_code: + mgm.setdefault("station_code", station_code) + if station_name: + mgm.setdefault("station_name", station_name) + + def collect_observation_source( weather: Any, source: Any, @@ -144,7 +175,16 @@ def collect_observation_source( ) results: dict[str, Any] = {} - attach(results, normalized_city, bool(use_fahrenheit)) + if normalized_source == "mgm": + attach( + results, + normalized_city, + include_mgm=True, + include_nearby=True, + ) + _enrich_mgm_results(results, normalized_city) + else: + attach(results, normalized_city, bool(use_fahrenheit)) if not results: return ObservationSourceResult( source=normalized_source,