diff --git a/src/database/db_manager.py b/src/database/db_manager.py index ee1dfbee..177ae264 100644 --- a/src/database/db_manager.py +++ b/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, *, diff --git a/tests/test_latest_observation_overlay.py b/tests/test_latest_observation_overlay.py index 41ccbc90..5cae3e05 100644 --- a/tests/test_latest_observation_overlay.py +++ b/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): diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index fa45b3bb..bd9718fe 100644 --- a/tests/test_observation_collector.py +++ b/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 diff --git a/web/services/latest_observation_overlay.py b/web/services/latest_observation_overlay.py index f6beed24..f6f72577 100644 --- a/web/services/latest_observation_overlay.py +++ b/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")