Inkycal/inkycal/custom/openweathermap_wrapper.py

"""
Inkycal OpenWeatherMap API abstraction module
- Retrieves free weather data from OWM 2.5/3.0 API endpoints (with provided API key)
- Handles temperature and wind unit conversions
- Converts data to a standardized timezone and language
- Returns ready-to-use weather structures for current, hourly, and daily forecasts
"""

import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Literal

import requests
from dateutil import tz

# Type annotations for strict typing
TEMP_UNITS = Literal["celsius", "fahrenheit"]
WIND_UNITS = Literal["meters_sec", "km_hour", "miles_hour", "knots", "beaufort"]
WEATHER_TYPE = Literal["current", "forecast"]
API_VERSIONS = Literal["2.5", "3.0"]

API_BASE_URL = "https://api.openweathermap.org/data"

logger = logging.getLogger(__name__)
logger.setLevel(level=logging.INFO)


def is_timestamp_within_range(timestamp: datetime, start_time: datetime, end_time: datetime) -> bool:
    """Check if the given timestamp lies between start_time and end_time."""
    return start_time <= timestamp <= end_time


def get_json_from_url(request_url):
    """Performs an HTTP GET request and returns the parsed JSON response."""
    response = requests.get(request_url)
    if not response.ok:
        raise AssertionError(
            f"Failure getting weather: code {response.status_code}. Reason: {response.text}"
        )
    return json.loads(response.text)


