Add KMA nearby weather support for Seoul and Busan

src/data_collection/country_networks.py

@@ -39,6 +39,8 @@ def _provider_code_for_city(city: str) -> str:
     settlement_source = str(meta.get("settlement_source") or "").strip().lower()
     if normalized in {"ankara", "istanbul"}:
         return "turkey_mgm"
+    if normalized in {"busan", "seoul"}:
+        return "korea_kma"
     if settlement_source == "hko":
         return "hongkong_hko"
     if settlement_source == "cwa":
@@ -205,6 +207,34 @@ def _jma_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
     return out
 
 
+def _kma_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
+    rows = raw.get("kma_official_nearby") or []
+    out: List[Dict[str, Any]] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        out.append(
+            _normalize_station_row(
+                station_code=row.get("station_code") or row.get("icao") or row.get("istNo"),
+                station_label=row.get("station_label") or row.get("name"),
+                temp=row.get("temp"),
+                lat=row.get("lat"),
+                lon=row.get("lon"),
+                obs_time=row.get("obs_time"),
+                source_code="kma",
+                source_label="KMA",
+                is_official=True,
+                is_airport_station=False,
+                is_settlement_anchor=False,
+                extra={
+                    "distance_km": _safe_float(row.get("distance_km")),
+                    "network_type": row.get("network_type"),
+                },
+            )
+        )
+    return out
+
+
 def _mgm_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
     meta = _city_meta(city)
     rows = raw.get("mgm_nearby") or []
@@ -445,6 +475,28 @@ class JapanJmaNetworkProvider(CountryNetworkProvider):
         }
 
 
+class KoreaKmaNetworkProvider(CountryNetworkProvider):
+    def __init__(self) -> None:
+        super().__init__("korea_kma", "KMA")
+
+    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
+        rows = _kma_rows(raw, city)
+        if rows:
+            return rows
+        return _metar_cluster_rows(raw)
+
+    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
+        rows = self.official_nearby_current(city, raw)
+        has_kma = bool(_kma_rows(raw, city))
+        return {
+            "provider_code": self.provider_code,
+            "provider_label": self.provider_label,
+            "available": has_kma,
+            "mode": "official_active" if has_kma else ("fallback_metar_cluster" if rows else "reference_only"),
+            "row_count": len(rows),
+        }
+
+
 class HongKongHkoNetworkProvider(CountryNetworkProvider):
     def __init__(self) -> None:
         super().__init__("hongkong_hko", "HKO")
@@ -467,6 +519,8 @@ def get_country_network_provider(city: str) -> CountryNetworkProvider:
     provider_code = _provider_code_for_city(city)
     if provider_code == "turkey_mgm":
         return TurkeyMgmNetworkProvider()
+    if provider_code == "korea_kma":
+        return KoreaKmaNetworkProvider()
     if provider_code == "japan_jma":
         return JapanJmaNetworkProvider()
     if provider_code == "china_cma":

src/data_collection/kma_station_sources.py

