PolyWeather/src/data_collection/amsc_awos_sources.py

"""AMSC AWOS runway observation source for China mainland airports.

The AMSC `getWindPlate` endpoint exposes runway-point air temperature fields:
TDZ_TEMP (touchdown zone), MID_TEMP (mid runway), and END_TEMP (runway end).
These values are air temperatures reported by runway observation positions,
not pavement/surface temperatures.
"""

from __future__ import annotations

import json
import os
import ssl
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
from urllib.parse import quote
from urllib.request import Request, urlopen

from loguru import logger

from src.utils.metrics import record_source_call
from src.utils.runtime_secrets import get_runtime_secret

AMSC_AWOS_BASE_URL = os.getenv("AMSC_AWOS_BASE_URL", "").strip()

AMSC_AWOS_AIRPORTS: Dict[str, Dict[str, str]] = {
    "shanghai": {"icao": "ZSPD", "label": "Shanghai Pudong"},
    "beijing": {"icao": "ZBAA", "label": "Beijing Capital"},
    "guangzhou": {"icao": "ZGGG", "label": "Guangzhou Baiyun"},
    "chengdu": {"icao": "ZUUU", "label": "Chengdu Shuangliu"},
    "chongqing": {"icao": "ZUCK", "label": "Chongqing Jiangbei"},
    "wuhan": {"icao": "ZHHH", "label": "Wuhan Tianhe"},
    "qingdao": {"icao": "ZSQD", "label": "Qingdao Jiaodong"},
}

AMSC_SETTLEMENT_RUNWAY_TARGETS: Dict[str, str] = {
    "shanghai": "35R",
    "chengdu": "02L",
    "chongqing": "02L",
    "guangzhou": "02L",
    "wuhan": "04",
    "beijing": "01",
    "qingdao": "34",
}


def _amsc_supported_city_codes() -> Dict[str, str]:
    return {city: meta["icao"] for city, meta in AMSC_AWOS_AIRPORTS.items()}


def _amsc_normalize_runway(value: Any) -> str:
    return str(value or "").strip().upper().replace(" ", "")


def _amsc_safe_float(value: Any) -> Optional[float]:
    if value is None:
        return None
    text = str(value).strip()
    if not text or text in {"-", "--", "null", "None"}:
        return None
    try:
        parsed = float(text)
    except (TypeError, ValueError):
        return None
    if not -80.0 < parsed < 80.0:
        return None
    return parsed


def _amsc_split_runway_pair(label: str) -> tuple[str, str]:
    parts = [_amsc_normalize_runway(part) for part in str(label or "").split("/") if part.strip()]
    if len(parts) >= 2:
        return parts[0], parts[1]
    runway = _amsc_normalize_runway(label) or "--"
    return runway, runway


def _amsc_settlement_endpoint(
    city_key: str,
    runway_pair: tuple[str, str],
    *,
    tdz_temp: Optional[float],
    end_temp: Optional[float],
) -> Optional[Dict[str, Any]]:
    target = _amsc_normalize_runway(AMSC_SETTLEMENT_RUNWAY_TARGETS.get(city_key))
    if not target:
        return None
    first = _amsc_normalize_runway(runway_pair[0])
    second = _amsc_normalize_runway(runway_pair[1])
    if target == first:
        temp = tdz_temp if tdz_temp is not None else end_temp
        position = "tdz" if tdz_temp is not None else "end_fallback"
    elif target == second:
        temp = end_temp if end_temp is not None else tdz_temp
        position = "end" if end_temp is not None else "tdz_fallback"
    else:
        return None
    if temp is None:
        return None
    return {
        "runway": target,
        "pair": f"{first}/{second}",
        "position": position,
        "temp": temp,
    }


def _amsc_wind_dir(*candidates: Any) -> Optional[int]:
    """Parse wind direction from AMSC fields (degrees)."""
    for value in candidates:
        parsed = _amsc_safe_float(value)
        if parsed is not None and 0 <= parsed <= 360:
            return int(round(parsed))
    return None


def _amsc_parse_rvr(value: Any) -> Optional[int]:
    """Parse RVR field like 'P2000' → 2000 (meters)."""
    text = str(value or "").strip().upper()
    if not text or text in {"--", "-", "NULL", "NONE"}:
        return None
    text = text.lstrip("PM")
    parsed = _amsc_safe_float(text)
    return int(round(parsed)) if parsed is not None else None


