From 9cb794655b52da425df354d43128d518d579836d Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 19:51:34 +0800 Subject: [PATCH] Emit realtime events from canonical updates --- tests/test_canonical_engine.py | 31 ++++++++++++++ tests/test_observation_collector.py | 63 ++++++++++++++++++++++++++++ web/observation_collector_service.py | 39 ++++++++++++++++- web/realtime_patch_schema.py | 8 ++++ web/services/canonical_engine.py | 42 +++++++++++++++++++ 5 files changed, 181 insertions(+), 2 deletions(-) diff --git a/tests/test_canonical_engine.py b/tests/test_canonical_engine.py index ba7637e8..cba4a34d 100644 --- a/tests/test_canonical_engine.py +++ b/tests/test_canonical_engine.py @@ -81,3 +81,34 @@ def test_canonical_engine_ignores_failed_latest_rows(): assert canonical is not None assert canonical["source"] == "madis_hfmetar" assert canonical["value"] == 24.0 + + +def test_canonical_engine_builds_realtime_event_from_canonical(): + from web.services.canonical_engine import build_realtime_event_from_canonical + + event = build_realtime_event_from_canonical( + { + "city": "shenzhen", + "value": 28.1, + "temp_symbol": "°C", + "source": "hko_obs", + "station_code": "LFS", + "station_name": "Lau Fau Shan", + "observed_at": "2026-06-14T01:00:00+00:00", + "freshness_sec": 300, + "freshness_status": "fresh", + "confidence": 0.9, + } + ) + + assert event is not None + assert event["type"] == "city_observation_patch.v1" + assert event["city"] == "shenzhen" + assert event["source"] == "hko_obs" + assert event["obs_time"] == "2026-06-14T01:00:00Z" + assert event["payload"]["temp"] == 28.1 + assert event["payload"]["station_code"] == "LFS" + assert event["payload"]["station_label"] == "Lau Fau Shan" + assert event["payload"]["unit"] == "celsius" + assert event["payload"]["freshness_status"] == "fresh" + assert event["payload"]["confidence"] == 0.9 diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index a2f5af3e..15fc9cd8 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -460,6 +460,69 @@ def test_observation_collector_recomputes_canonical_from_raw_latest_settlement_s assert canonical["payload"]["value"] == 28.1 +def test_observation_collector_appends_realtime_event_after_canonical_refresh(tmp_path): + from src.database.db_manager import DBManager + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + db = DBManager(str(tmp_path / "polyweather.db")) + appended = [] + broadcasted = [] + + class FakeEventStore: + uses_external_live_fanout = False + + def append_event(self, event): + appended.append(event) + return {**event, "revision": 7} + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_hko_obs_official_nearby(self, results, city, use_fahrenheit): + results["hko_obs_nearby"] = [ + { + "source": "hko_obs", + "source_label": "HKO", + "temperature_c": 28.1, + "observation_time": "2026-06-14T01:00:00+00:00", + "station_code": "LFS", + "station_name": "Lau Fau Shan", + }, + { + "source": "hko_obs", + "source_label": "HKO", + "temperature_c": 27.6, + "observation_time": "2026-06-14T01:00:00+00:00", + "station_code": "HKO", + "station_name": "Hong Kong Observatory", + }, + ] + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("hko_obs", ("shenzhen",), 600)], + observation_store=db, + realtime_event_store=FakeEventStore(), + realtime_broadcaster=lambda event: broadcasted.append(event), + async_cache_refresh=False, + ) + + assert collector.run_due_once(now_ts=1000.0) == 1 + + assert len(appended) == 1 + event = appended[0] + assert event["type"] == "city_observation_patch.v1" + assert event["city"] == "shenzhen" + assert event["source"] == "hko_obs" + assert event["payload"]["temp"] == 28.1 + assert event["payload"]["station_code"] == "LFS" + assert broadcasted == [{**event, "revision": 7}] + + def test_observation_collector_records_no_results_source_health(tmp_path): from src.database.db_manager import DBManager from web.observation_collector_service import ( diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index e2aff0ff..ab1cb993 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -18,13 +18,17 @@ 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_engine import refresh_canonical_temperature_from_latest +from web.services.canonical_engine import ( + build_realtime_event_from_canonical, + refresh_canonical_temperature_from_latest, +) from web.services.canonical_temperature import build_canonical_temperature from web.services.observation_freshness import build_observation_freshness from web.services.observation_source_adapters import ( ObservationRecord, collect_observation_source, ) +from web.realtime_event_store_factory import create_realtime_event_store def _env_bool(name: str, default: bool) -> bool: @@ -64,6 +68,8 @@ class ObservationCollector: cache_refresher: Optional[Callable[[str], Any]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, observation_store: Optional[Any] = None, + realtime_event_store: Optional[Any] = None, + realtime_broadcaster: Optional[Callable[[dict[str, Any]], Any]] = None, async_cache_refresh: Optional[bool] = None, cache_refresh_workers: Optional[int] = None, ) -> None: @@ -72,6 +78,8 @@ class ObservationCollector: self.cache_refresher = cache_refresher self.status_recorder = status_recorder self.observation_store = observation_store + self.realtime_event_store = realtime_event_store + self.realtime_broadcaster = realtime_broadcaster self._last_run_ts: dict[tuple[str, str], float] = {} self._lock = threading.Lock() self._cache_refresh_lock = threading.Lock() @@ -332,6 +340,7 @@ class ObservationCollector: return try: setter(record.city, canonical) + self._append_realtime_event(canonical) except Exception as exc: logger.debug( "canonical temperature write skipped source={} city={}: {}", @@ -340,6 +349,26 @@ class ObservationCollector: exc, ) + def _append_realtime_event(self, canonical: dict[str, Any]) -> None: + store = self.realtime_event_store + appender = getattr(store, "append_event", None) + if not callable(appender): + return + event = build_realtime_event_from_canonical(canonical) + if not event: + return + try: + stored_event = appender(event) + except Exception as exc: + logger.debug("realtime event append skipped city={}: {}", canonical.get("city"), exc) + return + broadcaster = self.realtime_broadcaster + if callable(broadcaster) and not bool(getattr(store, "uses_external_live_fanout", False)): + try: + broadcaster(stored_event) + except Exception as exc: + logger.debug("realtime event broadcast skipped city={}: {}", canonical.get("city"), exc) + def _store_raw_observation_status( self, *, @@ -409,8 +438,10 @@ class ObservationCollector: if wrote: refreshed_cities: set[str] = set() for city in {record.city for record in written_records}: - if refresh_canonical_temperature_from_latest(self.observation_store, city): + canonical = refresh_canonical_temperature_from_latest(self.observation_store, city) + if canonical: refreshed_cities.add(city) + self._append_realtime_event(canonical) for record in written_records: if record.city not in refreshed_cities: self._store_canonical_temperature_from_observation( @@ -508,6 +539,8 @@ def start_observation_collector_loop( profiles: Optional[Sequence[ObservationSourceProfile]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, observation_store: Optional[Any] = None, + realtime_event_store: Optional[Any] = None, + realtime_broadcaster: Optional[Callable[[dict[str, Any]], Any]] = None, ) -> Optional[threading.Thread]: if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True): return None @@ -523,6 +556,8 @@ def start_observation_collector_loop( cache_refresher=cache_refresher, status_recorder=status_recorder or ObservationCollectorStatusRepository(), observation_store=observation_store or DBManager(), + realtime_event_store=realtime_event_store or create_realtime_event_store(), + realtime_broadcaster=realtime_broadcaster, ) global _COLLECTOR_THREAD diff --git a/web/realtime_patch_schema.py b/web/realtime_patch_schema.py index 00dcdb55..d0bf5550 100644 --- a/web/realtime_patch_schema.py +++ b/web/realtime_patch_schema.py @@ -240,6 +240,14 @@ def _payload_from_v1(raw_payload: Any) -> Dict[str, Any]: payload["runway_points"] = runway_points if isinstance(raw_payload.get("hourly"), dict): payload["hourly"] = raw_payload["hourly"] + for key in ("freshness_status", "source_role"): + value = raw_payload.get(key) + if isinstance(value, str) and value.strip(): + payload[key] = value.strip() + for key in ("freshness_sec", "confidence"): + value = _finite_number(raw_payload.get(key)) + if value is not None: + payload[key] = int(value) if key == "freshness_sec" else value return payload diff --git a/web/services/canonical_engine.py b/web/services/canonical_engine.py index 792e5192..20813550 100644 --- a/web/services/canonical_engine.py +++ b/web/services/canonical_engine.py @@ -8,6 +8,7 @@ from src.data_collection.city_registry import CITY_REGISTRY 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 +from web.realtime_patch_schema import normalize_observation_patch _SETTLEMENT_SOURCE_ADAPTERS = { @@ -158,6 +159,47 @@ def build_canonical_temperature_from_observations( return candidates[0][1] +def build_realtime_event_from_canonical(canonical: dict[str, Any]) -> Optional[dict[str, Any]]: + if not isinstance(canonical, dict): + return None + try: + value = float(canonical.get("value")) + except (TypeError, ValueError): + return None + city = _normalized_city(canonical.get("city")) + source = _normalized_source(canonical.get("source")) + if not city or not source: + return None + unit = "fahrenheit" if str(canonical.get("temp_symbol") or "").upper().endswith("F") else "celsius" + payload: dict[str, Any] = { + "temp": value, + "unit": unit, + } + station_code = str(canonical.get("station_code") or "").strip() + if station_code: + payload["station_code"] = station_code + station_label = str(canonical.get("station_name") or canonical.get("source_label") or "").strip() + if station_label: + payload["station_label"] = station_label + for key in ("freshness_status", "source_role"): + text = str(canonical.get(key) or "").strip() + if text: + payload[key] = text + for key in ("freshness_sec", "confidence"): + value_meta = canonical.get(key) + if value_meta is not None and value_meta != "": + payload[key] = value_meta + return normalize_observation_patch( + { + "type": "city_observation_patch.v1", + "city": city, + "source": source, + "obs_time": canonical.get("observed_at"), + "payload": payload, + } + ) + + def refresh_canonical_temperature_from_latest(db: Any, city: str) -> Optional[dict[str, Any]]: getter = getattr(db, "list_latest_raw_observations_for_city", None) setter = getattr(db, "set_canonical_temperature", None)