From 24f2a82893c338bdf53be7c87185e39b32161c70 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Thu, 21 May 2026 19:50:54 +0800 Subject: [PATCH] feat: fetch and cache hourly multi-model forecast data in SQLite for all cities --- src/data_collection/nws_open_meteo_sources.py | 119 +++++++++++++++++- src/data_collection/weather_sources.py | 1 + tests/test_multi_model_sources.py | 61 +++++++++ 3 files changed, 177 insertions(+), 4 deletions(-) diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index 957970bc..1ad8f160 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -139,6 +139,45 @@ def _parse_open_meteo_multi_model_daily( return [str(d) for d in dates], daily_forecasts, model_metadata, model_keys +def _parse_open_meteo_multi_model_hourly( + hourly: Dict[str, Any], +) -> tuple[list[str], Dict[str, list[Optional[float]]]]: + times = hourly.get("time", []) + if not isinstance(times, list): + times = [] + + hourly_times = [str(t) for t in times] + hourly_forecasts: Dict[str, list[Optional[float]]] = {} + + for model_key, spec in OPEN_METEO_MULTI_MODEL_SPECS.items(): + label = str(spec["label"]) + key = f"temperature_2m_{model_key}" + values = hourly.get(key, []) + if not isinstance(values, list) or not values: + continue + + parsed_values = [] + has_valid = False + for v in values: + if v is None: + parsed_values.append(None) + else: + try: + parsed_values.append(round(float(v), 1)) + has_valid = True + except (TypeError, ValueError): + parsed_values.append(None) + + if has_valid: + if len(parsed_values) < len(hourly_times): + parsed_values.extend([None] * (len(hourly_times) - len(parsed_values))) + elif len(parsed_values) > len(hourly_times): + parsed_values = parsed_values[:len(hourly_times)] + hourly_forecasts[label] = parsed_values + + return hourly_times, hourly_forecasts + + def _count_models(values: Any) -> int: if not isinstance(values, dict): return 0 @@ -152,6 +191,38 @@ def _count_models(values: Any) -> int: return count +def _get_hourly_by_time( + times: list[str], + forecasts: Dict[str, list[Optional[float]]], +) -> Dict[str, Dict[str, float]]: + hourly_by_time: Dict[str, Dict[str, float]] = {} + for i, t_str in enumerate(times): + hour_data: Dict[str, float] = {} + for label, values in forecasts.items(): + if i < len(values) and values[i] is not None: + hour_data[label] = values[i] + if hour_data: + hourly_by_time[t_str] = hour_data + return hourly_by_time + + +def _reconstruct_hourly_forecasts( + merged_times: list[str], + merged_hourly_by_time: Dict[str, Dict[str, float]], +) -> Dict[str, list[Optional[float]]]: + labels = set() + for hour_data in merged_hourly_by_time.values(): + labels.update(hour_data.keys()) + + hourly_forecasts: Dict[str, list[Optional[float]]] = {} + for label in sorted(labels): + values = [] + for t_str in merged_times: + values.append(merged_hourly_by_time[t_str].get(label)) + hourly_forecasts[label] = values + return hourly_forecasts + + def _merge_multi_model_result_with_cache( cached: Dict[str, Any], fresh: Dict[str, Any], @@ -173,21 +244,51 @@ def _merge_multi_model_result_with_cache( merged_daily[str(date)] = ( dict(fresh_day) if _count_models(fresh_day) >= _count_models(cached_day) - else {**dict(fresh_day), **dict(cached_day)} + else {**dict(cached_day), **dict(fresh_day)} ) + # Merge hourly forecasts + cached_hourly_times = cached.get("hourly_times") or [] + cached_hourly_forecasts = cached.get("hourly_forecasts") or {} + fresh_hourly_times = fresh.get("hourly_times") or [] + fresh_hourly_forecasts = fresh.get("hourly_forecasts") or {} + + cached_hourly_by_time = _get_hourly_by_time(cached_hourly_times, cached_hourly_forecasts) + fresh_hourly_by_time = _get_hourly_by_time(fresh_hourly_times, fresh_hourly_forecasts) + + merged_hourly_by_time = {} + all_hourly_times = sorted({*cached_hourly_by_time.keys(), *fresh_hourly_by_time.keys()}) + + # Cutoff to keep cache size bounded: discard times before fresh starts + if fresh_hourly_times: + cutoff = fresh_hourly_times[0] + all_hourly_times = [t for t in all_hourly_times if t >= cutoff] + + for t_str in all_hourly_times: + c_hour = cached_hourly_by_time.get(t_str) or {} + f_hour = fresh_hourly_by_time.get(t_str) or {} + merged_hourly_by_time[t_str] = ( + dict(f_hour) + if _count_models(f_hour) >= _count_models(c_hour) + else {**dict(c_hour), **dict(f_hour)} + ) + + merged_hourly_forecasts = _reconstruct_hourly_forecasts(all_hourly_times, merged_hourly_by_time) + merged = { **cached, **fresh, - "forecasts": {**dict(fresh_forecasts), **dict(cached_forecasts)}, + "forecasts": {**dict(cached_forecasts), **dict(fresh_forecasts)}, "daily_forecasts": merged_daily or cached_daily or fresh_daily, + "hourly_times": all_hourly_times, + "hourly_forecasts": merged_hourly_forecasts, "model_metadata": { - **(fresh.get("model_metadata") if isinstance(fresh.get("model_metadata"), dict) else {}), **(cached.get("model_metadata") if isinstance(cached.get("model_metadata"), dict) else {}), + **(fresh.get("model_metadata") if isinstance(fresh.get("model_metadata"), dict) else {}), }, "model_keys": { - **(fresh.get("model_keys") if isinstance(fresh.get("model_keys"), dict) else {}), **(cached.get("model_keys") if isinstance(cached.get("model_keys"), dict) else {}), + **(fresh.get("model_keys") if isinstance(fresh.get("model_keys"), dict) else {}), }, "partial_refresh": True, "partial_refresh_model_count": _count_models(fresh_forecasts), @@ -728,6 +829,7 @@ class NwsOpenMeteoSourceMixin: "latitude": lat, "longitude": lon, "daily": "temperature_2m_max", + "hourly": "temperature_2m", "models": models, "timezone": "auto", "forecast_days": 3, @@ -751,6 +853,13 @@ class NwsOpenMeteoSourceMixin: _parse_open_meteo_multi_model_daily(daily) ) + hourly = data.get("hourly", {}) + if not isinstance(hourly, dict): + hourly = {} + hourly_times, hourly_forecasts = ( + _parse_open_meteo_multi_model_hourly(hourly) + ) + if not daily_forecasts: logger.warning("Multi-model: 无有效模型数据") record_source_call("open_meteo", "multi_model", "empty", (time.perf_counter() - started) * 1000.0) @@ -770,6 +879,8 @@ class NwsOpenMeteoSourceMixin: "provider": "open-meteo", "forecasts": forecasts, # 今天 {"ECMWF": 12.3, "GFS": 11.8, ...} (向后兼容) "daily_forecasts": daily_forecasts, # 按天 {"2026-02-23": {...}, "2026-02-24": {...}} + "hourly_times": hourly_times, + "hourly_forecasts": hourly_forecasts, "model_metadata": model_metadata, "model_keys": model_keys, "dates": dates, diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index dfbef895..b46f7f9e 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -156,6 +156,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self.session = httpx.Client( timeout=self.timeout, follow_redirects=True, + trust_env=False, headers={"User-Agent": self.user_agent}, limits=httpx.Limits(max_connections=50, max_keepalive_connections=20), ) diff --git a/tests/test_multi_model_sources.py b/tests/test_multi_model_sources.py index 5497a546..a2852fba 100644 --- a/tests/test_multi_model_sources.py +++ b/tests/test_multi_model_sources.py @@ -158,3 +158,64 @@ def test_persisted_open_meteo_cooldown_skips_outbound_request(monkeypatch, tmp_p result = collector.fetch_multi_model(40.1281, 32.9951, city="ankara") assert result is None + + +def test_multi_model_hourly_parser(): + from src.data_collection.nws_open_meteo_sources import _parse_open_meteo_multi_model_hourly + + hourly = { + "time": ["2026-05-21T00:00", "2026-05-21T01:00"], + "temperature_2m_ecmwf_ifs025": [15.2, 15.0], + "temperature_2m_gfs_seamless": [14.8, None], + "temperature_2m_icon_d2": [None, None], # completely empty, should be ignored + } + + times, forecasts = _parse_open_meteo_multi_model_hourly(hourly) + + assert times == ["2026-05-21T00:00", "2026-05-21T01:00"] + assert forecasts["ECMWF"] == [15.2, 15.0] + assert forecasts["GFS"] == [14.8, None] + assert "ICON-D2" not in forecasts + + +def test_merge_multi_model_result_with_cache_hourly(): + from src.data_collection.nws_open_meteo_sources import _merge_multi_model_result_with_cache + + cached = { + "forecasts": {"ECMWF": 20.0, "GFS": 19.5, "ICON-EU": 19.8}, + "daily_forecasts": { + "2026-05-21": {"ECMWF": 20.0, "GFS": 19.5, "ICON-EU": 19.8} + }, + "hourly_times": ["2026-05-21T00:00", "2026-05-21T01:00"], + "hourly_forecasts": { + "ECMWF": [15.2, 15.0], + "GFS": [14.8, 14.5], + "ICON-EU": [15.0, 14.8], + }, + } + + fresh = { + "forecasts": {"ECMWF": 20.2}, # Only ECMWF returned (e.g. subset refresh) + "daily_forecasts": { + "2026-05-21": {"ECMWF": 20.2} + }, + "hourly_times": ["2026-05-21T01:00", "2026-05-21T02:00"], # time shifts forward + "hourly_forecasts": { + "ECMWF": [15.1, 15.3], + }, + } + + merged = _merge_multi_model_result_with_cache(cached, fresh) + + # Standard checks + assert merged["forecasts"]["ECMWF"] == 20.2 + assert merged["forecasts"]["GFS"] == 19.5 + assert merged["forecasts"]["ICON-EU"] == 19.8 + + # Hourly merge and alignment checks + assert merged["hourly_times"] == ["2026-05-21T01:00", "2026-05-21T02:00"] + # At T01:00, fresh has ECMWF=15.1, cached has GFS=14.5, ICON-EU=14.8 + assert merged["hourly_forecasts"]["ECMWF"] == [15.1, 15.3] + assert merged["hourly_forecasts"]["GFS"] == [14.5, None] + assert merged["hourly_forecasts"]["ICON-EU"] == [14.8, None] +