def _amsc_parse_utc_time(value: Any) -> tuple[Optional[str], Optional[str]]:
    text = str(value or "").strip()
    if not text:
        return None, None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
        try:
            utc_dt = datetime.strptime(text, fmt).replace(tzinfo=timezone.utc)
            local_dt = utc_dt + timedelta(hours=8)
            return utc_dt.isoformat(), local_dt.strftime("%Y-%m-%d %H:%M:%S")
        except ValueError:
            continue
    return text, None


def _amsc_parse_wind_plate_payload(
    payload: Dict[str, Any],
    *,
    city_key: str,
    icao: str,
) -> Optional[Dict[str, Any]]:
    if not isinstance(payload, dict):
        return None
    if payload.get("errCode") is not None:
        return None
    if payload.get("code") not in (None, 200, "200"):
        return None

    data = payload.get("data")
    if not isinstance(data, dict) or not data:
        return None

    airport_meta = AMSC_AWOS_AIRPORTS.get(city_key, {"icao": icao, "label": icao})
    runway_pairs = []
    runway_temps = []
    point_temperatures = []
    valid_values = []
    observation_time = None
    observation_time_local = None
    raw_metar = None

    for key, raw_row in data.items():
        if not isinstance(raw_row, dict):
            continue
        runway_label = str(raw_row.get("RNO") or key or "").strip()
        if not runway_label:
            continue
        tdz = _amsc_safe_float(raw_row.get("TDZ_TEMP"))
        mid = _amsc_safe_float(raw_row.get("MID_TEMP"))
        end = _amsc_safe_float(raw_row.get("END_TEMP"))
        points = [value for value in (tdz, mid, end) if value is not None]
        if not points:
            continue

        if observation_time is None:
            observation_time, observation_time_local = _amsc_parse_utc_time(raw_row.get("OTIME"))
        if raw_metar is None and raw_row.get("METAR"):
            raw_metar = str(raw_row.get("METAR"))

        runway_pair = _amsc_split_runway_pair(runway_label)
        settlement_endpoint = _amsc_settlement_endpoint(
            city_key,
            runway_pair,
            tdz_temp=tdz,
            end_temp=end,
        )
        runway_pairs.append(runway_pair)
        target_temp = (
            settlement_endpoint["temp"]
            if settlement_endpoint is not None
            else max(points)
        )
        best_temp = (
            settlement_endpoint["temp"]
            if settlement_endpoint is not None
            else tdz if tdz is not None else max(points)
        )
        runway_temps.append((best_temp, None))
        valid_values.extend(points)

        # Wind: prefer TDZ point, fallback to END
        wind_dir = _amsc_wind_dir(
            raw_row.get("TDZ_WIND_D10"),
            raw_row.get("END_WIND_D10"),
        )
        wind_speed = _amsc_safe_float(
            raw_row.get("TDZ_WIND_F10") or raw_row.get("END_WIND_F10")
        )
        # RVR / MOR: prefer 1-min average
        raw_rvr = (
            raw_row.get("TDZ_RVR_1A") or raw_row.get("END_RVR_1A")
            or raw_row.get("TDZ_RVR_10A") or raw_row.get("END_RVR_10A")
        )
        rvr = _amsc_parse_rvr(raw_rvr)
        raw_mor = (
            raw_row.get("TDZ_MOR_1A") or raw_row.get("END_MOR_1A")
            or raw_row.get("TDZ_MOR_10A") or raw_row.get("END_MOR_10A")
        )
        mor = _amsc_safe_float(raw_mor)
        # Humidity: prefer TDZ, fallback to END, then MID
        humidity = _amsc_safe_float(
            raw_row.get("TDZ_HUMID") or raw_row.get("END_HUMID") or raw_row.get("MID_HUMID")
        )

        point = {
            "runway": f"{runway_pair[0]}/{runway_pair[1]}",
            "temp": target_temp,
            "tdz_temp": tdz,
            "mid_temp": mid,
            "end_temp": end,
            "target_runway_max": target_temp,
            "wind_dir": wind_dir,
            "wind_speed": wind_speed,
            "rvr": rvr,
            "mor": mor,
            "humidity": humidity,
        }
        if settlement_endpoint is not None:
            point.update(
                {
                    "is_settlement": True,
                    "settlement_runway": settlement_endpoint["runway"],
                    "settlement_runway_position": settlement_endpoint["position"],
                    "settlement_runway_temp": settlement_endpoint["temp"],
                }
            )
        point_temperatures.append(point)

    if not valid_values or not runway_pairs:
        return None

    max_temp = round(max(valid_values), 1)
    min_temp = round(min(valid_values), 1)
    avg_temp = round(sum(valid_values) / len(valid_values), 1)
    settlement_point = next(
        (
            point
            for point in point_temperatures
            if point.get("is_settlement") and point.get("settlement_runway_temp") is not None
        ),
        None,
    )
    display_temp = (
        round(float(settlement_point["settlement_runway_temp"]), 1)
        if settlement_point is not None
        else max_temp
    )

    return {
        "temp": display_temp,
        "temp_c": display_temp,
        "temp_source": "settlement_runway_endpoint" if settlement_point else "runway_max",
        "settlement_runway": settlement_point.get("settlement_runway") if settlement_point else None,
        "settlement_runway_pair": settlement_point.get("runway") if settlement_point else None,
        "settlement_runway_position": settlement_point.get("settlement_runway_position") if settlement_point else None,
        "settlement_runway_temp": display_temp if settlement_point else None,
        "runway_temps": runway_temps,
        "runway_temp_range": (min_temp, max_temp),
        "runway_temp_avg": avg_temp,
        "source": "amsc_awos",
        "source_label": f"AMSC AWOS {airport_meta.get('label', icao)} ({icao})",
        "source_code": "amsc_awos",
        "network_type": "amsc_awos",
        "icao": icao,
        "station_label": airport_meta.get("label") or icao,
        "raw_metar": raw_metar,
        "raw_taf": None,
        "wind_dir": next((pt.get("wind_dir") for pt in point_temperatures if pt.get("wind_dir") is not None), None) if point_temperatures else None,
        "wind_speed": point_temperatures[0].get("wind_speed") if point_temperatures else None,
        "runway_obs": {
            "runway_pairs": runway_pairs,
            "temperatures": runway_temps,
            "point_temperatures": point_temperatures,
        },
        "observation_source": "AMSC AWOS runway-point air temperature",
        "observation_source_zh": "AMSC AWOS 跑道观测气温",
        "observation_time": observation_time,
        "observation_time_local": observation_time_local,
    }


