Introduce observation source adapters

tests/test_observation_collector.py

@@ -310,6 +310,73 @@ def test_observation_collector_writes_raw_observation_store(tmp_path):
     assert latest["station_code"] == "ZSQD"
 
 
+def test_observation_collector_consumes_source_adapter_records(monkeypatch, tmp_path):
+    from src.database.db_manager import DBManager
+    import web.observation_collector_service as collector_service
+    from web.observation_collector_service import (
+        ObservationCollector,
+        ObservationSourceProfile,
+    )
+    from web.services.observation_source_adapters import (
+        ObservationRecord,
+        ObservationSourceResult,
+    )
+
+    db = DBManager(str(tmp_path / "polyweather.db"))
+    calls = []
+
+    def fake_collect_observation_source(weather, source, city, *, use_fahrenheit):
+        calls.append((weather.marker, source, city, use_fahrenheit))
+        return ObservationSourceResult(
+            source="amsc_awos",
+            city="qingdao",
+            status="ok",
+            error="",
+            records=(
+                ObservationRecord(
+                    source="amsc_awos",
+                    city="qingdao",
+                    value=24.5,
+                    observed_at="2026-06-14T01:00:00+00:00",
+                    observed_at_local="2026-06-14 09:00",
+                    station_code="ZSQD",
+                    station_name="Qingdao Jiaodong",
+                    runway="17L",
+                    value_unit="c",
+                    source_label="AMSC AWOS",
+                    payload={"temp_c": 24.5},
+                ),
+            ),
+        )
+
+    monkeypatch.setattr(
+        collector_service,
+        "collect_observation_source",
+        fake_collect_observation_source,
+    )
+
+    class FakeWeather:
+        marker = "weather"
+
+        def _uses_fahrenheit(self, city):
+            return False
+
+    collector = ObservationCollector(
+        weather=FakeWeather(),
+        profiles=[ObservationSourceProfile("amsc_awos", ("qingdao",), 180)],
+        observation_store=db,
+        async_cache_refresh=False,
+    )
+
+    assert collector.run_due_once(now_ts=1000.0) == 1
+    assert calls == [("weather", "amsc_awos", "qingdao", False)]
+
+    latest = db.get_latest_raw_observation("amsc_awos", "qingdao", station_code="ZSQD", runway="17L")
+    assert latest is not None
+    assert latest["value"] == 24.5
+    assert latest["payload"]["temp_c"] == 24.5
+
+
 def test_observation_collector_records_no_results_source_health(tmp_path):
     from src.database.db_manager import DBManager
     from web.observation_collector_service import (
@@ -387,6 +454,7 @@ def test_observation_collector_writes_canonical_latest_from_source(tmp_path):
 def test_observation_collector_canonical_uses_source_freshness_profile(tmp_path):
     from src.database.db_manager import DBManager
     from web.observation_collector_service import ObservationCollector
+    from web.services.observation_source_adapters import ObservationRecord
 
     db = DBManager(str(tmp_path / "polyweather.db"))
     collector = ObservationCollector(
@@ -397,17 +465,19 @@ def test_observation_collector_canonical_uses_source_freshness_profile(tmp_path)
     )
 
     collector._store_canonical_temperature_from_observation(
-        city="qingdao",
+        record=ObservationRecord(
-        source="amsc_awos",
+            source="amsc_awos",
-        row={
+            city="qingdao",
-            "source": "amsc_awos",
+            value=24.0,
-            "source_label": "AMSC AWOS",
+            observed_at="2026-06-14T01:00:00+00:00",
-            "temp_c": 24.0,
+            observed_at_local="",
-            "observation_time": "2026-06-14T01:00:00+00:00",
+            station_code="ZSQD",
-            "icao": "ZSQD",
+            station_name="",
-        },
+            runway="",
-        value=24.0,
+            value_unit="c",
-        observed_at="2026-06-14T01:00:00+00:00",
+            source_label="AMSC AWOS",
+            payload={"temp_c": 24.0},
+        ),
         fetched_at="2026-06-14T01:05:00+00:00",
     )
 

tests/test_observation_source_adapters.py

@@ -0,0 +1,119 @@
+def test_source_adapter_normalizes_amsc_awos_payload_to_observation_record():
+    from web.services.observation_source_adapters import collect_observation_source
+
+    calls = []
+
+    class FakeWeather:
+        def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit):
+            calls.append((city, use_fahrenheit))
+            results["amos"] = {
+                "source": "amsc_awos",
+                "source_label": "AMSC AWOS",
+                "temp_c": "24.3",
+                "observation_time": "2026-06-14T01:00:00+00:00",
+                "observation_time_local": "2026-06-14 09:00",
+                "icao": "ZSQD",
+                "station_label": "Qingdao Jiaodong",
+                "runway": "17L",
+            }
+
+    result = collect_observation_source(
+        FakeWeather(),
+        "AMSC_AWOS",
+        "Qingdao",
+        use_fahrenheit=False,
+    )
+
+    assert calls == [("qingdao", False)]
+    assert result.source == "amsc_awos"
+    assert result.city == "qingdao"
+    assert result.status == "ok"
+    assert result.error == ""
+    assert len(result.records) == 1
+
+    record = result.records[0]
+    assert record.source == "amsc_awos"
+    assert record.city == "qingdao"
+    assert record.value == 24.3
+    assert record.observed_at == "2026-06-14T01:00:00+00:00"
+    assert record.observed_at_local == "2026-06-14 09:00"
+    assert record.station_code == "ZSQD"
+    assert record.station_name == "Qingdao Jiaodong"
+    assert record.runway == "17L"
+    assert record.value_unit == "c"
+    assert record.source_label == "AMSC AWOS"
+    assert record.payload["temp_c"] == "24.3"
+
+
+def test_source_adapter_flattens_nearby_source_lists():
+    from web.services.observation_source_adapters import collect_observation_source
+
+    class FakeWeather:
+        def _attach_hko_obs_official_nearby(self, results, city, use_fahrenheit):
+            results["hko_obs_nearby"] = [
+                {
+                    "source": "hko_obs",
+                    "source_label": "HKO",
+                    "temperature_c": 28.1,
+                    "observation_time": "2026-06-14T01:00:00+00:00",
+                    "station_code": "LFS",
+                    "station_name": "Lau Fau Shan",
+                },
+                {
+                    "source": "hko_obs",
+                    "source_label": "HKO",
+                    "temperature_c": 27.6,
+                    "observation_time": "2026-06-14T01:00:00+00:00",
+                    "station_code": "HKO",
+                    "station_name": "Hong Kong Observatory",
+                },
+            ]
+
+    result = collect_observation_source(
+        FakeWeather(),
+        "hko_obs",
+        "shenzhen",
+        use_fahrenheit=False,
+    )
+
+    assert result.status == "ok"
+    assert [record.station_code for record in result.records] == ["LFS", "HKO"]
+    assert [record.value for record in result.records] == [28.1, 27.6]
+
+
+def test_source_adapter_reports_parse_error_for_unusable_source_rows():
+    from web.services.observation_source_adapters import collect_observation_source
+
+    class FakeWeather:
+        def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit):
+            results["bad"] = {
+                "source": "amsc_awos",
+                "observation_time": "2026-06-14T01:00:00+00:00",
+                "icao": "ZSQD",
+            }
+
+    result = collect_observation_source(
+        FakeWeather(),
+        "amsc_awos",
+        "qingdao",
+        use_fahrenheit=False,
+    )
+
+    assert result.status == "parse_error"
+    assert result.error == "source response had no usable temperature"
+    assert result.records == ()
+
+
+def test_source_adapter_reports_unsupported_source_without_calling_weather():
+    from web.services.observation_source_adapters import collect_observation_source
+
+    result = collect_observation_source(
+        object(),
+        "unknown_source",
+        "qingdao",
+        use_fahrenheit=False,
+    )
+
+    assert result.status == "unsupported"
+    assert result.error == "unsupported observation source"
+    assert result.records == ()

