From 56f397bb2e3c6972a0d526f217126f51fc784756 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 15 Jun 2026 18:59:41 +0800 Subject: [PATCH] Persist AMSC runway history from collector --- tests/test_latest_observation_overlay.py | 59 ++++++++++++++++ tests/test_observation_collector.py | 60 ++++++++++++++++ web/observation_collector_service.py | 78 ++++++++++++++++++++- web/services/latest_observation_overlay.py | 79 ++++++++++++++++++++++ 4 files changed, 275 insertions(+), 1 deletion(-) diff --git a/tests/test_latest_observation_overlay.py b/tests/test_latest_observation_overlay.py index bc4923f7..d5ce25d2 100644 --- a/tests/test_latest_observation_overlay.py +++ b/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): diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 1e2e969a..fa45b3bb 100644 --- a/tests/test_observation_collector.py +++ b/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 diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 7902d2c9..587c08ef 100644 --- a/web/observation_collector_service.py +++ b/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: diff --git a/web/services/latest_observation_overlay.py b/web/services/latest_observation_overlay.py index de915fb1..7c225bf0 100644 --- a/web/services/latest_observation_overlay.py +++ b/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: