From 53cf5b556bd8c4752c3fd8486ce63861ef7fd9b9 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 16 Jun 2026 23:01:29 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E9=BD=90=E5=AE=9E=E6=97=B6=E8=A7=82?= =?UTF-8?q?=E6=B5=8B=E9=87=87=E9=9B=86=E6=BA=90=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_observation_collector.py | 14 +++ tests/test_observation_source_adapters.py | 96 +++++++++++++++++++++ web/observation_collector_service.py | 45 ++++++++++ web/services/observation_source_adapters.py | 57 +++++++++++- 4 files changed, 209 insertions(+), 3 deletions(-) diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index a32a377c..eec387a3 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -88,14 +88,28 @@ def test_observation_collector_profiles_match_source_cadence(): assert profiles["cowin_obs"].interval_sec == 60 assert profiles["hko_obs"].interval_sec == 600 assert profiles["mgm"].interval_sec == 300 + assert profiles["jma_amedas"].interval_sec == 600 + assert profiles["singapore_mss"].interval_sec == 60 + assert profiles["fmi"].interval_sec == 600 + assert profiles["knmi"].interval_sec == 600 + assert profiles["ims"].interval_sec == 600 + assert profiles["aeroweb"].interval_sec == 900 + assert profiles["cwa"].interval_sec == 600 + assert profiles["metar"].interval_sec == 1800 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 profiles["jma_amedas"].cities == ("tokyo",) + assert profiles["aeroweb"].cities == ("paris",) + assert profiles["cwa"].cities == ("taipei",) + assert {"madrid", "milan", "tokyo", "paris"}.issubset(set(profiles["metar"].cities)) assert SOURCE_CADENCE_SECONDS["amsc_awos"] == 180 assert SOURCE_CADENCE_SECONDS["mgm"] == 300 + assert SOURCE_CADENCE_SECONDS["jma_amedas"] == 600 + assert SOURCE_CADENCE_SECONDS["metar"] == 1800 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 f53676a2..b23db34b 100644 --- a/tests/test_observation_source_adapters.py +++ b/tests/test_observation_source_adapters.py @@ -127,6 +127,102 @@ def test_source_adapter_collects_mgm_with_keyword_flags_and_station_metadata(): assert record.station_name == "Esenboga Airport" +def test_source_adapter_collects_jma_official_nearby_rows(): + from web.services.observation_source_adapters import collect_observation_source + + calls = [] + + class FakeWeather: + def _attach_japan_official_nearby(self, results, city, use_fahrenheit): + calls.append((city, use_fahrenheit)) + results["jma_current"] = { + "source": "jma_amedas", + "source_label": "JMA", + "temp": 23.4, + "obs_time": "2026-06-16T06:00:00+09:00", + "station_code": "44166", + "station_name": "Haneda", + } + + result = collect_observation_source( + FakeWeather(), + "jma_amedas", + "tokyo", + use_fahrenheit=False, + ) + + assert calls == [("tokyo", False)] + assert result.status == "ok" + assert len(result.records) == 1 + assert result.records[0].source == "jma_amedas" + assert result.records[0].value == 23.4 + assert result.records[0].observed_at == "2026-06-16T06:00:00+09:00" + assert result.records[0].station_code == "44166" + + +def test_source_adapter_collects_cwa_direct_settlement_payload(): + from web.services.observation_source_adapters import collect_observation_source + + calls = [] + + class FakeWeather: + def fetch_cwa_taipei_settlement_current(self): + calls.append("fetch_cwa") + return { + "source": "cwa", + "source_label": "CWA", + "station_code": "466920", + "station_name": "Taipei", + "observation_time": "2026-06-16T06:00:00+08:00", + "current": {"temp": 29.2}, + } + + result = collect_observation_source( + FakeWeather(), + "cwa", + "taipei", + use_fahrenheit=False, + ) + + assert calls == ["fetch_cwa"] + assert result.status == "ok" + assert result.records[0].source == "cwa" + assert result.records[0].value == 29.2 + assert result.records[0].observed_at == "2026-06-16T06:00:00+08:00" + assert result.records[0].station_code == "466920" + + +def test_source_adapter_collects_metar_for_low_frequency_cities(): + from web.services.observation_source_adapters import collect_observation_source + + calls = [] + + class FakeWeather: + def fetch_metar(self, city, use_fahrenheit=False, utc_offset=0): + calls.append((city, use_fahrenheit, utc_offset)) + return { + "source": "metar", + "icao": "LEMD", + "station_name": "Madrid Barajas", + "observation_time": "2026-06-16T12:00:00.000Z", + "current": {"temp": 28.0}, + } + + result = collect_observation_source( + FakeWeather(), + "metar", + "madrid", + use_fahrenheit=False, + ) + + assert calls == [("madrid", False, 3600)] + assert result.status == "ok" + assert result.records[0].source == "metar" + assert result.records[0].value == 28.0 + assert result.records[0].observed_at == "2026-06-16T12:00:00.000Z" + assert result.records[0].station_code == "LEMD" + + def test_source_adapter_reports_parse_error_for_unusable_source_rows(): from web.services.observation_source_adapters import collect_observation_source diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 36095b0f..41fd570c 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -591,6 +591,11 @@ def build_observation_source_profiles() -> List[ObservationSourceProfile]: 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", @@ -622,6 +627,46 @@ def build_observation_source_profiles() -> List[ObservationSourceProfile]: 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)), + ), ] diff --git a/web/services/observation_source_adapters.py b/web/services/observation_source_adapters.py index d4d7472c..d0d08139 100644 --- a/web/services/observation_source_adapters.py +++ b/web/services/observation_source_adapters.py @@ -5,6 +5,8 @@ from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Iterable +from src.data_collection.city_registry import CITY_REGISTRY + @dataclass(frozen=True) class ObservationRecord: @@ -37,8 +39,17 @@ _ATTACH_METHODS: dict[str, str] = { "hko_obs": "_attach_hko_obs_official_nearby", "cowin_obs": "_attach_cowin_official_nearby", "mgm": "_attach_turkish_mgm_data", + "jma_amedas": "_attach_japan_official_nearby", + "fmi": "_attach_fmi_official_nearby", + "knmi": "_attach_knmi_official_nearby", + "singapore_mss": "_attach_singapore_mss_data", + "ims": "_attach_israel_ims_data", + "ncm": "_attach_saudi_ncm_data", + "aeroweb": "_attach_paris_aeroweb_data", } +_NO_UNIT_ATTACH_SOURCES = {"singapore_mss", "ims", "ncm", "aeroweb"} + _TURKISH_MGM_STATION_CODES = { "ankara": "17128", "istanbul": "17058", @@ -70,6 +81,14 @@ def _text(row: dict[str, Any], keys: Iterable[str]) -> str: return "" +def _city_utc_offset_seconds(city: str) -> int: + meta = CITY_REGISTRY.get(city) or {} + try: + return int(meta.get("tz_offset") or meta.get("utc_offset_seconds") or 0) + except (TypeError, ValueError): + return 0 + + def _observation_value(row: dict[str, Any]) -> float | None: for key in ("temp_c", "temperature_c", "temp", "value"): value = _float_or_none(row.get(key)) @@ -156,7 +175,7 @@ def collect_observation_source( normalized_source = _normalize_source(source) normalized_city = _normalize_city(city) method_name = _ATTACH_METHODS.get(normalized_source) - if not method_name: + if not method_name and normalized_source not in {"cwa", "metar"}: return ObservationSourceResult( source=normalized_source, city=normalized_city, @@ -164,8 +183,8 @@ def collect_observation_source( error="unsupported observation source", records=(), ) - attach: Callable[[dict[str, Any], str, bool], Any] | None = getattr(weather, method_name, None) - if not callable(attach): + attach: Callable[..., Any] | None = getattr(weather, method_name, None) if method_name else None + if method_name and not callable(attach): return ObservationSourceResult( source=normalized_source, city=normalized_city, @@ -183,6 +202,38 @@ def collect_observation_source( include_nearby=True, ) _enrich_mgm_results(results, normalized_city) + elif normalized_source == "cwa": + fetch_cwa = getattr(weather, "fetch_cwa_taipei_settlement_current", None) + if not callable(fetch_cwa) or normalized_city != "taipei": + return ObservationSourceResult( + source=normalized_source, + city=normalized_city, + status="unsupported", + error="weather collector missing fetch_cwa_taipei_settlement_current", + records=(), + ) + cwa_payload = fetch_cwa() + if cwa_payload: + results["cwa"] = cwa_payload + elif normalized_source == "metar": + fetch_metar = getattr(weather, "fetch_metar", None) + if not callable(fetch_metar): + return ObservationSourceResult( + source=normalized_source, + city=normalized_city, + status="unsupported", + error="weather collector missing fetch_metar", + records=(), + ) + metar_payload = fetch_metar( + normalized_city, + use_fahrenheit=bool(use_fahrenheit), + utc_offset=_city_utc_offset_seconds(normalized_city), + ) + if metar_payload: + results["metar"] = metar_payload + elif normalized_source in _NO_UNIT_ATTACH_SOURCES: + attach(results, normalized_city) else: attach(results, normalized_city, bool(use_fahrenheit)) if not results: