diff --git a/config/config.yaml b/config/config.yaml index 5d04c2c4..8bb5c764 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -19,8 +19,8 @@ cities: - id: new_york city: New York country: USA - latitude: 40.7128 - longitude: -74.006 + latitude: 40.7769 + longitude: -73.874 - id: chicago city: Chicago country: USA diff --git a/docker-compose.yml b/docker-compose.yml index 106c1c04..bfcdae95 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,6 +29,7 @@ services: - .env environment: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: bot TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-180} @@ -98,6 +99,14 @@ services: env_file: *id001 environment: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} + POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180} + POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true} + POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600} + POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-5} + POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} + POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} POLYWEATHER_SERVICE_ROLE: web healthcheck: interval: 30s diff --git a/docs/AIRPORT_REALTIME_SOURCES.md b/docs/AIRPORT_REALTIME_SOURCES.md index 30eed830..8eb6e4aa 100644 --- a/docs/AIRPORT_REALTIME_SOURCES.md +++ b/docs/AIRPORT_REALTIME_SOURCES.md @@ -11,13 +11,13 @@ | 香港 | CoWIN 6087 | 6087 | CoWIN (`cowin.hku.hk`) | 1 分钟 | 参考站温度（保良局陈守仁小学） | 免费 | | 香港 | HKO | HKO | HKO 官方 CSV (`data.weather.gov.hk`) | 10 分钟 | 官方气象站温度 | 免费 | | 台北 | 松山/中央气象署 | 466920 | CWA 开放数据 | 10 分钟 | 官方站点温度 | 免费 | -| 北京 | 首都机场 | ZBAA | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 上海 | 浦东机场 | ZSPD | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 广州 | 白云机场 | ZGGG | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 成都 | 双流机场 | ZUUU | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 重庆 | 江北机场 | ZUCK | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 武汉 | 天河机场 | ZHHH | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | -| 青岛 | 胶东机场 | ZSQD | AMSC AWOS | 1 分钟 | 跑道端点气温 | 免费 | +| 北京 | 首都机场 | ZBAA | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 上海 | 浦东机场 | ZSPD | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 广州 | 白云机场 | ZGGG | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 成都 | 双流机场 | ZUUU | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 重庆 | 江北机场 | ZUCK | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 武汉 | 天河机场 | ZHHH | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | +| 青岛 | 胶东机场 | ZSQD | AMSC AWOS | 3 分钟 | 跑道端点气温 | 免费 | | 东京 | 羽田 | RJTT | JMA AMeDAS (`jma.go.jp`) | 10 分钟 | 机场站点实时温度 | 免费 | | 安卡拉 | Esenboğa | 17128 | MGM (`servis.mgm.gov.tr`) | 5-15 分钟 | 机场站点实时温度 | 免费 | | 伊斯坦布尔 | 伊斯坦布尔机场 | 17058 | MGM (`servis.mgm.gov.tr`) | 5-15 分钟 | 机场站点实时温度 | 免费 | @@ -56,11 +56,20 @@ > `MID_TEMP` / `END_TEMP`。这些字段是跑道观测位置气温，不是道面温度。 > 结算跑道展示使用配置的结算端点；辅助跑道只作为背景曲线。 -## 推送机制 +## 独立观测采集器 + +- Web/API 进程启动 `observation-collector` 后台线程，按源频率独立采集，不依赖 Telegram 推送循环 +- 默认频率：AMOS 60s、AMSC AWOS 180s、MADIS HFMETAR 300s、CoWIN 60s、HKO 600s +- 每次采集复用 `weather_sources.py` 现有 `_attach_*` 写入逻辑，负责写 `airport_obs_log` / `runway_obs_log` / 今日观测缓存，并通过 `/api/internal/collector-patch` 写 Redis Stream 或 SQLite event log 后广播 SSE +- 采集成功后刷新对应城市 `panel` cache；前端继续使用 HTTP snapshot + SSE patch，不需要依赖 Telegram 触发更新 +- `observation_source_gate.py` 对 AMSC、AMOS、MADIS、HKO、CoWIN 做 per-source/per-city singleflight 和 SQLite cooldown，防止 Web 请求、collector 和兜底分析同时打同一个外部源 + +## Telegram 推送机制 - 每城按原生频率独立推送，不捆绑 -- 首尔/釜山 60s，其余 600s +- 首尔/釜山 60s，中国 AMSC 城市 180s，其余 600s - 循环轮询 60s 以匹配最快频率 +- Telegram 推送优先读取网站侧 `full`/`panel` 城市缓存；缓存缺失时只做非强制 `panel` 分析兜底，不触发 `force_refresh_observations_only` - 仅当当前温度距 DEB 预测最高 ≤3°C 时推送 - 确认过峰值后自动停止 @@ -69,7 +78,7 @@ 为了向用户提供接近行情盘的实况响应并降低服务器负载，系统使用 **HTTP snapshot + Server-Sent Events (SSE) Patch + 可重放事件日志** 架构。生产环境推荐 Redis Stream；本地或单进程可回退 SQLite event log。 ### 1. 数据推送链路 (Data Pipeline) -1. **Collector 采集端触发**：在 `weather_sources.py` 中，当高频实况源（如 AMOS, CoWIN, MADIS 等）采集到温度更新或观测时间变更时，会调用 `_emit_temperature_patch_if_changed` 过滤重复值，并异步向 `/api/internal/collector-patch` 发送 POST 报文。 +1. **Observation Collector 采集端触发**：`web.observation_collector_service` 按源频率调用采集层；在 `weather_sources.py` 中，当高频实况源（如 AMOS, AMSC, CoWIN, HKO, MADIS 等）采集到温度更新或观测时间变更时，会调用 `_emit_temperature_patch_if_changed` 过滤重复值，并异步向 `/api/internal/collector-patch` 发送 POST 报文。 2. **标准化事件**：`realtime_patch_schema.py` 将旧 `city_patch` 或新 payload 统一成 `city_observation_patch.v1`。 3. **事件存储**：生产环境写入 Redis Stream（`stream:city_observation`）并生成全局递增 `revision`；SQLite `observation_patch_events` 保留为本地/兜底 replay。 4. **FastAPI SSE 广播**：FastAPI 后端的 `sse_router.py` 根据城市订阅集合向匹配连接推送 patch；断线重连时按 `since_revision` replay。 diff --git a/docs/CITY_DATA_SOURCES.md b/docs/CITY_DATA_SOURCES.md index 2e424491..a560148e 100644 --- a/docs/CITY_DATA_SOURCES.md +++ b/docs/CITY_DATA_SOURCES.md @@ -13,13 +13,13 @@ | hong kong | CoWIN 6087 | ~1 min | cowin.hku.hk, 保良局陳守仁小學，前端图表默认展示 | | hong kong | HKO 官方 CSV | ~10 min | data.weather.gov.hk（文件名虽含 1min，实际 10min 一报） | | singapore | MSS 官方 API | ~1 min | api.data.gov.sg, 站号 S24 | -| beijing | AMSC AWOS (ZBAA) | ~1 min | 中国 | -| shanghai | AMSC AWOS (ZSPD) | ~1 min | 中国 | -| guangzhou | AMSC AWOS (ZGGG) | ~1 min | 中国 | -| chengdu | AMSC AWOS (ZUUU) | ~1 min | 中国 | -| chongqing | AMSC AWOS (ZUCK) | ~1 min | 中国 | -| wuhan | AMSC AWOS (ZHHH) | ~1 min | 中国 | -| qingdao | AMSC AWOS (ZSQD) | ~1 min | 中国 | +| beijing | AMSC AWOS (ZBAA) | 3 min | 中国 | +| shanghai | AMSC AWOS (ZSPD) | 3 min | 中国 | +| guangzhou | AMSC AWOS (ZGGG) | 3 min | 中国 | +| chengdu | AMSC AWOS (ZUUU) | 3 min | 中国 | +| chongqing | AMSC AWOS (ZUCK) | 3 min | 中国 | +| wuhan | AMSC AWOS (ZHHH) | 3 min | 中国 | +| qingdao | AMSC AWOS (ZSQD) | 3 min | 中国 | ### Tier 2 — 5 分钟高频 (MADIS) @@ -123,7 +123,8 @@ ankara, istanbul, helsinki, amsterdam, paris 频率取决于源头： -- AMSC / AMOS / CoWIN / MSS：源头约 1 分钟，图表按 1 分钟粒度追加。 +- AMOS / CoWIN / MSS：源头约 1 分钟，图表按 1 分钟粒度追加。 +- AMSC：中国跑道观测城市按 3 分钟采集，不再强制 60 秒刷新。 - MADIS：源头约 5 分钟。 - HKO / CWA / JMA / FMI / KNMI：源头约 10 分钟。 - METAR-only 城市：按 METAR 可用频率和缓存 TTL，不伪装成 1 分钟实测。 diff --git a/requirements.txt b/requirements.txt index b93aee76..a0e5a101 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ httpx loguru pyTelegramBotAPI python-dotenv +PyYAML netCDF4 numpy web3 diff --git a/src/data_collection/amos_station_sources.py b/src/data_collection/amos_station_sources.py index c06eb919..5d92fe44 100644 --- a/src/data_collection/amos_station_sources.py +++ b/src/data_collection/amos_station_sources.py @@ -15,6 +15,7 @@ from typing import Any, Dict, Optional from loguru import logger +from src.data_collection.observation_source_gate import run_observation_source from src.utils.metrics import record_source_call AMOS_BASE_URL = os.getenv("AMOS_BASE_URL", "").strip() or "https://global.amo.go.kr/amosobsnew/AmosRealTimeImage.do" @@ -469,6 +470,26 @@ class AmosStationSourceMixin: self, city: str, use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + city_key = str(city or "").strip().lower() + interval_sec = max( + 30, + int(os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC", "60") or "60"), + ) + return run_observation_source( + "amos", + city_key, + interval_sec, + lambda: self._fetch_amos_official_current_uncached( + city_key, + use_fahrenheit=use_fahrenheit, + ), + ) + + def _fetch_amos_official_current_uncached( + self, + city: str, + use_fahrenheit: bool = False, ) -> Optional[Dict[str, Any]]: """Fetch AMOS runway-level observations for Seoul or Busan. diff --git a/src/data_collection/amsc_awos_sources.py b/src/data_collection/amsc_awos_sources.py index 3d03892a..33a7f7c3 100644 --- a/src/data_collection/amsc_awos_sources.py +++ b/src/data_collection/amsc_awos_sources.py @@ -19,6 +19,7 @@ from urllib.request import Request, urlopen from loguru import logger +from src.data_collection.observation_source_gate import run_observation_source from src.utils.metrics import record_source_call from src.utils.runtime_secrets import get_runtime_secret @@ -343,6 +344,27 @@ class AmscAwosSourceMixin: city_key: str, *, use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + normalized_city = str(city_key or "").strip().lower() + interval_sec = max( + 60, + int(os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC", "180") or "180"), + ) + return run_observation_source( + "amsc_awos", + normalized_city, + interval_sec, + lambda: self._fetch_amsc_awos_current_uncached( + normalized_city, + use_fahrenheit=use_fahrenheit, + ), + ) + + def _fetch_amsc_awos_current_uncached( + self, + city_key: str, + *, + use_fahrenheit: bool = False, ) -> Optional[Dict[str, Any]]: del use_fahrenheit # AMSC reports Celsius; project UI converts elsewhere if needed. normalized_city = str(city_key or "").strip().lower() diff --git a/src/data_collection/cowin_sources.py b/src/data_collection/cowin_sources.py index 0fd242c9..0fe6f958 100644 --- a/src/data_collection/cowin_sources.py +++ b/src/data_collection/cowin_sources.py @@ -16,6 +16,7 @@ from urllib.parse import urlencode import requests from loguru import logger +from src.data_collection.observation_source_gate import run_observation_source from src.utils.metrics import record_source_call COWIN_BASE_URL = os.getenv("COWIN_BASE_URL", "").strip() or "https://cowin.hku.hk" @@ -94,6 +95,26 @@ class CowinSourceMixin: self, city: str, use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + city_key = str(city or "").strip().lower() + interval_sec = max( + 30, + int(os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC", "60") or "60"), + ) + return run_observation_source( + "cowin_obs", + city_key, + interval_sec, + lambda: self._fetch_cowin_obs_current_uncached( + city_key, + use_fahrenheit=use_fahrenheit, + ), + ) + + def _fetch_cowin_obs_current_uncached( + self, + city: str, + use_fahrenheit: bool = False, ) -> Optional[Dict[str, Any]]: started = time.perf_counter() city_key = str(city or "").strip().lower() diff --git a/src/data_collection/hko_obs_sources.py b/src/data_collection/hko_obs_sources.py index c2d4792f..a7ad180e 100644 --- a/src/data_collection/hko_obs_sources.py +++ b/src/data_collection/hko_obs_sources.py @@ -17,6 +17,7 @@ from typing import Any, Dict, Optional from loguru import logger # pyrefly: ignore [missing-import] +from src.data_collection.observation_source_gate import run_observation_source from src.utils.metrics import record_source_call HKO_BASE_URL = os.getenv("HKO_BASE_URL", "").strip() or "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather" @@ -67,6 +68,26 @@ class HkoObsSourceMixin: self, city: str, use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + city_key = (city or "").strip().lower() + interval_sec = max( + 60, + int(os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC", "600") or "600"), + ) + return run_observation_source( + "hko_obs", + city_key, + interval_sec, + lambda: self._fetch_hko_obs_current_uncached( + city_key, + use_fahrenheit=use_fahrenheit, + ), + ) + + def _fetch_hko_obs_current_uncached( + self, + city: str, + use_fahrenheit: bool = False, ) -> Optional[Dict[str, Any]]: started = time.perf_counter() city_key = (city or "").strip().lower() diff --git a/src/data_collection/madis_sources.py b/src/data_collection/madis_sources.py index d19b5459..c9dfe2cd 100644 --- a/src/data_collection/madis_sources.py +++ b/src/data_collection/madis_sources.py @@ -10,12 +10,14 @@ URL: https://madis-data.ncep.noaa.gov/madisPublic1/data/LDAD/hfmetar/netCDF/ from __future__ import annotations import gzip +import os import time from datetime import datetime, timezone from typing import Any, Dict, Optional, List from loguru import logger +from src.data_collection.observation_source_gate import run_observation_source from src.utils.metrics import record_source_call MADIS_HFMETAR_URL = ( @@ -177,6 +179,19 @@ class MadisSourceMixin: def fetch_madis_hfmetar(self) -> List[Dict[str, Any]]: """Fetch latest MADIS HFMETAR data and return parsed observations.""" + interval_sec = max( + 60, + int(os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC", "300") or "300"), + ) + return run_observation_source( + "madis_hfmetar", + "global", + interval_sec, + self._fetch_madis_hfmetar_uncached, + ) or [] + + def _fetch_madis_hfmetar_uncached(self) -> List[Dict[str, Any]]: + """Fetch latest MADIS HFMETAR data without source-level gate.""" started = time.perf_counter() fname = self._madis_latest_file() diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index 3625e7e7..8ac2f101 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -34,6 +34,54 @@ OPEN_METEO_MULTI_MODEL_SPECS: Dict[str, Dict[str, Any]] = { "resolution_km": None, "horizon": "16d", }, + "gfs_global": { + "label": "GFS Global", + "provider": "NOAA", + "model": "GFS Global 0.11°/0.25°", + "tier": "global", + "resolution_km": 13, + "horizon": "16d", + }, + "ncep_hrrr_conus": { + "label": "HRRR", + "provider": "NOAA", + "model": "HRRR CONUS", + "tier": "short_range_north_america", + "resolution_km": 3, + "horizon": "18h/48h", + }, + "ncep_nbm_conus": { + "label": "NBM", + "provider": "NOAA", + "model": "National Blend of Models CONUS", + "tier": "regional_north_america", + "resolution_km": 2.5, + "horizon": "11d", + }, + "ncep_nam_conus": { + "label": "NAM", + "provider": "NOAA", + "model": "NAM CONUS", + "tier": "short_range_north_america", + "resolution_km": 3, + "horizon": "60h", + }, + "ncep_gfs_graphcast025": { + "label": "GFS GraphCast", + "provider": "Google/NOAA", + "model": "GFS GraphCast 0.25°", + "tier": "ai_global", + "resolution_km": 25, + "horizon": "16d", + }, + "ncep_aigfs025": { + "label": "AI-GFS", + "provider": "NOAA", + "model": "AIGFS 0.25°", + "tier": "ai_global", + "resolution_km": 25, + "horizon": "16d", + }, "icon_seamless": { "label": "ICON", "provider": "DWD", @@ -784,7 +832,7 @@ class NwsOpenMeteoSourceMixin: 模型列表: - ECMWF IFS / AIFS - - GFS (美国 NOAA) + - GFS / HRRR / NBM / NAM / AI-GFS (美国 NOAA) - ICON Seamless / ICON-EU / ICON-D2 (德国气象局 DWD) - GEM Seamless / GDPS / RDPS / HRDPS (加拿大 ECCC) - JMA (日本气象厅) @@ -893,7 +941,7 @@ class NwsOpenMeteoSourceMixin: "model_keys": model_keys, "dates": dates, "unit": "fahrenheit" if use_fahrenheit else "celsius", - "attribution": "Open-Meteo forecast model API; underlying models from ECMWF, DWD, ECCC, NOAA and JMA.", + "attribution": "Open-Meteo forecast model API; underlying models from ECMWF, DWD, ECCC, NOAA/NCEP, Google and JMA.", } with self._multi_model_cache_lock: previous = self._multi_model_cache.get(cache_key) diff --git a/src/data_collection/observation_source_gate.py b/src/data_collection/observation_source_gate.py new file mode 100644 index 00000000..752f0051 --- /dev/null +++ b/src/data_collection/observation_source_gate.py @@ -0,0 +1,147 @@ +"""Shared singleflight and cooldown guard for high-frequency observation sources.""" + +from __future__ import annotations + +import os +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable, Dict, Hashable, Optional, Tuple, TypeVar + +from loguru import logger + +from src.database.db_manager import DBManager + +T = TypeVar("T") + + +ObservationSourceKey = Tuple[str, str] + + +@dataclass +class _SourceState: + result: Any = None + has_result: bool = False + cooldown_until_ts: float = 0.0 + inflight: Optional[threading.Event] = None + error: Optional[BaseException] = None + + +_GATE_LOCK = threading.Lock() +_SOURCE_STATES: Dict[ObservationSourceKey, _SourceState] = {} + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def _normalize_part(value: Any) -> str: + return str(value or "").strip().lower() + + +def _state_for(key: ObservationSourceKey) -> _SourceState: + state = _SOURCE_STATES.get(key) + if state is None: + state = _SourceState() + _SOURCE_STATES[key] = state + return state + + +def _acquire_cross_process_cooldown(key: ObservationSourceKey, ttl_sec: int) -> bool: + if not _env_bool("POLYWEATHER_OBSERVATION_SOURCE_DB_LOCK_ENABLED", True): + return True + cache_key = f"observation-source:{key[0]}:{key[1]}" + try: + owner = DBManager().acquire_cache_refresh_lock( + cache_key, + ttl_sec=max(15, int(ttl_sec or 60)), + ) + except Exception as exc: + logger.debug("observation source DB cooldown skipped key={}: {}", cache_key, exc) + return True + return bool(owner) + + +def run_observation_source( + source: Hashable, + city_or_scope: Hashable, + interval_sec: int, + fetcher: Callable[[], T], + *, + failure_cooldown_sec: int = 30, +) -> Optional[T]: + """Run a source fetch once per source/city interval and share in-flight work. + + The in-process gate returns the previous result during cooldown. The SQLite + lock is intentionally left to expire instead of being released so multiple + service processes also observe a coarse source cooldown. + """ + if not _env_bool("POLYWEATHER_OBSERVATION_SOURCE_GATE_ENABLED", True): + return fetcher() + + source_key = _normalize_part(source) + scope_key = _normalize_part(city_or_scope) + if not source_key or not scope_key: + return fetcher() + interval = max(1, int(interval_sec or 60)) + key: ObservationSourceKey = (source_key, scope_key) + + while True: + wait_event: Optional[threading.Event] = None + now_ts = time.time() + with _GATE_LOCK: + state = _state_for(key) + if now_ts < state.cooldown_until_ts: + return state.result if state.has_result else None + if state.inflight is None: + event = threading.Event() + state.inflight = event + state.error = None + break + wait_event = state.inflight + if wait_event is not None: + wait_event.wait(timeout=max(5.0, float(interval))) + with _GATE_LOCK: + state = _state_for(key) + if state.error is not None: + raise state.error + if state.has_result: + return state.result + if state.inflight is None: + continue + return None + + owner_event = event + try: + if not _acquire_cross_process_cooldown(key, interval): + with _GATE_LOCK: + state = _state_for(key) + return state.result if state.has_result else None + result = fetcher() + with _GATE_LOCK: + state = _state_for(key) + state.result = result + state.has_result = True + state.cooldown_until_ts = time.time() + interval + state.error = None + return result + except BaseException as exc: + with _GATE_LOCK: + state = _state_for(key) + state.error = exc + state.cooldown_until_ts = time.time() + max(1, int(failure_cooldown_sec or 30)) + raise + finally: + with _GATE_LOCK: + state = _state_for(key) + if state.inflight is owner_event: + state.inflight = None + owner_event.set() + + +def reset_observation_source_gate_for_tests() -> None: + with _GATE_LOCK: + _SOURCE_STATES.clear() diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 6da550a5..c1e5bd28 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -183,8 +183,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) self.open_meteo_multi_model_cache_ttl_sec = min(self.open_meteo_multi_model_cache_ttl_sec, MODEL_CACHE_TTL_SEC) self.multi_model_cache_version = str( - os.getenv("OPEN_METEO_MULTI_MODEL_CACHE_VERSION", "v3") - ).strip() or "v3" + os.getenv("OPEN_METEO_MULTI_MODEL_CACHE_VERSION", "v4") + ).strip() or "v4" self._open_meteo_cache: Dict[str, Dict] = {} self._ensemble_cache: Dict[str, Dict] = {} self._multi_model_cache: Dict[str, Dict] = {} @@ -263,8 +263,9 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._madis_cache_lock = threading.Lock() self._hko_obs_cache_lock = threading.Lock() self.cowin_obs_cache_ttl_sec = int( - os.getenv("COWIN_OBS_CACHE_TTL_SEC", "120") + os.getenv("COWIN_OBS_CACHE_TTL_SEC", str(OBSERVATION_REFRESH_SEC)) ) + self.cowin_obs_cache_ttl_sec = min(self.cowin_obs_cache_ttl_sec, OBSERVATION_REFRESH_SEC) self._cowin_obs_cache: Dict[str, Dict] = {} self._cowin_obs_cache_lock = threading.Lock() self.cwa_open_data_auth = ( @@ -1732,10 +1733,10 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_wunderground_historical(results, city_lower, use_fahrenheit) if lat and lon: - # 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. + # When force_refresh_observations_only is set by an explicit + # observation refresh caller, skip the OM fetch entirely if cached + # data exists. Stale model data is fine; the caller 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() diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index bf0ce954..da2eba88 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -1373,18 +1373,89 @@ def _get_airport_daily_high(city_weather: Dict[str, Any]): return max_so_far, max_time -# Per-city push interval — unified to 60s, obs_time dedup prevents spam +# Per-city push interval. The loop wakes every minute, but Telegram should +# read recent city cache instead of acting as an upstream observation producer. _AIRPORT_PUSH_INTERVAL = { - "seoul": 60, "busan": 60, "tokyo": 60, "ankara": 60, - "helsinki": 60, "amsterdam": 60, "istanbul": 60, "paris": 60, - "hong kong": 60, "shenzhen": 60, "singapore": 60, "taipei": 60, - "beijing": 60, "shanghai": 60, "guangzhou": 60, "qingdao": 60, - "chengdu": 60, "chongqing": 60, "wuhan": 60, - "new york": 60, "los angeles": 60, "chicago": 60, "denver": 60, - "atlanta": 60, "miami": 60, "san francisco": 60, "houston": 60, - "dallas": 60, "austin": 60, "seattle": 60, - "tel aviv": 60, + city: 600 for city in HIGH_FREQ_AIRPORT_CITIES } +_AIRPORT_PUSH_INTERVAL.update({ + "seoul": 60, + "busan": 60, + "beijing": 180, + "shanghai": 180, + "guangzhou": 180, + "qingdao": 180, + "chengdu": 180, + "chongqing": 180, + "wuhan": 180, +}) + + +def _airport_push_cache_max_age_sec(city: str) -> int: + interval = int(_AIRPORT_PUSH_INTERVAL.get((city or "").strip().lower(), 600) or 600) + return max(90, interval * 2) + + +def _read_cached_airport_city_weather(city: str, max_age_sec: Optional[int] = None) -> Optional[Dict[str, Any]]: + """Read web city cache for Telegram without triggering collection.""" + normalized_city = (city or "").strip().lower() + if not normalized_city: + return None + max_age = int(max_age_sec if max_age_sec is not None else _airport_push_cache_max_age_sec(normalized_city)) + now_ts = time.time() + stale_candidate: Optional[Tuple[float, Dict[str, Any]]] = None + try: + db = DBManager() + for kind in ("full", "panel"): + entry = db.get_city_cache(kind, normalized_city) + if not isinstance(entry, dict): + continue + updated_at_ts = float(entry.get("updated_at_ts") or 0.0) + payload = entry.get("payload") + if updated_at_ts <= 0 or not isinstance(payload, dict): + continue + age_sec = now_ts - updated_at_ts + if age_sec > max_age: + logger.debug( + "airport push cache stale city={} kind={} age_sec={} max_age_sec={}", + normalized_city, + kind, + round(age_sec, 1), + max_age, + ) + if stale_candidate is None or updated_at_ts > stale_candidate[0]: + stale_candidate = (updated_at_ts, dict(payload)) + continue + logger.debug( + "airport push cache hit city={} kind={} age_sec={}", + normalized_city, + kind, + round(max(0.0, age_sec), 1), + ) + return dict(payload) + except Exception as exc: + logger.debug("airport push city cache read failed city={}: {}", normalized_city, exc) + if stale_candidate is not None: + logger.debug("airport push using stale cache city={}", normalized_city) + return stale_candidate[1] + return None + + +def _load_airport_city_weather_for_push(city: str) -> Dict[str, Any]: + cached = _read_cached_airport_city_weather(city) + if cached is not None: + return cached + + from web.app import _analyze # lazy import - only the bot process needs it + + return _analyze( + city, + force_refresh=False, + force_refresh_observations_only=False, + detail_mode="panel", + ) + + # Per-city temperature window threshold (°C below DEB predicted high) # Continental airports: wider window (temp rises steadily over land) # Maritime airports: narrower (sea breeze moderates temp) @@ -1467,12 +1538,10 @@ def _process_airport_city( if now_ts - last_city_ts < city_interval: return None - from web.app import _analyze # lazy import — only the bot process needs it - city_weather: Dict[str, Any] = {} deb_pred: Optional[float] = None try: - city_weather = _analyze(city, force_refresh_observations_only=True) + city_weather = _load_airport_city_weather_for_push(city) deb_raw = (city_weather.get("deb") or {}).get("prediction") if deb_raw is not None: deb_pred = float(deb_raw) @@ -1564,7 +1633,7 @@ def _process_airport_city( and now_ts - last_city_ts > 540): time.sleep(4) try: - city_weather = _analyze(city, force_refresh_observations_only=True) + city_weather = _load_airport_city_weather_for_push(city) deb_raw2 = (city_weather.get("deb") or {}).get("prediction") if deb_raw2 is not None: deb_pred = float(deb_raw2) @@ -1620,6 +1689,21 @@ def _process_airport_city( return None +def _due_airport_cities( + cities: Set[str], + now_ts: int, + last_by_city: Dict[str, Any], +) -> List[str]: + due: List[str] = [] + for city in sorted(cities): + last_city = last_by_city.get(city) or {} + last_city_ts = int(last_city.get("ts") or 0) + city_interval = _AIRPORT_PUSH_INTERVAL.get(city, 600) + if now_ts - last_city_ts >= city_interval: + due.append(city) + return due + + def _run_high_freq_airport_cycle( bot: Any, config: Dict[str, Any], @@ -1630,9 +1714,16 @@ def _run_high_freq_airport_cycle( now_ts = int(time.time()) last_by_city = state.setdefault("last_by_city", {}) max_workers = max(1, min(4, _env_int("TELEGRAM_AIRPORT_PUSH_MAX_WORKERS", 1))) - logger.info("airport cycle tick cities={} max_workers={}", len(HIGH_FREQ_AIRPORT_CITIES), max_workers) + cities = _due_airport_cities(HIGH_FREQ_AIRPORT_CITIES, now_ts, last_by_city) + logger.info( + "airport cycle tick cities={} due={} max_workers={}", + len(HIGH_FREQ_AIRPORT_CITIES), + len(cities), + max_workers, + ) + if not cities: + return False - cities = sorted(HIGH_FREQ_AIRPORT_CITIES) pool = _get_airport_executor(max_workers) futures = { pool.submit( diff --git a/tests/test_amsc_awos_sources.py b/tests/test_amsc_awos_sources.py index 0fd1ca19..fa8acaab 100644 --- a/tests/test_amsc_awos_sources.py +++ b/tests/test_amsc_awos_sources.py @@ -146,6 +146,10 @@ def test_parse_wind_plate_payload_rejects_unauthorized_or_empty_payloads(): def test_fetch_amsc_official_current_uses_domestic_city_whitelist(monkeypatch): import src.data_collection.amsc_awos_sources as _amsc + from src.data_collection.observation_source_gate import reset_observation_source_gate_for_tests + + monkeypatch.setenv("POLYWEATHER_OBSERVATION_SOURCE_DB_LOCK_ENABLED", "false") + reset_observation_source_gate_for_tests() monkeypatch.setattr(_amsc, "AMSC_AWOS_BASE_URL", "https://www.amsc.net.cn/gateway/api/saas/rest/amc/AwosController/getWindPlate") assert _amsc_supported_city_codes()["beijing"] == "ZBAA" assert "new york" not in _amsc_supported_city_codes() diff --git a/tests/test_country_networks.py b/tests/test_country_networks.py index c2d6d1b2..9db5167b 100644 --- a/tests/test_country_networks.py +++ b/tests/test_country_networks.py @@ -1,7 +1,11 @@ import threading from datetime import datetime, timedelta, timezone +from pathlib import Path from typing import Any +import pytest +import yaml + from src.data_collection.country_networks import build_country_network_snapshot from src.data_collection.city_registry import ALIASES, CITY_REGISTRY from src.data_collection.city_time import CITY_TIME_ZONES, get_city_utc_offset_seconds @@ -76,6 +80,17 @@ def test_paris_registry_uses_le_bourget_anchor(): assert _DummyMetarSource.CITY_TO_ICAO["paris"] == "LFPB" +def test_new_york_config_uses_registry_airport_anchor(): + config_path = Path(__file__).resolve().parents[1] / "config" / "config.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + + city_config = next(city for city in config["cities"] if city["id"] == "new_york") + registry = CITY_REGISTRY["new york"] + + assert city_config["latitude"] == pytest.approx(registry["lat"]) + assert city_config["longitude"] == pytest.approx(registry["lon"]) + + def test_turkey_metar_uses_fast_cache_ttl(): source = _DummyMetarSource() diff --git a/tests/test_cowin_sources.py b/tests/test_cowin_sources.py index 5587694e..cb9158f9 100644 --- a/tests/test_cowin_sources.py +++ b/tests/test_cowin_sources.py @@ -5,6 +5,7 @@ import httpx from src.data_collection import cowin_sources from src.data_collection.cowin_sources import CowinSourceMixin +from src.data_collection.observation_source_gate import reset_observation_source_gate_for_tests class _FakeResponse: @@ -33,6 +34,8 @@ class _FakeCowinCollector(CowinSourceMixin): def test_cowin_current_retries_without_tls_verification_when_chain_is_incomplete(monkeypatch): + monkeypatch.setenv("POLYWEATHER_OBSERVATION_SOURCE_DB_LOCK_ENABLED", "false") + reset_observation_source_gate_for_tests() calls = [] def fake_get(url, **kwargs): diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 3f7fc425..d70bc4ab 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -52,12 +52,33 @@ def test_scan_terminal_prewarm_only_runs_for_web_service(monkeypatch): assert app_factory._scan_terminal_prewarm_enabled() is True +def test_observation_collector_only_runs_for_web_service(monkeypatch): + from web import app_factory + + monkeypatch.delenv("POLYWEATHER_SERVICE_ROLE", raising=False) + monkeypatch.delenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", raising=False) + assert app_factory._observation_collector_enabled() is False + + monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "bot") + assert app_factory._observation_collector_enabled() is False + + monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "web") + assert app_factory._observation_collector_enabled() is True + + monkeypatch.setenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", "false") + assert app_factory._observation_collector_enabled() is False + + def test_docker_compose_isolates_scan_terminal_prewarm_to_web_service(): compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") assert "POLYWEATHER_SERVICE_ROLE: web" in compose assert "POLYWEATHER_SERVICE_ROLE: bot" in compose assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in compose + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in compose + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true}" in compose + assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in compose + assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in compose def test_scan_terminal_backend_timeout_returns_before_next_proxy_abort(): diff --git a/tests/test_multi_model_sources.py b/tests/test_multi_model_sources.py index 5c1e74f2..3cc9acb8 100644 --- a/tests/test_multi_model_sources.py +++ b/tests/test_multi_model_sources.py @@ -39,6 +39,32 @@ def test_multi_model_parser_exposes_open_recommended_models(): assert model_keys["RDPS"] == "gem_regional" +def test_multi_model_parser_exposes_north_america_open_meteo_models(): + daily = { + "time": ["2026-06-05", "2026-06-06"], + "temperature_2m_max_gfs_global": [92.1, 91.4], + "temperature_2m_max_ncep_hrrr_conus": [92.3, None], + "temperature_2m_max_ncep_nbm_conus": [87.2, 88.7], + "temperature_2m_max_ncep_nam_conus": [87.9, 91.7], + "temperature_2m_max_ncep_gfs_graphcast025": [84.4, 86.1], + "temperature_2m_max_ncep_aigfs025": [85.9, 88.3], + } + + dates, forecasts, metadata, model_keys = _parse_open_meteo_multi_model_daily(daily) + + assert dates == ["2026-06-05", "2026-06-06"] + assert forecasts["2026-06-05"]["GFS Global"] == 92.1 + assert forecasts["2026-06-05"]["HRRR"] == 92.3 + assert forecasts["2026-06-05"]["NBM"] == 87.2 + assert forecasts["2026-06-05"]["NAM"] == 87.9 + assert forecasts["2026-06-05"]["GFS GraphCast"] == 84.4 + assert forecasts["2026-06-05"]["AI-GFS"] == 85.9 + assert "HRRR" not in forecasts["2026-06-06"] + assert metadata["HRRR"]["provider"] == "NOAA" + assert metadata["NBM"]["tier"] == "regional_north_america" + assert model_keys["AI-GFS"] == "ncep_aigfs025" + + def test_multi_model_order_includes_legacy_and_new_sources(): assert "ecmwf_ifs025" in OPEN_METEO_MULTI_MODEL_ORDER assert "ecmwf_aifs025_single" in OPEN_METEO_MULTI_MODEL_ORDER @@ -50,9 +76,23 @@ def test_multi_model_order_includes_legacy_and_new_sources(): assert "gem_global" in OPEN_METEO_MULTI_MODEL_ORDER assert "gem_regional" in OPEN_METEO_MULTI_MODEL_ORDER assert "gem_hrdps_continental" in OPEN_METEO_MULTI_MODEL_ORDER + assert "gfs_global" in OPEN_METEO_MULTI_MODEL_ORDER + assert "ncep_hrrr_conus" in OPEN_METEO_MULTI_MODEL_ORDER + assert "ncep_nbm_conus" in OPEN_METEO_MULTI_MODEL_ORDER + assert "ncep_nam_conus" in OPEN_METEO_MULTI_MODEL_ORDER + assert "ncep_gfs_graphcast025" in OPEN_METEO_MULTI_MODEL_ORDER + assert "ncep_aigfs025" in OPEN_METEO_MULTI_MODEL_ORDER assert "jma_seamless" in OPEN_METEO_MULTI_MODEL_ORDER +def test_multi_model_default_cache_version_refreshes_noaa_model_set(monkeypatch): + monkeypatch.delenv("OPEN_METEO_MULTI_MODEL_CACHE_VERSION", raising=False) + + collector = WeatherDataCollector({}) + + assert collector.multi_model_cache_version == "v4" + + def test_madis_patch_uses_city_display_unit_for_us(monkeypatch): collector = WeatherDataCollector({}) emitted = [] diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py new file mode 100644 index 00000000..c7f61801 --- /dev/null +++ b/tests/test_observation_collector.py @@ -0,0 +1,132 @@ +from concurrent.futures import ThreadPoolExecutor +import time + + +def test_observation_source_gate_shares_inflight_and_cooldown(monkeypatch): + from src.data_collection.observation_source_gate import ( + reset_observation_source_gate_for_tests, + run_observation_source, + ) + + monkeypatch.setenv("POLYWEATHER_OBSERVATION_SOURCE_DB_LOCK_ENABLED", "false") + reset_observation_source_gate_for_tests() + calls = 0 + + def fetcher(): + nonlocal calls + calls += 1 + time.sleep(0.05) + return {"temp": 23.4} + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit(run_observation_source, "amsc_awos", "qingdao", 180, fetcher), + executor.submit(run_observation_source, "amsc_awos", "qingdao", 180, fetcher), + ] + results = [future.result(timeout=2) for future in futures] + + assert results == [{"temp": 23.4}, {"temp": 23.4}] + assert calls == 1 + + cached = run_observation_source("amsc_awos", "qingdao", 180, fetcher) + + assert cached == {"temp": 23.4} + assert calls == 1 + + +def test_observation_source_gate_respects_failure_cooldown(monkeypatch): + from src.data_collection.observation_source_gate import ( + reset_observation_source_gate_for_tests, + run_observation_source, + ) + + monkeypatch.setenv("POLYWEATHER_OBSERVATION_SOURCE_DB_LOCK_ENABLED", "false") + reset_observation_source_gate_for_tests() + calls = 0 + + def fetcher(): + nonlocal calls + calls += 1 + raise RuntimeError("upstream down") + + try: + run_observation_source( + "cowin_obs", + "hong kong", + 60, + fetcher, + failure_cooldown_sec=60, + ) + except RuntimeError: + pass + + skipped = run_observation_source( + "cowin_obs", + "hong kong", + 60, + fetcher, + failure_cooldown_sec=60, + ) + + assert skipped is None + assert calls == 1 + + +def test_observation_collector_profiles_match_source_cadence(): + from web.observation_collector_service import build_observation_source_profiles + from web.realtime_patch_schema import SOURCE_CADENCE_SECONDS + + profiles = {profile.source: profile for profile in build_observation_source_profiles()} + + assert profiles["amsc_awos"].interval_sec == 180 + assert profiles["amos"].interval_sec == 60 + assert profiles["madis_hfmetar"].interval_sec == 300 + assert profiles["cowin_obs"].interval_sec == 60 + assert profiles["hko_obs"].interval_sec == 600 + assert "qingdao" in profiles["amsc_awos"].cities + assert {"seoul", "busan"}.issubset(set(profiles["amos"].cities)) + assert "new york" in profiles["madis_hfmetar"].cities + assert "hong kong" in profiles["cowin_obs"].cities + assert {"hong kong", "shenzhen"}.issubset(set(profiles["hko_obs"].cities)) + assert SOURCE_CADENCE_SECONDS["amsc_awos"] == 180 + + +def test_observation_collector_run_due_once_refreshes_panel_cache(): + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + calls = [] + refreshed = [] + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + calls.append((city, use_fahrenheit)) + results["amos"] = {"source": "amsc_awos", "temp_c": 24.0} + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ + ObservationSourceProfile( + source="amsc_awos", + cities=("qingdao",), + interval_sec=180, + ) + ], + cache_refresher=lambda city: refreshed.append(city), + ) + + assert collector.run_due_once(now_ts=1000.0) == 1 + assert calls == [("qingdao", False)] + assert refreshed == ["qingdao"] + + assert collector.run_due_once(now_ts=1100.0) == 0 + assert calls == [("qingdao", False)] + + assert collector.run_due_once(now_ts=1180.0) == 1 + assert calls == [("qingdao", False), ("qingdao", False)] + assert refreshed == ["qingdao", "qingdao"] diff --git a/tests/test_refresh_policy.py b/tests/test_refresh_policy.py index d81807b3..dc50df5e 100644 --- a/tests/test_refresh_policy.py +++ b/tests/test_refresh_policy.py @@ -28,6 +28,7 @@ def test_backend_defaults_use_refresh_policy(): source = weather_sources.WeatherDataCollector({}) assert source.metar_cache_ttl_sec == METAR_POLL_TTL_SEC assert source.hko_obs_cache_ttl_sec == OBSERVATION_REFRESH_SEC + assert source.cowin_obs_cache_ttl_sec == OBSERVATION_REFRESH_SEC assert source.settlement_cache_ttl_sec == OBSERVATION_REFRESH_SEC assert source.open_meteo_cache_ttl_sec == MODEL_CACHE_TTL_SEC assert source.open_meteo_multi_model_cache_ttl_sec == MODEL_CACHE_TTL_SEC diff --git a/tests/test_telegram_hashtags.py b/tests/test_telegram_hashtags.py index 2b78979b..7b834a9f 100644 --- a/tests/test_telegram_hashtags.py +++ b/tests/test_telegram_hashtags.py @@ -1,6 +1,7 @@ from src.utils.telegram_push import ( HIGH_FREQ_AIRPORT_CITIES, HIGH_FREQ_AIRPORT_ICAO, + _AIRPORT_PUSH_INTERVAL, _build_airport_status_message, _compute_slope_15m, _run_high_freq_airport_cycle, @@ -169,23 +170,41 @@ def test_shenzhen_is_in_high_freq_push_as_hko_station(): assert HIGH_FREQ_AIRPORT_ICAO["shenzhen"] == "LFS" -def test_high_freq_airport_push_forces_analysis_refresh(monkeypatch): +def test_china_airport_push_defaults_to_three_minute_city_interval(): + assert _AIRPORT_PUSH_INTERVAL["seoul"] == 60 + assert _AIRPORT_PUSH_INTERVAL["busan"] == 60 + assert _AIRPORT_PUSH_INTERVAL["shanghai"] == 180 + assert _AIRPORT_PUSH_INTERVAL["beijing"] == 180 + assert _AIRPORT_PUSH_INTERVAL["guangzhou"] == 180 + assert _AIRPORT_PUSH_INTERVAL["qingdao"] == 180 + assert _AIRPORT_PUSH_INTERVAL["chengdu"] == 180 + assert _AIRPORT_PUSH_INTERVAL["chongqing"] == 180 + assert _AIRPORT_PUSH_INTERVAL["wuhan"] == 180 + + +def test_high_freq_airport_push_prefers_fresh_city_cache(monkeypatch): import src.utils.telegram_push as telegram_push import web.app as web_app - calls = [] + def fail_analyze(*_args, **_kwargs): + raise AssertionError("airport Telegram push should read fresh city cache before _analyze") - def fake_analyze(city, force_refresh=False, force_refresh_observations_only=False, **_kwargs): - calls.append((city, force_refresh, force_refresh_observations_only)) - return { - "local_time": "12:00", - "current": {"temp": 31.0}, - "deb": {"prediction": 29.0}, - "airport_current": {"max_so_far": 30.0, "max_temp_time": "11:50", "obs_time": "12:00"}, - "mgm_nearby": [ - {"icao": "ZSQD", "temp": 31.0, "obs_time": "2026-05-17T04:00:00Z"}, - ], - } + class FakeDB: + def get_city_cache(self, kind, city): + if city != "qingdao" or kind != "full": + return None + return { + "updated_at_ts": telegram_push.time.time(), + "payload": { + "local_time": "12:00", + "current": {"temp": 31.0}, + "deb": {"prediction": 29.0}, + "airport_current": {"max_so_far": 30.0, "max_temp_time": "11:50", "obs_time": "12:00"}, + "mgm_nearby": [ + {"icao": "ZSQD", "temp": 31.0, "obs_time": "2026-05-17T04:00:00Z"}, + ], + }, + } class Bot: def __init__(self): @@ -196,7 +215,13 @@ def test_high_freq_airport_push_forces_analysis_refresh(monkeypatch): bot = Bot() monkeypatch.setattr(telegram_push, "HIGH_FREQ_AIRPORT_CITIES", {"qingdao"}) - monkeypatch.setattr(web_app, "_analyze", fake_analyze) + monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) + monkeypatch.setattr( + telegram_push, + "_rate_limited_send", + lambda bot, chat_id, message, **_kwargs: bot.send_message(chat_id, message), + ) + monkeypatch.setattr(web_app, "_analyze", fail_analyze) sent = _run_high_freq_airport_cycle( bot=bot, @@ -206,13 +231,96 @@ def test_high_freq_airport_push_forces_analysis_refresh(monkeypatch): ) assert sent is True - # All cities processed with force_refresh_observations_only=True - assert len(calls) >= 1 - assert all(c[2] is True for c in calls) - assert ("qingdao", False, True) in calls assert bot.messages +def test_airport_push_fallback_analysis_does_not_force_observation_refresh(monkeypatch): + import src.utils.telegram_push as telegram_push + import web.app as web_app + + calls = [] + + class FakeDB: + def get_city_cache(self, kind, city): + return None + + def fake_analyze(city, force_refresh=False, force_refresh_observations_only=False, detail_mode="full", **_kwargs): + calls.append((city, force_refresh, force_refresh_observations_only, detail_mode)) + return { + "local_time": "12:00", + "current": {"temp": 31.0}, + "deb": {"prediction": 29.0}, + "airport_current": {"max_so_far": 30.0, "max_temp_time": "11:50", "obs_time": "12:00"}, + "mgm_nearby": [ + {"icao": "ZSQD", "temp": 31.0, "obs_time": "2026-05-17T04:00:00Z"}, + ], + } + + monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) + monkeypatch.setattr(web_app, "_analyze", fake_analyze) + + city_weather = telegram_push._load_airport_city_weather_for_push("qingdao") + + assert city_weather["current"]["temp"] == 31.0 + assert ("qingdao", False, False, "panel") in calls + assert not any(city == "qingdao" and force_obs for city, _force, force_obs, _mode in calls) + + +def test_airport_push_uses_stale_cache_before_fallback_analysis(monkeypatch): + import src.utils.telegram_push as telegram_push + import web.app as web_app + + def fail_analyze(*_args, **_kwargs): + raise AssertionError("stale city cache should still prevent Telegram fallback analysis") + + class FakeDB: + def get_city_cache(self, kind, city): + if kind != "panel": + return None + return { + "updated_at_ts": 1.0, + "payload": { + "local_time": "12:00", + "current": {"temp": 31.0}, + "deb": {"prediction": 29.0}, + "airport_current": {"max_so_far": 30.0, "max_temp_time": "11:50", "obs_time": "12:00"}, + "mgm_nearby": [ + {"icao": "ZSQD", "temp": 31.0, "obs_time": "2026-05-17T04:00:00Z"}, + ], + }, + } + + monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) + monkeypatch.setattr(web_app, "_analyze", fail_analyze) + + city_weather = telegram_push._load_airport_city_weather_for_push("qingdao") + + assert city_weather["current"]["temp"] == 31.0 + + +def test_high_freq_airport_cycle_skips_cities_before_interval(monkeypatch): + import src.utils.telegram_push as telegram_push + + calls = [] + monkeypatch.setattr(telegram_push, "HIGH_FREQ_AIRPORT_CITIES", {"shanghai"}) + monkeypatch.setattr(telegram_push.time, "time", lambda: 1000.0) + monkeypatch.setattr( + telegram_push, + "_process_airport_city", + lambda *args, **kwargs: calls.append((args, kwargs)), + ) + + dirty = telegram_push._run_high_freq_airport_cycle( + bot=object(), + config={}, + chat_ids=["chat-1"], + state={"last_by_city": {"shanghai": {"ts": 999}}}, + ) + + assert dirty is False + assert calls == [] + + def test_high_freq_airport_push_workers_default_to_one_for_shared_cpu(monkeypatch): source = Path("src/utils/telegram_push.py").read_text(encoding="utf-8") assert 'TELEGRAM_AIRPORT_PUSH_MAX_WORKERS", 1' in source diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 7071c073..083664f5 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -2626,6 +2626,52 @@ def test_scan_terminal_timeout_does_not_replace_better_cached_snapshot(monkeypat assert [row["id"] for row in cached["rows"]] == ["old-1", "old-2"] +def test_scan_terminal_cold_requests_share_inflight_build(monkeypatch): + import time + from concurrent.futures import ThreadPoolExecutor + + filters = {"scan_mode": "tradable", "limit": 17, "min_edge_pct": 6.75} + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + scan_terminal_cache._SCAN_TERMINAL_REFRESHING.clear() + monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "false") + + calls = 0 + + def _fake_uncached(filters_arg, *, force_refresh=False, timeout_sec=None): + nonlocal calls + calls += 1 + time.sleep(0.05) + return { + "generated_at": "2026-06-05T00:00:00Z", + "filters": dict(filters_arg), + "summary": {"candidate_total": 1}, + "top_signal": None, + "rows": [{"id": "shared-row"}], + "status": "ready", + "stale": False, + "stale_reason": None, + "last_success_at": None, + "last_failed_at": None, + } + + monkeypatch.setattr( + scan_terminal_service, + "_build_scan_terminal_payload_uncached", + _fake_uncached, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit(scan_terminal_service.build_scan_terminal_payload, filters), + executor.submit(scan_terminal_service.build_scan_terminal_payload, filters), + ] + results = [future.result(timeout=2) for future in futures] + + assert calls == 1 + assert results[0]["rows"] == [{"id": "shared-row"}] + assert results[1]["rows"] == [{"id": "shared-row"}] + + def test_scan_terminal_prewarm_builds_default_terminal_payload(monkeypatch): calls = [] diff --git a/web/app_factory.py b/web/app_factory.py index 8bedb032..48764ad3 100644 --- a/web/app_factory.py +++ b/web/app_factory.py @@ -23,6 +23,7 @@ from web.routes import router as legacy_router from web.scan_terminal_service import start_scan_terminal_prewarm _ROUTES_REGISTERED_FLAG = "_polyweather_routes_registered" +_OBSERVATION_COLLECTOR_STARTED_FLAG = "_polyweather_observation_collector_started" def _service_role() -> str: @@ -36,6 +37,13 @@ def _scan_terminal_prewarm_enabled() -> bool: return enabled and _service_role() in {"web", "api", "backend"} +def _observation_collector_enabled() -> bool: + enabled = str( + os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED") or "true" + ).strip().lower() in {"1", "true", "yes", "on"} + return enabled and _service_role() in {"web", "api", "backend"} + + def create_app() -> FastAPI: """Return the configured FastAPI app with routers registered once.""" if not bool(getattr(core_app.state, _ROUTES_REGISTERED_FLAG, False)): @@ -52,4 +60,17 @@ def create_app() -> FastAPI: setattr(core_app.state, _ROUTES_REGISTERED_FLAG, True) if _scan_terminal_prewarm_enabled(): start_scan_terminal_prewarm() + if ( + _observation_collector_enabled() + and not bool(getattr(core_app.state, _OBSERVATION_COLLECTOR_STARTED_FLAG, False)) + ): + from web.core import _weather + from web.services.city_runtime import _refresh_city_panel_cache + from web.observation_collector_service import start_observation_collector_loop + + thread = start_observation_collector_loop( + weather=_weather, + cache_refresher=lambda city: _refresh_city_panel_cache(city, force_refresh=False), + ) + setattr(core_app.state, _OBSERVATION_COLLECTOR_STARTED_FLAG, bool(thread)) return core_app diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py new file mode 100644 index 00000000..357e18a7 --- /dev/null +++ b/web/observation_collector_service.py @@ -0,0 +1,205 @@ +"""Independent high-frequency observation collector for the web runtime.""" + +from __future__ import annotations + +import os +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable, Iterable, List, Optional, Sequence, Tuple + +from loguru import logger + +from src.data_collection.amos_station_sources import AMOS_AIRPORT_CODES +from src.data_collection.amsc_awos_sources import AMSC_AWOS_AIRPORTS +from src.data_collection.city_registry import CITY_REGISTRY +from src.data_collection.hko_obs_sources import HKO_STATIONS + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(raw) + except Exception: + return default + + +def _normalized_cities(cities: Iterable[str]) -> Tuple[str, ...]: + return tuple(sorted({str(city or "").strip().lower() for city in cities if str(city or "").strip()})) + + +@dataclass(frozen=True) +class ObservationSourceProfile: + source: str + cities: Tuple[str, ...] + interval_sec: int + + +class ObservationCollector: + def __init__( + self, + *, + weather: Any, + profiles: Sequence[ObservationSourceProfile], + cache_refresher: Optional[Callable[[str], Any]] = None, + ) -> None: + self.weather = weather + self.profiles = list(profiles) + self.cache_refresher = cache_refresher + self._last_run_ts: dict[tuple[str, str], float] = {} + self._lock = threading.Lock() + + def run_due_once(self, *, now_ts: Optional[float] = None) -> int: + now = float(time.time() if now_ts is None else now_ts) + due: List[tuple[ObservationSourceProfile, str]] = [] + with self._lock: + for profile in self.profiles: + interval = max(1, int(profile.interval_sec or 60)) + for city in profile.cities: + key = (profile.source, city) + last_ts = float(self._last_run_ts.get(key) or 0.0) + if now - last_ts >= interval: + self._last_run_ts[key] = now + due.append((profile, city)) + + completed = 0 + for profile, city in due: + try: + if self._collect_city_source(profile.source, city): + completed += 1 + self._refresh_city_cache(city) + except Exception as exc: + logger.warning( + "observation collector source failed source={} city={}: {}", + profile.source, + city, + exc, + ) + return completed + + def _collect_city_source(self, source: str, city: str) -> bool: + normalized_source = str(source or "").strip().lower() + normalized_city = str(city or "").strip().lower() + if not normalized_source or not normalized_city: + return False + use_fahrenheit = bool(self.weather._uses_fahrenheit(normalized_city)) + results: dict[str, Any] = {} + + if normalized_source == "amsc_awos": + self.weather._attach_china_amsc_awos_data(results, normalized_city, use_fahrenheit) + elif normalized_source == "amos": + self.weather._attach_korean_amos_data(results, normalized_city, use_fahrenheit) + elif normalized_source == "madis_hfmetar": + self.weather._attach_madis_hfmetar_data(results, normalized_city, use_fahrenheit) + elif normalized_source == "hko_obs": + self.weather._attach_hko_obs_official_nearby(results, normalized_city, use_fahrenheit) + elif normalized_source == "cowin_obs": + self.weather._attach_cowin_official_nearby(results, normalized_city, use_fahrenheit) + else: + logger.debug("observation collector skipped unknown source={}", normalized_source) + return False + return bool(results) + + def _refresh_city_cache(self, city: str) -> None: + if not callable(self.cache_refresher): + return + try: + self.cache_refresher(city) + except Exception as exc: + logger.warning("observation collector cache refresh failed city={}: {}", city, exc) + + +def build_observation_source_profiles() -> List[ObservationSourceProfile]: + us_madis_cities = [ + city + for city, meta in CITY_REGISTRY.items() + if str((meta or {}).get("icao") or "").strip().upper().startswith("K") + ] + return [ + ObservationSourceProfile( + source="amos", + cities=_normalized_cities(AMOS_AIRPORT_CODES.keys()), + interval_sec=max(30, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC", 60)), + ), + ObservationSourceProfile( + source="amsc_awos", + cities=_normalized_cities(AMSC_AWOS_AIRPORTS.keys()), + interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC", 180)), + ), + ObservationSourceProfile( + source="madis_hfmetar", + cities=_normalized_cities(us_madis_cities), + interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC", 300)), + ), + ObservationSourceProfile( + source="cowin_obs", + cities=("hong kong",), + interval_sec=max(30, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC", 60)), + ), + ObservationSourceProfile( + source="hko_obs", + cities=_normalized_cities(HKO_STATIONS.keys()), + interval_sec=max(60, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC", 600)), + ), + ] + + +_COLLECTOR_THREAD: Optional[threading.Thread] = None +_COLLECTOR_LOCK = threading.Lock() + + +def start_observation_collector_loop( + *, + weather: Any, + cache_refresher: Optional[Callable[[str], Any]] = None, + profiles: Optional[Sequence[ObservationSourceProfile]] = None, +) -> Optional[threading.Thread]: + if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True): + return None + tick_sec = max(5, _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC", 30)) + initial_delay_sec = max( + 0, + _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC", 5), + ) + selected_profiles = list(profiles or build_observation_source_profiles()) + collector = ObservationCollector( + weather=weather, + profiles=selected_profiles, + cache_refresher=cache_refresher, + ) + + global _COLLECTOR_THREAD + with _COLLECTOR_LOCK: + if _COLLECTOR_THREAD is not None and _COLLECTOR_THREAD.is_alive(): + return _COLLECTOR_THREAD + + def _runner() -> None: + logger.info( + "observation collector started profiles={} tick_sec={}", + len(selected_profiles), + tick_sec, + ) + if initial_delay_sec: + time.sleep(initial_delay_sec) + while True: + started = time.time() + collector.run_due_once(now_ts=started) + elapsed = time.time() - started + time.sleep(max(1.0, tick_sec - elapsed)) + + _COLLECTOR_THREAD = threading.Thread( + target=_runner, + name="observation-collector", + daemon=True, + ) + _COLLECTOR_THREAD.start() + return _COLLECTOR_THREAD diff --git a/web/realtime_patch_schema.py b/web/realtime_patch_schema.py index 4f07b87d..259bcca1 100644 --- a/web/realtime_patch_schema.py +++ b/web/realtime_patch_schema.py @@ -18,7 +18,7 @@ SCHEMA_VERSION = 1 EVENT_TYPE = "city_observation_patch.v1" SOURCE_CADENCE_SECONDS = { "amos": 60, - "amsc_awos": 60, + "amsc_awos": 180, "cowin_obs": 60, "hko_obs": 600, "singapore_mss": 60, diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index 0f32b535..1e1d3de9 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import re import threading import time +from concurrent.futures import Future from concurrent.futures import TimeoutError as FutureTimeoutError from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime @@ -24,6 +25,7 @@ from web.scan_terminal_cache import ( get_cached_scan_terminal_payload, get_scan_terminal_cache_entry, mark_scan_terminal_refreshing, + scan_terminal_cache_key, set_cached_scan_terminal_payload, set_scan_terminal_failure_state, ) @@ -38,6 +40,11 @@ from web.scan_terminal_payloads import ( compact_ranked_scan_rows_for_payload, ) from web.scan_terminal_ranker import build_ranked_scan_terminal_result + +_SCAN_TERMINAL_INFLIGHT_BUILD_LOCK = threading.Lock() +_SCAN_TERMINAL_INFLIGHT_BUILDS: Dict[str, Future] = {} + + def _normalize_locale(value: Any) -> str: text = str(value or "").strip().lower() return "en-US" if text.startswith("en") else "zh-CN" @@ -84,7 +91,7 @@ def _start_scan_terminal_background_refresh(filters: Dict[str, Any]) -> bool: def _runner() -> None: try: - _build_scan_terminal_payload_uncached(filters, force_refresh=True) + _build_scan_terminal_payload_singleflight(filters, force_refresh=True) except Exception as exc: # pragma: no cover - defensive background guard logger.warning("scan terminal background refresh failed: {}", exc) finally: @@ -279,6 +286,41 @@ def _build_scan_terminal_payload_uncached( ) +def _build_scan_terminal_payload_singleflight( + filters: Dict[str, Any], + *, + force_refresh: bool = False, +) -> Dict[str, Any]: + cache_key = scan_terminal_cache_key(filters) + owner = False + with _SCAN_TERMINAL_INFLIGHT_BUILD_LOCK: + future = _SCAN_TERMINAL_INFLIGHT_BUILDS.get(cache_key) + if future is None: + owner = True + future = Future() + _SCAN_TERMINAL_INFLIGHT_BUILDS[cache_key] = future + + if not owner: + logger.debug("scan terminal joining in-flight build key={}", cache_key) + return future.result() + + try: + payload = _build_scan_terminal_payload_uncached( + filters, + force_refresh=force_refresh, + ) + except Exception as exc: + future.set_exception(exc) + raise + else: + future.set_result(payload) + return payload + finally: + with _SCAN_TERMINAL_INFLIGHT_BUILD_LOCK: + if _SCAN_TERMINAL_INFLIGHT_BUILDS.get(cache_key) is future: + _SCAN_TERMINAL_INFLIGHT_BUILDS.pop(cache_key, None) + + def build_scan_terminal_payload( raw_filters: Optional[Dict[str, Any]] = None, *, @@ -333,13 +375,13 @@ def build_scan_terminal_payload( return ( timing_recorder.measure( "uncached_build", - lambda: _build_scan_terminal_payload_uncached( + lambda: _build_scan_terminal_payload_singleflight( filters, force_refresh=force_refresh, ), ) if timing_recorder is not None - else _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh) + else _build_scan_terminal_payload_singleflight(filters, force_refresh=force_refresh) )