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

feat: implement METAR data collection service and dashboard infrastructure

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-11 20:11:01 +08:00
parent 69168d2fdf
commit 77b4d7b341
14 changed files with 660 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_collection/city_registry.py
+8 -9
View File
@@ -149,21 +149,20 @@ CITY_REGISTRY = {
},
"taipei": {
"name": "Taipei",
"lat": 25.0670,
"lon": 121.5525,
"lat": 25.0377,
"lon": 121.5149,
"icao": "RCSS",
"settlement_source": "wunderground",
"settlement_station_code": "RCSS",
"settlement_station_label": "Taipei Songshan Airport Station",
"settlement_url": "https://www.wunderground.com/history/daily/tw/taipei/RCSS",
"settlement_source": "cwa",
"settlement_station_code": "466920",
"settlement_station_label": "中央气象署台北站",
"tz_offset": 28800,
"use_fahrenheit": False,
"is_major": True,
"risk_level": "low",
"risk_emoji": "🟢",
"airport_name": "台北松山机场",
"distance_km": 4.1,
"warning": "市场现按 Wunderground 台北松山机场站整度°C口径结算；以历史页当日最终完成后的最高整度摄氏值为准。",
"airport_name": "中央气象署台北站",
"distance_km": 0.0,
"warning": "结算按交通部中央气象署台北站口径，不应混用松山机场 METAR 作为结算主源。",
},
"shanghai": {
"name": "Shanghai",
src/data_collection/metar_sources.py
+10 -2
View File
@@ -248,7 +248,11 @@ class MetarSourceMixin:
"hours": 24,
"_t": int(time.time()),
}
response = self.session.get(url, params=params, timeout=self.timeout)
response = self.session.get(
url,
params=params,
timeout=getattr(self, "metar_timeout_sec", self.timeout),
)
response.raise_for_status()
data = response.json()
if not data:
@@ -295,7 +299,11 @@ class MetarSourceMixin:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
}
resp = self.session.get(url, headers=headers, timeout=self.timeout)
resp = self.session.get(
url,
headers=headers,
timeout=getattr(self, "metar_cluster_timeout_sec", self.timeout),
)
if resp.status_code != 200:
logger.warning(f"METAR cluster fetch HTTP {resp.status_code} for {icaos}")
return []
src/data_collection/nws_open_meteo_sources.py
+3 -3
View File
@@ -208,7 +208,7 @@ class NwsOpenMeteoSourceMixin:
response = self._http_get(
url,
params=params,
timeout=self.timeout,
timeout=getattr(self, "open_meteo_timeout_sec", self.timeout),
)
response.raise_for_status()
data = response.json()
@@ -369,7 +369,7 @@ class NwsOpenMeteoSourceMixin:
response = self._http_get(
url,
params=params,
timeout=self.timeout,
timeout=getattr(self, "open_meteo_timeout_sec", self.timeout),
)
response.raise_for_status()
data = response.json()
@@ -526,7 +526,7 @@ class NwsOpenMeteoSourceMixin:
response = self._http_get(
url,
params=params,
timeout=self.timeout,
timeout=getattr(self, "open_meteo_timeout_sec", self.timeout),
)
response.raise_for_status()
data = response.json()
src/data_collection/settlement_sources.py
+28 -16
View File
@@ -4,6 +4,7 @@ import csv
import math
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
@@ -226,19 +227,31 @@ class SettlementSourceMixin:
try:
base = "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather"
temp_csv = self._http_get(f"{base}/latest_1min_temperature.csv", timeout=self.timeout)
temp_csv.raise_for_status()
maxmin_csv = self._http_get(f"{base}/latest_since_midnight_maxmin.csv", timeout=self.timeout)
maxmin_csv.raise_for_status()
humidity_csv = self._http_get(f"{base}/latest_1min_humidity.csv", timeout=self.timeout)
humidity_csv.raise_for_status()
wind_csv = self._http_get(f"{base}/latest_10min_wind.csv", timeout=self.timeout)
wind_csv.raise_for_status()
csv_urls = {
"temp": f"{base}/latest_1min_temperature.csv",
"maxmin": f"{base}/latest_since_midnight_maxmin.csv",
"humidity": f"{base}/latest_1min_humidity.csv",
"wind": f"{base}/latest_10min_wind.csv",
}
temp_rows = self._csv_rows(temp_csv.text)
maxmin_rows = self._csv_rows(maxmin_csv.text)
humidity_rows = self._csv_rows(humidity_csv.text)
wind_rows = self._csv_rows(wind_csv.text)
def _fetch_csv(url: str):
response = self._http_get(url, timeout=self.timeout)
response.raise_for_status()
return response
fetched_csv = {}
with ThreadPoolExecutor(max_workers=4) as executor:
future_map = {
executor.submit(_fetch_csv, url): key
for key, url in csv_urls.items()
}
for future, key in future_map.items():
fetched_csv[key] = future.result()
temp_rows = self._csv_rows(fetched_csv["temp"].text)
maxmin_rows = self._csv_rows(fetched_csv["maxmin"].text)
humidity_rows = self._csv_rows(fetched_csv["humidity"].text)
wind_rows = self._csv_rows(fetched_csv["wind"].text)
temp_row = self._pick_station_row(temp_rows, candidate_names)
maxmin_row = self._pick_station_row(maxmin_rows, candidate_names)
@@ -626,6 +639,8 @@ class SettlementSourceMixin:
station_name=station_name,
station_candidates=station_candidates,
)
if settlement_source == "cwa":
return self.fetch_cwa_taipei_settlement_current()
if settlement_source == "noaa":
station_code = (
str(city_meta.get("settlement_station_code") or "").strip()
@@ -643,8 +658,5 @@ class SettlementSourceMixin:
except Exception as exc:
logger.warning(f"Settlement source dispatch failed city={city}: {exc}")
if normalized == "taipei":
return self.fetch_noaa_station_settlement_current(
station_code="RCTP",
station_name="Taiwan Taoyuan International Airport",
)
return self.fetch_cwa_taipei_settlement_current()
return None
src/data_collection/weather_sources.py
+101 -41
View File
@@ -113,12 +113,24 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
def __init__(self, config: dict):
self.config = config
self.timeout = 30 # 增加超时以支持高延迟 VPS
# Keep external calls short so one degraded source cannot block the whole city pipeline.
self.timeout = max(
2.0, float(os.getenv("POLYWEATHER_HTTP_TIMEOUT_SEC", "8"))
)
self.http_retry_count = max(
0, int(os.getenv("POLYWEATHER_HTTP_RETRY_COUNT", "1"))
0, int(os.getenv("POLYWEATHER_HTTP_RETRY_COUNT", "0"))
)
self.http_retry_backoff_sec = max(
0.0, float(os.getenv("POLYWEATHER_HTTP_RETRY_BACKOFF_SEC", "0.35"))
0.0, float(os.getenv("POLYWEATHER_HTTP_RETRY_BACKOFF_SEC", "0.2"))
)
self.open_meteo_timeout_sec = max(
2.0, float(os.getenv("POLYWEATHER_OPEN_METEO_TIMEOUT_SEC", "5"))
)
self.metar_timeout_sec = max(
2.0, float(os.getenv("POLYWEATHER_METAR_TIMEOUT_SEC", "4"))
)
self.metar_cluster_timeout_sec = max(
2.0, float(os.getenv("POLYWEATHER_METAR_CLUSTER_TIMEOUT_SEC", "3.5"))
)
self.session = httpx.Client(
timeout=self.timeout,
@@ -151,7 +163,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
self._open_meteo_rl_lock = threading.Lock()
# Open-Meteo burst control: avoid hammering API with many cities at once.
self._open_meteo_min_interval_sec: float = float(
os.getenv("OPEN_METEO_MIN_CALL_INTERVAL_SEC", "3")
os.getenv("OPEN_METEO_MIN_CALL_INTERVAL_SEC", "1")
)
self._open_meteo_last_call_ts: float = 0.0
self._open_meteo_call_lock = threading.Lock()
@@ -728,8 +740,18 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
hko_forecast = self.fetch_hko_forecast()
if hko_forecast:
results["hko_forecast"] = hko_forecast
elif settlement_source == "cwa":
cwa_forecast = self.fetch_cwa_taipei_forecast()
if cwa_forecast is not None:
results["cwa_forecast"] = cwa_forecast
def _attach_turkish_mgm_data(self, results: Dict, city_lower: str) -> None:
def _attach_turkish_mgm_data(
self,
results: Dict,
city_lower: str,
*,
include_nearby: bool = True,
) -> None:
if city_lower not in self.TURKISH_PROVINCES:
return
istno, province = self.TURKISH_PROVINCES[city_lower]
@@ -737,15 +759,22 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
if not mgm_data:
return
results["mgm"] = mgm_data
results["nearby_source"] = "mgm"
nearby = self.fetch_mgm_nearby_stations(province, root_ist_no=istno)
if nearby:
results["mgm_nearby"] = nearby
if include_nearby:
results["nearby_source"] = "mgm"
nearby = self.fetch_mgm_nearby_stations(province, root_ist_no=istno)
if nearby:
results["mgm_nearby"] = nearby
def _attach_global_nearby_cluster(
self, results: Dict, city_lower: str, use_fahrenheit: bool
) -> None:
if city_lower not in self.CITY_METAR_CLUSTERS or "mgm_nearby" in results:
city_meta = self.CITY_REGISTRY.get(str(city_lower or "").strip().lower()) or {}
settlement_source = str(city_meta.get("settlement_source") or "").strip().lower()
if (
city_lower not in self.CITY_METAR_CLUSTERS
or "mgm_nearby" in results
or settlement_source in {"hko", "cwa"}
):
return
cluster_icaos = self.CITY_METAR_CLUSTERS[city_lower]
cluster_data = self.fetch_metar_nearby_cluster(
@@ -854,21 +883,26 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
lat: float,
lon: float,
use_fahrenheit: bool,
*,
include_ensemble: bool = True,
include_multi_model: bool = True,
) -> None:
if use_fahrenheit:
nws_data = self.fetch_nws(lat, lon)
if nws_data:
results["nws"] = nws_data
ensemble_data = self.fetch_ensemble(lat, lon, use_fahrenheit=use_fahrenheit)
if ensemble_data:
results["ensemble"] = ensemble_data
if include_ensemble:
ensemble_data = self.fetch_ensemble(lat, lon, use_fahrenheit=use_fahrenheit)
if ensemble_data:
results["ensemble"] = ensemble_data
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
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
def fetch_all_sources(
self,
@@ -877,6 +911,10 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
lon: float = None,
country: str = None,
force_refresh: bool = False,
include_taf: bool = True,
include_nearby: bool = True,
include_ensemble: bool = True,
include_multi_model: bool = True,
) -> Dict:
"""
Fetch weather data from all available sources
@@ -910,23 +948,34 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
)
if metar_data:
results["metar"] = metar_data
if supports_aviationweather and city_lower != "hong kong":
if include_taf and supports_aviationweather and city_lower != "hong kong":
taf_data = self.fetch_taf(city, utc_offset=utc_offset)
if taf_data:
results["taf"] = taf_data
self._attach_turkish_mgm_data(results, city_lower)
self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
self._attach_korea_official_nearby(results, city_lower, use_fahrenheit)
self._attach_russia_official_nearby(results, city_lower, use_fahrenheit)
if city_lower == "warsaw":
self._attach_warsaw_official_nearby(results, use_fahrenheit)
self._attach_global_nearby_cluster(
results, city_lower, use_fahrenheit
self._attach_turkish_mgm_data(
results,
city_lower,
include_nearby=include_nearby,
)
if include_nearby:
self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
self._attach_korea_official_nearby(results, city_lower, use_fahrenheit)
self._attach_russia_official_nearby(results, city_lower, use_fahrenheit)
if city_lower == "warsaw":
self._attach_warsaw_official_nearby(results, use_fahrenheit)
self._attach_global_nearby_cluster(
results, city_lower, use_fahrenheit
)
self._attach_nws_and_models(
results, city, lat, lon, use_fahrenheit
results,
city,
lat,
lon,
use_fahrenheit,
include_ensemble=include_ensemble,
include_multi_model=include_multi_model,
)
else:
fallback_utc_offset = int(
@@ -940,30 +989,41 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
)
if metar_data:
results["metar"] = metar_data
if supports_aviationweather and city_lower != "hong kong":
if include_taf and supports_aviationweather and city_lower != "hong kong":
taf_data = self.fetch_taf(city, utc_offset=fallback_utc_offset)
if taf_data:
results["taf"] = taf_data
self._attach_turkish_mgm_data(results, city_lower)
self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
self._attach_korea_official_nearby(results, city_lower, use_fahrenheit)
self._attach_russia_official_nearby(results, city_lower, use_fahrenheit)
if city_lower == "warsaw":
self._attach_warsaw_official_nearby(results, use_fahrenheit)
self._attach_global_nearby_cluster(
results, city_lower, use_fahrenheit
self._attach_turkish_mgm_data(
results,
city_lower,
include_nearby=include_nearby,
)
if include_nearby:
self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
self._attach_korea_official_nearby(results, city_lower, use_fahrenheit)
self._attach_russia_official_nearby(results, city_lower, use_fahrenheit)
if city_lower == "warsaw":
self._attach_warsaw_official_nearby(results, use_fahrenheit)
self._attach_global_nearby_cluster(
results, city_lower, use_fahrenheit
)
self._attach_nws_and_models(
results, city, lat, lon, use_fahrenheit
results,
city,
lat,
lon,
use_fahrenheit,
include_ensemble=include_ensemble,
include_multi_model=include_multi_model,
)
else:
if supports_aviationweather:
metar_data = self.fetch_metar(city, use_fahrenheit=use_fahrenheit)
if metar_data:
results["metar"] = metar_data
if supports_aviationweather and city_lower != "hong kong":
if include_taf and supports_aviationweather and city_lower != "hong kong":
taf_data = self.fetch_taf(city)
if taf_data:
results["taf"] = taf_data