web/observation_collector_service.py

@@ -20,6 +20,10 @@ from src.database.runtime_state import ObservationCollectorStatusRepository
 from web.services.analysis_utils import parse_utc_datetime
 from web.services.canonical_temperature import build_canonical_temperature
 from web.services.observation_freshness import build_observation_freshness
+from web.services.observation_source_adapters import (
+    ObservationRecord,
+    collect_observation_source,
+)
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -249,34 +253,28 @@ class ObservationCollector:
         if not normalized_source or not normalized_city:
             return False
         use_fahrenheit = bool(self.weather._uses_fahrenheit(normalized_city))
-        results: dict[str, Any] = {}
+        result = collect_observation_source(
-
+            self.weather,
-        if normalized_source == "amsc_awos":
+            normalized_source,
-            self.weather._attach_china_amsc_awos_data(results, normalized_city, use_fahrenheit)
+            normalized_city,
-        elif normalized_source == "amos":
+            use_fahrenheit=use_fahrenheit,
-            self.weather._attach_korean_amos_data(results, normalized_city, use_fahrenheit)
+        )
-        elif normalized_source == "madis_hfmetar":
+        if result.status == "unsupported":
-            self.weather._attach_madis_hfmetar_data(results, normalized_city, use_fahrenheit)
-        elif normalized_source == "hko_obs":
-            self.weather._attach_hko_obs_official_nearby(results, normalized_city, use_fahrenheit)
-        elif normalized_source == "cowin_obs":
-            self.weather._attach_cowin_official_nearby(results, normalized_city, use_fahrenheit)
-        else:
             logger.debug("observation collector skipped unknown source={}", normalized_source)
             return False