@@ -0,0 +1,269 @@
+from __future__ import annotations
+
+import math
+import time
+from datetime import datetime
+from threading import Lock
+from typing import Any, Dict, List, Optional
+
+from loguru import logger
+
+from src.utils.metrics import record_source_call
+
+
+KMA_CITY_STATIONS: Dict[str, Dict[str, Any]] = {
+    "busan": {
+        "airport_station_code": "153",
+        "airport_label": "김해공항",
+        "airport_icao": "RKPK",
+    },
+    "seoul": {
+        "airport_station_code": "113",
+        "airport_label": "인천공항",
+        "airport_icao": "RKSI",
+    },
+}
+
+KMA_GEOJSON_URLS: Dict[str, str] = {
+    "air": "https://www.weather.go.kr/wgis-nuri/js/info/air.geojson",
+    "sfc": "https://www.weather.go.kr/wgis-nuri/js/info/sfc.geojson",
+    "aws": "https://www.weather.go.kr/wgis-nuri/js/info/aws.geojson",
+}
+
+KMA_OBSERVATION_URLS: Dict[str, str] = {
+    "air": "https://www.weather.go.kr/wgis-nuri/aws/air?date=",
+    "sfc": "https://www.weather.go.kr/wgis-nuri/aws/sfc?date=",
+    "aws": "https://www.weather.go.kr/wgis-nuri/aws/aws?date=",
+}
+
+
+class KmaStationSourceMixin:
+    def _kma_http_get_json(self, url: str):
+        getter = getattr(self, "_http_get_json", None)
+        if callable(getter):
+            return getter(url)
+        response = self.session.get(url, timeout=self.timeout)
+        response.raise_for_status()
+        return response.json()
+
+    @staticmethod
+    def _kma_safe_float(value: Any) -> Optional[float]:
+        try:
+            if value in (None, "", "-", "null"):
+                return None
+            numeric = float(value)
+            if math.isnan(numeric):
+                return None
+            return numeric
+        except Exception:
+            return None
+
+    @staticmethod
+    def _kma_distance_km(
+        lat1: Optional[float],
+        lon1: Optional[float],
+        lat2: Optional[float],
+        lon2: Optional[float],
+    ) -> Optional[float]:
+        if None in (lat1, lon1, lat2, lon2):
+            return None
+        try:
+            r = 6371.0
+            d_lat = math.radians(float(lat2) - float(lat1))
+            d_lon = math.radians(float(lon2) - float(lon1))
+            a = (
+                math.sin(d_lat / 2) ** 2
+                + math.cos(math.radians(float(lat1)))
+                * math.cos(math.radians(float(lat2)))
+                * math.sin(d_lon / 2) ** 2
+            )
+            c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
+            return round(r * c, 2)
+        except Exception:
+            return None
+
+    @staticmethod
+    def _kma_to_local_obs_time(tm: Any) -> Optional[str]:
+        text = str(tm or "").strip()
+        if len(text) != 12 or not text.isdigit():
+            return None
+        try:
+            parsed = datetime.strptime(text, "%Y%m%d%H%M")
+            return parsed.strftime("%H:%M")
+        except Exception:
+            return None
+
+    def _kma_cached_fetch(self, kind: str, url: str) -> Any:
+        now_ts = time.time()
+        cache_key = f"{kind}:{url}"
+        with self._kma_cache_lock:
+            cached = self._kma_cache.get(cache_key)
+            if cached and now_ts - cached["t"] < self.kma_cache_ttl_sec:
+                return cached["d"]
+        data = self._kma_http_get_json(url)
+        with self._kma_cache_lock:
+            self._kma_cache[cache_key] = {"d": data, "t": now_ts}
+        return data
+
+    def _kma_load_geo_features(self, layer: str) -> List[Dict[str, Any]]:
+        payload = self._kma_cached_fetch("geo", KMA_GEOJSON_URLS[layer])
+        if not isinstance(payload, dict):
+            return []
+        features = payload.get("features")
+        return features if isinstance(features, list) else []
+
+    def _kma_load_observations(self, layer: str) -> List[Dict[str, Any]]:
+        payload = self._kma_cached_fetch("obs", KMA_OBSERVATION_URLS[layer])
+        return payload if isinstance(payload, list) else []
+
+    def _kma_station_feature_map(self, layer: str) -> Dict[str, Dict[str, Any]]:
+        mapping: Dict[str, Dict[str, Any]] = {}
+        for feature in self._kma_load_geo_features(layer):
+            if not isinstance(feature, dict):
+                continue
+            props = feature.get("properties") or {}
+            geometry = feature.get("geometry") or {}
+            coords = geometry.get("coordinates") or []
+            if not isinstance(props, dict) or len(coords) < 2:
+                continue
+            station_id = str(props.get("stnId") or "").strip()
+            if not station_id:
+                continue
+            mapping[station_id] = {
+                "lat": self._kma_safe_float(coords[1]),
+                "lon": self._kma_safe_float(coords[0]),
+                "stn_ko": props.get("stnKo"),
+                "stn_en": props.get("stnEn"),
+                "type": props.get("type"),
+            }
+        return mapping
+
+    def _kma_anchor(
+        self,
+        city: str,
+    ) -> tuple[Optional[float], Optional[float], Optional[str]]:
+        city_key = str(city or "").strip().lower()
+        city_meta = self.CITY_REGISTRY.get(city_key) or {}
+        kma_meta = KMA_CITY_STATIONS.get(city_key) or {}
+        airport_station_code = str(kma_meta.get("airport_station_code") or "").strip()
+        if airport_station_code:
+            air_map = self._kma_station_feature_map("air")
+            airport_feature = air_map.get(airport_station_code) or {}
+            lat = self._kma_safe_float(airport_feature.get("lat"))
+            lon = self._kma_safe_float(airport_feature.get("lon"))
+            if lat is not None and lon is not None:
+                return lat, lon, airport_station_code
+        return (
+            self._kma_safe_float(city_meta.get("lat")),
+            self._kma_safe_float(city_meta.get("lon")),
+            airport_station_code or None,
+        )
+
+    def fetch_kma_official_nearby(
+        self,
+        city: str,
+        use_fahrenheit: bool = False,
+    ) -> List[Dict[str, Any]]:
+        started = time.perf_counter()
+        city_key = str(city or "").strip().lower()
+        city_meta = KMA_CITY_STATIONS.get(city_key) or {}
+        if not city_meta:
+            record_source_call(
+                "kma",
+                "nearby",
+                "unsupported_city",
+                (time.perf_counter() - started) * 1000.0,
+            )
+            return []
+
+        try:
+            anchor_lat, anchor_lon, airport_station_code = self._kma_anchor(city_key)
+            feature_maps = {
+                "air": self._kma_station_feature_map("air"),
+                "sfc": self._kma_station_feature_map("sfc"),
+                "aws": self._kma_station_feature_map("aws"),
+            }
+            observation_layers = {
+                "air": self._kma_load_observations("air"),
+                "sfc": self._kma_load_observations("sfc"),
+                "aws": self._kma_load_observations("aws"),
+            }
+
+            by_station: Dict[str, Dict[str, Any]] = {}
+            layer_priority = {"aws": 0, "sfc": 1}
+            for layer in ("aws", "sfc"):
+                for row in observation_layers[layer]:
+                    if not isinstance(row, dict):
+                        continue
+                    station_id = str(row.get("stnId") or "").strip()
+                    if not station_id or station_id == airport_station_code:
+                        continue
+                    temp_c = self._kma_safe_float(row.get("ta"))
+                    if temp_c is None:
+                        continue
+                    feature_meta = feature_maps[layer].get(station_id) or {}
+                    lat = self._kma_safe_float(feature_meta.get("lat"))
+                    lon = self._kma_safe_float(feature_meta.get("lon"))
+                    distance_km = self._kma_distance_km(anchor_lat, anchor_lon, lat, lon)
+                    temp = (
+                        round(temp_c * 9 / 5 + 32, 1)
+                        if use_fahrenheit
+                        else round(temp_c, 1)
+                    )
+                    candidate = {
+                        "name": feature_meta.get("stn_ko")
+                        or feature_meta.get("stn_en")
+                        or f"KMA {station_id}",
+                        "station_label": feature_meta.get("stn_ko")
+                        or feature_meta.get("stn_en")
+                        or f"KMA {station_id}",
+                        "station_code": station_id,
+                        "icao": station_id,
+                        "istNo": station_id,
+                        "lat": lat,
+                        "lon": lon,
+                        "temp": temp,
+                        "obs_time": self._kma_to_local_obs_time(row.get("tm")),
+                        "source": "kma",
+                        "source_label": "KMA",
+                        "source_code": "kma",
+                        "is_official": True,
+                        "is_airport_station": False,
+                        "is_settlement_anchor": False,
+                        "network_type": layer,
+                        "distance_km": distance_km,
+                    }
+                    existing = by_station.get(station_id)
+                    if (
+                        existing is None
+                        or layer_priority.get(layer, 99)
+                        < layer_priority.get(str(existing.get("network_type")), 99)
+                    ):
+                        by_station[station_id] = candidate
+
+            rows = sorted(
+                by_station.values(),
+                key=lambda item: (
+                    item.get("distance_km") is None,
+                    item.get("distance_km") if item.get("distance_km") is not None else 9999,
+                    item.get("station_label") or "",
+                ),
+            )
+            trimmed = rows[:6]
+            record_source_call(
+                "kma",
+                "nearby",
+                "success" if trimmed else "empty",
+                (time.perf_counter() - started) * 1000.0,
+            )
+            return trimmed
+        except Exception as exc:
+            logger.warning("KMA nearby fetch failed city={} error={}", city_key, exc)
+            record_source_call(
+                "kma",
+                "nearby",
+                "error",
+                (time.perf_counter() - started) * 1000.0,
+            )
+            return []
+

