From 9e3a4e2f45b992878f8893d2a1ed1f6f5b12ad98 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Fri, 29 May 2026 17:20:07 +0800 Subject: [PATCH] Fix WU historical current observation merge --- .../LiveTemperatureThresholdChart.tsx | 7 +- ...temperatureDefaultVisibilityPolicy.test.ts | 17 + .../scan-terminal/temperature-chart-logic.ts | 3 + frontend/lib/dashboard-types.ts | 10 + src/data_collection/weather_sources.py | 51 ++- src/data_collection/wunderground_sources.py | 405 ++++++++++++++++++ tests/test_city_payloads.py | 77 +++- tests/test_wunderground_historical.py | 211 +++++++++ web/analysis_service.py | 13 + web/scan_terminal_city_row.py | 2 + web/services/city_api.py | 20 +- web/services/city_payloads.py | 4 + web/services/city_runtime.py | 48 +++ 13 files changed, 859 insertions(+), 9 deletions(-) create mode 100644 src/data_collection/wunderground_sources.py create mode 100644 tests/test_wunderground_historical.py diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 404c1a44..4b8adb9a 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -89,6 +89,10 @@ function formatCityLocalDate(tzOffsetSeconds: number | null | undefined) { return `${y}-${m}-${d}`; } +function getWundergroundDailyHigh(hourly: HourlyForecast) { + return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null; +} + // ── Main component ───────────────────────────────────────────────────── export function LiveTemperatureThresholdChart({ @@ -490,7 +494,7 @@ export function LiveTemperatureThresholdChart({ [row, chartHourly, settlementPlate], ); const displayRunwayTemp = selectDisplayRunwayTemp(liveTemp, currentRunwayTemp, hasRunwayData); - const wundergroundDailyHigh = validNumber(chartHourly?.airportCurrent?.max_so_far ?? chartHourly?.airportPrimary?.max_so_far) ?? null; + const wundergroundDailyHigh = getWundergroundDailyHigh(chartHourly); const localDateStr = chartLocalDate || new Date().toISOString().slice(0, 10); const modelSources = (row?.model_cluster_sources && Object.keys(row.model_cluster_sources).length > 0) @@ -795,6 +799,7 @@ export const __getDebPeakWindowRangeForTest = getDebPeakWindowRange; export const __getLiveObservationLabelsForTest = getLiveObservationLabels; export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics; export const __getPeakGlowStateForTest = getPeakGlowState; +export const __getWundergroundDailyHighForTest = getWundergroundDailyHigh; export const __shouldPollLiveChartForTest = shouldPollLiveChart; export const __mergePatchIntoHourlyForTest = mergePatchIntoHourly; export const __selectDisplayRunwayTempForTest = selectDisplayRunwayTemp; diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts index a3ed4c9f..e609b572 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts @@ -5,6 +5,7 @@ import { __getLiveObservationLabelsForTest, __getObservationDisplayMetricsForTest, __getPeakGlowStateForTest, + __getWundergroundDailyHighForTest, __getVisibleTemperatureSeriesForTest, __isTemperatureSeriesVisibleByDefaultForTest, __mergePatchIntoHourlyForTest, @@ -102,6 +103,22 @@ export function runTests() { "morning observations near the intraday observed high should not trigger peak glow before the forecast hot window", ); + assert( + __getWundergroundDailyHighForTest({ + wundergroundCurrent: { max_so_far: 26 }, + airportCurrent: { max_so_far: 27 }, + airportPrimary: { max_so_far: 27 }, + } as any) === 26, + "WU display should use Wunderground historical.json high before airport/METAR highs", + ); + assert( + __getWundergroundDailyHighForTest({ + airportCurrent: { max_so_far: 27 }, + airportPrimary: { max_so_far: 27 }, + } as any) === null, + "WU display should not fall back to airport/METAR highs when historical.json is missing", + ); + const guangzhou = { city: "guangzhou", local_date: "2026-05-25", diff --git a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts index ca0c3262..c8451840 100644 --- a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts +++ b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts @@ -784,6 +784,7 @@ type HourlyForecast = { amos?: AmosData | null; airportCurrent?: AirportCurrentConditions | null; airportPrimary?: AirportCurrentConditions | null; + wundergroundCurrent?: AirportCurrentConditions | null; forecastDaily?: ForecastDay[]; multiModelDaily?: Record; probabilities?: LegacyGaussianProbabilitySource | null; @@ -809,6 +810,7 @@ function seedHourlyForecastFromRow(row: ScanOpportunityRow | null): HourlyForeca amos: null, airportCurrent: null, airportPrimary: null, + wundergroundCurrent: (row as any)?.wunderground_current || null, forecastDaily: [], multiModelDaily: {}, probabilities: { @@ -844,6 +846,7 @@ function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForec amos: json.amos || null, airportCurrent: json.airport_current || null, airportPrimary: json.airport_primary || null, + wundergroundCurrent: (json as any).wunderground_current || (json as any)?.official?.wunderground_current || null, forecastDaily: json.forecast?.daily || [], multiModelDaily: json.multi_model_daily || {}, probabilities: json.probabilities || null, diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index 8c4d0cc4..9e2da045 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -130,6 +130,13 @@ export interface AirportCurrentConditions { last_observation_local_date?: string | null; current_local_date?: string | null; freshness?: ObservationFreshness | null; + source?: string | null; + location_id?: string | null; + local_date?: string | null; + temp_symbol?: string | null; + api_units?: string | null; + observation_count?: number | null; + today_obs?: Array<{ time?: string; temp?: number | null }>; } export interface NearbyStation { @@ -242,6 +249,7 @@ export interface CitySummary { deb?: { prediction?: number | null; }; + wunderground_current?: AirportCurrentConditions; deviation_monitor?: DeviationMonitor; risk?: { level?: RiskLevel; @@ -502,6 +510,7 @@ export interface ScanOpportunityRow { temp_symbol?: string | null; current_temp?: number | null; current_max_so_far?: number | null; + wunderground_current?: AirportCurrentConditions; metar_context?: { source?: string | null; station?: string | null; @@ -730,6 +739,7 @@ export interface CityDetail { }; airport_current?: AirportCurrentConditions; airport_primary?: AirportCurrentConditions; + wunderground_current?: AirportCurrentConditions; airport_primary_today_obs?: Array<{ time?: string; temp?: number | null; diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 0b61b55c..0b7849b5 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -29,10 +29,12 @@ from src.data_collection.singapore_mss_sources import SingaporeMssSourceMixin from src.data_collection.ims_sources import ImsSourceMixin 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.database.db_manager import DBManager -class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, AmscAwosSourceMixin, FmiSourceMixin, KnmiSourceMixin, HkoObsSourceMixin, CowinSourceMixin, MadisSourceMixin, SingaporeMssSourceMixin, ImsSourceMixin, NcmSourceMixin, AerowebSourceMixin): +class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, AmscAwosSourceMixin, FmiSourceMixin, KnmiSourceMixin, HkoObsSourceMixin, CowinSourceMixin, MadisSourceMixin, SingaporeMssSourceMixin, ImsSourceMixin, NcmSourceMixin, AerowebSourceMixin, WundergroundHistoricalMixin): """ Multi-source weather data collector @@ -227,6 +229,17 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self.settlement_cache_ttl_sec = min(self.settlement_cache_ttl_sec, OBSERVATION_REFRESH_SEC) self._settlement_cache: Dict[str, Dict] = {} self._settlement_cache_lock = threading.Lock() + self.wunderground_historical_cache_ttl_sec = max( + 30, + int( + os.getenv( + "WUNDERGROUND_HISTORICAL_CACHE_TTL_SEC", + str(OBSERVATION_REFRESH_SEC), + ) + ), + ) + self._wunderground_historical_cache: Dict[str, Dict] = {} + self._wunderground_historical_cache_lock = threading.Lock() self.fmi_cache_ttl_sec = int( os.getenv("FMI_CACHE_TTL_SEC", "120") ) @@ -337,6 +350,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour (self._knmi_cache, self._knmi_cache_lock, float(self.knmi_cache_ttl_sec * 2)), (self._hko_obs_cache, self._hko_obs_cache_lock, float(self.hko_obs_cache_ttl_sec * 2)), (self._cowin_obs_cache, self._cowin_obs_cache_lock, float(self.cowin_obs_cache_ttl_sec * 5)), + ( + self._wunderground_historical_cache, + self._wunderground_historical_cache_lock, + float(self.wunderground_historical_cache_ttl_sec * 2), + ), ]: stale = [ key @@ -921,6 +939,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour or normalized ) self._settlement_cache.pop(f"cwa:{station_code.lower()}", None) + location_id = self._wunderground_location_id(normalized) + if location_id: + prefix = f"wu:{location_id}:" + with self._wunderground_historical_cache_lock: + for key in list(self._wunderground_historical_cache.keys()): + if key.startswith(prefix): + self._wunderground_historical_cache.pop(key, None) def _uses_fahrenheit(self, city_lower: str) -> bool: return city_lower in self.US_CITIES @@ -972,6 +997,29 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if cwa_forecast is not None: results["cwa_forecast"] = cwa_forecast + def _attach_wunderground_historical( + self, + results: Dict, + city_lower: str, + use_fahrenheit: bool, + ) -> None: + try: + utc_offset = get_city_utc_offset_seconds(city_lower) + payload = self.fetch_wunderground_historical( + city_lower, + use_fahrenheit=use_fahrenheit, + utc_offset=utc_offset, + ) + except Exception as exc: + logger.warning( + "Wunderground historical attach failed city={} error={}", + city_lower, + exc, + ) + return + if payload: + results["wunderground_current"] = payload + def _attach_turkish_mgm_data( self, results: Dict, @@ -1666,6 +1714,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) self._log_temperature_unit(city, use_fahrenheit) self._attach_settlement_sources(results, city_lower) + self._attach_wunderground_historical(results, city_lower, use_fahrenheit) if lat and lon: # When force_refresh_observations_only is set (airport push loop), diff --git a/src/data_collection/wunderground_sources.py b/src/data_collection/wunderground_sources.py new file mode 100644 index 00000000..21726d5d --- /dev/null +++ b/src/data_collection/wunderground_sources.py @@ -0,0 +1,405 @@ +from __future__ import annotations + +import os +import re +import threading +import time +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional +from urllib.parse import quote + +import httpx +from loguru import logger + +from src.utils.metrics import record_source_call +from src.utils.refresh_policy import OBSERVATION_REFRESH_SEC + + +DEFAULT_WEATHER_COM_API_KEY = "6532d6454b8aa370768e63d6ba5a832e" + +ICAO_COUNTRY_PREFIXES = { + "C": "CA", + "E": "GB", + "K": "US", + "L": "FR", + "M": "MX", + "O": "SA", + "R": "JP", + "S": "BR", + "V": "HK", + "W": "SG", + "Z": "CN", +} + +ICAO_COUNTRY_EXACT_PREFIXES = { + "ED": "DE", + "EF": "FI", + "EG": "GB", + "EH": "NL", + "EP": "PL", + "LE": "ES", + "LF": "FR", + "LI": "IT", + "LL": "IL", + "LT": "TR", + "MM": "MX", + "MP": "PA", + "OE": "SA", + "OP": "PK", + "RC": "TW", + "RJ": "JP", + "RK": "KR", + "RP": "PH", + "SA": "AR", + "SB": "BR", + "VE": "IN", + "VH": "HK", + "VI": "IN", + "WM": "MY", + "WS": "SG", + "WI": "ID", + "FA": "ZA", + "CY": "CA", +} + +WU_PATH_COUNTRY_ALIASES = { + "uk": "GB", +} + + +class WundergroundHistoricalMixin: + def _ensure_wunderground_historical_cache(self) -> None: + if not hasattr(self, "_wunderground_historical_cache"): + self._wunderground_historical_cache = {} + if not hasattr(self, "_wunderground_historical_cache_lock"): + self._wunderground_historical_cache_lock = threading.Lock() + if not hasattr(self, "wunderground_historical_cache_ttl_sec"): + self.wunderground_historical_cache_ttl_sec = max( + 30, + int( + os.getenv( + "WUNDERGROUND_HISTORICAL_CACHE_TTL_SEC", + str(OBSERVATION_REFRESH_SEC), + ) + ), + ) + + def _wunderground_api_key(self) -> str: + return ( + os.getenv("POLYWEATHER_WUNDERGROUND_API_KEY") + or os.getenv("WUNDERGROUND_API_KEY") + or DEFAULT_WEATHER_COM_API_KEY + ).strip() + + def _wunderground_country_from_url(self, city_meta: Dict[str, Any]) -> Optional[str]: + url = str(city_meta.get("settlement_url") or "").strip().lower() + match = re.search(r"/history/daily/([^/]+)/", url) + if not match: + return None + raw = match.group(1).strip().upper() + return WU_PATH_COUNTRY_ALIASES.get(raw.lower(), raw) + + def _wunderground_country_from_icao(self, icao: str) -> Optional[str]: + icao = str(icao or "").strip().upper() + if len(icao) >= 2 and icao[:2] in ICAO_COUNTRY_EXACT_PREFIXES: + return ICAO_COUNTRY_EXACT_PREFIXES[icao[:2]] + if icao[:1] in ICAO_COUNTRY_PREFIXES: + return ICAO_COUNTRY_PREFIXES[icao[:1]] + return None + + def _wunderground_location_id(self, city: str) -> Optional[str]: + normalized = str(city or "").strip().lower() + city_meta = (getattr(self, "CITY_REGISTRY", {}) or {}).get(normalized) or {} + station_code = ( + str(city_meta.get("wunderground_station_code") or "").strip() + or str(city_meta.get("settlement_station_code") or "").strip() + or str(city_meta.get("icao") or "").strip() + ) + if not station_code and hasattr(self, "get_icao_code"): + station_code = str(self.get_icao_code(city) or "").strip() + station_code = station_code.upper() + if not station_code: + return None + + country_code = ( + str(city_meta.get("wunderground_country_code") or "").strip().upper() + or self._wunderground_country_from_url(city_meta) + or self._wunderground_country_from_icao(station_code) + ) + if not country_code: + return None + return f"{station_code}:9:{country_code}" + + @staticmethod + def _wu_local_date(utc_offset: int, local_date: Optional[str]) -> str: + if local_date: + return str(local_date).replace("-", "")[:8] + now_local = datetime.now(timezone.utc) + timedelta(seconds=int(utc_offset or 0)) + return now_local.strftime("%Y%m%d") + + @staticmethod + def _wu_float(value: Any) -> Optional[float]: + try: + numeric = float(value) + except (TypeError, ValueError): + return None + if not (numeric == numeric): + return None + return numeric + + @staticmethod + def _wu_temp_value(value: float) -> int | float: + return round(value, 1) if isinstance(value, float) and value % 1 else int(value) + + def _wu_request_headers(self) -> Dict[str, str]: + return { + "Accept": "application/json", + "User-Agent": getattr(self, "user_agent", "PolyWeather/1.0"), + "Cache-Control": "no-cache", + "Pragma": "no-cache", + } + + def _parse_wunderground_current_point( + self, + data: Dict[str, Any], + *, + units: str, + utc_offset: int, + local_date_iso: str, + ) -> Optional[Dict[str, Any]]: + observation = data.get("observation") if isinstance(data, dict) else None + if not isinstance(observation, dict): + return None + + unit_bucket_name = "imperial" if units == "e" else "metric" + unit_bucket = observation.get(unit_bucket_name) + if not isinstance(unit_bucket, dict): + unit_bucket = observation.get("metric") if isinstance(observation.get("metric"), dict) else {} + temp = self._wu_float(unit_bucket.get("temp") if isinstance(unit_bucket, dict) else None) + if temp is None: + return None + + ts = observation.get("obs_time") or observation.get("valid_time_gmt") + local_dt = None + raw_local = str(observation.get("obs_time_local") or "").strip() + if raw_local: + try: + local_dt = datetime.strptime(raw_local, "%Y-%m-%dT%H:%M:%S%z") + ts = int(local_dt.timestamp()) + except ValueError: + local_dt = None + if local_dt is None: + try: + utc_dt = datetime.fromtimestamp(int(ts), tz=timezone.utc) + except (TypeError, ValueError, OSError): + return None + local_dt = utc_dt + timedelta(seconds=int(utc_offset or 0)) + if local_dt.strftime("%Y-%m-%d") != local_date_iso: + return None + if units == "m" and hasattr(self, "_is_plausible_metar_temp_c"): + if not self._is_plausible_metar_temp_c(temp, "", str(observation.get("obs_id") or "")): + return None + + max_24h = self._wu_float( + unit_bucket.get("temp_max_24hour") if isinstance(unit_bucket, dict) else None + ) + return { + "ts": int(ts), + "time": local_dt.strftime("%H:%M"), + "temp": self._wu_temp_value(temp), + "max_24h": self._wu_temp_value(max_24h) if max_24h is not None else None, + "wx_desc": observation.get("phrase_32char") + or observation.get("phrase_22char") + or observation.get("phrase_12char"), + } + + def fetch_wunderground_historical( + self, + city: str, + *, + use_fahrenheit: bool = False, + utc_offset: int = 0, + local_date: Optional[str] = None, + ) -> Optional[Dict[str, Any]]: + """Fetch Wunderground/weather.com intraday history for the city-local day.""" + started = time.perf_counter() + self._ensure_wunderground_historical_cache() + + api_key = self._wunderground_api_key() + location_id = self._wunderground_location_id(city) + if not api_key or not location_id: + record_source_call("wunderground", "historical", "missing_location", 0.0) + return None + + date_key = self._wu_local_date(utc_offset, local_date) + units = "e" if use_fahrenheit else "m" + cache_key = f"wu:{location_id}:{units}:{date_key}" + now_ts = time.time() + with self._wunderground_historical_cache_lock: + cached = self._wunderground_historical_cache.get(cache_key) + if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec: + return dict(cached["d"]) + + current_url = ( + "https://api.weather.com/v1/location/" + f"{quote(location_id, safe=':')}/observations/current.json" + f"?apiKey={quote(api_key)}&units={units}" + ) + url = ( + "https://api.weather.com/v1/location/" + f"{quote(location_id, safe=':')}/observations/historical.json" + f"?apiKey={quote(api_key)}&units={units}&startDate={date_key}&endDate={date_key}" + ) + + current_data: Optional[Dict[str, Any]] = None + try: + current_response = self._http_get( + current_url, + timeout=float(os.getenv("WUNDERGROUND_CURRENT_TIMEOUT_SEC", "4")), + headers=self._wu_request_headers(), + ) + if current_response.status_code != 204: + current_response.raise_for_status() + parsed_current = current_response.json() + if isinstance(parsed_current, dict): + current_data = parsed_current + except (httpx.HTTPError, ValueError) as exc: + logger.warning( + "Wunderground current fetch failed city={} location={} error={}", + city, + location_id, + exc, + ) + + try: + response = self._http_get( + url, + timeout=float(os.getenv("WUNDERGROUND_HISTORICAL_TIMEOUT_SEC", "4")), + headers=self._wu_request_headers(), + ) + response.raise_for_status() + data = response.json() + except (httpx.HTTPError, ValueError) as exc: + logger.warning( + "Wunderground historical fetch failed city={} location={} error={}", + city, + location_id, + exc, + ) + record_source_call( + "wunderground", + "historical", + "error", + (time.perf_counter() - started) * 1000.0, + ) + return None + + observations = data.get("observations") if isinstance(data, dict) else [] + if not isinstance(observations, list) or not observations: + record_source_call( + "wunderground", + "historical", + "empty", + (time.perf_counter() - started) * 1000.0, + ) + return None + + local_date_iso = f"{date_key[:4]}-{date_key[4:6]}-{date_key[6:8]}" + points = [] + for row in observations: + if not isinstance(row, dict): + continue + temp = self._wu_float(row.get("temp")) + ts = row.get("valid_time_gmt") + if temp is None or ts is None: + continue + try: + utc_dt = datetime.fromtimestamp(int(ts), tz=timezone.utc) + except (TypeError, ValueError, OSError): + continue + local_dt = utc_dt + timedelta(seconds=int(utc_offset or 0)) + if local_dt.strftime("%Y-%m-%d") != local_date_iso: + continue + if not use_fahrenheit and hasattr(self, "_is_plausible_metar_temp_c"): + if not self._is_plausible_metar_temp_c(temp, city, str(row.get("obs_id") or location_id)): + continue + points.append( + { + "ts": int(ts), + "time": local_dt.strftime("%H:%M"), + "temp": self._wu_temp_value(temp), + "wx_desc": row.get("wx_phrase"), + } + ) + + current_point = ( + self._parse_wunderground_current_point( + current_data, + units=units, + utc_offset=utc_offset, + local_date_iso=local_date_iso, + ) + if isinstance(current_data, dict) + else None + ) + if current_point: + points = [point for point in points if point.get("ts") != current_point["ts"]] + points.append(current_point) + + if not points: + record_source_call( + "wunderground", + "historical", + "empty_local_day", + (time.perf_counter() - started) * 1000.0, + ) + return None + + points.sort(key=lambda item: item["ts"]) + latest = points[-1] + max_point = points[0] + for point in points[1:]: + if float(point["temp"]) > float(max_point["temp"]): + max_point = point + if current_point and current_point.get("max_24h") is not None: + current_max = current_point["max_24h"] + if float(current_max) > float(max_point["temp"]): + max_point = { + "ts": current_point["ts"], + "time": current_point["time"], + "temp": current_max, + "wx_desc": current_point.get("wx_desc"), + } + + station_code = location_id.split(":", 1)[0] + payload = { + "source": "wunderground_historical", + "source_code": "wunderground", + "source_label": "Wunderground historical/current.json", + "station_code": station_code, + "location_id": location_id, + "local_date": local_date_iso, + "temp_symbol": "°F" if use_fahrenheit else "°C", + "temp": latest["temp"], + "latest_temp": latest["temp"], + "obs_time": latest["time"], + "max_so_far": max_point["temp"], + "daily_high": max_point["temp"], + "max_temp_time": max_point["time"], + "today_obs": [{"time": p["time"], "temp": p["temp"]} for p in points], + "observation_count": len(points), + "api_units": units, + "current_json_used": bool(current_point), + } + with self._wunderground_historical_cache_lock: + self._wunderground_historical_cache[cache_key] = { + "t": time.time(), + "d": payload, + } + record_source_call( + "wunderground", + "historical", + "success", + (time.perf_counter() - started) * 1000.0, + ) + return dict(payload) diff --git a/tests/test_city_payloads.py b/tests/test_city_payloads.py index 772459af..641b4119 100644 --- a/tests/test_city_payloads.py +++ b/tests/test_city_payloads.py @@ -1,4 +1,5 @@ -from web.services.city_payloads import build_city_summary_payload +from web.services.city_payloads import build_city_detail_payload, build_city_summary_payload +from web.services import city_runtime def test_city_summary_payload_preserves_deb_metadata(): @@ -26,3 +27,77 @@ def test_city_summary_payload_preserves_deb_metadata(): "bias_adjustment": 1.3, "bias_samples": 18, } + + +def test_city_payloads_expose_wunderground_current(): + data = { + "name": "shanghai", + "display_name": "Shanghai", + "temp_symbol": "°C", + "current": {}, + "risk": {}, + "wunderground_current": { + "source": "wunderground_historical", + "station_code": "ZSPD", + "max_so_far": 26, + "today_obs": [{"time": "13:30", "temp": 26}], + }, + } + + summary = build_city_summary_payload(data) + detail = build_city_detail_payload(data) + + assert summary["wunderground_current"]["max_so_far"] == 26 + assert detail["wunderground_current"]["max_so_far"] == 26 + assert detail["official"]["wunderground_current"]["station_code"] == "ZSPD" + assert detail["timeseries"]["wunderground_today_obs"] == [{"time": "13:30", "temp": 26}] + + +def test_api_payload_overlays_latest_wunderground_state(monkeypatch): + stale_payload = { + "name": "guangzhou", + "temp_symbol": "°C", + "utc_offset_seconds": 28800, + "wunderground_current": { + "source": "wunderground_historical", + "station_code": "ZGGG", + "temp": 36, + "max_so_far": 36, + "today_obs": [{"time": "14:00", "temp": 36}], + }, + "official": { + "wunderground_current": { + "source": "wunderground_historical", + "station_code": "ZGGG", + "temp": 36, + "max_so_far": 36, + }, + }, + "timeseries": { + "wunderground_today_obs": [{"time": "14:00", "temp": 36}], + }, + } + + def fake_fetch(city: str, *, use_fahrenheit: bool, utc_offset: int): + assert city == "guangzhou" + assert use_fahrenheit is False + assert utc_offset == 28800 + return { + "source": "wunderground_historical", + "station_code": "ZGGG", + "temp": 38, + "max_so_far": 38, + "today_obs": [{"time": "15:17", "temp": 38}], + } + + monkeypatch.setattr(city_runtime._weather, "fetch_wunderground_historical", fake_fetch) + + overlay = getattr(city_runtime, "_overlay_latest_wunderground_current", None) + assert callable(overlay), "city API must overlay cached payloads with latest WU state" + payload = overlay("guangzhou", stale_payload) + + assert payload["wunderground_current"]["temp"] == 38 + assert payload["wunderground_current"]["max_so_far"] == 38 + assert payload["official"]["wunderground_current"]["max_so_far"] == 38 + assert payload["timeseries"]["wunderground_today_obs"] == [{"time": "15:17", "temp": 38}] + assert stale_payload["wunderground_current"]["max_so_far"] == 36 diff --git a/tests/test_wunderground_historical.py b/tests/test_wunderground_historical.py new file mode 100644 index 00000000..871ed14f --- /dev/null +++ b/tests/test_wunderground_historical.py @@ -0,0 +1,211 @@ +from __future__ import annotations + +from urllib.parse import parse_qs, urlparse + +import httpx + +from src.data_collection.weather_sources import WeatherDataCollector + + +def _collector(monkeypatch, tmp_path) -> WeatherDataCollector: + monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) + return WeatherDataCollector({"weather": {}}) + + +def test_fetch_wunderground_historical_uses_weather_com_historical_json(monkeypatch, tmp_path): + collector = _collector(monkeypatch, tmp_path) + called = {} + + def fake_get(url: str, **_kwargs): + called["url"] = url + return httpx.Response( + 200, + json={ + "metadata": { + "status_code": 200, + "location_id": "ZSPD:9:CN", + "units": "m", + }, + "observations": [ + { + "obs_id": "ZSPD", + "obs_name": "Shanghai/Pudong", + "valid_time_gmt": 1780027200, + "temp": 25, + "wx_phrase": "Fair", + }, + { + "obs_id": "ZSPD", + "obs_name": "Shanghai/Pudong", + "valid_time_gmt": 1780029000, + "temp": 26, + "wx_phrase": "Fair", + }, + { + "obs_id": "ZSPD", + "obs_name": "Shanghai/Pudong", + "valid_time_gmt": 1780032600, + "temp": 26, + "wx_phrase": "Fair", + }, + ], + }, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr(collector, "_http_get", fake_get) + + payload = collector.fetch_wunderground_historical( + "shanghai", + use_fahrenheit=False, + utc_offset=28800, + local_date="2026-05-29", + ) + + parsed = urlparse(called["url"]) + query = parse_qs(parsed.query) + assert parsed.path == "/v1/location/ZSPD:9:CN/observations/historical.json" + assert query["units"] == ["m"] + assert query["startDate"] == ["20260529"] + assert query["endDate"] == ["20260529"] + assert payload["source"] == "wunderground_historical" + assert payload["station_code"] == "ZSPD" + assert payload["location_id"] == "ZSPD:9:CN" + assert payload["temp"] == 26 + assert payload["max_so_far"] == 26 + assert payload["max_temp_time"] == "12:30" + assert payload["obs_time"] == "13:30" + + +def test_fetch_wunderground_historical_uses_current_json_to_lift_latest_and_high(monkeypatch, tmp_path): + collector = _collector(monkeypatch, tmp_path) + calls = [] + + def fake_get(url: str, **kwargs): + calls.append({"url": url, "headers": kwargs.get("headers") or {}}) + if urlparse(url).path.endswith("/current.json"): + return httpx.Response( + 200, + json={ + "metadata": { + "status_code": 200, + "location_id": "ZGGG:9:CN", + "units": "m", + }, + "observation": { + "obs_time": 1780039020, + "obs_time_local": "2026-05-29T15:17:00+0800", + "metric": { + "temp": 38, + "temp_max_24hour": 38, + }, + }, + }, + request=httpx.Request("GET", url), + ) + return httpx.Response( + 200, + json={ + "metadata": { + "status_code": 200, + "location_id": "ZGGG:9:CN", + "units": "m", + }, + "observations": [ + { + "obs_id": "ZGGG", + "valid_time_gmt": 1780034400, + "temp": 36, + }, + ], + }, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr(collector, "_http_get", fake_get) + + payload = collector.fetch_wunderground_historical( + "guangzhou", + use_fahrenheit=False, + utc_offset=28800, + local_date="2026-05-29", + ) + + paths = [urlparse(call["url"]).path for call in calls] + assert "/v1/location/ZGGG:9:CN/observations/historical.json" in paths + assert "/v1/location/ZGGG:9:CN/observations/current.json" in paths + assert all(call["headers"].get("Cache-Control") == "no-cache" for call in calls) + assert all(call["headers"].get("Pragma") == "no-cache" for call in calls) + assert payload["temp"] == 38 + assert payload["obs_time"] == "15:17" + assert payload["max_so_far"] == 38 + assert payload["max_temp_time"] == "15:17" + + +def test_fetch_wunderground_historical_keeps_fahrenheit_for_us_cities(monkeypatch, tmp_path): + collector = _collector(monkeypatch, tmp_path) + called = {} + + def fake_get(url: str, **_kwargs): + called["url"] = url + return httpx.Response( + 200, + json={ + "metadata": { + "status_code": 200, + "location_id": "KLGA:9:US", + "units": "e", + }, + "observations": [ + {"obs_id": "KLGA", "valid_time_gmt": 1780027200, "temp": 73}, + {"obs_id": "KLGA", "valid_time_gmt": 1780030800, "temp": 75}, + ], + }, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr(collector, "_http_get", fake_get) + + payload = collector.fetch_wunderground_historical( + "new york", + use_fahrenheit=True, + utc_offset=-14400, + local_date="2026-05-29", + ) + + parsed = urlparse(called["url"]) + assert parsed.path == "/v1/location/KLGA:9:US/observations/historical.json" + assert parse_qs(parsed.query)["units"] == ["e"] + assert payload["temp_symbol"] == "°F" + assert payload["max_so_far"] == 75 + + +def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path): + collector = _collector(monkeypatch, tmp_path) + called = {} + + monkeypatch.setattr(collector, "fetch_settlement_current", lambda _city: None) + monkeypatch.setattr(collector, "_supports_aviationweather", lambda _city: False) + + def fake_fetch(city: str, *, use_fahrenheit: bool, utc_offset: int, local_date=None): + called["city"] = city + called["use_fahrenheit"] = use_fahrenheit + called["utc_offset"] = utc_offset + called["local_date"] = local_date + return { + "source": "wunderground_historical", + "location_id": "ZSPD:9:CN", + "max_so_far": 26, + } + + monkeypatch.setattr(collector, "fetch_wunderground_historical", fake_fetch) + + payload = collector.fetch_all_sources("shanghai") + + assert payload["wunderground_current"]["max_so_far"] == 26 + assert called == { + "city": "shanghai", + "use_fahrenheit": False, + "utc_offset": 28800, + "local_date": None, + } diff --git a/web/analysis_service.py b/web/analysis_service.py index 2ed68c40..1b2bb173 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -790,6 +790,7 @@ def _analyze( taf = raw.get("taf", {}) mgm = raw.get("mgm") or {} settlement_current = raw.get("settlement_current") or {} + wunderground_current = raw.get("wunderground_current") or {} ens_raw = raw.get("ensemble", {}) mm = raw.get("multi_model", {}) if not isinstance(om, dict): @@ -800,6 +801,8 @@ def _analyze( mgm = {} if not isinstance(settlement_current, dict): settlement_current = {} + if not isinstance(wunderground_current, dict): + wunderground_current = {} if not isinstance(ens_raw, dict): ens_raw = {} if not isinstance(mm, dict): @@ -1795,6 +1798,7 @@ def _analyze( "last_observation_local_date": metar.get("observation_local_date") if metar else None, "current_local_date": local_date_str, }, + "wunderground_current": wunderground_current, "settlement_station": network_snapshot.get("settlement_station") or {}, "airport_primary": airport_primary_current, "airport_primary_today_obs": network_snapshot.get("airport_primary_today_obs") or [], @@ -1941,6 +1945,11 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: jobs: Dict[str, Any] = { "settlement_current": lambda: _weather.fetch_settlement_current(city) or {}, + "wunderground_current": lambda: _weather.fetch_wunderground_historical( + city, + use_fahrenheit=is_f, + utc_offset=default_utc_offset, + ) or {}, "open_meteo": lambda: _weather.fetch_from_open_meteo(lat, lon, use_fahrenheit=is_f) or {}, "multi_model": lambda: _weather.fetch_multi_model(lat, lon, city=city, use_fahrenheit=is_f) or {}, } @@ -1969,8 +1978,11 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: fetched[key] = future.result() settlement_current = fetched.get("settlement_current") or {} + wunderground_current = fetched.get("wunderground_current") or {} open_meteo = fetched.get("open_meteo") or {} mm = fetched.get("multi_model") or {} + if not isinstance(wunderground_current, dict): + wunderground_current = {} utc_offset = open_meteo.get("utc_offset") if utc_offset is None: utc_offset = default_utc_offset @@ -2233,6 +2245,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: "obs_age_min": obs_age_min, "observation_status": "live" if cur_temp is not None else "missing", }, + "wunderground_current": wunderground_current, "deb": { "prediction": _sf(deb_val), "raw_prediction": _sf(deb_raw_val), diff --git a/web/scan_terminal_city_row.py b/web/scan_terminal_city_row.py index 60e3ca5b..fe79ee0c 100644 --- a/web/scan_terminal_city_row.py +++ b/web/scan_terminal_city_row.py @@ -96,6 +96,7 @@ def _build_terminal_row( "temp_symbol": data.get("temp_symbol"), "current_temp": current.get("temp"), "current_max_so_far": current.get("max_so_far"), + "wunderground_current": data.get("wunderground_current") or {}, "metar_context": metar_context, "metar_today_obs": metar_context.get("today_obs") or [], "metar_recent_obs": metar_context.get("recent_obs") or [], @@ -207,6 +208,7 @@ def _build_quick_row( "risk_level": risk.get("level"), "current_temp": curr.get("temp"), "current_max_so_far": curr.get("max_so_far"), + "wunderground_current": data.get("wunderground_current") or {}, "deb_prediction": deb.get("prediction"), "model_cluster_sources": ( daily_entry.get("models") diff --git a/web/services/city_api.py b/web/services/city_api.py index 26d09a94..1ea469a3 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -11,6 +11,14 @@ from loguru import logger import web.routes as legacy_routes +async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return await run_in_threadpool( + legacy_routes._overlay_latest_wunderground_current, + city, + payload, + ) + + def _build_cities_payload() -> Dict[str, Any]: out = [] deb_recent_index = legacy_routes._build_recent_deb_performance_index() @@ -86,7 +94,7 @@ async def get_city_detail_payload( if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) - return cached_entry.get("payload") or {} + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) if detail_mode == "panel": if force_refresh: @@ -95,7 +103,7 @@ async def get_city_detail_payload( if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_PANEL_CACHE_TTL_SEC): return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) - return cached_entry.get("payload") or {} + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) if detail_mode == "nearby": if force_refresh: @@ -104,7 +112,7 @@ async def get_city_detail_payload( if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_NEARBY_CACHE_TTL_SEC): return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) - return cached_entry.get("payload") or {} + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) if detail_mode == "market": if force_refresh: @@ -113,7 +121,7 @@ async def get_city_detail_payload( if cached_entry: if not legacy_routes._market_analysis_cache_is_fresh(cached_entry): return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) - return cached_entry.get("payload") or {} + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) return await run_in_threadpool(legacy_routes._analyze, city, force_refresh, False, detail_mode) @@ -131,7 +139,7 @@ async def get_city_summary_payload( if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC): return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False) - return cached_entry.get("payload") or {} + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False) @@ -154,7 +162,7 @@ async def get_city_detail_aggregate_payload( if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) else: - data = cached_entry.get("payload") or {} + data = await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) else: data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) diff --git a/web/services/city_payloads.py b/web/services/city_payloads.py index 081123ef..92d0e868 100644 --- a/web/services/city_payloads.py +++ b/web/services/city_payloads.py @@ -28,6 +28,7 @@ def build_city_summary_payload(data: Dict[str, Any]) -> Dict[str, Any]: ), }, "deb": dict(deb) if deb else {"prediction": None}, + "wunderground_current": data.get("wunderground_current") or {}, "deviation_monitor": data.get("deviation_monitor") or {}, "risk": { "level": data.get("risk", {}).get("level"), @@ -212,6 +213,7 @@ def build_city_detail_payload( ), "airport_primary": data.get("airport_primary") or {}, "airport_primary_today_obs": data.get("airport_primary_today_obs") or [], + "wunderground_current": data.get("wunderground_current") or {}, "official_nearby": data.get("official_nearby") or [], "official_network_source": data.get("official_network_source"), "official_network_status": data.get("official_network_status") or {}, @@ -224,6 +226,7 @@ def build_city_detail_payload( "metar_recent_obs": data.get("metar_recent_obs") or [], "metar_today_obs": data.get("metar_today_obs") or [], "settlement_today_obs": data.get("settlement_today_obs") or [], + "wunderground_today_obs": (data.get("wunderground_current") or {}).get("today_obs") or [], "hourly": data.get("hourly") or {}, "mgm_hourly": (data.get("mgm") or {}).get("hourly", []), "forecast_daily": (data.get("forecast") or {}).get("daily", []), @@ -266,6 +269,7 @@ def build_city_detail_payload( "center_station_candidate": data.get("center_station_candidate"), "airport_vs_network_delta": data.get("airport_vs_network_delta"), "airport_current": data.get("airport_current") or {}, + "wunderground_current": data.get("wunderground_current") or {}, "amos": data.get("amos") or {}, "nearby_source": data.get("nearby_source") or ( diff --git a/web/services/city_runtime.py b/web/services/city_runtime.py index 63e87dbf..3c02e0bd 100644 --- a/web/services/city_runtime.py +++ b/web/services/city_runtime.py @@ -2,6 +2,7 @@ from __future__ import annotations import os import time +from copy import deepcopy from datetime import datetime from typing import Optional @@ -53,6 +54,7 @@ from web.core import ( _resolve_auth_points, # noqa: F401 - compatibility export for tests and transitional routers _resolve_weekly_profile, # noqa: F401 - compatibility export for tests and transitional routers _sf, + _weather, ) router = APIRouter() @@ -329,6 +331,52 @@ def _refresh_city_full_cache(city: str, force_refresh: bool = False) -> dict: return payload +def _overlay_wunderground_current(payload: dict, wunderground: dict) -> dict: + if not isinstance(payload, dict) or not isinstance(wunderground, dict) or not wunderground: + return payload + next_payload = deepcopy(payload) + next_payload["wunderground_current"] = dict(wunderground) + + official = next_payload.get("official") + if isinstance(official, dict): + official["wunderground_current"] = dict(wunderground) + + timeseries = next_payload.get("timeseries") + if isinstance(timeseries, dict): + timeseries["wunderground_today_obs"] = list(wunderground.get("today_obs") or []) + + return next_payload + + +def _overlay_latest_wunderground_current(city: str, payload: dict) -> dict: + if not isinstance(payload, dict): + return payload + city_key = str(city or payload.get("name") or payload.get("city") or "").strip().lower() + if city_key not in CITIES: + return deepcopy(payload) + try: + utc_offset = int( + payload.get("utc_offset_seconds") + or (payload.get("overview") or {}).get("utc_offset_seconds") + or get_city_utc_offset_seconds(city_key) + or 0 + ) + except Exception: + utc_offset = get_city_utc_offset_seconds(city_key) + try: + latest_wu = _weather.fetch_wunderground_historical( + city_key, + use_fahrenheit=bool(CITIES[city_key].get("f")), + utc_offset=utc_offset, + ) + except Exception as exc: + logger.warning("latest WU overlay failed city={} error={}", city_key, exc) + return deepcopy(payload) + if not isinstance(latest_wu, dict) or not latest_wu: + return deepcopy(payload) + return _overlay_wunderground_current(payload, latest_wu) + + def _schedule_cache_refresh( background_tasks: BackgroundTasks,