astrape-oracle/classes/weather/ecowitt.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any
from urllib.request import urlopen
import json
import re

from gibil.classes.models import WeatherStationObservation


class EcowittGw3000Client:
    """Fetches local live data from an Ecowitt GW3000/GW3000A gateway."""

    def __init__(self, host: str, timeout_seconds: float = 5) -> None:
        self.host = host.removeprefix("http://").removeprefix("https://").rstrip("/")
        self.timeout_seconds = timeout_seconds

    def fetch_observation(self) -> WeatherStationObservation:
        received_at = datetime.now(timezone.utc)
        with urlopen(self._live_data_url(), timeout=self.timeout_seconds) as response:
            payload = json.loads(response.read().decode("utf-8"))

        return EcowittGw3000Parser().parse(
            payload=payload,
            received_at=received_at,
        )

    def _live_data_url(self) -> str:
        return f"http://{self.host}/get_livedata_info"


class EcowittGw3000Parser:
    """Normalizes GW3000 live-data JSON into a station observation."""

    common_id_names = {
        "0x02": "temperature_c",
        "0x07": "relative_humidity_pct",
        "0x03": "dew_point_c",
        "0x0B": "wind_speed_m_s",
        "0x0C": "wind_gust_m_s",
        "0x19": "wind_max_m_s",
        "0x0A": "wind_direction_deg",
        "0x15": "solar_radiation_w_m2",
        "0x17": "uv_index",
    }
    rain_id_names = {
        "0x0D": "rain_event_mm",
        "0x0E": "rain_rate_mm_h",
        "0x7C": "rain_hourly_mm",
        "0x10": "rain_daily_mm",
        "0x11": "rain_weekly_mm",
        "0x12": "rain_monthly_mm",
        "0x13": "rain_yearly_mm",
    }

    def parse(
        self,
        payload: dict[str, Any],
        received_at: datetime | None = None,
    ) -> WeatherStationObservation:
        if received_at is None:
            received_at = datetime.now(timezone.utc)

        common = self._parse_id_list(payload.get("common_list", []))
        rain = self._parse_id_list(payload.get("rain", []))
        indoor = self._parse_indoor(payload.get("wh25", []))

        return WeatherStationObservation(
            observed_at=received_at,
            received_at=received_at,
            source="ecowitt_gw3000",
            temperature_c=common.get("temperature_c"),
            relative_humidity_pct=common.get("relative_humidity_pct"),
            dew_point_c=common.get("dew_point_c"),
            wind_speed_m_s=common.get("wind_speed_m_s"),
            wind_gust_m_s=common.get("wind_gust_m_s"),
            wind_max_m_s=common.get("wind_max_m_s"),
            wind_direction_deg=common.get("wind_direction_deg"),
            solar_radiation_w_m2=common.get("solar_radiation_w_m2"),
            uv_index=common.get("uv_index"),
            rain_rate_mm_h=rain.get("rain_rate_mm_h"),
            rain_event_mm=rain.get("rain_event_mm"),
            rain_hourly_mm=rain.get("rain_hourly_mm"),
            rain_daily_mm=rain.get("rain_daily_mm"),
            rain_weekly_mm=rain.get("rain_weekly_mm"),
            rain_monthly_mm=rain.get("rain_monthly_mm"),
            rain_yearly_mm=rain.get("rain_yearly_mm"),
            indoor_temperature_c=indoor.get("indoor_temperature_c"),
            indoor_humidity_pct=indoor.get("indoor_humidity_pct"),
            absolute_pressure_hpa=indoor.get("absolute_pressure_hpa"),
            relative_pressure_hpa=indoor.get("relative_pressure_hpa"),
            metadata={
                "raw_payload": payload,
                "debug": payload.get("debug", []),
            },
        )

    def _parse_id_list(self, rows: object) -> dict[str, float]:
        parsed: dict[str, float] = {}
        if not isinstance(rows, list):
            return parsed

        for row in rows:
            if not isinstance(row, dict):
                continue
            row_id = str(row.get("id", ""))
            key = self.common_id_names.get(row_id) or self.rain_id_names.get(row_id)
            if key is None:
                continue
            value = self._parse_number(row.get("val"))
            if value is None:
                continue
            if key in {"wind_speed_m_s", "wind_gust_m_s", "wind_max_m_s"}:
                value = self._knots_to_m_s(value)
            parsed[key] = value

        return parsed

    def _parse_indoor(self, rows: object) -> dict[str, float]:
        if not isinstance(rows, list) or not rows:
            return {}
        row = rows[0]
        if not isinstance(row, dict):
            return {}

        values = {
            "indoor_temperature_c": self._parse_number(row.get("intemp")),
            "indoor_humidity_pct": self._parse_number(row.get("inhumi")),
            "absolute_pressure_hpa": self._parse_number(row.get("abs")),
            "relative_pressure_hpa": self._parse_number(row.get("rel")),
        }
        return {
            key: value
            for key, value in values.items()
            if value is not None
        }

    def _parse_number(self, value: object) -> float | None:
        if value is None:
            return None
        if isinstance(value, (int, float)):
            return float(value)

        match = re.search(r"-?\d+(?:\.\d+)?", str(value))
        if match is None:
            return None
        return float(match.group(0))

    def _knots_to_m_s(self, value: float) -> float:
        return value * 0.514444