src/data_collection/weather_sources.py

@@ -10,12 +10,13 @@ from src.data_collection.open_meteo_cache import OpenMeteoCacheMixin
 from src.data_collection.settlement_sources import SettlementSourceMixin
 from src.data_collection.metar_sources import MetarSourceMixin
 from src.data_collection.mgm_sources import MgmSourceMixin
+from src.data_collection.kma_station_sources import KmaStationSourceMixin
 from src.data_collection.jma_amedas_sources import JmaAmedasSourceMixin
 from src.data_collection.nmc_sources import NmcSourceMixin
 from src.data_collection.nws_open_meteo_sources import NwsOpenMeteoSourceMixin
 
 
-class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin):
+class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, KmaStationSourceMixin, JmaAmedasSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin):
     """
     Multi-source weather data collector
 
@@ -173,6 +174,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         )
         self._jma_cache: Dict[str, Dict] = {}
         self._jma_cache_lock = threading.Lock()
+        self.kma_cache_ttl_sec = int(
+            os.getenv("KMA_STATION_CACHE_TTL_SEC", "300")
+        )
+        self._kma_cache: Dict[str, Dict] = {}
+        self._kma_cache_lock = threading.Lock()
         self.settlement_cache_ttl_sec = int(
             os.getenv("SETTLEMENT_SOURCE_CACHE_TTL_SEC", "120")
         )
@@ -777,6 +783,21 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             results["mgm_nearby"] = official_rows
         results["nearby_source"] = "jma"
 
+    def _attach_korea_official_nearby(
+        self, results: Dict, city_lower: str, use_fahrenheit: bool
+    ) -> None:
+        if city_lower not in {"busan", "seoul"}:
+            return
+        official_rows = self.fetch_kma_official_nearby(
+            city_lower, use_fahrenheit=use_fahrenheit
+        )
+        if not official_rows:
+            return
+        results["kma_official_nearby"] = official_rows
+        if "mgm_nearby" not in results:
+            results["mgm_nearby"] = official_rows
+        results["nearby_source"] = "kma"
+
     def _attach_warsaw_official_nearby(
         self, results: Dict, use_fahrenheit: bool
     ) -> None:
@@ -873,6 +894,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                 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)
                 if city_lower == "warsaw":
                     self._attach_warsaw_official_nearby(results, use_fahrenheit)
                 self._attach_global_nearby_cluster(
@@ -901,6 +923,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                 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)
                 if city_lower == "warsaw":
                     self._attach_warsaw_official_nearby(results, use_fahrenheit)
                 self._attach_global_nearby_cluster(