Persist AMSC runway history from collector

tests/test_latest_observation_overlay.py

@@ -48,6 +48,65 @@ def test_overlay_replaces_amos_when_old_local_time_string_looks_later_than_new_u
     assert result["amos"]["runway_obs"]["point_temperatures"][0]["runway"] == "17L/35R"
 
 
+def test_overlay_appends_latest_amsc_points_to_runway_history():
+    class FakeDB:
+        def get_latest_raw_observation(self, source, city):
+            assert (source, city) == ("amsc_awos", "chengdu")
+            return {
+                "observed_at": "2026-06-15T10:51:00+00:00",
+                "fetched_at": "2026-06-15T10:51:30+00:00",
+                "station_code": "ZUUU",
+                "station_name": "Chengdu Shuangliu",
+                "payload": {
+                    "source": "amsc_awos",
+                    "source_label": "AMSC AWOS Chengdu Shuangliu (ZUUU)",
+                    "icao": "ZUUU",
+                    "temp_c": 28.4,
+                    "observation_time": "2026-06-15T10:51:00+00:00",
+                    "observation_time_local": "2026-06-15 18:51:00",
+                    "runway_obs": {
+                        "point_temperatures": [
+                            {
+                                "runway": "02L/20R",
+                                "tdz_temp": 28.4,
+                                "end_temp": 28.2,
+                                "target_runway_max": 28.4,
+                            },
+                            {
+                                "runway": "02R/20L",
+                                "tdz_temp": 28.1,
+                                "end_temp": 28.0,
+                                "target_runway_max": 28.1,
+                            },
+                        ],
+                    },
+                },
+            }
+
+    stale_payload = {
+        "name": "chengdu",
+        "temp_symbol": "°C",
+        "runway_plate_history": {
+            "02L/20R": [{"time": "2026-06-14T10:44:00+00:00", "temp": 31.6}],
+        },
+        "amos": {
+            "source": "amsc_awos",
+            "observation_time": "2026-06-14T10:44:00+00:00",
+        },
+    }
+
+    result = overlay_latest_amsc_observation(FakeDB(), "chengdu", stale_payload)
+
+    assert result["runway_plate_history"]["02L/20R"][-1] == {
+        "time": "2026-06-15T10:51:00+00:00",
+        "temp": 28.4,
+    }
+    assert result["runway_plate_history"]["02R/20L"][-1] == {
+        "time": "2026-06-15T10:51:00+00:00",
+        "temp": 28.1,
+    }
+
+
 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

