diff --git a/scripts/check_city_cache.py b/scripts/check_city_cache.py new file mode 100644 index 00000000..ed203948 --- /dev/null +++ b/scripts/check_city_cache.py @@ -0,0 +1,23 @@ +"""Check which cities have hourly / multi-model data in the analysis cache.""" +import urllib.request, json + +TOK = "Bearer ZLuvwSsQvHSj2geIWfX7gjn9rJU_heOlgq0FBp7cUOgXaB2eHbE20R4PKwonP2FP" +CITIES = [ + "london", "paris", "amsterdam", "helsinki", "istanbul", "ankara", + "moscow", "new york", "los angeles", "chicago", "miami", "houston", + "lagos", "jeddah", "karachi", "tel aviv", +] + +for city in CITIES: + url = f"http://localhost:8000/api/city/{city}/detail?force_refresh=false" + req = urllib.request.Request(url, headers={"Authorization": TOK}) + try: + r = urllib.request.urlopen(req, timeout=30) + d = json.loads(r.read()) + except Exception as e: + print(f"{city}: FAILED {e}") + continue + h = d.get("timeseries", {}).get("hourly", {}) + m = d.get("models", {}) + deb = d.get("deb", {}) + print(f"{city}: hourly={len(h.get('times',[]))}p models={len(m)} deb={deb.get('prediction')}") diff --git a/scripts/check_om_cache.py b/scripts/check_om_cache.py new file mode 100644 index 00000000..8382c233 --- /dev/null +++ b/scripts/check_om_cache.py @@ -0,0 +1,14 @@ +"""Check which cities have Open-Meteo cache entries in the web container.""" +import time +from src.data_collection.weather_sources import WeatherDataCollector +from src.utils.config_loader import load_config + +w = WeatherDataCollector(load_config()) +print("Multi-model cache entries:") +for k in sorted(w._multi_model_cache.keys()): + age = int(time.time() - w._multi_model_cache[k].get("t", 0)) + print(f" {k} age={age}s") + +print(f"\nOM cache entries: {len(w._open_meteo_cache)}") +print(f"Multi-model cache entries: {len(w._multi_model_cache)}") +print(f"Rate limit until: {w._open_meteo_rate_limit_until}") diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index f2f5e4d9..30c59c64 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -359,6 +359,13 @@ class NwsOpenMeteoSourceMixin: if stale and isinstance(stale.get("data"), dict): record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0) 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): + record_source_call("open_meteo", "forecast", "disk_fallback", (time.perf_counter() - started) * 1000.0) + 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: @@ -528,9 +535,15 @@ class NwsOpenMeteoSourceMixin: if stale and isinstance(stale.get("data"), dict): record_source_call("open_meteo", "ensemble", "stale_cache", (time.perf_counter() - started) * 1000.0) return dict(stale["data"]) + self._load_open_meteo_disk_cache() + with self._ensemble_cache_lock: + stale2 = self._ensemble_cache.get(cache_key) + if stale2 and isinstance(stale2.get("data"), dict): + record_source_call("open_meteo", "ensemble", "disk_fallback", (time.perf_counter() - started) * 1000.0) + return dict(stale2["data"]) record_source_call("open_meteo", "ensemble", "cooldown_skip", (time.perf_counter() - started) * 1000.0) return None - + with self._ensemble_cache_lock: cached = self._ensemble_cache.get(cache_key) if ( @@ -685,6 +698,12 @@ class NwsOpenMeteoSourceMixin: if stale and isinstance(stale.get("data"), dict): record_source_call("open_meteo", "multi_model", "stale_cache", (time.perf_counter() - started) * 1000.0) return dict(stale["data"]) + self._load_open_meteo_disk_cache() + with self._multi_model_cache_lock: + stale2 = self._multi_model_cache.get(cache_key) + if stale2 and isinstance(stale2.get("data"), dict): + record_source_call("open_meteo", "multi_model", "disk_fallback", (time.perf_counter() - started) * 1000.0) + return dict(stale2["data"]) record_source_call("open_meteo", "multi_model", "cooldown_skip", (time.perf_counter() - started) * 1000.0) return None diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index d0645e07..722bc5bd 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -1322,24 +1322,48 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_settlement_sources(results, city_lower) if lat and lon: - # 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 + # When force_refresh_observations_only is set (airport push loop), + # skip the OM fetch entirely if cached data exists — the 60 s cycle + # must not hammer the Open-Meteo API. Stale model data is fine; + # the loop 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 ) - if multi_model_data: - results["multi_model"] = multi_model_data - open_meteo = self.fetch_from_open_meteo( - lat, lon, use_fahrenheit=use_fahrenheit - ) if open_meteo: results["open-meteo"] = open_meteo # 获取时区偏移以过滤 METAR