class OpenWeatherMap:
    def __init__(self, api_key: str, city_id: int = None, lat: float = None, lon: float = None,
                 api_version: API_VERSIONS = "2.5", temp_unit: TEMP_UNITS = "celsius",
                 wind_unit: WIND_UNITS = "meters_sec", language: str = "en", tz_name: str = "UTC") -> None:
        """
        Initializes the OWM wrapper with localization settings.
        Chooses API version, units, location and timezone preferences.
        """
        self.api_key = api_key
        self.temp_unit = temp_unit
        self.wind_unit = wind_unit
        self.language = language
        self._api_version = api_version

        if self._api_version == "3.0":
            assert isinstance(lat, float) and isinstance(lon, float)

        self.location_substring = (
            f"lat={lat}&lon={lon}" if (lat and lon) else f"id={city_id}"
        )
        self.tz_zone = tz.gettz(tz_name)
        logger.info(f"OWM wrapper initialized with API v{self._api_version}, lang={language}, tz={tz_name}.")

    def get_weather_data_from_owm(self, weather: WEATHER_TYPE):
        """Gets either current or forecast weather data."""
        if weather == "current":
            url = f"{API_BASE_URL}/2.5/weather?{self.location_substring}&appid={self.api_key}&units=Metric&lang={self.language}"
            data = get_json_from_url(url)
            if self._api_version == "3.0":
                uvi_url = f"{API_BASE_URL}/3.0/onecall?{self.location_substring}&appid={self.api_key}&exclude=minutely,hourly,daily&units=Metric&lang={self.language}"
                data["uvi"] = get_json_from_url(uvi_url)["current"].get("uvi")
        elif weather == "forecast":
            url = f"{API_BASE_URL}/2.5/forecast?{self.location_substring}&appid={self.api_key}&units=Metric&lang={self.language}"
            data = get_json_from_url(url)["list"]
        return data

    def get_current_weather(self) -> Dict:
        """
        Fetches and processes current weather data.
        Includes gust fallback and unit conversions.
        """
        data = self.get_weather_data_from_owm("current")
        wind_data = data.get("wind", {})
        base_speed = wind_data.get("speed", 0.0)
        gust_speed = wind_data.get("gust", base_speed)
        converted_gust = self.get_converted_windspeed(gust_speed)

        weather = {
            "detailed_status": data["weather"][0]["description"],
            "weather_icon_name": data["weather"][0]["icon"],
            "temp": self.get_converted_temperature(data["main"]["temp"]),
            "temp_feels_like": self.get_converted_temperature(data["main"]["feels_like"]),
            "min_temp": self.get_converted_temperature(data["main"]["temp_min"]),
            "max_temp": self.get_converted_temperature(data["main"]["temp_max"]),
            "humidity": data["main"]["humidity"],
            "wind": converted_gust,
            "wind_gust": converted_gust,
            "uvi": data.get("uvi"),
            "sunrise": datetime.fromtimestamp(data["sys"]["sunrise"], tz=self.tz_zone),
            "sunset": datetime.fromtimestamp(data["sys"]["sunset"], tz=self.tz_zone),
        }

        return weather

    def get_weather_forecast(self) -> List[Dict]:
        """
        Parses OWM 5-day / 3-hour forecast into a list of hourly dictionaries.
        """
        forecasts = self.get_weather_data_from_owm("forecast")
        hourly = []

        for f in forecasts:
            rain = f.get("rain", {}).get("3h", 0.0)
            snow = f.get("snow", {}).get("3h", 0.0)
            precip_mm = rain + snow

            hourly.append({
                "temp": self.get_converted_temperature(f["main"]["temp"]),
                "min_temp": self.get_converted_temperature(f["main"]["temp_min"]),
                "max_temp": self.get_converted_temperature(f["main"]["temp_max"]),
                "precip_3h_mm": precip_mm,
                "wind": self.get_converted_windspeed(f["wind"]["speed"]),
                "wind_gust": self.get_converted_windspeed(f["wind"].get("gust", f["wind"]["speed"])),
                "pressure": f["main"]["pressure"],
                "humidity": f["main"]["humidity"],
                "precip_probability": f.get("pop", 0.0) * 100.0,
                "icon": f["weather"][0]["icon"],
                "datetime": datetime.fromtimestamp(f["dt"], tz=self.tz_zone),
            })

        return hourly

    def get_forecast_for_day(self, days_from_today: int) -> Dict:
        """
        Aggregates hourly data into daily summary with min/max temp, precip and icon.
        """
        forecasts = self.get_weather_forecast()
        now = datetime.now(tz=self.tz_zone)
        start = now.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=days_from_today)
        end = start + timedelta(days=1)

        daily = [f for f in forecasts if start <= f["datetime"] < end]
        if not daily:
            daily.append(forecasts[0])  # fallback to first forecast

        temps = [f["temp"] for f in daily]
        rain = sum(f["precip_3h_mm"] for f in daily)
        icons = [f["icon"] for f in daily if f["icon"]]
        icon = max(set(icons), key=icons.count)

        return {
            "datetime": start,
            "icon": icon,
            "temp_min": min(temps),
            "temp_max": max(temps),
            "precip_mm": rain
        }

    def get_converted_temperature(self, value: float) -> float:
        return self.celsius_to_fahrenheit(value) if self.temp_unit == "fahrenheit" else value

    def get_converted_windspeed(self, value: float) -> float:
        if self.wind_unit == "km_hour":
            return self.mps_to_kph(value)
        if self.wind_unit == "miles_hour":
            return self.mps_to_mph(value)
        if self.wind_unit == "knots":
            return self.mps_to_knots(value)
        if self.wind_unit == "beaufort":
            return self.mps_to_beaufort(value)
        return value  # default is meters/sec

    @staticmethod
    def mps_to_beaufort(mps: float) -> int:
        thresholds = [0.3, 1.6, 3.4, 5.5, 8.0, 10.8, 13.9, 17.2, 20.8, 24.5, 28.5, 32.7]
        return next((i for i, t in enumerate(thresholds) if mps < t), 12)

    @staticmethod
    def mps_to_mph(mps: float) -> float:
        return mps * 2.23694

    @staticmethod
    def mps_to_kph(mps: float) -> float:
        return mps * 3.6

    @staticmethod
    def mps_to_knots(mps: float) -> float:
        return mps * 1.94384

    @staticmethod
    def celsius_to_fahrenheit(c: float) -> float:
        return c * 9.0 / 5.0 + 32.0


def main():
    # Simple test entry point
    key = ""
    city = 2643743  # London
    owm = OpenWeatherMap(api_key=key, city_id=city, language="de", tz_name="Europe/Berlin")
    print(owm.get_current_weather())
    print(owm.get_forecast_for_day(days_from_today=2))


if __name__ == "__main__":
    main()