From 43aeb6047e3134e4af742520c47c5d3836f4d249 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 19:33:09 +0800 Subject: [PATCH] Select canonical temperature from raw observations --- src/database/db_manager.py | 46 ++++++++ tests/test_canonical_engine.py | 83 +++++++++++++ tests/test_observation_collector.py | 83 +++++++++++++ web/observation_collector_service.py | 17 ++- web/services/canonical_engine.py | 170 +++++++++++++++++++++++++++ 5 files changed, 395 insertions(+), 4 deletions(-) create mode 100644 tests/test_canonical_engine.py create mode 100644 web/services/canonical_engine.py diff --git a/src/database/db_manager.py b/src/database/db_manager.py index 106a7e5b..ee1dfbee 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -1247,6 +1247,52 @@ class DBManager: "updated_at_ts": float(row["updated_at_ts"] or 0.0), } + def list_latest_raw_observations_for_city(self, city: str, *, limit: int = 100) -> List[Dict[str, Any]]: + normalized_city = str(city or "").strip().lower() + if not normalized_city: + return [] + safe_limit = max(1, min(int(limit or 100), 500)) + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT * + FROM raw_observation_latest + WHERE city = ? + ORDER BY updated_at_ts DESC + LIMIT ? + """, + (normalized_city, safe_limit), + ).fetchall() + out: List[Dict[str, Any]] = [] + for row in rows: + try: + payload = json.loads(str(row["payload_json"] or "{}")) + except Exception: + payload = {} + if not isinstance(payload, dict): + payload = {} + out.append( + { + "source": str(row["source"] or ""), + "city": str(row["city"] or ""), + "station_code": str(row["station_code"] or ""), + "station_name": str(row["station_name"] or ""), + "runway": str(row["runway"] or ""), + "value": self._float_or_none(row["value"]), + "value_unit": str(row["value_unit"] or ""), + "observed_at": str(row["observed_at"] or ""), + "fetched_at": str(row["fetched_at"] or ""), + "source_latency_sec": self._float_or_none(row["source_latency_sec"]), + "status": str(row["status"] or ""), + "error_count": int(row["error_count"] or 0), + "last_success_at": str(row["last_success_at"] or ""), + "payload": payload, + "updated_at_ts": float(row["updated_at_ts"] or 0.0), + } + ) + return out + def enqueue_observation_refresh_request( self, *, diff --git a/tests/test_canonical_engine.py b/tests/test_canonical_engine.py new file mode 100644 index 00000000..ba7637e8 --- /dev/null +++ b/tests/test_canonical_engine.py @@ -0,0 +1,83 @@ +def test_canonical_engine_prefers_settlement_station_over_later_nearby_row(): + from web.services.canonical_engine import build_canonical_temperature_from_observations + + canonical = build_canonical_temperature_from_observations( + "shenzhen", + [ + { + "source": "hko_obs", + "city": "shenzhen", + "station_code": "HKO", + "station_name": "Hong Kong Observatory", + "value": 27.6, + "value_unit": "c", + "observed_at": "2026-06-14T01:00:00+00:00", + "fetched_at": "2026-06-14T01:05:00+00:00", + "status": "ok", + "payload": {"source_label": "HKO"}, + "updated_at_ts": 2000.0, + }, + { + "source": "hko_obs", + "city": "shenzhen", + "station_code": "LFS", + "station_name": "Lau Fau Shan", + "value": 28.1, + "value_unit": "c", + "observed_at": "2026-06-14T01:00:00+00:00", + "fetched_at": "2026-06-14T01:05:00+00:00", + "status": "ok", + "payload": {"source_label": "HKO"}, + "updated_at_ts": 1000.0, + }, + ], + ) + + assert canonical is not None + assert canonical["city"] == "shenzhen" + assert canonical["value"] == 28.1 + assert canonical["source"] == "hko_obs" + assert canonical["source_role"] == "settlement_official" + assert canonical["station_code"] == "LFS" + assert canonical["station_name"] == "Lau Fau Shan" + assert canonical["freshness_status"] == "fresh" + assert canonical["freshness_sec"] == 300 + + +def test_canonical_engine_ignores_failed_latest_rows(): + from web.services.canonical_engine import build_canonical_temperature_from_observations + + canonical = build_canonical_temperature_from_observations( + "qingdao", + [ + { + "source": "amsc_awos", + "city": "qingdao", + "station_code": "ZSQD", + "value": None, + "value_unit": "c", + "observed_at": "", + "fetched_at": "2026-06-14T01:05:00+00:00", + "status": "timeout", + "payload": {"error": "timeout"}, + "updated_at_ts": 2000.0, + }, + { + "source": "madis_hfmetar", + "city": "qingdao", + "station_code": "ZSQD", + "station_name": "Qingdao Jiaodong", + "value": 24.0, + "value_unit": "c", + "observed_at": "2026-06-14T01:00:00+00:00", + "fetched_at": "2026-06-14T01:05:00+00:00", + "status": "ok", + "payload": {"source_label": "MADIS HFMETAR"}, + "updated_at_ts": 1000.0, + }, + ], + ) + + assert canonical is not None + assert canonical["source"] == "madis_hfmetar" + assert canonical["value"] == 24.0 diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 267d4ba1..a2f5af3e 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -218,6 +218,40 @@ def test_raw_observation_store_computes_source_latency_when_times_are_known(tmp_ assert latest["source_latency_sec"] == 90.0 +def test_raw_observation_store_lists_latest_observations_for_city(tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + + db.append_raw_observation( + source="hko_obs", + city="Shenzhen", + value=28.1, + observed_at="2026-06-14T01:00:00+00:00", + fetched_at="2026-06-14T01:05:00+00:00", + station_code="LFS", + station_name="Lau Fau Shan", + status="ok", + payload={"source_label": "HKO"}, + ) + db.append_raw_observation( + source="hko_obs", + city="shenzhen", + value=27.6, + observed_at="2026-06-14T01:00:00+00:00", + fetched_at="2026-06-14T01:05:30+00:00", + station_code="HKO", + station_name="Hong Kong Observatory", + status="ok", + payload={"source_label": "HKO"}, + ) + + rows = db.list_latest_raw_observations_for_city("shenzhen") + + assert [row["station_code"] for row in rows] == ["HKO", "LFS"] + assert [row["value"] for row in rows] == [27.6, 28.1] + + def test_observation_refresh_request_queue_claims_pending_requests(tmp_path): from src.database.db_manager import DBManager @@ -377,6 +411,55 @@ def test_observation_collector_consumes_source_adapter_records(monkeypatch, tmp_ assert latest["payload"]["temp_c"] == 24.5 +def test_observation_collector_recomputes_canonical_from_raw_latest_settlement_station(tmp_path): + from src.database.db_manager import DBManager + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + db = DBManager(str(tmp_path / "polyweather.db")) + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + 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", + }, + ] + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("hko_obs", ("shenzhen",), 600)], + observation_store=db, + async_cache_refresh=False, + ) + + assert collector.run_due_once(now_ts=1000.0) == 1 + + canonical = db.get_canonical_temperature("shenzhen") + assert canonical is not None + assert canonical["payload"]["station_code"] == "LFS" + assert canonical["payload"]["station_name"] == "Lau Fau Shan" + assert canonical["payload"]["value"] == 28.1 + + def test_observation_collector_records_no_results_source_health(tmp_path): from src.database.db_manager import DBManager from web.observation_collector_service import ( diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index ef512248..e2aff0ff 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -18,6 +18,7 @@ from src.data_collection.hko_obs_sources import HKO_STATIONS from src.database.db_manager import DBManager from src.database.runtime_state import ObservationCollectorStatusRepository from web.services.analysis_utils import parse_utc_datetime +from web.services.canonical_engine import refresh_canonical_temperature_from_latest 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 ( @@ -380,6 +381,7 @@ class ObservationCollector: return len(records) fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) wrote = 0 + written_records: list[ObservationRecord] = [] for record in records: try: writer( @@ -395,10 +397,7 @@ class ObservationCollector: status="ok", payload=dict(record.payload), ) - self._store_canonical_temperature_from_observation( - record=record, - fetched_at=fetched_at, - ) + written_records.append(record) wrote += 1 except Exception as exc: logger.debug( @@ -408,6 +407,16 @@ class ObservationCollector: exc, ) if wrote: + refreshed_cities: set[str] = set() + for city in {record.city for record in written_records}: + if refresh_canonical_temperature_from_latest(self.observation_store, city): + refreshed_cities.add(city) + for record in written_records: + if record.city not in refreshed_cities: + self._store_canonical_temperature_from_observation( + record=record, + fetched_at=fetched_at, + ) logger.debug("raw observations stored count={}", wrote) return wrote diff --git a/web/services/canonical_engine.py b/web/services/canonical_engine.py new file mode 100644 index 00000000..792e5192 --- /dev/null +++ b/web/services/canonical_engine.py @@ -0,0 +1,170 @@ +"""Canonical temperature selection from normalized raw observations.""" + +from __future__ import annotations + +from typing import Any, Iterable, Optional + +from src.data_collection.city_registry import CITY_REGISTRY +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 + + +_SETTLEMENT_SOURCE_ADAPTERS = { + "hko": {"hko_obs", "cowin_obs"}, + "cwa": {"cwa"}, + "noaa": {"madis_hfmetar", "metar", "noaa"}, + "wunderground": {"amsc_awos", "amos", "madis_hfmetar", "metar", "wunderground"}, +} + +_SOURCE_WEIGHTS = { + "amsc_awos": 720, + "amos": 700, + "hko_obs": 680, + "cowin_obs": 660, + "madis_hfmetar": 500, + "metar": 460, +} + +_FRESHNESS_WEIGHTS = { + "fresh": 80, + "expected_wait": 60, + "delayed": 35, + "unknown": 20, + "stale": 5, +} + + +def _normalized_city(city: Any) -> str: + return str(city or "").strip().lower() + + +def _normalized_source(source: Any) -> str: + return str(source or "").strip().lower() + + +def _station_code(value: Any) -> str: + return str(value or "").strip().upper() + + +def _source_label(row: dict[str, Any]) -> str: + payload = row.get("payload") if isinstance(row.get("payload"), dict) else {} + for source in (row, payload): + for key in ("source_label", "label", "source_name"): + value = str(source.get(key) or "").strip() + if value: + return value + source = _normalized_source(row.get("source")) + return source.replace("_", " ").upper() if source else "Observation" + + +def _observed_at_local(row: dict[str, Any]) -> str: + payload = row.get("payload") if isinstance(row.get("payload"), dict) else {} + for source in (row, payload): + for key in ("observed_at_local", "observation_time_local", "obs_time_local"): + value = str(source.get(key) or "").strip() + if value: + return value + return "" + + +def _freshness(row: dict[str, Any]) -> dict[str, Any]: + source = _normalized_source(row.get("source")) + source_label = _source_label(row) + fetched_at = str(row.get("fetched_at") or "").strip() + return build_observation_freshness( + source_code=source, + source_label=source_label, + observed_at=row.get("observed_at"), + observed_at_local=_observed_at_local(row), + ingested_at=fetched_at, + now_utc=parse_utc_datetime(fetched_at), + ) + + +def _candidate_canonical(city: str, row: dict[str, Any]) -> Optional[dict[str, Any]]: + value = row.get("value") + if value is None or str(row.get("status") or "").strip().lower() != "ok": + return None + source = _normalized_source(row.get("source")) + source_label = _source_label(row) + observed_at = str(row.get("observed_at") or "").strip() + observed_at_local = _observed_at_local(row) + payload = { + "name": city, + "temp_symbol": "°F" if str(row.get("value_unit") or "").lower().startswith("f") else "°C", + "updated_at": str(row.get("fetched_at") or ""), + "current": { + "temp": value, + "source_code": source, + "source_label": source_label, + "settlement_source": source, + "settlement_source_label": source_label, + "station_code": row.get("station_code"), + "station_name": row.get("station_name"), + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + "obs_time": observed_at_local or observed_at, + "freshness": _freshness(row), + "observation_status": "live", + }, + } + return build_canonical_temperature(city, payload, fetched_at=str(row.get("fetched_at") or "")) + + +def _score(city: str, row: dict[str, Any], canonical: dict[str, Any]) -> tuple[int, float]: + meta = CITY_REGISTRY.get(city) or {} + station_code = _station_code(row.get("station_code")) + settlement_station_code = _station_code(meta.get("settlement_station_code") or meta.get("icao")) + settlement_source = _normalized_source(meta.get("settlement_source")) + expected_sources = _SETTLEMENT_SOURCE_ADAPTERS.get(settlement_source, {settlement_source}) + station_name = str(row.get("station_name") or "").strip().lower() + candidates = { + str(candidate or "").strip().lower() + for candidate in (meta.get("settlement_station_candidates") or []) + if str(candidate or "").strip() + } + score = _SOURCE_WEIGHTS.get(_normalized_source(row.get("source")), 100) + if _normalized_source(row.get("source")) in expected_sources: + score += 300 + if settlement_station_code and station_code == settlement_station_code: + score += 1000 + if candidates and station_name in candidates: + score += 750 + score += _FRESHNESS_WEIGHTS.get(str(canonical.get("freshness_status") or ""), 0) + try: + updated_at_ts = float(row.get("updated_at_ts") or 0.0) + except (TypeError, ValueError): + updated_at_ts = 0.0 + return score, updated_at_ts + + +def build_canonical_temperature_from_observations( + city: str, + observations: Iterable[dict[str, Any]], +) -> Optional[dict[str, Any]]: + normalized_city = _normalized_city(city) + candidates: list[tuple[tuple[int, float], dict[str, Any]]] = [] + for row in observations or []: + if not isinstance(row, dict): + continue + canonical = _candidate_canonical(normalized_city, row) + if not canonical: + continue + candidates.append((_score(normalized_city, row, canonical), canonical)) + if not candidates: + return None + candidates.sort(key=lambda item: item[0], reverse=True) + return candidates[0][1] + + +def refresh_canonical_temperature_from_latest(db: Any, city: str) -> Optional[dict[str, Any]]: + getter = getattr(db, "list_latest_raw_observations_for_city", None) + setter = getattr(db, "set_canonical_temperature", None) + if not callable(getter) or not callable(setter): + return None + canonical = build_canonical_temperature_from_observations(city, getter(city)) + if not canonical: + return None + setter(city, canonical) + return canonical