gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复 Open-Meteo 冷却期无限循环导致多模型数据缺失

Browse Source 
- fetch_all_sources 新增 om_from_cache_only 模式：force_refresh_observations_only
  时直接从内存缓存取 OM 数据，不发起 HTTP 请求。缓存未命中则跳过，
  避免机场推送 60s 周期持续触发 429 限流
- nws_open_meteo_sources 三类 fetch 函数冷却期加磁盘缓存兜底：
  内存未命中时 force-reload SQLite 磁盘缓存再查一次
- 预热停止后冷却期自然过期（900s），下一次正常请求会填充缓存
This commit is contained in:
2569718930@qq.com
2026-05-17 22:31:10 +08:00
parent 415f03466d
commit 22e2409e13
4 changed files with 98 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_collection/nws_open_meteo_sources.py
+20 -1
View File
@@ -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
src/data_collection/weather_sources.py
+41 -17
View File
@@ -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