Select canonical temperature from raw observations

2026-06-14 19:33:09 +08:00
parent 00b163a8ec
commit 43aeb6047e
5 changed files with 395 additions and 4 deletions
@@ -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,
        *,
@@ -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
@@ -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 (
@@ -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

@@ -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