diff --git a/src/data_collection/forecast_source_bundle.py b/src/data_collection/forecast_source_bundle.py new file mode 100644 index 00000000..03589dd6 --- /dev/null +++ b/src/data_collection/forecast_source_bundle.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +from typing import Any, Dict + + +def _open_meteo_cache_key( + lat: float, + lon: float, + *, + forecast_days: int = 14, + use_fahrenheit: bool = False, +) -> str: + return ( + f"{round(float(lat), 4)}:{round(float(lon), 4)}:" + f"{forecast_days}:{'f' if use_fahrenheit else 'c'}" + ) + + +def _multi_model_cache_key( + collector: Any, + city: str, + lat: float, + lon: float, + *, + use_fahrenheit: bool = False, +) -> str: + cache_city = str(city or "").strip().lower() + return ( + f"{round(float(lat), 4)}:{round(float(lon), 4)}:{cache_city}:" + f"{'f' if use_fahrenheit else 'c'}:{collector.multi_model_cache_version}" + ) + + +def _read_open_meteo_bundle_from_cache( + collector: Any, + *, + city: str, + lat: float, + lon: float, + use_fahrenheit: bool, + include_multi_model: bool, +) -> Dict[str, Any]: + collector._maybe_reload_open_meteo_disk_cache() + results: Dict[str, Any] = {} + om_key = _open_meteo_cache_key(lat, lon, use_fahrenheit=use_fahrenheit) + with collector._open_meteo_cache_lock: + om_cached = collector._open_meteo_cache.get(om_key) + + if not om_cached or not isinstance(om_cached.get("data"), dict): + return results + + results["open-meteo"] = dict(om_cached["data"]) + if include_multi_model: + mm_key = _multi_model_cache_key( + collector, + city, + lat, + lon, + use_fahrenheit=use_fahrenheit, + ) + with collector._multi_model_cache_lock: + mm_cached = collector._multi_model_cache.get(mm_key) + if mm_cached and isinstance(mm_cached.get("data"), dict): + results["multi_model"] = dict(mm_cached["data"]) + return results + + +def fetch_open_meteo_forecast_bundle( + collector: Any, + *, + city: str, + lat: float, + lon: float, + use_fahrenheit: bool, + include_multi_model: bool = True, + cache_only: bool = False, +) -> Dict[str, Any]: + """Fetch non-high-frequency Open-Meteo forecast payloads for a city. + + Observation-only refreshes can set *cache_only* so high-frequency callers + reuse existing model data without making outbound Open-Meteo requests. + """ + if lat is None or lon is None: + return {} + + if cache_only: + return _read_open_meteo_bundle_from_cache( + collector, + city=city, + lat=lat, + lon=lon, + use_fahrenheit=use_fahrenheit, + include_multi_model=include_multi_model, + ) + + results: Dict[str, Any] = {} + # Populate the richer multi-model cache before the regular forecast + # endpoint can trip the shared Open-Meteo cooldown. + if include_multi_model: + multi_model_data = collector.fetch_multi_model( + lat, + lon, + city=city, + use_fahrenheit=use_fahrenheit, + ) + if multi_model_data: + results["multi_model"] = multi_model_data + + open_meteo = collector.fetch_from_open_meteo( + lat, + lon, + use_fahrenheit=use_fahrenheit, + ) + if open_meteo: + results["open-meteo"] = open_meteo + + return results + + +def ensure_multi_model_hourly_payload( + collector: Any, + current: Any, + *, + city: str, + lat: float, + lon: float, + use_fahrenheit: bool, +) -> Dict[str, Any]: + """Return a multi-model payload with hourly curves when available.""" + current_payload = current if isinstance(current, dict) else {} + if current_payload.get("hourly_times"): + return current_payload + + hourly_payload = collector.fetch_multi_model( + lat, + lon, + city=city, + use_fahrenheit=use_fahrenheit, + ) + if hourly_payload and hourly_payload.get("hourly_times"): + return {**current_payload, **hourly_payload} + return current_payload diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 4ba54098..d60ef7f4 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -31,6 +31,7 @@ from src.data_collection.ncm_sources import NcmSourceMixin from src.data_collection.aeroweb_sources import AerowebSourceMixin from src.data_collection.wunderground_sources import WundergroundHistoricalMixin from src.data_collection.city_time import get_city_utc_offset_seconds +from src.data_collection.forecast_source_bundle import fetch_open_meteo_forecast_bundle from src.database.db_manager import DBManager @@ -1794,48 +1795,17 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_wunderground_historical(results, city_lower, use_fahrenheit) if lat and lon: - # When force_refresh_observations_only is set by an explicit - # observation refresh caller, skip the OM fetch entirely if cached - # data exists. Stale model data is fine; the caller only needs - # fresh METAR / AMOS observations. - om_from_cache_only = force_refresh_observations_only - if om_from_cache_only: - self._maybe_reload_open_meteo_disk_cache() - base = f"{round(float(lat), 4)}:{round(float(lon), 4)}" - unit = "f" if use_fahrenheit else "c" - cache_city = city_lower - om_key = f"{base}:14:{unit}" - with self._open_meteo_cache_lock: - om_cached = self._open_meteo_cache.get(om_key) - if om_cached and isinstance(om_cached.get("data"), dict): - open_meteo = dict(om_cached["data"]) - if include_multi_model: - mm_key = f"{base}:{cache_city}:{unit}:{self.multi_model_cache_version}" - with self._multi_model_cache_lock: - mm_cached = self._multi_model_cache.get(mm_key) - if mm_cached and isinstance(mm_cached.get("data"), dict): - results["multi_model"] = dict(mm_cached["data"]) - else: - open_meteo = None - else: - # Prioritize the model cluster before the regular Open-Meteo - # forecast. The regular forecast endpoint can set the shared - # Open-Meteo 429 cooldown; if that happens first, cities with no - # existing multi-model cache (notably Ankara after a deploy) fall - # back to a single Open-Meteo/DEB line and the decision card loses - # most of its model support. Fetching the multi-model payload - # first gives the richer, longer-lived model cache the first chance - # to populate; the regular forecast can still use its stale cache - # if Open-Meteo rate-limits the cycle. - if include_multi_model: - multi_model_data = self.fetch_multi_model( - lat, lon, city=city, use_fahrenheit=use_fahrenheit - ) - if multi_model_data: - results["multi_model"] = multi_model_data - open_meteo = self.fetch_from_open_meteo( - lat, lon, use_fahrenheit=use_fahrenheit - ) + forecast_bundle = fetch_open_meteo_forecast_bundle( + self, + city=city_lower, + lat=lat, + lon=lon, + use_fahrenheit=use_fahrenheit, + include_multi_model=include_multi_model, + cache_only=force_refresh_observations_only, + ) + results.update(forecast_bundle) + open_meteo = forecast_bundle.get("open-meteo") if open_meteo: results["open-meteo"] = open_meteo # 获取时区偏移以过滤 METAR diff --git a/tests/test_multi_model_sources.py b/tests/test_multi_model_sources.py index 3cc9acb8..bac17685 100644 --- a/tests/test_multi_model_sources.py +++ b/tests/test_multi_model_sources.py @@ -2,7 +2,9 @@ from src.data_collection.nws_open_meteo_sources import ( OPEN_METEO_MULTI_MODEL_ORDER, _parse_open_meteo_multi_model_daily, ) +from src.data_collection.forecast_source_bundle import ensure_multi_model_hourly_payload import src.data_collection.open_meteo_cache as open_meteo_cache_module +import src.data_collection.weather_sources as weather_sources_module from src.data_collection.weather_sources import WeatherDataCollector from src.database.runtime_state import ( OpenMeteoCacheRepository, @@ -195,6 +197,138 @@ def test_fetch_all_sources_prioritizes_multi_model_before_forecast(monkeypatch, assert result["multi_model"]["forecasts"]["ECMWF"] == 24.0 +def test_fetch_all_sources_delegates_non_hf_forecast_bundle(monkeypatch, tmp_path): + monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) + collector = WeatherDataCollector({}) + calls = [] + + monkeypatch.setattr(collector, "_log_temperature_unit", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_evict_city_caches", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_settlement_sources", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_wunderground_historical", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_supports_aviationweather", lambda city: False) + monkeypatch.setattr(collector, "_attach_turkish_mgm_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_korean_amos_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_china_amsc_awos_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_madis_hfmetar_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_singapore_mss_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_israel_ims_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_saudi_ncm_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_paris_aeroweb_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_china_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_japan_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_fmi_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_knmi_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_cowin_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_hko_obs_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_cwa_settlement_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_global_nearby_cluster", lambda *args, **kwargs: None) + + def fake_forecast_bundle(collector_arg, **kwargs): + calls.append({"collector": collector_arg, **kwargs}) + return { + "open-meteo": { + "utc_offset": 10800, + "daily": {"temperature_2m_max": [24.0]}, + }, + "multi_model": {"forecasts": {"ECMWF": 24.5}}, + } + + monkeypatch.setattr( + weather_sources_module, + "fetch_open_meteo_forecast_bundle", + fake_forecast_bundle, + ) + + result = collector.fetch_all_sources( + "ankara", + lat=40.1281, + lon=32.9951, + force_refresh_observations_only=True, + include_ensemble=False, + include_nearby=False, + include_taf=False, + include_mgm=False, + ) + + assert calls == [ + { + "collector": collector, + "city": "ankara", + "lat": 40.1281, + "lon": 32.9951, + "use_fahrenheit": False, + "include_multi_model": True, + "cache_only": True, + } + ] + assert result["open-meteo"]["utc_offset"] == 10800 + assert result["multi_model"]["forecasts"]["ECMWF"] == 24.5 + + +def test_ensure_multi_model_hourly_payload_fetches_missing_hourly_outside_analysis_layer(): + calls = [] + + class FakeCollector: + def fetch_multi_model(self, lat, lon, *, city, use_fahrenheit): + calls.append( + { + "lat": lat, + "lon": lon, + "city": city, + "use_fahrenheit": use_fahrenheit, + } + ) + return { + "hourly_times": ["2026-06-14T10:00"], + "hourly_forecasts": {"ECMWF": [24.1]}, + "forecasts": {"ECMWF": 26.0}, + } + + result = ensure_multi_model_hourly_payload( + FakeCollector(), + {"forecasts": {"GFS": 25.0}}, + city="ankara", + lat=40.1281, + lon=32.9951, + use_fahrenheit=False, + ) + + assert calls == [ + { + "lat": 40.1281, + "lon": 32.9951, + "city": "ankara", + "use_fahrenheit": False, + } + ] + assert result["forecasts"] == {"ECMWF": 26.0} + assert result["hourly_times"] == ["2026-06-14T10:00"] + assert result["hourly_forecasts"]["ECMWF"] == [24.1] + + +def test_ensure_multi_model_hourly_payload_reuses_existing_hourly(): + class FakeCollector: + def fetch_multi_model(self, *_args, **_kwargs): + raise AssertionError("existing hourly payload should not fetch again") + + result = ensure_multi_model_hourly_payload( + FakeCollector(), + { + "forecasts": {"GFS": 25.0}, + "hourly_times": ["2026-06-14T10:00"], + "hourly_forecasts": {"GFS": [24.0]}, + }, + city="ankara", + lat=40.1281, + lon=32.9951, + use_fahrenheit=False, + ) + + assert result["forecasts"]["GFS"] == 25.0 + assert result["hourly_forecasts"]["GFS"] == [24.0] + + def test_force_refresh_preserves_open_meteo_model_caches_by_default(monkeypatch, tmp_path): monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) collector = WeatherDataCollector({}) diff --git a/web/analysis_service.py b/web/analysis_service.py index f0ba30a6..b36f54ea 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -35,6 +35,7 @@ from src.analysis.trend_engine import _resolve_peak_hours from src.data_collection.country_networks import build_country_network_snapshot from src.data_collection.city_registry import ALIASES, CITY_REGISTRY from src.data_collection.city_time import get_city_utc_offset_seconds +from src.data_collection.forecast_source_bundle import ensure_multi_model_hourly_payload from src.database.runtime_state import IntradayPathSnapshotRepository from web.services.city_payloads import ( build_city_chart_detail_payload as _city_chart_payload_detail, @@ -835,10 +836,14 @@ def _analyze( ens_raw = {} if not isinstance(mm, dict): mm = {} - if not mm.get("hourly_times"): - mm_hourly = _weather.fetch_multi_model(lat, lon, city=city, use_fahrenheit=is_f) - if mm_hourly and mm_hourly.get("hourly_times"): - mm = {**mm, **mm_hourly} + mm = ensure_multi_model_hourly_payload( + _weather, + mm, + city=city, + lat=lat, + lon=lon, + use_fahrenheit=is_f, + ) raw["multi_model"] = mm risk = CITY_RISK_PROFILES.get(city, {}) network_snapshot = (