Track observation source health states

src/database/db_manager.py

@@ -6,7 +6,7 @@ import secrets
 import threading
 import time
 from collections import Counter
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Optional, Dict, Any, List, Set, Tuple
 from urllib.parse import urlparse
 
@@ -850,6 +850,27 @@ class DBManager:
         except (TypeError, ValueError):
             return None
 
+    @staticmethod
+    def _parse_datetime_or_none(value: Any) -> Optional[datetime]:
+        raw = str(value or "").strip()
+        if not raw:
+            return None
+        try:
+            parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+        if parsed.tzinfo is None:
+            return parsed.replace(tzinfo=timezone.utc)
+        return parsed.astimezone(timezone.utc)
+
+    @classmethod
+    def _source_latency_or_none(cls, observed_at: Any, fetched_at: Any) -> Optional[float]:
+        observed = cls._parse_datetime_or_none(observed_at)
+        fetched = cls._parse_datetime_or_none(fetched_at)
+        if observed is None or fetched is None:
+            return None
+        return max(0.0, round((fetched - observed).total_seconds(), 3))
+
     def _cache_table_name(self, kind: str) -> Optional[str]:
         normalized = str(kind or "").strip().lower()
         if normalized == "summary":
@@ -1071,10 +1092,35 @@ class DBManager:
         safe_status = str(status or "ok").strip().lower() or "ok"
         value_float = self._float_or_none(value)
         latency_float = self._float_or_none(source_latency_sec)
+        if latency_float is None:
+            latency_float = self._source_latency_or_none(safe_observed_at, safe_fetched_at)
         payload_json = json.dumps(payload or {}, ensure_ascii=False)
         created_at_ts = now_dt.timestamp()
-        success_at = str(last_success_at or (safe_fetched_at if safe_status == "ok" else "")).strip()
         with self._get_connection() as conn:
