补齐实时观测采集源覆盖

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():

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
 

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)),
+        ),
     ]
 
 

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)
+    attach: Callable[..., Any] | None = getattr(weather, method_name, None) if method_name else None
-    if not callable(attach):
+    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: