From 8bf7cdd6870a4995243fe7f09bef11316e9fd114 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 15 Jun 2026 00:29:43 +0800 Subject: [PATCH] Overlay latest AMSC observations in city and Telegram payloads --- src/utils/telegram_push.py | 24 +-- tests/test_canonical_temperature.py | 113 +++++++++++ tests/test_telegram_hashtags.py | 86 +++++++++ tests/test_web_observability.py | 2 +- web/services/city_api.py | 24 ++- web/services/city_runtime.py | 8 +- web/services/latest_observation_overlay.py | 206 +++++++++++++++++++++ 7 files changed, 441 insertions(+), 22 deletions(-) create mode 100644 web/services/latest_observation_overlay.py diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index b7ffe09c..d7814c96 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -28,6 +28,7 @@ from src.utils.telegram_i18n import ( telegram_push_language as _resolve_telegram_push_language, ) from web.services.canonical_temperature import build_city_weather_from_canonical +from web.services.latest_observation_overlay import overlay_latest_amsc_observation # Forum topic routing: maps city_key -> message_thread_id for the push forum group. # Created by scripts/create_forum_topics.py, stored in the runtime data dir. @@ -1582,23 +1583,11 @@ def _attach_latest_raw_observation_payload( city: str, payload: Dict[str, Any], ) -> Dict[str, Any]: - canonical = payload.get("canonical_temperature") or {} - source = str(canonical.get("source") or "").strip().lower() - if source != "amsc_awos": - return payload - getter = getattr(db, "get_latest_raw_observation", None) - if not callable(getter): - return payload try: - row = getter("amsc_awos", city) + return overlay_latest_amsc_observation(db, city, payload) except Exception as exc: - logger.debug("airport push latest raw observation read failed city={}: {}", city, exc) + logger.debug("airport push latest observation overlay failed city={}: {}", city, exc) return payload - raw_payload = row.get("payload") if isinstance(row, dict) else None - if isinstance(raw_payload, dict) and raw_payload: - payload = dict(payload) - payload["amos"] = dict(raw_payload) - return payload def _read_canonical_airport_city_weather(city: str, db: Optional[Any] = None) -> Optional[Dict[str, Any]]: @@ -1661,8 +1650,11 @@ def _merge_airport_push_context( def _load_airport_city_weather_for_push(city: str) -> Dict[str, Any]: - cached = _read_cached_airport_city_weather(city) - canonical = _read_canonical_airport_city_weather(city) + normalized_city = (city or "").strip().lower() + cached = _read_cached_airport_city_weather(normalized_city) + if cached is not None: + cached = _attach_latest_raw_observation_payload(DBManager(), normalized_city, cached) + canonical = _read_canonical_airport_city_weather(normalized_city) if canonical is not None: cached_ts = _cached_payload_observation_epoch(cached or {}) canonical_ts = _cached_payload_observation_epoch(canonical) diff --git a/tests/test_canonical_temperature.py b/tests/test_canonical_temperature.py index 871cec29..5e396b83 100644 --- a/tests/test_canonical_temperature.py +++ b/tests/test_canonical_temperature.py @@ -406,6 +406,119 @@ def test_force_refresh_panel_returns_canonical_latest_without_waiting_for_sync_r assert payload["detail_depth"] == "panel" +def test_force_refresh_panel_overlays_latest_amsc_raw_over_stale_cache(monkeypatch): + import web.services.city_api as city_api + + enqueued = [] + + class FakeDB: + def get_city_cache(self, kind, city): + assert kind == "panel" + assert city == "shanghai" + return { + "updated_at_ts": 1000.0, + "payload": { + "detail_depth": "panel", + "current": { + "temp": 21.8, + "source_code": "metar", + "observed_at": "2026-06-14T10:30:00+00:00", + }, + "airport_primary": { + "temp": 21.8, + "source_code": "metar", + "obs_time": "2026-06-14T10:30:00+00:00", + }, + "amos": { + "source": "amsc_awos", + "temp_c": 22.2, + "observation_time": "2026-06-14T10:44:00+00:00", + }, + "deb": {"prediction": 24.5}, + }, + } + + def get_canonical_temperature(self, city): + assert city == "shanghai" + return { + "payload": { + "city": "shanghai", + "value": 21.8, + "source": "metar", + "source_label": "METAR", + "source_role": "airport_official", + "observed_at": "2026-06-14T10:30:00+00:00", + "observed_at_local": "18:30", + "freshness_sec": 120, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T10:30:05+00:00", + "confidence": 0.78, + }, + } + + def get_latest_raw_observation(self, source, city): + assert (source, city) == ("amsc_awos", "shanghai") + return { + "observed_at": "2026-06-14T15:43:00+00:00", + "fetched_at": "2026-06-14T15:43:30+00:00", + "payload": { + "source": "amsc_awos", + "source_label": "AMSC AWOS Shanghai Pudong (ZSPD)", + "icao": "ZSPD", + "temp_c": 25.4, + "observation_time": "2026-06-14T15:43:00+00:00", + "observation_time_local": "2026-06-14 23:43:00", + "runway_obs": { + "runway_pairs": [("17L", "35R")], + "temperatures": [(25.4, None)], + "point_temperatures": [ + { + "runway": "17L/35R", + "tdz_temp": 25.0, + "mid_temp": 25.2, + "end_temp": 25.4, + "target_runway_max": 25.4, + } + ], + }, + }, + } + + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True + + async def identity_overlay(_city, payload): + return payload + + monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) + monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeDB()) + monkeypatch.setattr(city_api, "_overlay_cached_wunderground", identity_overlay) + + payload = asyncio.run( + city_api.get_city_detail_payload( + object(), + "Shanghai", + force_refresh=True, + depth="panel", + ) + ) + + assert payload["amos"]["observation_time"] == "2026-06-14T15:43:00+00:00" + assert payload["current"]["source_code"] == "amsc_awos" + assert payload["current"]["temp"] == 25.4 + assert payload["canonical_temperature"]["source"] == "amsc_awos" + assert payload["deb"]["prediction"] == 24.5 + assert enqueued == [ + { + "city": "shanghai", + "kind": "panel", + "priority": "high", + "reason": "force_refresh", + } + ] + + def test_city_panel_canonical_fallback_enqueues_collector_refresh_when_queue_exists(monkeypatch): import web.services.city_api as city_api diff --git a/tests/test_telegram_hashtags.py b/tests/test_telegram_hashtags.py index 138f6e30..80ef3989 100644 --- a/tests/test_telegram_hashtags.py +++ b/tests/test_telegram_hashtags.py @@ -413,6 +413,92 @@ def test_airport_push_prefers_newer_amsc_canonical_over_stale_cache(monkeypatch) assert city_weather["amos"]["runway_obs"]["point_temperatures"][0]["runway"] == "16/34" +def test_airport_push_overlays_latest_amsc_raw_when_canonical_is_stale(monkeypatch): + import src.utils.telegram_push as telegram_push + + class FakeDB: + def get_city_cache(self, kind, city): + if kind != "panel": + return None + assert city == "shanghai" + return { + "updated_at_ts": 1000.0, + "payload": { + "current": { + "temp": 21.8, + "source_code": "metar", + "observed_at": "2026-06-14T10:30:00+00:00", + }, + "airport_primary": { + "temp": 21.8, + "source_code": "metar", + "obs_time": "2026-06-14T10:30:00+00:00", + }, + "amos": { + "source": "amsc_awos", + "temp_c": 22.2, + "observation_time": "2026-06-14T10:44:00+00:00", + }, + "deb": {"prediction": 24.5}, + }, + } + + def get_canonical_temperature(self, city): + assert city == "shanghai" + return { + "payload": { + "city": "shanghai", + "value": 21.8, + "source": "metar", + "source_label": "METAR", + "source_role": "airport_official", + "observed_at": "2026-06-14T10:30:00+00:00", + "observed_at_local": "18:30", + "freshness_sec": 120, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T10:30:05+00:00", + "confidence": 0.78, + }, + } + + def get_latest_raw_observation(self, source, city): + assert (source, city) == ("amsc_awos", "shanghai") + return { + "observed_at": "2026-06-14T15:43:00+00:00", + "fetched_at": "2026-06-14T15:43:30+00:00", + "payload": { + "source": "amsc_awos", + "source_label": "AMSC AWOS Shanghai Pudong (ZSPD)", + "icao": "ZSPD", + "temp_c": 25.4, + "observation_time": "2026-06-14T15:43:00+00:00", + "observation_time_local": "2026-06-14 23:43:00", + "runway_obs": { + "runway_pairs": [("17L", "35R")], + "temperatures": [(25.4, None)], + "point_temperatures": [ + { + "runway": "17L/35R", + "tdz_temp": 25.0, + "mid_temp": 25.2, + "end_temp": 25.4, + "target_runway_max": 25.4, + } + ], + }, + }, + } + + monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) + + city_weather = telegram_push._load_airport_city_weather_for_push("shanghai") + + assert city_weather["amos"]["observation_time"] == "2026-06-14T15:43:00+00:00" + assert city_weather["current"]["source_code"] == "amsc_awos" + assert city_weather["current"]["temp"] == 25.4 + assert city_weather["deb"]["prediction"] == 24.5 + + def test_airport_push_without_cache_or_canonical_skips_analysis(monkeypatch): import src.utils.telegram_push as telegram_push import web.app as web_app diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 2ecc0c7d..3869ddd9 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -3782,7 +3782,7 @@ def test_scan_terminal_nonforce_ignores_ancient_success_snapshot(monkeypatch): monkeypatch.setattr( scan_terminal_service.time, "time", - lambda: old_success_t + scan_terminal_service.SCAN_TERMINAL_PAYLOAD_TTL_SEC * 3, + lambda: old_success_t + scan_terminal_service.SCAN_TERMINAL_STALE_SUCCESS_MAX_AGE_SEC + 1, ) monkeypatch.setattr( scan_terminal_service, diff --git a/web/services/city_api.py b/web/services/city_api.py index a754c8f7..a252d6e8 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -16,6 +16,7 @@ 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.request_timing import ServerTimingRecorder _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None @@ -150,7 +151,12 @@ async def _get_canonical_city_payload(city: str, *, detail_depth: str = "panel") "probabilities": {"mu": None, "distribution": []}, } ) - return payload + return await run_in_threadpool( + overlay_latest_amsc_observation, + legacy_routes._CACHE_DB, + city, + payload, + ) def _enqueue_collector_refresh_request( @@ -260,7 +266,13 @@ async def _refresh_city_payload_with_stale_timeout( city, kind, ) - return await _overlay_cached_wunderground(city, cached_before_refresh) + latest_payload = await run_in_threadpool( + overlay_latest_amsc_observation, + legacy_routes._CACHE_DB, + city, + cached_before_refresh, + ) + return await _overlay_cached_wunderground(city, latest_payload) async def _refresh_city_cache_with_stale_timeout( @@ -285,10 +297,16 @@ def _start_city_cache_stale_refresh( async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: + latest_payload = await run_in_threadpool( + overlay_latest_amsc_observation, + legacy_routes._CACHE_DB, + city, + payload, + ) return await run_in_threadpool( legacy_routes._overlay_latest_wunderground_current, city, - payload, + latest_payload, ) diff --git a/web/services/city_runtime.py b/web/services/city_runtime.py index 16d9d2b5..dffbf0fd 100644 --- a/web/services/city_runtime.py +++ b/web/services/city_runtime.py @@ -30,6 +30,7 @@ from web.analysis_service import ( from web.services.canonical_temperature import ( build_city_weather_from_canonical, ) +from web.services.latest_observation_overlay import overlay_latest_amsc_observation from web.scan_terminal_service import build_scan_terminal_payload # noqa: F401 - compatibility export for tests and transitional routers from web.core import ( CITIES, @@ -299,7 +300,10 @@ def _cached_city_payload(kind: str, city: str) -> dict: if not isinstance(entry, dict): return {} payload = entry.get("payload") - return _strip_wunderground_current(payload) if isinstance(payload, dict) else {} + if not isinstance(payload, dict): + return {} + payload = overlay_latest_amsc_observation(_CACHE_DB, city, payload) + return _strip_wunderground_current(payload) def _canonical_city_payload(city: str, *, detail_depth: str) -> dict: @@ -339,7 +343,7 @@ def _canonical_city_payload(city: str, *, detail_depth: str) -> dict: "probabilities": payload.get("probabilities") or {"mu": None, "distribution": []}, } ) - return payload + return overlay_latest_amsc_observation(_CACHE_DB, city, payload) def _initializing_city_payload(city: str, *, detail_depth: str) -> dict: diff --git a/web/services/latest_observation_overlay.py b/web/services/latest_observation_overlay.py new file mode 100644 index 00000000..5b017217 --- /dev/null +++ b/web/services/latest_observation_overlay.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timezone +from typing import Any, Optional + +from loguru import logger + +from web.services.canonical_temperature import build_canonical_temperature + + +def parse_observation_epoch(value: Any) -> Optional[int]: + if value is None or value == "": + return None + if isinstance(value, datetime): + dt = value + else: + text = str(value).strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(text) + except ValueError: + for fmt in ("%Y-%m-%d %H:%M:%S%z", "%Y-%m-%d %H:%M:%S"): + try: + dt = datetime.strptime(text, fmt) + break + except ValueError: + continue + else: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return int(dt.timestamp()) + + +def _payload_latest_epoch(payload: dict[str, Any], keys: tuple[str, ...]) -> Optional[int]: + values = [payload.get(key) for key in keys] + parsed = [epoch for epoch in (parse_observation_epoch(value) for value in values) if epoch is not None] + return max(parsed) if parsed else None + + +def _block_epoch(block: Any) -> Optional[int]: + if not isinstance(block, dict): + return None + return _payload_latest_epoch( + block, + ( + "observed_at", + "observation_time", + "obs_time", + "observed_at_local", + "observation_time_local", + ), + ) + + +def _raw_amsc_epoch(row: dict[str, Any], raw_payload: dict[str, Any]) -> Optional[int]: + values = ( + raw_payload.get("observation_time"), + raw_payload.get("observed_at"), + row.get("observed_at"), + ) + parsed = [epoch for epoch in (parse_observation_epoch(value) for value in values) if epoch is not None] + return max(parsed) if parsed else None + + +def _to_float(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _latest_amsc_row(db: Any, city: str) -> tuple[Optional[dict[str, Any]], Optional[dict[str, Any]]]: + getter = getattr(db, "get_latest_raw_observation", None) + if not callable(getter): + return None, None + try: + row = getter("amsc_awos", city) + except Exception as exc: + logger.debug("latest AMSC raw overlay read failed city={}: {}", city, exc) + return None, None + if not isinstance(row, dict): + return None, None + raw_payload = row.get("payload") + if not isinstance(raw_payload, dict) or not raw_payload: + return row, None + return row, raw_payload + + +def _raw_observation_update( + city: str, + row: dict[str, Any], + raw_payload: dict[str, Any], +) -> Optional[dict[str, Any]]: + temp = _to_float(raw_payload.get("temp_c") if raw_payload.get("temp_c") is not None else raw_payload.get("temp")) + if temp is None: + return None + observed_at = str( + raw_payload.get("observation_time") + or raw_payload.get("observed_at") + or row.get("observed_at") + or "" + ).strip() + observed_at_local = str( + raw_payload.get("observation_time_local") + or raw_payload.get("observed_at_local") + or "" + ).strip() + source_label = str(raw_payload.get("source_label") or "AMSC AWOS").strip() + station_code = str(raw_payload.get("icao") or row.get("station_code") or "").strip().upper() or None + station_name = str(raw_payload.get("station_label") or row.get("station_name") or source_label).strip() + freshness = { + "freshness_status": "fresh", + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + "source_code": "amsc_awos", + "source_label": source_label, + } + return { + "temp": round(temp, 1), + "source_code": "amsc_awos", + "source_label": source_label, + "settlement_source": "amsc_awos", + "settlement_source_label": source_label, + "station_code": station_code, + "station_name": station_name, + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + "obs_time": observed_at or observed_at_local, + "freshness": freshness, + "observation_status": "live", + "city": city, + } + + +def _merge_observation_block( + payload: dict[str, Any], + key: str, + update: dict[str, Any], + raw_epoch: int, +) -> bool: + current = payload.get(key) + if not isinstance(current, dict): + current = {} + current_epoch = _block_epoch(current) + if current_epoch is not None and current_epoch >= raw_epoch: + return False + merged = dict(current) + merged.update(update) + payload[key] = merged + return True + + +def overlay_latest_amsc_observation( + db: Any, + city: str, + payload: dict[str, Any], +) -> dict[str, Any]: + normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower() + if not normalized_city or not isinstance(payload, dict) or not payload: + return payload + row, raw_payload = _latest_amsc_row(db, normalized_city) + if not isinstance(row, dict) or not isinstance(raw_payload, dict): + return payload + raw_epoch = _raw_amsc_epoch(row, raw_payload) + if raw_epoch is None: + return payload + update = _raw_observation_update(normalized_city, row, raw_payload) + if not update: + return payload + + next_payload = deepcopy(payload) + changed = False + amos = next_payload.get("amos") + amos_epoch = _block_epoch(amos) + if amos_epoch is None or raw_epoch > amos_epoch: + next_payload["amos"] = dict(raw_payload) + changed = True + + for key in ("current", "airport_primary", "airport_current"): + changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed + + canonical = next_payload.get("canonical_temperature") + canonical_epoch = _block_epoch(canonical) + if canonical_epoch is None or raw_epoch > canonical_epoch: + canonical_payload = build_canonical_temperature( + normalized_city, + { + "name": normalized_city, + "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C", + "updated_at": row.get("fetched_at"), + "current": update, + }, + fetched_at=str(row.get("fetched_at") or ""), + ) + if canonical_payload: + next_payload["canonical_temperature"] = canonical_payload + changed = True + + return next_payload if changed else payload