""" Open-Meteo weather API client. Free weather API with no API key required. https://open-meteo.com/en/docs Uses Open-Meteo Geocoding API for city name to coordinate conversion. """ import httpx import logging from datetime import datetime from typing import Optional from .base import ( WeatherProvider, AirQualityProvider, WeatherCondition, CurrentWeather, DayForecast, WeatherForecast, GeoLocation, SunTimes, AirQuality, ) logger = logging.getLogger(__name__) # WMO Weather interpretation codes to our standardized conditions # https://open-meteo.com/en/docs#weathervariables WMO_CODE_MAP: dict[int, WeatherCondition] = { 0: WeatherCondition.CLEAR, # Clear sky 1: WeatherCondition.CLEAR, # Mainly clear 2: WeatherCondition.PARTLY_CLOUDY, # Partly cloudy 3: WeatherCondition.CLOUDY, # Overcast 45: WeatherCondition.FOG, # Fog 48: WeatherCondition.FOG, # Depositing rime fog 51: WeatherCondition.DRIZZLE, # Light drizzle 53: WeatherCondition.DRIZZLE, # Moderate drizzle 55: WeatherCondition.DRIZZLE, # Dense drizzle 56: WeatherCondition.DRIZZLE, # Light freezing drizzle 57: WeatherCondition.DRIZZLE, # Dense freezing drizzle 61: WeatherCondition.RAIN, # Slight rain 63: WeatherCondition.RAIN, # Moderate rain 65: WeatherCondition.HEAVY_RAIN, # Heavy rain 66: WeatherCondition.RAIN, # Light freezing rain 67: WeatherCondition.HEAVY_RAIN, # Heavy freezing rain 71: WeatherCondition.SNOW, # Slight snow fall 73: WeatherCondition.SNOW, # Moderate snow fall 75: WeatherCondition.HEAVY_SNOW, # Heavy snow fall 77: WeatherCondition.SNOW, # Snow grains 80: WeatherCondition.RAIN, # Slight rain showers 81: WeatherCondition.RAIN, # Moderate rain showers 82: WeatherCondition.HEAVY_RAIN, # Violent rain showers 85: WeatherCondition.SNOW, # Slight snow showers 86: WeatherCondition.HEAVY_SNOW, # Heavy snow showers 95: WeatherCondition.THUNDERSTORM, # Thunderstorm 96: WeatherCondition.THUNDERSTORM, # Thunderstorm with slight hail 99: WeatherCondition.THUNDERSTORM, # Thunderstorm with heavy hail } # Human-readable descriptions for WMO codes WMO_DESCRIPTIONS: dict[int, str] = { 0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast", 45: "Fog", 48: "Depositing rime fog", 51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle", 56: "Light freezing drizzle", 57: "Dense freezing drizzle", 61: "Slight rain", 63: "Moderate rain", 65: "Heavy rain", 66: "Light freezing rain", 67: "Heavy freezing rain", 71: "Slight snow fall", 73: "Moderate snow fall", 75: "Heavy snow fall", 77: "Snow grains", 80: "Slight rain showers", 81: "Moderate rain showers", 82: "Violent rain showers", 85: "Slight snow showers", 86: "Heavy snow showers", 95: "Thunderstorm", 96: "Thunderstorm with slight hail", 99: "Thunderstorm with heavy hail", } class OpenMeteoProvider(WeatherProvider, AirQualityProvider): """Open-Meteo weather and air quality API implementation.""" GEOCODING_URL = "https://geocoding-api.open-meteo.com/v1/search" WEATHER_URL = "https://api.open-meteo.com/v1/forecast" AIR_QUALITY_URL = "https://air-quality-api.open-meteo.com/v1/air-quality" def __init__( self, timezone: str = "Europe/Amsterdam", timeout: int = 10 ): """ Initialize Open-Meteo client. Args: timezone: Default timezone for weather data timeout: HTTP request timeout in seconds """ self.timezone = timezone self.timeout = timeout self._client: Optional[httpx.AsyncClient] = None @property def client(self) -> httpx.AsyncClient: """Lazy-initialize HTTP client.""" if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient(timeout=self.timeout) return self._client async def close(self): """Close HTTP client.""" if self._client and not self._client.is_closed: await self._client.aclose() self._client = None async def geocode(self, city: str) -> Optional[GeoLocation]: """ Convert city name to coordinates. Args: city: City name (can include country, e.g., "Amsterdam, Netherlands") Returns: GeoLocation with coordinates or None if not found """ try: response = await self.client.get( self.GEOCODING_URL, params={ "name": city, "count": 1, "language": "en", "format": "json" } ) response.raise_for_status() data = response.json() results = data.get("results", []) if not results: logger.warning(f"No geocoding results for: {city}") return None result = results[0] return GeoLocation( name=result.get("name", city), latitude=result["latitude"], longitude=result["longitude"], country=result.get("country"), admin_area=result.get("admin1") # State/province ) except httpx.HTTPError as e: logger.error(f"Geocoding request failed for '{city}': {e}") return None except (KeyError, IndexError) as e: logger.error(f"Invalid geocoding response for '{city}': {e}") return None async def get_current(self, location: GeoLocation) -> CurrentWeather: """ Get current weather for a location. Args: location: GeoLocation with lat/long Returns: CurrentWeather with standardized data Raises: ValueError: If API request fails """ try: response = await self.client.get( self.WEATHER_URL, params={ "latitude": location.latitude, "longitude": location.longitude, "current": [ "temperature_2m", "apparent_temperature", "relative_humidity_2m", "weather_code", "wind_speed_10m", "wind_direction_10m" ], "daily": ["uv_index_max"], "timezone": self.timezone, "temperature_unit": "celsius", "wind_speed_unit": "kmh", "forecast_days": 1 } ) response.raise_for_status() data = response.json() current = data.get("current", {}) weather_code = current.get("weather_code", 0) # Get today's UV index from daily data daily = data.get("daily", {}) uv_index = None if daily.get("uv_index_max"): uv_index = daily["uv_index_max"][0] return CurrentWeather( temperature=current.get("temperature_2m", 0.0), feels_like=current.get("apparent_temperature"), humidity=int(current.get("relative_humidity_2m", 0)), wind_speed=current.get("wind_speed_10m", 0.0), wind_direction=current.get("wind_direction_10m"), condition=WMO_CODE_MAP.get(weather_code, WeatherCondition.UNKNOWN), condition_text=WMO_DESCRIPTIONS.get(weather_code, "Unknown"), timestamp=datetime.now(), location=location.name, uv_index=uv_index ) except httpx.HTTPError as e: logger.error(f"Weather request failed for {location.name}: {e}") raise ValueError(f"Failed to get weather: {e}") async def get_forecast( self, location: GeoLocation, days: int = 7 ) -> WeatherForecast: """ Get weather forecast for a location. Args: location: GeoLocation with lat/long days: Number of forecast days (1-16) Returns: WeatherForecast with current and daily data Raises: ValueError: If API request fails """ days = min(max(days, 1), 16) # Open-Meteo supports 1-16 days try: response = await self.client.get( self.WEATHER_URL, params={ "latitude": location.latitude, "longitude": location.longitude, "current": [ "temperature_2m", "apparent_temperature", "relative_humidity_2m", "weather_code", "wind_speed_10m", "wind_direction_10m" ], "daily": [ "weather_code", "temperature_2m_max", "temperature_2m_min", "precipitation_sum", "precipitation_probability_max", "uv_index_max" ], "timezone": self.timezone, "temperature_unit": "celsius", "wind_speed_unit": "kmh", "forecast_days": days } ) response.raise_for_status() data = response.json() # Parse current weather current_data = data.get("current", {}) daily_data = data.get("daily", {}) weather_code = current_data.get("weather_code", 0) # Get today's UV from daily data uv_index = None if daily_data.get("uv_index_max"): uv_index = daily_data["uv_index_max"][0] current = CurrentWeather( temperature=current_data.get("temperature_2m", 0.0), feels_like=current_data.get("apparent_temperature"), humidity=int(current_data.get("relative_humidity_2m", 0)), wind_speed=current_data.get("wind_speed_10m", 0.0), wind_direction=current_data.get("wind_direction_10m"), condition=WMO_CODE_MAP.get(weather_code, WeatherCondition.UNKNOWN), condition_text=WMO_DESCRIPTIONS.get(weather_code, "Unknown"), timestamp=datetime.now(), location=location.name, uv_index=uv_index ) # Parse daily forecast daily = [] dates = daily_data.get("time", []) for i, date_str in enumerate(dates): code = daily_data.get("weather_code", [])[i] if i < len(daily_data.get("weather_code", [])) else 0 uv_max = daily_data.get("uv_index_max", [])[i] if i < len(daily_data.get("uv_index_max", [])) else None daily.append(DayForecast( date=datetime.fromisoformat(date_str), temp_high=daily_data.get("temperature_2m_max", [])[i] if i < len(daily_data.get("temperature_2m_max", [])) else 0.0, temp_low=daily_data.get("temperature_2m_min", [])[i] if i < len(daily_data.get("temperature_2m_min", [])) else 0.0, condition=WMO_CODE_MAP.get(code, WeatherCondition.UNKNOWN), condition_text=WMO_DESCRIPTIONS.get(code, "Unknown"), precipitation_chance=daily_data.get("precipitation_probability_max", [])[i] if i < len(daily_data.get("precipitation_probability_max", [])) else None, precipitation_mm=daily_data.get("precipitation_sum", [])[i] if i < len(daily_data.get("precipitation_sum", [])) else None, uv_index_max=uv_max )) return WeatherForecast( location=location.name, current=current, daily=daily ) except httpx.HTTPError as e: logger.error(f"Forecast request failed for {location.name}: {e}") raise ValueError(f"Failed to get forecast: {e}") async def get_sun_times(self, location: GeoLocation) -> SunTimes: """ Get sunrise/sunset times for today. Args: location: GeoLocation with lat/long Returns: SunTimes with sunrise, sunset, and daylight duration Raises: ValueError: If API request fails """ try: response = await self.client.get( self.WEATHER_URL, params={ "latitude": location.latitude, "longitude": location.longitude, "daily": [ "sunrise", "sunset", "daylight_duration" ], "timezone": self.timezone, "forecast_days": 1 } ) response.raise_for_status() data = response.json() daily = data.get("daily", {}) date_str = daily.get("time", [""])[0] sunrise_str = daily.get("sunrise", [""])[0] sunset_str = daily.get("sunset", [""])[0] daylight = daily.get("daylight_duration", [0])[0] return SunTimes( location=location.name, date=datetime.fromisoformat(date_str) if date_str else datetime.now(), sunrise=datetime.fromisoformat(sunrise_str) if sunrise_str else datetime.now(), sunset=datetime.fromisoformat(sunset_str) if sunset_str else datetime.now(), daylight_duration=int(daylight) if daylight else 0 ) except httpx.HTTPError as e: logger.error(f"Sun times request failed for {location.name}: {e}") raise ValueError(f"Failed to get sun times: {e}") async def get_air_quality(self, location: GeoLocation) -> AirQuality: """ Get current air quality for a location. Args: location: GeoLocation with lat/long Returns: AirQuality with pollutant measurements and AQI Raises: ValueError: If API request fails """ try: response = await self.client.get( self.AIR_QUALITY_URL, params={ "latitude": location.latitude, "longitude": location.longitude, "current": [ "european_aqi", "us_aqi", "pm2_5", "pm10", "ozone", "nitrogen_dioxide", "sulphur_dioxide", "carbon_monoxide", "grass_pollen", "birch_pollen", "alder_pollen" ], "timezone": self.timezone } ) response.raise_for_status() data = response.json() current = data.get("current", {}) return AirQuality( location=location.name, timestamp=datetime.now(), aqi_european=current.get("european_aqi"), aqi_us=current.get("us_aqi"), pm2_5=current.get("pm2_5"), pm10=current.get("pm10"), ozone=current.get("ozone"), nitrogen_dioxide=current.get("nitrogen_dioxide"), sulphur_dioxide=current.get("sulphur_dioxide"), carbon_monoxide=current.get("carbon_monoxide"), pollen_grass=current.get("grass_pollen"), pollen_birch=current.get("birch_pollen"), pollen_alder=current.get("alder_pollen") ) except httpx.HTTPError as e: logger.error(f"Air quality request failed for {location.name}: {e}") raise ValueError(f"Failed to get air quality: {e}")