From 482dc3c9e60664bfde1f3f28636c4bd40ecd703d Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 16 Jun 2026 17:08:31 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85=20CWA=20=E5=AE=98=E6=96=B9?= =?UTF-8?q?=E6=97=A5=E5=86=85=E8=A7=82=E6=B5=8B=E5=85=9C=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_latest_observation_overlay.py | 55 ++++++++++++++++++++++ web/services/latest_observation_overlay.py | 46 +++++++++++++++++- 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/tests/test_latest_observation_overlay.py b/tests/test_latest_observation_overlay.py index 7af78b58..bf4dd48b 100644 --- a/tests/test_latest_observation_overlay.py +++ b/tests/test_latest_observation_overlay.py @@ -553,3 +553,58 @@ def test_overlay_latest_cwa_resets_stale_taipei_detail_to_latest_local_day(): } ] assert result["timeseries"]["metar_today_obs"] == result["metar_today_obs"] + + +def test_overlay_latest_cwa_uses_official_intraday_history_when_fetcher_is_empty(monkeypatch): + from datetime import datetime, timedelta, timezone + + expected_date = (datetime.now(timezone.utc) + timedelta(hours=8)).strftime("%Y-%m-%d") + + class FakeRepo: + def load_points(self, *, source_code, station_code, target_date): + assert source_code == "cwa" + assert station_code == "466920" + return [ + {"time": "14:30", "temp": 28.8}, + {"time": "15:30", "temp": 29.4}, + ] + + class FakeWeather: + def fetch_cwa_taipei_settlement_current(self): + return None + + monkeypatch.setattr( + observation_overlay, + "OfficialIntradayObservationRepository", + lambda: FakeRepo(), + ) + + result = observation_overlay.overlay_latest_cwa_observation( + FakeWeather(), + "taipei", + { + "name": "taipei", + "display_name": "Taipei", + "temp_symbol": "°C", + "local_date": "2026-06-14", + "local_time": "18:00", + "current": { + "temp": 26.0, + "source_code": "cwa", + "obs_time": "2026-06-14T10:00:00+00:00", + }, + "airport_current": { + "temp": 26.0, + "source_code": "cwa", + "obs_time": "2026-06-14T10:00:00+00:00", + }, + "metar_today_obs": [{"time": "18:00", "temp": 26.0}], + "timeseries": {"metar_today_obs": [{"time": "18:00", "temp": 26.0}]}, + }, + ) + + assert result["local_date"] == expected_date + assert result["local_time"] == "15:30" + assert result["airport_current"]["temp"] == 29.4 + assert result["airport_current"]["source_code"] == "cwa" + assert result["airport_current"]["obs_time"] == f"{expected_date}T15:30:00+08:00" diff --git a/web/services/latest_observation_overlay.py b/web/services/latest_observation_overlay.py index 4eab400a..89ccaa87 100644 --- a/web/services/latest_observation_overlay.py +++ b/web/services/latest_observation_overlay.py @@ -1,12 +1,13 @@ from __future__ import annotations from copy import deepcopy -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import time from typing import Any, Optional from loguru import logger +from src.database.runtime_state import OfficialIntradayObservationRepository from web.services.canonical_temperature import build_canonical_temperature _RAW_AMSC_RUNWAY_HISTORY_CACHE: dict[tuple[str, bool], tuple[float, dict[str, list[dict[str, Any]]]]] = {} @@ -905,6 +906,48 @@ def _latest_cwa_data_from_airport_obs_log(db: Any, city: str, use_fahrenheit: bo } +def _latest_cwa_data_from_official_intraday_history(use_fahrenheit: bool) -> Optional[dict[str, Any]]: + try: + local_now = datetime.now(timezone.utc).astimezone(timezone(timedelta(hours=8))) + target_date = local_now.strftime("%Y-%m-%d") + points = OfficialIntradayObservationRepository().load_points( + source_code="cwa", + station_code="466920", + target_date=target_date, + ) + except Exception as exc: + logger.debug("latest CWA official intraday history read failed: {}", exc) + return None + + latest: Optional[dict[str, Any]] = None + latest_epoch: Optional[int] = None + for point in points if isinstance(points, list) else []: + if not isinstance(point, dict): + continue + time_text = str(point.get("time") or "").strip() + temp = _to_float(point.get("temp")) + if len(time_text) != 5 or ":" not in time_text or temp is None: + continue + obs_time = f"{target_date}T{time_text}:00+08:00" + epoch = parse_observation_epoch(obs_time) + if epoch is None: + continue + if latest_epoch is None or epoch > latest_epoch: + latest_epoch = epoch + latest = { + "source": "cwa", + "source_label": "CWA", + "station_code": "466920", + "station_name": "\u81fa\u5317", + "observation_time": obs_time, + "current": { + "temp": round(float(temp) * 9 / 5 + 32, 1) if use_fahrenheit else round(float(temp), 1), + }, + "unit": "fahrenheit" if use_fahrenheit else "celsius", + } + return latest + + def _newer_cwa_payload(left: Optional[dict[str, Any]], right: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]: if not isinstance(left, dict): return right if isinstance(right, dict) else None @@ -924,6 +967,7 @@ def overlay_latest_cwa_observation(weather, city, payload, db=None): use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper() cwa_data = _latest_cwa_data_from_airport_obs_log(db, normalized_city, use_fahrenheit) + cwa_data = _newer_cwa_payload(cwa_data, _latest_cwa_data_from_official_intraday_history(use_fahrenheit)) fetcher = getattr(weather, "fetch_cwa_taipei_settlement_current", None) if not callable(fetcher) and not cwa_data: