接入 KNMI 阿姆斯特丹史基浦机场 10 分钟数据源

通过 KNMI Open Data API 获取 Schiphol 机场 10 分钟观测数据（NetCDF 格式）。需设置 KNMI_API_KEY 环境变量。Docker 镜像新增 libhdf5-dev/netCDF4 依赖。 Constraint: KNMI API key 为 JWT 格式，通过 Authorization header 传递 Scope-risk: 中高，新增系统依赖 libhdf5-dev + netCDF4 Tested: pytest 176 passed, KNMI API 连通性验证通过
2026-05-12 20:10:06 +08:00
parent a32a49f3c7
commit 5d630bb910
7 changed files with 295 additions and 8 deletions
@@ -10,8 +10,10 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
    PIP_ROOT_USER_ACTION=ignore \
    TZ=UTC

-# 安装系统依赖 (如果有必要的包可以取消注释)
-# RUN apt-get update && apt-get install -y --no-install-recommends gcc && rm -rf /var/lib/apt/lists/*
+# 安装系统依赖 (netCDF4 需要 HDF5 + 编译工具)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc libhdf5-dev libnetcdf-dev && \
+    rm -rf /var/lib/apt/lists/*

 # 复制 requirements 文件
 COPY requirements.txt .
@@ -3,6 +3,7 @@ httpx
 loguru
 pyTelegramBotAPI
 python-dotenv
+netCDF4
 pytz
 numpy
 lightgbm
@@ -813,7 +813,7 @@ def _build_advice_cn(
    return "，".join(parts) + "。"


-_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"}
+_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM"}


 def _calc_airport_rapid_temp_change(
@@ -190,7 +190,7 @@ class StartupCoordinator:
        details = {
            "mode": "airport-periodic",
            "interval_sec": interval,
-            "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki"],
+            "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam"],
            "chat_targets": len(chat_ids),
            "window": "DEB proximity ≤3°C",
        }
@@ -0,0 +1,249 @@
+"""KNMI (Royal Netherlands Meteorological Institute) data source.
+
+Fetches 10-minute weather observations from the KNMI Data Platform
+for Amsterdam Schiphol airport (WMO 06240, station 240).
+Uses netCDF4; requires libhdf5-dev on Linux.
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from loguru import logger
+
+from src.utils.metrics import record_source_call
+
+KNMI_DATASET = "10-minute-in-situ-meteorological-observations"
+KNMI_VERSION = "1.0"
+KNMI_API_BASE = "https://api.dataplatform.knmi.nl/open-data/v1"
+KNMI_STATION = {
+    "amsterdam": {
+        "station": "240",
+        "icao": "EHAM",
+        "label": "Schiphol 10min (KNMI)",
+    },
+}
+
+
+class KnmiSourceMixin:
+    def _knmi_api_key(self) -> str:
+        import os
+        return str(os.getenv("KNMI_API_KEY") or "").strip()
+
+    def _knmi_http_get(self, url: str, api_key: str) -> bytes:
+        headers = {"Authorization": api_key}
+        getter = getattr(self, "_http_get", None)
+        if callable(getter):
+            resp = getter(url)
+            return resp.content if hasattr(resp, "content") else resp
+        resp = self.session.get(url, timeout=self.timeout, headers=headers)
+        resp.raise_for_status()
+        return resp.content
+
+    def _knmi_http_get_json(self, url: str, api_key: str) -> Dict:
+        headers = {"Authorization": api_key}
+        getter = getattr(self, "_http_get_json", None)
+        if callable(getter):
+            return getter(url)
+        resp = self.session.get(url, timeout=self.timeout, headers=headers)
+        resp.raise_for_status()
+        return resp.json()
+
+    def fetch_knmi_current(
+        self,
+        city: str,
+        use_fahrenheit: bool = False,
+    ) -> Optional[Dict[str, Any]]:
+        started = time.perf_counter()
+        city_key = str(city or "").strip().lower()
+        meta = KNMI_STATION.get(city_key) or {}
+        if not meta:
+            record_source_call("knmi", "current", "unsupported_city",
+                               (time.perf_counter() - started) * 1000.0)
+            return None
+
+        api_key = self._knmi_api_key()
+        if not api_key:
+            logger.warning("KNMI_API_KEY not set, skipping fetch")
+            return None
+
+        cache_key = f"knmi:{city_key}:{use_fahrenheit}"
+        now_ts = time.time()
+        with self._knmi_cache_lock:
+            cached = self._knmi_cache.get(cache_key)
+            if cached and now_ts - cached["t"] < self.knmi_cache_ttl_sec:
+                record_source_call("knmi", "current", "cache_hit",
+                                   (time.perf_counter() - started) * 1000.0)
+                return cached["d"]
+
+        base = f"{KNMI_API_BASE}/datasets/{KNMI_DATASET}/versions/{KNMI_VERSION}"
+
+        try:
+            # Get latest file
+            resp = self._knmi_http_get_json(
+                f"{base}/files?maxKeys=1&sorting=desc&orderBy=created",
+                api_key,
+            )
+            files = resp.get("files") or []
+            if not files:
+                record_source_call("knmi", "current", "no_files",
+                                   (time.perf_counter() - started) * 1000.0)
+                return None
+
+            fname = files[0]["filename"]
+
+            # Get download URL
+            url_resp = self._knmi_http_get_json(f"{base}/files/{fname}/url", api_key)
+            download_url = url_resp.get("temporaryDownloadUrl")
+            if not download_url:
+                record_source_call("knmi", "current", "no_download_url",
+                                   (time.perf_counter() - started) * 1000.0)
+                return None
+
+            # Download and parse NetCDF
+            nc_bytes = self._knmi_http_get(download_url, api_key)
+
+            try:
+                from netCDF4 import Dataset
+                nc = Dataset(fname, memory=nc_bytes)
+            except ImportError:
+                logger.error("netCDF4 not installed; KNMI data unavailable")
+                record_source_call("knmi", "current", "netcdf4_missing",
+                                   (time.perf_counter() - started) * 1000.0)
+                return None
+
+            try:
+                stn = meta["station"]
+                # Find station index
+                station_ids = list(nc.variables.get("station_id", [])[:])
+                if not station_ids:
+                    station_codes = []
+                    for s in nc.variables.get("station", []):
+                        try:
+                            station_codes.append(str(int(s)))
+                        except Exception:
+                            station_codes.append("")
+                    station_ids = station_codes
+
+                try:
+                    idx = station_ids.index(stn)
+                except ValueError:
+                    idx = station_ids.index(int(stn)) if stn.isdigit() else -1
+
+                if idx < 0 or idx >= len(station_ids):
+                    logger.warning("KNMI station {} not found in file", stn)
+                    nc.close()
+                    return None
+
+                # Extract latest timestep for temperature
+                ta = nc.variables.get("ta", None)
+                if ta is None:
+                    nc.close()
+                    return None
+
+                data = ta[:]
+                latest_temp = float(data[-1, idx]) if data.ndim == 2 else float(data[-1])
+
+                # Wind speed
+                ff = nc.variables.get("ff", None)
+                wind_ms = None
+                if ff is not None:
+                    wind_data = ff[:]
+                    wind_ms = float(wind_data[-1, idx] if wind_data.ndim == 2 else wind_data[-1])
+
+                # Pressure
+                p0 = nc.variables.get("p0", None)
+                pressure_hpa = None
+                if p0 is not None:
+                    p_data = p0[:]
+                    pressure_hpa = float(p_data[-1, idx] if p_data.ndim == 2 else p_data[-1])
+                    if pressure_hpa > 5000:
+                        pressure_hpa = pressure_hpa / 100.0
+
+                nc.close()
+
+                if latest_temp < -90 or latest_temp > 60:
+                    record_source_call("knmi", "current", "bad_value",
+                                       (time.perf_counter() - started) * 1000.0)
+                    return None
+
+                temp_c = round(float(latest_temp), 1)
+                wind_kt = round(wind_ms * 1.94384, 1) if wind_ms is not None else None
+                if pressure_hpa is not None:
+                    pressure_hpa = round(float(pressure_hpa), 1)
+
+                temp = round(temp_c * 9 / 5 + 32, 1) if use_fahrenheit else temp_c
+
+                # Extract obs time from filename: KMDS__OPER_P___10M_OBS_L2_202605121150.nc
+                import re
+                obs_time = None
+                time_match = re.search(r"(\d{12})", fname)
+                if time_match:
+                    ts = time_match.group(1)
+                    obs_dt = datetime.strptime(ts, "%Y%m%d%H%M")
+                    obs_time = obs_dt.isoformat()
+
+                result = {
+                    "source": "knmi",
+                    "timestamp": datetime.utcnow().isoformat(),
+                    "station_code": stn,
+                    "station_name": meta["label"],
+                    "icao": meta["icao"],
+                    "obs_time": obs_time or datetime.utcnow().isoformat(),
+                    "current": {
+                        "temp": temp,
+                    },
+                    "temp_c": temp_c,
+                    "wind_kt": wind_kt,
+                    "pressure_hpa": pressure_hpa,
+                }
+
+                with self._knmi_cache_lock:
+                    self._knmi_cache[cache_key] = {"d": result, "t": now_ts}
+                record_source_call("knmi", "current", "success",
+                                   (time.perf_counter() - started) * 1000.0)
+                return result
+
+            finally:
+                try:
+                    nc.close()
+                except Exception:
+                    pass
+
+        except Exception as exc:
+            logger.warning("KNMI current fetch failed city={} error={}", city_key, exc)
+            with self._knmi_cache_lock:
+                stale = self._knmi_cache.get(cache_key)
+                if stale:
+                    record_source_call("knmi", "current", "stale_cache",
+                                       (time.perf_counter() - started) * 1000.0)
+                    return stale["d"]
+            record_source_call("knmi", "current", "error",
+                               (time.perf_counter() - started) * 1000.0)
+            return None
+
+    def fetch_knmi_official_nearby(
+        self,
+        city: str,
+        use_fahrenheit: bool = False,
+    ) -> list[Dict[str, Any]]:
+        current = self.fetch_knmi_current(city, use_fahrenheit=use_fahrenheit)
+        if not current:
+            return []
+        meta = KNMI_STATION.get(str(city or "").strip().lower()) or {}
+        return [
+            {
+                "name": meta.get("label") or "Schiphol 10min (KNMI)",
+                "station_label": meta.get("label"),
+                "lat": 52.3081,
+                "lon": 4.7642,
+                "temp": (current.get("current") or {}).get("temp"),
+                "icao": meta.get("icao"),
+                "istNo": meta.get("station"),
+                "source": "knmi",
+                "source_label": "KNMI",
+                "obs_time": current.get("obs_time"),
+            }
+        ]
@@ -16,10 +16,11 @@ from src.data_collection.nmc_sources import NmcSourceMixin
 from src.data_collection.nws_open_meteo_sources import NwsOpenMeteoSourceMixin
 from src.data_collection.amos_station_sources import AmosStationSourceMixin
 from src.data_collection.fmi_sources import FmiSourceMixin
+from src.data_collection.knmi_sources import KnmiSourceMixin
 from src.database.db_manager import DBManager


-class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, FmiSourceMixin):
+class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, FmiSourceMixin, KnmiSourceMixin):
    """
    Multi-source weather data collector

@@ -227,6 +228,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
        )
        self._fmi_cache: Dict[str, Dict] = {}
        self._fmi_cache_lock = threading.Lock()
+        self.knmi_cache_ttl_sec = int(
+            os.getenv("KNMI_CACHE_TTL_SEC", "120")
+        )
+        self._knmi_cache: Dict[str, Dict] = {}
+        self._knmi_cache_lock = threading.Lock()
        self.cwa_open_data_auth = (
            os.getenv("CWA_OPEN_DATA_AUTH")
            or os.getenv("CWA_OPEN_DATA_API_KEY")
@@ -737,6 +743,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
            self._jma_cache.pop(f"{normalized}:{use_fahrenheit}", None)
        with self._fmi_cache_lock:
            self._fmi_cache.pop(f"fmi:{normalized}:{use_fahrenheit}", None)
+        with self._knmi_cache_lock:
+            self._knmi_cache.pop(f"knmi:{normalized}:{use_fahrenheit}", None)
        with self._nmc_cache_lock:
            self._nmc_cache.pop(f"{normalized}:{use_fahrenheit}", None)
        with self._ru_station_cache_lock:
@@ -899,6 +907,31 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
        except Exception:
            logger.exception("airport_obs_log append failed for jma city={}", city_lower)

+    def _attach_knmi_official_nearby(
+        self, results: Dict, city_lower: str, use_fahrenheit: bool
+    ) -> None:
+        if city_lower != "amsterdam":
+            return
+        rows = self.fetch_knmi_official_nearby(
+            city_lower, use_fahrenheit=use_fahrenheit
+        )
+        if not rows:
+            return
+        results["knmi_official_nearby"] = rows
+        if "mgm_nearby" not in results:
+            results["mgm_nearby"] = rows
+        results["nearby_source"] = "knmi"
+        try:
+            row = rows[0] if rows else {}
+            DBManager().append_airport_obs(
+                icao=str(row.get("icao") or "EHAM"),
+                city=city_lower,
+                temp_c=row.get("temp"),
+                obs_time=str(row.get("obs_time") or datetime.now().isoformat()),
+            )
+        except Exception:
+            logger.exception("airport_obs_log append failed for knmi city={}", city_lower)
+
    def _attach_fmi_official_nearby(
        self, results: Dict, city_lower: str, use_fahrenheit: bool
    ) -> None:
@@ -1086,6 +1119,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                    self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
                    self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
                    self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit)
+                    self._attach_knmi_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)
@@ -1129,6 +1163,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                    self._attach_china_official_nearby(results, city_lower, use_fahrenheit)
                    self._attach_japan_official_nearby(results, city_lower, use_fahrenheit)
                    self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit)
+                    self._attach_knmi_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)
@@ -690,8 +690,8 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th

 # ── high-freq airport push loop ──

-HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki"}
-HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"}
+HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam"}
+HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM"}

 _AIRPORT_PUSH_STATE_PATH = os.path.join(
    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
@@ -723,7 +723,7 @@ def _save_airport_state(state: Dict[str, Any]) -> None:
    os.replace(tmp, path)


-_AIRPORT_CITY_LABEL = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa", "helsinki": "赫尔辛基/Vantaa"}
+_AIRPORT_CITY_LABEL = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa", "helsinki": "赫尔辛基/Vantaa", "amsterdam": "阿姆斯特丹/Schiphol"}


 def _build_airport_status_message(