+            previous_latest = conn.execute(
+                """
+                SELECT status, error_count, last_success_at, fetched_at
+                FROM raw_observation_latest
+                WHERE source = ? AND city = ?
+                ORDER BY updated_at_ts DESC
+                LIMIT 1
+                """,
+                (normalized_source, normalized_city),
+            ).fetchone()
+            previous_error_count = int(previous_latest[1] or 0) if previous_latest else 0
+            previous_last_success = str(previous_latest[2] or "").strip() if previous_latest else ""
+            previous_status = str(previous_latest[0] or "").strip().lower() if previous_latest else ""
+            previous_fetched_at = str(previous_latest[3] or "").strip() if previous_latest else ""
+            if safe_status == "ok":
+                safe_error_count = 0
+                success_at = str(last_success_at or safe_fetched_at).strip()
+            else:
+                safe_error_count = max(1, int(error_count or 0), previous_error_count + 1)
+                success_at = str(
+                    last_success_at
+                    or previous_last_success
+                    or (previous_fetched_at if previous_status == "ok" else "")
+                ).strip()
             conn.execute(
                 """
                 INSERT INTO raw_observation_store (
@@ -1096,7 +1142,7 @@ class DBManager:
                     safe_fetched_at,
                     latency_float,
                     safe_status,
-                    max(0, int(error_count or 0)),
+                    safe_error_count,
                     success_at,
                     payload_json,
                     created_at_ts,
@@ -1135,7 +1181,7 @@ class DBManager:
                     safe_fetched_at,
                     latency_float,
                     safe_status,
-                    max(0, int(error_count or 0)),
+                    safe_error_count,
                     success_at,
                     payload_json,
                     created_at_ts,

tests/test_observation_collector.py

@@ -162,6 +162,62 @@ def test_raw_observation_store_records_latest_observation(tmp_path):
     assert latest["payload"]["temp_c"] == 24.0
 
 
+def test_raw_observation_failure_preserves_last_success_and_increments_errors(tmp_path):
+    from src.database.db_manager import DBManager
+
+    db = DBManager(str(tmp_path / "polyweather.db"))
+
+    db.append_raw_observation(
+        source="amsc_awos",
+        city="Qingdao",
+        value=24.0,
+        observed_at="2026-06-14T01:00:00+00:00",
+        fetched_at="2026-06-14T01:01:00+00:00",
+        station_code="ZSQD",
+        station_name="Qingdao Jiaodong",
+        status="ok",
+        payload={"temp_c": 24.0},
+    )
+    db.append_raw_observation(
+        source="amsc_awos",
+        city="qingdao",
+        fetched_at="2026-06-14T01:02:00+00:00",
+        status="timeout",
+        payload={"error": "upstream timeout"},
+    )
+
+    latest = db.get_latest_raw_observation("amsc_awos", "qingdao")
+
+    assert latest is not None
+    assert latest["status"] == "timeout"
+    assert latest["value"] is None
+    assert latest["error_count"] == 1
+    assert latest["last_success_at"] == "2026-06-14T01:01:00+00:00"
+    assert latest["payload"]["error"] == "upstream timeout"
+
+
+def test_raw_observation_store_computes_source_latency_when_times_are_known(tmp_path):
+    from src.database.db_manager import DBManager
+
+    db = DBManager(str(tmp_path / "polyweather.db"))
+
+    db.append_raw_observation(
+        source="amsc_awos",
+        city="Qingdao",
+        value=24.0,
+        observed_at="2026-06-14T01:00:00+00:00",
+        fetched_at="2026-06-14T01:01:30+00:00",
+        station_code="ZSQD",
+        status="ok",
+        payload={"temp_c": 24.0},
+    )
+
+    latest = db.get_latest_raw_observation("amsc_awos", "qingdao")
+
+    assert latest is not None
+    assert latest["source_latency_sec"] == 90.0
+
+
 def test_observation_refresh_request_queue_claims_pending_requests(tmp_path):
     from src.database.db_manager import DBManager
 
@@ -254,6 +310,40 @@ def test_observation_collector_writes_raw_observation_store(tmp_path):
     assert latest["station_code"] == "ZSQD"
 
 
+def test_observation_collector_records_no_results_source_health(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_china_amsc_awos_data(self, results, city, use_fahrenheit):
+            return None
+
+    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) == 0
+
+    latest = db.get_latest_raw_observation("amsc_awos", "qingdao")
+    assert latest is not None
+    assert latest["status"] == "no_results"
+    assert latest["value"] is None
+    assert latest["error_count"] == 1
+    assert latest["last_success_at"] == ""
+    assert latest["payload"]["status"] == "no_results"
+
+
 def test_observation_collector_writes_canonical_latest_from_source(tmp_path):
     from src.database.db_manager import DBManager
     from web.observation_collector_service import (
@@ -294,6 +384,41 @@ def test_observation_collector_writes_canonical_latest_from_source(tmp_path):
     assert canonical["payload"]["observed_at"] == "2026-06-14T01:00:00+00:00"
 
 
+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
+
+    db = DBManager(str(tmp_path / "polyweather.db"))
+    collector = ObservationCollector(
+        weather=object(),
+        profiles=[],
+        observation_store=db,
+        async_cache_refresh=False,
+    )
+
+    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",
+        fetched_at="2026-06-14T01:05:00+00:00",
+    )
+
+    canonical = db.get_canonical_temperature("qingdao")
+
+    assert canonical is not None
+    assert canonical["payload"]["freshness_status"] == "expected_wait"
+    assert canonical["payload"]["freshness_sec"] == 300
+    assert canonical["payload"]["confidence"] < 0.92
+
+
 def test_observation_collector_consumes_refresh_request_queue(tmp_path):
     from src.database.db_manager import DBManager
     from web.observation_collector_service import (

web/observation_collector_service.py

@@ -17,7 +17,9 @@ from src.data_collection.city_registry import CITY_REGISTRY
 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_temperature import build_canonical_temperature
+from web.services.observation_freshness import build_observation_freshness
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -121,6 +123,12 @@ class ObservationCollector:
                     error = "no_results"
             except Exception as exc:
                 error = str(exc) or exc.__class__.__name__
+                self._store_raw_observation_status(
+                    source=profile.source,
+                    city=city,
+                    status=self._failure_status_from_exception(exc),
+                    error=error,
+                )
                 logger.warning(
                     "observation collector source failed source={} city={}: {}",
                     profile.source,
@@ -256,10 +264,33 @@ class ObservationCollector:
         else:
             logger.debug("observation collector skipped unknown source={}", normalized_source)
             return False
-        ok = bool(results)
-        if ok:
-            self._store_raw_observations(normalized_source, normalized_city, results)
-        return ok
+        if not results:
+            self._store_raw_observation_status(
+                source=normalized_source,
+                city=normalized_city,
+                status="no_results",
+                error="source returned no observation rows",
+            )
+            return False
+        wrote = self._store_raw_observations(normalized_source, normalized_city, results)
+        if wrote <= 0:
+            self._store_raw_observation_status(
+                source=normalized_source,
+                city=normalized_city,
+                status="parse_error",
+                error="source response had no usable temperature",
+            )
+            return False
+        return True
+
+    @staticmethod
+    def _failure_status_from_exception(exc: Exception) -> str:
+        text = f"{exc.__class__.__name__} {exc}".lower()
+        if "timeout" in text or "timed out" in text:
+            return "timeout"
+        if "401" in text or "403" in text or "auth" in text or "unauthor" in text:
+            return "auth_error"
+        return "error"
 
     @staticmethod
     def _observation_value(row: dict[str, Any]) -> Optional[float]:
@@ -326,6 +357,15 @@ class ObservationCollector:
         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"),
+            ingested_at=fetched_at,
+            now_utc=parse_utc_datetime(fetched_at),
+        )
         payload = {
             "name": city,
             "temp_symbol": "°F" if value_unit.startswith("f") else "°C",
@@ -333,19 +373,15 @@ class ObservationCollector:
             "current": {
                 "temp": value,
                 "source_code": source,
-                "source_label": self._source_label(row, source),
+                "source_label": source_label,
                 "settlement_source": source,
-                "settlement_source_label": self._source_label(row, 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,
-                "freshness": {
-                    "freshness_status": "fresh",
-                    "observed_at": observed_at or None,
-                    "observed_at_local": row.get("observation_time_local"),
-                },
+                "freshness": freshness,
                 "observation_status": "live",
             },
         }
@@ -371,14 +407,49 @@ class ObservationCollector:
                     if isinstance(item, dict):
                         yield item
 
-    def _store_raw_observations(self, source: str, city: str, results: dict[str, Any]) -> None:
+    def _store_raw_observation_status(
+        self,
+        *,
+        source: str,
+        city: str,
+        status: str,
+        error: str = "",
+    ) -> None:
         store = self.observation_store
         writer = getattr(store, "append_raw_observation", None)
         if not callable(writer):
             return
         fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())
+        try:
+            writer(
+                source=source,
+                city=city,
+                fetched_at=fetched_at,
+                status=status,
+                payload={
+                    "source": source,
+                    "city": city,
+                    "status": status,
+                    "error": str(error or "").strip(),
+                },
+            )
+        except Exception as exc:
+            logger.debug(
+                "raw observation status write skipped source={} city={}: {}",
+                source,
+                city,
+                exc,
+            )
+
+    def _store_raw_observations(self, source: str, city: str, results: dict[str, Any]) -> 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)
+        fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())
         wrote = 0
-        for row in self._iter_raw_observation_rows(source, results):
+        for row in rows:
             value = self._observation_value(row)
             if value is None:
                 continue
@@ -416,6 +487,7 @@ class ObservationCollector:
                 )
         if wrote:
             logger.debug("raw observations stored source={} city={} count={}", source, city, wrote)
+        return wrote
 
     def _refresh_city_cache(self, city: str) -> None:
         if not callable(self.cache_refresher):