-        if not results:
+        if result.status != "ok":
             self._store_raw_observation_status(
-                source=normalized_source,
+                source=result.source or normalized_source,
-                city=normalized_city,
+                city=result.city or normalized_city,
-                status="no_results",
+                status=result.status,
-                error="source returned no observation rows",
+                error=result.error,
             )
             return False
-        wrote = self._store_raw_observations(normalized_source, normalized_city, results)
+        wrote = self._store_raw_observations(result.records)
         if wrote <= 0:
             self._store_raw_observation_status(
-                source=normalized_source,
+                source=result.source or normalized_source,
-                city=normalized_city,
+                city=result.city or normalized_city,
                 status="parse_error",
                 error="source response had no usable temperature",
             )
@@ -292,120 +290,54 @@ class ObservationCollector:
             return "auth_error"
         return "error"
 
-    @staticmethod
-    def _observation_value(row: dict[str, Any]) -> Optional[float]:
-        for key in ("temp_c", "temperature_c", "temp", "value"):
-            try:
-                value = row.get(key)
-                if value is not None and value != "":
-                    return float(value)
-            except (TypeError, ValueError):
-                continue
-        current = row.get("current")
-        if isinstance(current, dict):
-            try:
-                value = current.get("temp")
-                if value is not None and value != "":
-                    return float(value)
-            except (TypeError, ValueError):
-                return None
-        return None
-
-    @staticmethod
-    def _observation_time(row: dict[str, Any]) -> str:
-        for key in ("observation_time", "observed_at", "obs_time", "time_utc", "time"):
-            value = str(row.get(key) or "").strip()
-            if value:
-                return value
-        return ""
-
-    @staticmethod
-    def _station_code(row: dict[str, Any]) -> str:
-        for key in ("station_code", "icao", "istNo", "station_id", "code"):
-            value = str(row.get(key) or "").strip()
-            if value:
-                return value
-        return ""
-
-    @staticmethod
-    def _station_name(row: dict[str, Any]) -> str:
-        for key in ("station_name", "station_label", "name", "label"):
-            value = str(row.get(key) or "").strip()
-            if value:
-                return value
-        return ""
-
-    @staticmethod
-    def _source_label(row: dict[str, Any], source: str) -> str:
-        for key in ("source_label", "label", "source_name"):
-            value = str(row.get(key) or "").strip()
-            if value:
-                return value
-        return str(source or "").replace("_", " ").upper()
-
     def _store_canonical_temperature_from_observation(
         self,
         *,
-        city: str,
+        record: ObservationRecord,
-        source: str,
-        row: dict[str, Any],
-        value: float,
-        observed_at: str,
         fetched_at: str,
     ) -> None:
         setter = getattr(self.observation_store, "set_canonical_temperature", None)
         if not callable(setter):
             return
-        value_unit = str(row.get("unit") or row.get("temp_unit") or "c").strip().lower()
-        source_label = self._source_label(row, source)
         freshness = build_observation_freshness(
-            source_code=source,
+            source_code=record.source,
-            source_label=source_label,
+            source_label=record.source_label,
-            observed_at=observed_at or None,
+            observed_at=record.observed_at or None,
-            observed_at_local=row.get("observation_time_local"),
+            observed_at_local=record.observed_at_local or None,
             ingested_at=fetched_at,
             now_utc=parse_utc_datetime(fetched_at),
         )
         payload = {
-            "name": city,
+            "name": record.city,
-            "temp_symbol": "°F" if value_unit.startswith("f") else "°C",
+            "temp_symbol": "°F" if record.value_unit.startswith("f") else "°C",
             "updated_at": fetched_at,
             "current": {
-                "temp": value,
+                "temp": record.value,
-                "source_code": source,
+                "source_code": record.source,
-                "source_label": source_label,
+                "source_label": record.source_label,
-                "settlement_source": source,
+                "settlement_source": record.source,
-                "settlement_source_label": source_label,
+                "settlement_source_label": record.source_label,
-                "station_code": self._station_code(row),
+                "station_code": record.station_code,
-                "station_name": self._station_name(row),
+                "station_name": record.station_name,
-                "observed_at": observed_at or None,
+                "observed_at": record.observed_at or None,
-                "observed_at_local": row.get("observation_time_local"),
+                "observed_at_local": record.observed_at_local or None,
-                "obs_time": row.get("observation_time_local") or observed_at,
+                "obs_time": record.observed_at_local or record.observed_at,
                 "freshness": freshness,
                 "observation_status": "live",
             },
         }