@@ -222,6 +222,66 @@ def test_observation_collector_aligns_amsc_payload_time_with_record_observed_at(
     assert latest["payload"]["observation_time_local"] == "2026-06-14 23:43:00"
 
 
+def test_observation_collector_persists_amsc_runway_history(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(
+        weather=object(),
+        profiles=[],
+        observation_store=db,
+    )
+
+    record = ObservationRecord(
+        source="amsc_awos",
+        city="chengdu",
+        value=28.4,
+        observed_at="2026-06-15T10:51:00+00:00",
+        observed_at_local="2026-06-15 18:51:00",
+        station_code="ZUUU",
+        station_name="Chengdu Shuangliu",
+        runway="",
+        value_unit="c",
+        source_label="AMSC AWOS Chengdu Shuangliu (ZUUU)",
+        payload={
+            "source": "amsc_awos",
+            "icao": "ZUUU",
+            "temp_c": 28.4,
+            "runway_obs": {
+                "point_temperatures": [
+                    {
+                        "runway": "02L/20R",
+                        "tdz_temp": 28.4,
+                        "mid_temp": None,
+                        "end_temp": 28.2,
+                        "target_runway_max": 28.4,
+                        "wind_dir": 130,
+                        "wind_speed": 4.0,
+                    },
+                    {
+                        "runway": "02R/20L",
+                        "tdz_temp": 28.1,
+                        "mid_temp": None,
+                        "end_temp": 28.0,
+                        "target_runway_max": 28.1,
+                    },
+                ],
+            },
+        },
+    )
+
+    assert collector._store_raw_observations([record]) == 1
+
+    rows = db.get_runway_obs_recent("ZUUU", minutes=60)
+
+    assert [row["runway"] for row in rows] == ["02L/20R", "02R/20L"]
+    assert rows[0]["otime_utc"] == "2026-06-15T10:51:00+00:00"
+    assert rows[0]["target_runway_max"] == 28.4
+    assert rows[0]["wind_dir"] == 130
+
+
 def test_raw_observation_failure_preserves_last_success_and_increments_errors(tmp_path):
     from src.database.db_manager import DBManager
 

web/observation_collector_service.py

@@ -414,6 +414,80 @@ class ObservationCollector:
                 payload["observation_time_local"] = record.observed_at_local
         return payload
 
+    @staticmethod
+    def _float_or_none(value: Any) -> Optional[float]:
+        try:
+            if value is None or value == "":
+                return None
+            return float(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _int_or_none(value: Any) -> Optional[int]:
+        try:
+            if value is None or value == "":
+                return None
+            return int(float(value))
+        except (TypeError, ValueError):
+            return None
+
+    def _store_amsc_runway_observations(
+        self,
+        record: ObservationRecord,
+        payload: dict[str, Any],
+    ) -> None:
+        source = str(record.source or payload.get("source") or "").strip().lower()
+        if source != "amsc_awos":
+            return
+        appender = getattr(self.observation_store, "append_runway_obs", None)
+        if not callable(appender):
+            return
+        runway_obs = payload.get("runway_obs")
+        if not isinstance(runway_obs, dict):
+            return
+        point_temperatures = runway_obs.get("point_temperatures")
+        if not isinstance(point_temperatures, list) or not point_temperatures:
+            return
+        icao = str(payload.get("icao") or record.station_code or "").strip().upper()
+        obs_time = str(
+            payload.get("observation_time")
+            or payload.get("observed_at")
+            or record.observed_at
+            or ""
+        ).strip()
+        if not icao or not obs_time:
+            return
+        for point in point_temperatures:
+            if not isinstance(point, dict):
+                continue
+            runway = str(point.get("runway") or "").strip().upper()
+            if not runway:
+                continue
+            try:
+                appender(
+                    icao=icao,
+                    city=record.city,
+                    runway=runway,
+                    tdz_temp=self._float_or_none(point.get("tdz_temp")),
+                    mid_temp=self._float_or_none(point.get("mid_temp")),
+                    end_temp=self._float_or_none(point.get("end_temp")),
+                    target_runway_max=self._float_or_none(point.get("target_runway_max")),
+                    wind_dir=self._int_or_none(point.get("wind_dir")),
+                    wind_speed=self._float_or_none(point.get("wind_speed")),
+                    rvr=self._int_or_none(point.get("rvr")),
+                    mor=self._float_or_none(point.get("mor")),
+                    humidity=self._float_or_none(point.get("humidity")),
+                    otime_utc=obs_time,
+                )
+            except Exception as exc:
+                logger.debug(
+                    "AMSC runway observation write skipped city={} runway={}: {}",
+                    record.city,
+                    runway,
+                    exc,
+                )
+
     def _store_raw_observations(self, records: Sequence[ObservationRecord]) -> int:
         store = self.observation_store
         writer = getattr(store, "append_raw_observation", None)
@@ -424,6 +498,7 @@ class ObservationCollector:
         written_records: list[ObservationRecord] = []
         for record in records:
             try:
+                payload = self._raw_payload_for_record(record)
                 writer(
                     source=record.source,
                     city=record.city,
@@ -435,8 +510,9 @@ class ObservationCollector:
                     runway=record.runway,
                     value_unit=record.value_unit,
                     status="ok",
-                    payload=self._raw_payload_for_record(record),
+                    payload=payload,
                 )
+                self._store_amsc_runway_observations(record, payload)
                 written_records.append(record)
                 wrote += 1
             except Exception as exc:

web/services/latest_observation_overlay.py

@@ -191,6 +191,83 @@ def _merge_observation_block(
     return True
 
 
+def _runway_history_temp(point: dict[str, Any]) -> Optional[float]:
+    for key in ("target_runway_max", "temp", "tdz_temp", "end_temp", "mid_temp"):
+        temp = _to_float(point.get(key))
+        if temp is not None:
+            return temp
+    return None
+
+
+def _append_latest_amsc_runway_history(
+    payload: dict[str, Any],
+    row: dict[str, Any],
+    raw_payload: dict[str, Any],
+) -> bool:
+    runway_obs = raw_payload.get("runway_obs")
+    if not isinstance(runway_obs, dict):
+        return False
+    point_temperatures = runway_obs.get("point_temperatures")
+    if not isinstance(point_temperatures, list) or not point_temperatures:
+        return False
+    observed_at = str(
+        raw_payload.get("observation_time")
+        or raw_payload.get("observed_at")
+        or row.get("observed_at")
+        or ""
+    ).strip()
+    if not observed_at:
+        return False
+
+    history = payload.get("runway_plate_history")
+    if not isinstance(history, dict):
+        history = {}
+    else:
+        history = deepcopy(history)
+
+    use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper()
+    changed = False
+    for point in point_temperatures:
+        if not isinstance(point, dict):
+            continue
+        runway = str(point.get("runway") or "").strip().upper()
+        temp = _runway_history_temp(point)
+        if not runway or temp is None:
+            continue
+        if use_fahrenheit:
+            temp = temp * 9.0 / 5.0 + 32.0
+        entry = {
+            "time": observed_at,
+            "temp": round(float(temp), 1),
+        }
+        existing = history.get(runway)
+        runway_history = list(existing) if isinstance(existing, list) else []
+        replaced = False
+        for index, current in enumerate(runway_history):
+            if not isinstance(current, dict):
+                continue
+            current_time = str(
+                current.get("time")
+                or current.get("timestamp")
+                or current.get("observed_at")
+                or ""
+            ).strip()
+            if current_time == observed_at:
+                if current != entry:
+                    runway_history[index] = entry
+                    changed = True
+                replaced = True
+                break
+        if not replaced:
+            runway_history.append(entry)
+            changed = True
+        history[runway] = runway_history
+
+    if changed:
+        payload["runway_plate_history"] = history
+    return changed
+
+
 def overlay_latest_amsc_observation(
     db: Any,
     city: str,
@@ -220,6 +297,8 @@ 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_latest_amsc_runway_history(next_payload, row, raw_payload) or changed
+
     canonical = next_payload.get("canonical_temperature")
     canonical_epoch = _block_epoch(canonical)
     if canonical_epoch is None or raw_epoch > canonical_epoch: