Build runway history from raw observations

src/database/db_manager.py

@@ -1293,6 +1293,64 @@ class DBManager:
             )
         return out
 
+    def list_raw_observation_history(
+        self,
+        source: str,
+        city: str,
+        *,
+        minutes: int = 60,
+        limit: int = 1000,
+    ) -> List[Dict[str, Any]]:
+        normalized_source = str(source or "").strip().lower()
+        normalized_city = str(city or "").strip().lower()
+        if not normalized_source or not normalized_city:
+            return []
+        safe_limit = max(1, min(int(limit or 1000), 5000))
+        safe_minutes = max(1, min(int(minutes or 60), 7 * 24 * 60))
+        cutoff_ts = datetime.now().timestamp() - safe_minutes * 60
+        with self._get_connection() as conn:
+            conn.row_factory = sqlite3.Row
+            rows = conn.execute(
+                """
+                SELECT *
+                FROM raw_observation_store
+                WHERE source = ?
+                  AND city = ?
+                  AND created_at_ts >= ?
+                ORDER BY observed_at ASC, fetched_at ASC, created_at_ts ASC
+                LIMIT ?
+                """,
+                (normalized_source, normalized_city, cutoff_ts, 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,
+                    "created_at_ts": float(row["created_at_ts"] or 0.0),
+                }
+            )
+        return out
+
     def enqueue_observation_refresh_request(
         self,
         *,

tests/test_latest_observation_overlay.py

@@ -117,6 +117,50 @@ def test_overlay_appends_latest_amsc_points_to_runway_history():
     assert runway_writes[0]["target_runway_max"] == 28.4
 
 
+def test_overlay_builds_amsc_runway_history_from_raw_store():
+    def raw_row(observed_at, temp):
+        return {
+            "observed_at": observed_at,
+            "payload": {
+                "source": "amsc_awos",
+                "icao": "ZUUU",
+                "temp_c": temp,
+                "observation_time": observed_at,
+                "runway_obs": {
+                    "point_temperatures": [
+                        {
+                            "runway": "02L/20R",
+                            "target_runway_max": temp,
+                        },
+                    ],
+                },
+            },
+        }
+
+    class FakeDB:
+        def get_latest_raw_observation(self, source, city):
+            assert (source, city) == ("amsc_awos", "chengdu")
+            return raw_row("2026-06-15T11:08:00+00:00", 25.4)
+
+        def list_raw_observation_history(self, source, city, *, minutes=60, limit=1000):
+            assert (source, city) == ("amsc_awos", "chengdu")
+            return [
+                raw_row("2026-06-15T11:04:00+00:00", 25.6),
+                raw_row("2026-06-15T11:08:00+00:00", 25.4),
+            ]
+
+    result = overlay_latest_amsc_observation(
+        FakeDB(),
+        "chengdu",
+        {"name": "chengdu", "temp_symbol": "°C", "runway_plate_history": {}},
+    )
+
+    assert result["runway_plate_history"]["02L/20R"][-2:] == [
+        {"time": "2026-06-15T11:04:00+00:00", "temp": 25.6},
+        {"time": "2026-06-15T11:08:00+00:00", "temp": 25.4},
+    ]
+
+
 def test_overlay_uses_latest_success_when_newer_status_row_has_no_observation():
     class FakeDB:
         def get_latest_raw_observation(self, source, city):

tests/test_observation_collector.py

@@ -181,6 +181,38 @@ def test_raw_observation_store_records_latest_observation(tmp_path):
     assert latest["payload"]["temp_c"] == 24.0
 
 
+def test_raw_observation_store_lists_source_city_history(tmp_path):
+    from src.database.db_manager import DBManager
+
+    db = DBManager(str(tmp_path / "polyweather.db"))
+    db.append_raw_observation(
+        source="amsc_awos",
+        city="Chengdu",
+        value=25.6,
+        observed_at="2026-06-15T11:04:00+00:00",
+        fetched_at="2026-06-15T11:04:30+00:00",
+        station_code="ZUUU",
+        payload={"temp_c": 25.6},
+    )
+    db.append_raw_observation(
+        source="amsc_awos",
+        city="chengdu",
+        value=25.4,
+        observed_at="2026-06-15T11:08:00+00:00",
+        fetched_at="2026-06-15T11:08:30+00:00",
+        station_code="ZUUU",
+        payload={"temp_c": 25.4},
+    )
+
+    rows = db.list_raw_observation_history("amsc_awos", "chengdu", minutes=60, limit=10)
+
+    assert [row["observed_at"] for row in rows] == [
+        "2026-06-15T11:04:00+00:00",
+        "2026-06-15T11:08:00+00:00",
+    ]
+    assert rows[-1]["payload"]["temp_c"] == 25.4
+
+
 def test_observation_collector_aligns_amsc_payload_time_with_record_observed_at(tmp_path):
     from src.database.db_manager import DBManager
     from web.observation_collector_service import ObservationCollector

web/services/latest_observation_overlay.py

@@ -214,6 +214,8 @@ def _append_latest_amsc_runway_history(
     payload: dict[str, Any],
     row: dict[str, Any],
     raw_payload: dict[str, Any],
+    *,
+    persist: bool = True,
 ) -> bool:
     runway_obs = raw_payload.get("runway_obs")
     if not isinstance(runway_obs, dict):
@@ -237,7 +239,7 @@ def _append_latest_amsc_runway_history(
         history = deepcopy(history)
 
     use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper()
-    appender = getattr(db, "append_runway_obs", None)
+    appender = getattr(db, "append_runway_obs", None) if persist else None
     icao = str(raw_payload.get("icao") or row.get("station_code") or "").strip().upper()
     changed = False
     for point in point_temperatures:
@@ -305,6 +307,37 @@ def _append_latest_amsc_runway_history(
     return changed
 
 
+def _append_amsc_runway_history_from_raw_store(
+    db: Any,
+    city: str,
+    payload: dict[str, Any],
+) -> bool:
+    lister = getattr(db, "list_raw_observation_history", None)
+    if not callable(lister):
+        return False
+    try:
+        rows = lister("amsc_awos", city, minutes=36 * 60, limit=3000)
+    except Exception as exc:
+        logger.debug("latest AMSC raw history overlay skipped city={}: {}", city, exc)
+        return False
+    changed = False
+    for row in rows if isinstance(rows, list) else []:
+        if not isinstance(row, dict):
+            continue
+        raw_payload = row.get("payload")
+        if not isinstance(raw_payload, dict) or not _amsc_payload_has_observation(raw_payload):
+            continue
+        changed = _append_latest_amsc_runway_history(
+            db,
+            city,
+            payload,
+            row,
+            raw_payload,
+            persist=False,
+        ) or changed
+    return changed
+
+
 def overlay_latest_amsc_observation(
     db: Any,
     city: str,
@@ -334,6 +367,7 @@ def overlay_latest_amsc_observation(
     for key in ("current", "airport_primary", "airport_current"):
         changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed
 
+    changed = _append_amsc_runway_history_from_raw_store(db, normalized_city, next_payload) or changed
     changed = _append_latest_amsc_runway_history(db, normalized_city, next_payload, row, raw_payload) or changed
 
     canonical = next_payload.get("canonical_temperature")