From 00b163a8ec4bdfe71023e4387c71f0e0f9c887f4 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 19:20:43 +0800 Subject: [PATCH] Introduce observation source adapters --- tests/test_observation_collector.py | 92 +++++++-- tests/test_observation_source_adapters.py | 119 ++++++++++++ web/observation_collector_service.py | 198 ++++++-------------- web/services/observation_source_adapters.py | 179 ++++++++++++++++++ 4 files changed, 439 insertions(+), 149 deletions(-) create mode 100644 tests/test_observation_source_adapters.py create mode 100644 web/services/observation_source_adapters.py diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 34234bdd..267d4ba1 100644 --- a/tests/test_observation_collector.py +++ b/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", - source="amsc_awos", - row={ - "source": "amsc_awos", - "source_label": "AMSC AWOS", - "temp_c": 24.0, - "observation_time": "2026-06-14T01:00:00+00:00", - "icao": "ZSQD", - }, - value=24.0, - observed_at="2026-06-14T01:00:00+00:00", + record=ObservationRecord( + source="amsc_awos", + city="qingdao", + value=24.0, + observed_at="2026-06-14T01:00:00+00:00", + observed_at_local="", + station_code="ZSQD", + station_name="", + runway="", + value_unit="c", + source_label="AMSC AWOS", + payload={"temp_c": 24.0}, + ), fetched_at="2026-06-14T01:05:00+00:00", ) diff --git a/tests/test_observation_source_adapters.py b/tests/test_observation_source_adapters.py new file mode 100644 index 00000000..494a672a --- /dev/null +++ b/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 == () diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index a165bdfa..ef512248 100644 --- a/web/observation_collector_service.py +++ b/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] = {} - - if normalized_source == "amsc_awos": - self.weather._attach_china_amsc_awos_data(results, normalized_city, use_fahrenheit) - elif normalized_source == "amos": - self.weather._attach_korean_amos_data(results, normalized_city, use_fahrenheit) - elif normalized_source == "madis_hfmetar": - 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: + result = collect_observation_source( + self.weather, + normalized_source, + normalized_city, + use_fahrenheit=use_fahrenheit, + ) + if result.status == "unsupported": 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, - city=normalized_city, - status="no_results", - error="source returned no observation rows", + source=result.source or normalized_source, + city=result.city or normalized_city, + status=result.status, + 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, - city=normalized_city, + source=result.source or normalized_source, + 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, - source: str, - row: dict[str, Any], - value: float, - observed_at: str, + record: ObservationRecord, 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_label=source_label, - observed_at=observed_at or None, - observed_at_local=row.get("observation_time_local"), + source_code=record.source, + source_label=record.source_label, + observed_at=record.observed_at or None, + observed_at_local=record.observed_at_local or None, ingested_at=fetched_at, now_utc=parse_utc_datetime(fetched_at), ) payload = { - "name": city, - "temp_symbol": "°F" if value_unit.startswith("f") else "°C", + "name": record.city, + "temp_symbol": "°F" if record.value_unit.startswith("f") else "°C", "updated_at": fetched_at, "current": { - "temp": value, - "source_code": source, - "source_label": source_label, - "settlement_source": source, - "settlement_source_label": source_label, - "station_code": self._station_code(row), - "station_name": self._station_name(row), - "observed_at": observed_at or None, - "observed_at_local": row.get("observation_time_local"), - "obs_time": row.get("observation_time_local") or observed_at, + "temp": record.value, + "source_code": record.source, + "source_label": record.source_label, + "settlement_source": record.source, + "settlement_source_label": record.source_label, + "station_code": record.station_code, + "station_name": record.station_name, + "observed_at": record.observed_at or None, + "observed_at_local": record.observed_at_local or None, + "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) - - def _iter_raw_observation_rows( - self, - source: str, - 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 + logger.debug( + "canonical temperature write skipped source={} city={}: {}", + record.source, + record.city, + exc, + ) 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: - rows = list(self._iter_raw_observation_rows(source, results)) + def _store_raw_observations(self, records: Sequence[ObservationRecord]) -> int: 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: - 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() + for record in records: try: - observed_at = self._observation_time(row) writer( - source=payload_source or source, - city=city, - value=value, - observed_at=observed_at, + source=record.source, + city=record.city, + value=record.value, + observed_at=record.observed_at, fetched_at=fetched_at, - station_code=self._station_code(row), - station_name=self._station_name(row), - runway=str(row.get("runway") or "").strip(), - value_unit=str(row.get("unit") or row.get("temp_unit") or "c").strip().lower(), + station_code=record.station_code, + station_name=record.station_name, + runway=record.runway, + value_unit=record.value_unit, status="ok", - payload=dict(row), + payload=dict(record.payload), ) self._store_canonical_temperature_from_observation( - city=city, - source=payload_source or source, - row=row, - value=value, - observed_at=observed_at, + record=record, fetched_at=fetched_at, ) wrote += 1 except Exception as exc: logger.debug( "raw observation store write skipped source={} city={}: {}", - source, - city, + record.source, + 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: diff --git a/web/services/observation_source_adapters.py b/web/services/observation_source_adapters.py new file mode 100644 index 00000000..73baa513 --- /dev/null +++ b/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, + )