diff --git a/src/data_collection/forecast_source_bundle.py b/src/data_collection/forecast_source_bundle.py index 5dac8621..a870ddc0 100644 --- a/src/data_collection/forecast_source_bundle.py +++ b/src/data_collection/forecast_source_bundle.py @@ -2,7 +2,10 @@ from __future__ import annotations from typing import Any, Dict -from src.data_collection.multi_model_freshness import multi_model_has_current_window +from src.data_collection.multi_model_freshness import ( + multi_model_has_current_window, + open_meteo_forecast_has_current_window, +) def _open_meteo_cache_key( @@ -48,8 +51,9 @@ def _read_open_meteo_bundle_from_cache( with collector._open_meteo_cache_lock: om_cached = collector._open_meteo_cache.get(om_key) - if om_cached and isinstance(om_cached.get("data"), dict): - results["open-meteo"] = dict(om_cached["data"]) + om_data = om_cached.get("data") if isinstance(om_cached, dict) else None + if isinstance(om_data, dict) and open_meteo_forecast_has_current_window(om_data): + results["open-meteo"] = dict(om_data) if include_multi_model: mm_key = _multi_model_cache_key( collector, diff --git a/src/data_collection/multi_model_freshness.py b/src/data_collection/multi_model_freshness.py index 471b029f..34119d4b 100644 --- a/src/data_collection/multi_model_freshness.py +++ b/src/data_collection/multi_model_freshness.py @@ -63,6 +63,26 @@ def multi_model_has_current_window(multi_model: Any, *, today: Optional[date] = return max(parsed_dates) >= today +def open_meteo_forecast_has_current_window( + forecast: Any, + *, + today: Optional[date] = None, +) -> bool: + """Return False when cached Open-Meteo forecast data is definitely older than today.""" + if not isinstance(forecast, dict): + return False + today = today or datetime.now(timezone.utc).date() + raw_date_values = [] + daily = forecast.get("daily") if isinstance(forecast.get("daily"), dict) else {} + raw_date_values.extend(daily.get("time") or []) + hourly = forecast.get("hourly") if isinstance(forecast.get("hourly"), dict) else {} + raw_date_values.extend(hourly.get("time") or []) + parsed_dates = [parsed for raw in raw_date_values for parsed in [_parse_date(raw)] if parsed] + if not parsed_dates: + return True + return max(parsed_dates) >= today + + def multi_model_covers_local_date(multi_model: Any, local_date: str) -> bool: if not isinstance(multi_model, dict): return False diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index ee56aaa5..ae0f6fd0 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -6,7 +6,10 @@ from typing import Any, Dict, Optional from loguru import logger -from src.data_collection.multi_model_freshness import multi_model_has_current_window +from src.data_collection.multi_model_freshness import ( + multi_model_has_current_window, + open_meteo_forecast_has_current_window, +) from src.utils.metrics import record_source_call @@ -528,16 +531,18 @@ class NwsOpenMeteoSourceMixin: logger.debug(f"Open-Meteo 冷却期中，跳过请求，还需 {remaining}s") with self._open_meteo_cache_lock: stale = self._open_meteo_cache.get(cache_key) - if stale and isinstance(stale.get("data"), dict): + stale_data = stale.get("data") if isinstance(stale, dict) else None + if isinstance(stale_data, dict) and open_meteo_forecast_has_current_window(stale_data): record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0) - return dict(stale["data"]) + return dict(stale_data) # Memory miss: force-reload from disk and retry once self._load_open_meteo_disk_cache() with self._open_meteo_cache_lock: stale2 = self._open_meteo_cache.get(cache_key) - if stale2 and isinstance(stale2.get("data"), dict): + stale2_data = stale2.get("data") if isinstance(stale2, dict) else None + if isinstance(stale2_data, dict) and open_meteo_forecast_has_current_window(stale2_data): record_source_call("open_meteo", "forecast", "disk_fallback", (time.perf_counter() - started) * 1000.0) - return dict(stale2["data"]) + return dict(stale2_data) record_source_call("open_meteo", "forecast", "cooldown_skip", (time.perf_counter() - started) * 1000.0) return None with self._open_meteo_cache_lock: @@ -547,6 +552,11 @@ class NwsOpenMeteoSourceMixin: and now_ts - float(cached.get("t", 0)) < self.open_meteo_cache_ttl_sec ): cached_data = cached.get("data") + if isinstance(cached_data, dict): + if not open_meteo_forecast_has_current_window(cached_data): + self._open_meteo_cache.pop(cache_key, None) + record_source_call("open_meteo", "forecast", "expired_cache_skip", (time.perf_counter() - started) * 1000.0) + cached_data = None if isinstance(cached_data, dict): record_source_call("open_meteo", "forecast", "cache_hit", (time.perf_counter() - started) * 1000.0) return dict(cached_data) @@ -672,8 +682,9 @@ class NwsOpenMeteoSourceMixin: logger.error(f"Open-Meteo forecast failed: {e}") with self._open_meteo_cache_lock: stale = self._open_meteo_cache.get(cache_key) - if stale and isinstance(stale.get("data"), dict): - fallback = dict(stale["data"]) + stale_data = stale.get("data") if isinstance(stale, dict) else None + if isinstance(stale_data, dict) and open_meteo_forecast_has_current_window(stale_data): + fallback = dict(stale_data) fallback["stale_cache"] = True record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0) return fallback diff --git a/tests/test_multi_model_sources.py b/tests/test_multi_model_sources.py index 17361b92..9bf045ab 100644 --- a/tests/test_multi_model_sources.py +++ b/tests/test_multi_model_sources.py @@ -513,6 +513,65 @@ def test_fetch_multi_model_ignores_cache_when_dates_are_stale(monkeypatch, tmp_p assert result["forecasts"]["GFS"] == 38.6 +def test_fetch_open_meteo_ignores_cache_when_dates_are_stale(monkeypatch, tmp_path): + monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) + collector = WeatherDataCollector({}) + today = datetime.now(timezone.utc).date() + old_dates = [ + (today - timedelta(days=11)).isoformat(), + (today - timedelta(days=10)).isoformat(), + ] + fresh_dates = [today.isoformat(), (today + timedelta(days=1)).isoformat()] + cache_key = f"{round(float(48.9694), 4)}:{round(float(2.4414), 4)}:14:c" + collector._open_meteo_cache[cache_key] = { + "t": time.time(), + "data": { + "source": "open-meteo", + "daily": { + "time": old_dates, + "temperature_2m_max": [24.5, 26.6], + }, + "hourly": {"time": [f"{old_dates[0]}T15:00"], "temperature_2m": [24.0]}, + }, + } + calls = [] + + class FakeResponse: + def raise_for_status(self): + return None + + def json(self): + return { + "current_weather": {"temperature": 32.0}, + "utc_offset_seconds": 7200, + "timezone": "Europe/Paris", + "daily": { + "time": fresh_dates, + "temperature_2m_max": [39.2, 33.0], + "sunrise": [f"{fresh_dates[0]}T05:30", f"{fresh_dates[1]}T05:31"], + "sunset": [f"{fresh_dates[0]}T21:55", f"{fresh_dates[1]}T21:55"], + "sunshine_duration": [36000, 33000], + }, + "hourly": { + "time": [f"{fresh_dates[0]}T15:00", f"{fresh_dates[0]}T16:00"], + "temperature_2m": [38.8, 39.2], + }, + } + + monkeypatch.setattr(collector, "_wait_open_meteo_slot", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + collector, + "_http_get", + lambda *args, **kwargs: calls.append((args, kwargs)) or FakeResponse(), + ) + + result = collector.fetch_from_open_meteo(48.9694, 2.4414) + + assert len(calls) == 1 + assert result["daily"]["time"][:2] == fresh_dates + assert result["daily"]["temperature_2m_max"][0] == 39.2 + + def test_multi_model_hourly_parser(): from src.data_collection.nws_open_meteo_sources import _parse_open_meteo_multi_model_hourly