PolyWeather/src/data_collection/settlement_sources.py

from __future__ import annotations

import csv
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from loguru import logger


class SettlementSourceMixin:
    IMGW_METEO_API_BASE = "https://meteo.imgw.pl/api/v1"
    IMGW_METEO_API_TOKEN = "p4DXKjsYadfBV21TYrDk"

    def _get_settlement_cache(self, key: str) -> Optional[Dict[str, Any]]:
        now_ts = time.time()
        with self._settlement_cache_lock:
            cached = self._settlement_cache.get(key)
            if cached and now_ts - float(cached.get("t", 0)) < self.settlement_cache_ttl_sec:
                return cached.get("d")
        return None

    def _set_settlement_cache(self, key: str, payload: Dict[str, Any]) -> None:
        with self._settlement_cache_lock:
            self._settlement_cache[key] = {"t": time.time(), "d": payload}

    @staticmethod
    def _csv_rows(text: str) -> List[Dict[str, str]]:
        normalized = str(text or "").lstrip("")
        if not normalized.strip():
            return []
        return [row for row in csv.DictReader(normalized.splitlines()) if isinstance(row, dict)]

    @staticmethod
    def _hko_parse_local_iso(raw_yyyymmddhhmm: Optional[str]) -> Optional[str]:
        raw = str(raw_yyyymmddhhmm or "").strip()
        if len(raw) != 12 or not raw.isdigit():
            return None
        try:
            dt = datetime.strptime(raw, "%Y%m%d%H%M").replace(
                tzinfo=timezone(timedelta(hours=8))
            )
            return dt.isoformat()
        except Exception:
            return None

    @staticmethod
    def _hko_compass_to_deg(compass: Optional[str]) -> Optional[float]:
        value = str(compass or "").strip().upper()
        if not value:
            return None
        mapping = {
            "N": 0.0,
            "NNE": 22.5,
            "NE": 45.0,
            "ENE": 67.5,
            "E": 90.0,
            "ESE": 112.5,
            "SE": 135.0,
            "SSE": 157.5,
            "S": 180.0,
            "SSW": 202.5,
            "SW": 225.0,
            "WSW": 247.5,
            "W": 270.0,
            "WNW": 292.5,
            "NW": 315.0,
            "NNW": 337.5,
        }
        return mapping.get(value)

    @staticmethod
    def _safe_float(value: Any) -> Optional[float]:
        if value is None:
            return None
        text = str(value).strip()
        if not text or text in {"***", "N/A", "NA"}:
            return None
        try:
            return float(text)
        except Exception:
            return None

    @staticmethod
    def _pick_station_row(
        rows: List[Dict[str, str]], candidates: List[str]
    ) -> Optional[Dict[str, str]]:
        if not rows:
            return None
        normalized_map = {
            str(name).strip().lower(): row
            for row in rows
            for name in [row.get("Automatic Weather Station")]
            if isinstance(row, dict) and name
        }
        for name in candidates:
            hit = normalized_map.get(str(name).strip().lower())
            if hit:
                return hit
        for row in rows:
            station = str(row.get("Automatic Weather Station") or "").strip().lower()
            if "observatory" in station:
                return row
        return rows[0] if rows else None

    def fetch_hko_settlement_current(self) -> Optional[Dict[str, Any]]:
        cache_key = "hko:hong_kong"
        cached = self._get_settlement_cache(cache_key)
        if cached:
            return cached

        try:
            base = "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather"
            temp_csv = self.session.get(f"{base}/latest_1min_temperature.csv", timeout=self.timeout)
            temp_csv.raise_for_status()
            maxmin_csv = self.session.get(f"{base}/latest_since_midnight_maxmin.csv", timeout=self.timeout)
            maxmin_csv.raise_for_status()
            humidity_csv = self.session.get(f"{base}/latest_1min_humidity.csv", timeout=self.timeout)
            humidity_csv.raise_for_status()
            wind_csv = self.session.get(f"{base}/latest_10min_wind.csv", timeout=self.timeout)
            wind_csv.raise_for_status()

            temp_rows = self._csv_rows(temp_csv.text)
            maxmin_rows = self._csv_rows(maxmin_csv.text)
            humidity_rows = self._csv_rows(humidity_csv.text)
            wind_rows = self._csv_rows(wind_csv.text)

            station_candidates = ["HK Observatory", "Hong Kong Observatory"]
            temp_row = self._pick_station_row(temp_rows, station_candidates)
            maxmin_row = self._pick_station_row(maxmin_rows, station_candidates)
            humidity_row = self._pick_station_row(humidity_rows, station_candidates)
            wind_row = self._pick_station_row(wind_rows, station_candidates)
            if not temp_row or not maxmin_row:
                return None

            obs_raw = temp_row.get("Date time") or maxmin_row.get("Date time")
            obs_iso = self._hko_parse_local_iso(obs_raw)
            current_temp = self._safe_float(temp_row.get("Air Temperature(degree Celsius)"))
            max_so_far = self._safe_float(maxmin_row.get("Maximum Air Temperature Since Midnight(degree Celsius)"))
            min_so_far = self._safe_float(maxmin_row.get("Minimum Air Temperature Since Midnight(degree Celsius)"))
            humidity = self._safe_float(humidity_row.get("Relative Humidity(percent)")) if humidity_row else None
            wind_speed_kmh = self._safe_float(wind_row.get("10-Minute Mean Speed(km/hour)")) if wind_row else None
            wind_speed_kt = round(float(wind_speed_kmh) / 1.852, 1) if wind_speed_kmh is not None else None
            wind_dir = self._hko_compass_to_deg(
                wind_row.get("10-Minute Mean Wind Direction(Compass points)") if wind_row else None
            )

            payload: Dict[str, Any] = {
                "source": "hko",
                "source_label": "HKO",
                "station_code": "HKO",
                "station_name": "HK Observatory",
                "observation_time": obs_iso,
                "current": {
                    "temp": round(current_temp, 1) if current_temp is not None else None,
                    "max_temp_so_far": round(max_so_far, 1) if max_so_far is not None else None,
                    "max_temp_time": None,
                    "today_low": round(min_so_far, 1) if min_so_far is not None else None,
                    "humidity": round(humidity, 1) if humidity is not None else None,
                    "wind_speed_kt": wind_speed_kt,
                    "wind_dir": wind_dir,
                },
                "unit": "celsius",
            }
            self._set_settlement_cache(cache_key, payload)
            return payload
        except Exception as exc:
            logger.warning(f"HKO settlement fetch failed: {exc}")
            return None

    def fetch_cwa_taipei_settlement_current(self) -> Optional[Dict[str, Any]]:
        cache_key = "cwa:taipei:466920"
        cached = self._get_settlement_cache(cache_key)
        if cached:
            return cached

        try:
            url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001"
            response = self.session.get(
                url,
                params={"Authorization": self.cwa_open_data_auth, "format": "JSON", "StationId": "466920"},
                timeout=self.timeout,
            )
            response.raise_for_status()
            data = response.json() if response.content else {}
            stations = (data.get("records") or {}).get("Station") or []
            if isinstance(stations, dict):
                station = stations
            elif isinstance(stations, list) and stations:
                station = stations[0]
            else:
                station = None
            if not isinstance(station, dict):
                return None

            wx = station.get("WeatherElement") or {}
            if not isinstance(wx, dict):
                wx = {}
            daily_extreme = wx.get("DailyExtreme") or {}
            high_info = (((daily_extreme.get("DailyHigh") or {}).get("TemperatureInfo") or {}))
            low_info = (((daily_extreme.get("DailyLow") or {}).get("TemperatureInfo") or {}))
            high_time_raw = (((high_info.get("Occurred_at") or {}).get("DateTime")))
            high_hhmm = None
            if high_time_raw and "T" in str(high_time_raw):
                try:
                    high_hhmm = datetime.fromisoformat(str(high_time_raw)).strftime("%H:%M")
                except Exception:
                    high_hhmm = str(high_time_raw).split("T")[1][:5]

            obs_time_raw = (station.get("ObsTime") or {}).get("DateTime")
            wind_speed_ms = self._safe_float(wx.get("WindSpeed"))
            payload: Dict[str, Any] = {
                "source": "cwa",
                "source_label": "CWA",
                "station_code": str(station.get("StationId") or "466920"),
                "station_name": str(station.get("StationName") or "臺北"),
                "observation_time": str(obs_time_raw or "").strip() or None,
                "current": {
                    "temp": self._safe_float(wx.get("AirTemperature")),
                    "max_temp_so_far": self._safe_float(high_info.get("AirTemperature")),
                    "max_temp_time": high_hhmm,
                    "today_low": self._safe_float(low_info.get("AirTemperature")),
                    "humidity": self._safe_float(wx.get("RelativeHumidity")),
                    "wind_speed_kt": round(float(wind_speed_ms) * 1.943844, 1) if wind_speed_ms is not None else None,
                    "wind_dir": self._safe_float(wx.get("WindDirection")),
                },
                "unit": "celsius",
            }
            self._set_settlement_cache(cache_key, payload)
            return payload
        except Exception as exc:
            logger.warning(f"CWA settlement fetch failed: {exc}")
            return None

    def fetch_hko_forecast(self) -> Optional[float]:
        try:
            url = "https://data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType=fnd&lang=tc"
            res = self.session.get(url, timeout=self.timeout).json()
            return float(res["weatherForecast"][0]["forecastMaxtemp"]["value"])
        except Exception as exc:
            logger.warning(f"HKO Forecast request failed: {exc}")
            return None

    def fetch_cwa_taipei_forecast(self) -> Optional[float]:
        try:
            if not self.cwa_open_data_auth:
                return None
            url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-D0047-061"
            res = self.session.get(
                url,
                params={"Authorization": self.cwa_open_data_auth, "format": "JSON", "elementName": "MaxT"},
                timeout=self.timeout,
            ).json()
            locs = res.get("records", {}).get("Locations", [])[0].get("Location", [])
            if not locs:
                return None
            loc = locs[0]
            for weather_element in loc.get("WeatherElement", []):
                if weather_element.get("ElementName") == "MaxT":
                    return float(weather_element["Time"][0]["ElementValue"][0]["Temperature"])
            return None
        except Exception as exc:
            logger.warning(f"CWA Forecast request failed: {exc}")
            return None

    def _imgw_api_get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
        try:
            query = {"token": self.IMGW_METEO_API_TOKEN}
            if isinstance(params, dict):
                query.update(params)
            response = self.session.get(
                f"{self.IMGW_METEO_API_BASE}/{path}",
                params=query,
                timeout=self.timeout,
            )
            response.raise_for_status()
            return response.json() if response.content else None
        except Exception as exc:
            logger.warning(f"IMGW API request failed path={path}: {exc}")
            return None

    def fetch_imgw_synoptic_station_current(
        self,
        localization: str,
        *,
        display_name: Optional[str] = None,
        use_fahrenheit: bool = False,
    ) -> Optional[Dict[str, Any]]:
        cache_key = f"imgw:synoptic:{str(localization or '').strip().lower()}"
        cached = self._get_settlement_cache(cache_key)
        if cached:
            return cached

        location_payload = self._imgw_api_get("geo/search", {"name": localization})
        location_rows = (location_payload or {}).get("data") or []
        if not isinstance(location_rows, list) or not location_rows:
            return None
        location_row = location_rows[0] if isinstance(location_rows[0], dict) else None
        if not isinstance(location_row, dict):
            return None

        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        synoptic_payload = self._imgw_api_get(
            "forecast/synoptic",
            {"date": date_str, "localization": localization},
        )
        data = (synoptic_payload or {}).get("data") or {}
        if not isinstance(data, dict) or not data:
            return None

        latest_row = None
        for key in sorted(data.keys()):
            row = data.get(key)
            if isinstance(row, dict) and row.get("temp") is not None:
                latest_row = row
        if not isinstance(latest_row, dict):
            return None

        temp_kelvin = self._safe_float(latest_row.get("temp"))
        if temp_kelvin is None:
            return None
        temp_c = temp_kelvin - 273.15
        temp_value = (temp_c * 9 / 5) + 32 if use_fahrenheit else temp_c
        wind_speed_ms = self._safe_float(latest_row.get("ws"))

        payload = {
            "name": display_name or str(location_row.get("name") or localization),
            "lat": self._safe_float(location_row.get("lat")),
            "lon": self._safe_float(location_row.get("lon")),
            "temp": round(temp_value, 1),
            "icao": "IMGW",
            "istNo": "IMGW",
            "wind_dir": self._safe_float(latest_row.get("wd")),
            "wind_speed": wind_speed_ms,
            "wind_speed_kt": round(float(wind_speed_ms) * 1.943844, 1) if wind_speed_ms is not None else None,
            "source": "imgw_synoptic",
            "raw_time": latest_row.get("fd"),
        }
        if payload["lat"] is None or payload["lon"] is None:
            return None
        self._set_settlement_cache(cache_key, payload)
        return payload

    def fetch_settlement_current(self, city: str) -> Optional[Dict[str, Any]]:
        normalized = str(city or "").strip().lower()
        if normalized == "hong kong":
            return self.fetch_hko_settlement_current()
        if normalized == "taipei":
            return self.fetch_cwa_taipei_settlement_current()
        return None