class AmscAwosSourceMixin:
    """Mixin that adds AMSC AWOS runway-point air temperatures."""

    def _amsc_headers(self) -> Dict[str, str]:
        headers = {
            "Accept": "application/json, text/plain, */*",
            "Referer": os.getenv("AMSC_AWOS_REFERER", "https://www.amsc.net.cn/"),
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
            ),
        }
        cookie = get_runtime_secret("POLYWEATHER_AMSC_COOKIE")
        session_id = get_runtime_secret("POLYWEATHER_AMSC_SESSION_ID")
        if cookie:
            headers["Cookie"] = cookie
        elif session_id:
            headers["sessionId"] = session_id
            headers["app"] = "AMS"
        return headers

    def _amsc_http_get_json(self, url: str, *, headers: Optional[Dict[str, str]] = None) -> Optional[Dict[str, Any]]:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        req = Request(url, headers=headers or {})
        with urlopen(req, timeout=getattr(self, "timeout", 10.0), context=ctx) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return None

    def fetch_amsc_awos_current(
        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()
        airport_meta = AMSC_AWOS_AIRPORTS.get(normalized_city)
        if not airport_meta:
            return None
        icao = airport_meta["icao"]
        url = f"{AMSC_AWOS_BASE_URL}?cccc={quote(icao)}"
        started = time.perf_counter()
        try:
            payload = self._amsc_http_get_json(url, headers=self._amsc_headers())
            result = _amsc_parse_wind_plate_payload(
                payload or {},
                city_key=normalized_city,
                icao=icao,
            )
            if result:
                record_source_call(
                    "amsc_awos",
                    "current",
                    "success",
                    (time.perf_counter() - started) * 1000.0,
                )
            else:
                record_source_call(
                    "amsc_awos",
                    "current",
                    "empty",
                    (time.perf_counter() - started) * 1000.0,
                )
            return result
        except Exception as exc:
            logger.warning("AMSC AWOS fetch failed city={} icao={}: {}", normalized_city, icao, exc)
            record_source_call(
                "amsc_awos",
                "current",
                "error",
                (time.perf_counter() - started) * 1000.0,
            )
            return None