-        canonical = build_canonical_temperature(city, payload, fetched_at=fetched_at)
+        canonical = build_canonical_temperature(record.city, payload, fetched_at=fetched_at)
         if not canonical:
             return
         try:
-            setter(city, canonical)
+            setter(record.city, canonical)
         except Exception as exc:
-            logger.debug("canonical temperature write skipped source={} city={}: {}", source, city, exc)
+            logger.debug(
-
+                "canonical temperature write skipped source={} city={}: {}",
-    def _iter_raw_observation_rows(
+                record.source,
-        self,
+                record.city,
-        source: str,
+                exc,
-        results: dict[str, Any],
+            )
-    ) -> Iterable[dict[str, Any]]:
-        for value in (results or {}).values():
-            if isinstance(value, dict):
-                yield value
-                continue
-            if isinstance(value, list):
-                for item in value:
-                    if isinstance(item, dict):
-                        yield item
 
     def _store_raw_observation_status(
         self,
@@ -441,52 +373,42 @@ class ObservationCollector:
                 exc,
             )
 
-    def _store_raw_observations(self, source: str, city: str, results: dict[str, Any]) -> int:
+    def _store_raw_observations(self, records: Sequence[ObservationRecord]) -> int:
-        rows = list(self._iter_raw_observation_rows(source, results))
         store = self.observation_store
         writer = getattr(store, "append_raw_observation", None)
         if not callable(writer):
-            return sum(1 for row in rows if self._observation_value(row) is not None)
+            return len(records)
         fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())
         wrote = 0
-        for row in rows:
+        for record in records:
-            value = self._observation_value(row)
-            if value is None:
-                continue
-            payload_source = str(row.get("source") or row.get("source_code") or source).strip().lower()
             try:
-                observed_at = self._observation_time(row)
                 writer(
-                    source=payload_source or source,
+                    source=record.source,
-                    city=city,
+                    city=record.city,
-                    value=value,
+                    value=record.value,
-                    observed_at=observed_at,
+                    observed_at=record.observed_at,
                     fetched_at=fetched_at,
-                    station_code=self._station_code(row),
+                    station_code=record.station_code,
-                    station_name=self._station_name(row),
+                    station_name=record.station_name,
-                    runway=str(row.get("runway") or "").strip(),
+                    runway=record.runway,
-                    value_unit=str(row.get("unit") or row.get("temp_unit") or "c").strip().lower(),
+                    value_unit=record.value_unit,
                     status="ok",
-                    payload=dict(row),
+                    payload=dict(record.payload),
                 )
                 self._store_canonical_temperature_from_observation(
-                    city=city,
+                    record=record,
-                    source=payload_source or source,
-                    row=row,
-                    value=value,
-                    observed_at=observed_at,
                     fetched_at=fetched_at,
                 )
                 wrote += 1
             except Exception as exc:
                 logger.debug(
                     "raw observation store write skipped source={} city={}: {}",
-                    source,
+                    record.source,
-                    city,
+                    record.city,
                     exc,
                 )
         if wrote:
-            logger.debug("raw observations stored source={} city={} count={}", source, city, wrote)
+            logger.debug("raw observations stored count={}", wrote)
         return wrote
 
     def _refresh_city_cache(self, city: str) -> None:

web/services/observation_source_adapters.py

@@ -0,0 +1,179 @@
+"""Observation source adapters with a normalized collector-facing output."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable, Iterable
+
+
+@dataclass(frozen=True)
+class ObservationRecord:
+    source: str
+    city: str
+    value: float
+    observed_at: str
+    observed_at_local: str
+    station_code: str
+    station_name: str
+    runway: str
+    value_unit: str
+    source_label: str
+    payload: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class ObservationSourceResult:
+    source: str
+    city: str
+    status: str
+    error: str
+    records: tuple[ObservationRecord, ...]
+
+
+_ATTACH_METHODS: dict[str, str] = {
+    "amsc_awos": "_attach_china_amsc_awos_data",
+    "amos": "_attach_korean_amos_data",
+    "madis_hfmetar": "_attach_madis_hfmetar_data",
+    "hko_obs": "_attach_hko_obs_official_nearby",
+    "cowin_obs": "_attach_cowin_official_nearby",
+}
+
+
+def _normalize_source(source: Any) -> str:
+    return str(source or "").strip().lower()
+
+
+def _normalize_city(city: Any) -> str:
+    return str(city or "").strip().lower()
+
+
+def _float_or_none(value: Any) -> float | None:
+    try:
+        if value is None or value == "":
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _text(row: dict[str, Any], keys: Iterable[str]) -> str:
+    for key in keys:
+        value = str(row.get(key) or "").strip()
+        if value:
+            return value
+    return ""
+
+
+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))
+        if value is not None:
+            return value
+    current = row.get("current")
+    if isinstance(current, dict):
+        return _float_or_none(current.get("temp"))
+    return None
+
+
+def _iter_result_rows(results: dict[str, Any]) -> Iterable[dict[str, Any]]:
+    for value in (results or {}).values():
+        if isinstance(value, dict):
+            yield value
+            continue
+        if isinstance(value, list):
+            for item in value:
+                if isinstance(item, dict):
+                    yield item
+
+
+def _record_from_row(
+    *,
+    source: str,
+    city: str,
+    row: dict[str, Any],
+) -> ObservationRecord | None:
+    value = _observation_value(row)
+    if value is None:
+        return None
+    record_source = _normalize_source(row.get("source") or row.get("source_code") or source)
+    return ObservationRecord(
+        source=record_source or source,
+        city=city,
+        value=value,
+        observed_at=_text(row, ("observation_time", "observed_at", "obs_time", "time_utc", "time")),
+        observed_at_local=_text(
+            row,
+            ("observation_time_local", "observed_at_local", "obs_time_local", "local_time"),
+        ),
+        station_code=_text(row, ("station_code", "icao", "istNo", "station_id", "code")).upper(),
+        station_name=_text(row, ("station_name", "station_label", "name")),
+        runway=_text(row, ("runway",)).upper(),
+        value_unit=str(row.get("unit") or row.get("temp_unit") or "c").strip().lower(),
+        source_label=_text(row, ("source_label", "label", "source_name"))
+        or (record_source or source).replace("_", " ").upper(),
+        payload=dict(row),
+    )
+
+
+def collect_observation_source(
+    weather: Any,
+    source: Any,
+    city: Any,
+    *,
+    use_fahrenheit: bool,
+) -> ObservationSourceResult:
+    normalized_source = _normalize_source(source)
+    normalized_city = _normalize_city(city)
+    method_name = _ATTACH_METHODS.get(normalized_source)
+    if not method_name:
+        return ObservationSourceResult(
+            source=normalized_source,
+            city=normalized_city,
+            status="unsupported",
+            error="unsupported observation source",
+            records=(),
+        )
+    attach: Callable[[dict[str, Any], str, bool], Any] | None = getattr(weather, method_name, None)
+    if not callable(attach):
+        return ObservationSourceResult(
+            source=normalized_source,
+            city=normalized_city,
+            status="unsupported",
+            error=f"weather collector missing {method_name}",
+            records=(),
+        )
+
+    results: dict[str, Any] = {}
+    attach(results, normalized_city, bool(use_fahrenheit))
+    if not results:
+        return ObservationSourceResult(
+            source=normalized_source,
+            city=normalized_city,
+            status="no_results",
+            error="source returned no observation rows",
+            records=(),
+        )
+
+    records = tuple(
+        record
+        for record in (
+            _record_from_row(source=normalized_source, city=normalized_city, row=row)
+            for row in _iter_result_rows(results)
+        )
+        if record is not None
+    )
+    if not records:
+        return ObservationSourceResult(
+            source=normalized_source,
+            city=normalized_city,
+            status="parse_error",
+            error="source response had no usable temperature",
+            records=(),
+        )
+    return ObservationSourceResult(
+        source=normalized_source,
+        city=normalized_city,
+        status="ok",
+        error="",
+        records=records,
+    )