"""Weather service — fetches weather data from Open-Meteo (free, no API key). Uses the Open-Meteo geocoding API to resolve city names to coordinates, then fetches current weather, hourly forecast, and 3-day daily forecast. """ from __future__ import annotations import logging from datetime import datetime from typing import Any, Dict, List, Optional, Tuple import httpx logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # WMO Weather Code → Emoji + German description # https://open-meteo.com/en/docs#weathervariables # --------------------------------------------------------------------------- WMO_CODES: Dict[int, Tuple[str, str]] = { 0: ("☀️", "Klar"), 1: ("🌤️", "Überwiegend klar"), 2: ("⛅", "Teilweise Bewölkt"), 3: ("☁️", "Bewölkt"), 45: ("🌫️", "Nebel"), 48: ("🌫️", "Reifnebel"), 51: ("🌦️", "Leichter Nieselregen"), 53: ("🌧️", "Mäßiger Nieselregen"), 55: ("🌧️", "Starker Nieselregen"), 56: ("🌨️", "Gefrierender Nieselregen"), 57: ("🌨️", "Starker gefr. Nieselregen"), 61: ("🌦️", "Leichter Regen"), 63: ("🌧️", "Mäßiger Regen"), 65: ("🌧️", "Starker Regen"), 66: ("🌨️", "Gefrierender Regen"), 67: ("🌨️", "Starker gefr. Regen"), 71: ("❄️", "Leichter Schneefall"), 73: ("❄️", "Mäßiger Schneefall"), 75: ("❄️", "Starker Schneefall"), 77: ("🌨️", "Schneekörner"), 80: ("🌦️", "Leichte Regenschauer"), 81: ("🌧️", "Mäßige Regenschauer"), 82: ("🌧️", "Starke Regenschauer"), 85: ("❄️", "Leichte Schneeschauer"), 86: ("❄️", "Starke Schneeschauer"), 95: ("⛈️", "Gewitter"), 96: ("⛈️", "Gewitter mit leichtem Hagel"), 99: ("⛈️", "Gewitter mit starkem Hagel"), } def _wmo_icon(code: int) -> str: """Map WMO weather code to an emoji.""" return WMO_CODES.get(code, ("🌤️", "Unbekannt"))[0] def _wmo_description(code: int) -> str: """Map WMO weather code to a German description.""" return WMO_CODES.get(code, ("🌤️", "Unbekannt"))[1] # --------------------------------------------------------------------------- # Geocoding cache (in-memory, process-lifetime) # --------------------------------------------------------------------------- _geocode_cache: Dict[str, Dict[str, Any]] = {} async def _geocode(location: str) -> Optional[Dict[str, Any]]: """Resolve a city name to lat/lon using Open-Meteo Geocoding API. Supports "City,Country" format (e.g. "Rab,Croatia") — the country part is used to filter results when multiple matches exist. Returns dict with keys: latitude, longitude, name, timezone """ cache_key = location.lower().strip() if cache_key in _geocode_cache: return _geocode_cache[cache_key] # Split "City,Country" into search name + country filter city_name = location country_hint = "" if "," in location: parts = [p.strip() for p in location.split(",", 1)] city_name = parts[0] country_hint = parts[1].lower() if len(parts) > 1 else "" try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( "https://geocoding-api.open-meteo.com/v1/search", params={"name": city_name, "count": 10, "language": "de"}, ) resp.raise_for_status() data = resp.json() except Exception as exc: logger.error("[WEATHER] Geocoding failed for '%s': %s", location, exc) return None results = data.get("results", []) if not results: logger.warning("[WEATHER] Geocoding: no results for '%s'", location) return None # If country hint provided, try to find a matching result best = results[0] if country_hint: for r in results: country = (r.get("country", "") or "").lower() country_code = (r.get("country_code", "") or "").lower() if country_hint in country or country_hint == country_code: best = r break else: logger.warning("[WEATHER] Country '%s' not found in results for '%s', using first match", country_hint, city_name) geo = { "latitude": best["latitude"], "longitude": best["longitude"], "name": best.get("name", location), "timezone": best.get("timezone", "Europe/Berlin"), } _geocode_cache[cache_key] = geo logger.info("[WEATHER] Geocoded '%s' → %s (%s, %.2f, %.2f)", location, geo["name"], best.get("country", "?"), geo["latitude"], geo["longitude"]) return geo # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- async def fetch_weather(location: str) -> Dict[str, Any]: """Fetch current weather and 3-day forecast from Open-Meteo. Args: location: City name (e.g. "Berlin", "Leverkusen", "Rab,Croatia"). Returns: Dict with current conditions, 3-day forecast, and optional error. """ fallback: Dict[str, Any] = { "location": location, "temp": 0, "feels_like": 0, "humidity": 0, "wind_kmh": 0, "description": "Nicht verfügbar", "icon": "❓", "forecast_3day": [], "error": None, } # Step 1: Geocode geo = await _geocode(location) if geo is None: fallback["error"] = f"Standort '{location}' nicht gefunden" logger.error("[WEATHER] Cannot fetch weather — geocoding failed for '%s'", location) return fallback # Step 2: Fetch weather try: async with httpx.AsyncClient(timeout=15) as client: resp = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": geo["latitude"], "longitude": geo["longitude"], "current": "temperature_2m,relative_humidity_2m,apparent_temperature,weather_code,wind_speed_10m", "daily": "weather_code,temperature_2m_max,temperature_2m_min,sunrise,sunset", "timezone": geo["timezone"], "forecast_days": 3, }, ) resp.raise_for_status() data = resp.json() except httpx.HTTPStatusError as exc: msg = f"HTTP {exc.response.status_code}" logger.error("[WEATHER] Open-Meteo HTTP error for '%s': %s", location, msg) fallback["error"] = msg return fallback except Exception as exc: msg = f"{type(exc).__name__}: {exc}" logger.error("[WEATHER] Open-Meteo request failed for '%s': %s", location, msg) fallback["error"] = msg return fallback # Step 3: Parse current conditions current = data.get("current", {}) wmo_code = int(current.get("weather_code", 0)) result: Dict[str, Any] = { "location": geo["name"], "temp": round(current.get("temperature_2m", 0)), "feels_like": round(current.get("apparent_temperature", 0)), "humidity": round(current.get("relative_humidity_2m", 0)), "wind_kmh": round(current.get("wind_speed_10m", 0)), "description": _wmo_description(wmo_code), "icon": _wmo_icon(wmo_code), "error": None, } # Step 4: Parse 3-day forecast daily = data.get("daily", {}) dates = daily.get("time", []) forecast_3day: List[Dict[str, Any]] = [] for i, date_str in enumerate(dates[:3]): code = int(daily.get("weather_code", [0] * 3)[i]) forecast_3day.append({ "date": date_str, "max_temp": round(daily.get("temperature_2m_max", [0] * 3)[i]), "min_temp": round(daily.get("temperature_2m_min", [0] * 3)[i]), "icon": _wmo_icon(code), "description": _wmo_description(code), "sunrise": daily.get("sunrise", [""] * 3)[i].split("T")[-1] if daily.get("sunrise") else "", "sunset": daily.get("sunset", [""] * 3)[i].split("T")[-1] if daily.get("sunset") else "", }) result["forecast_3day"] = forecast_3day logger.info("[WEATHER] Fetched '%s': %d°C, %s", geo["name"], result["temp"], result["description"]) return result async def fetch_hourly_forecast(location: str) -> List[Dict[str, Any]]: """Fetch hourly forecast for the next 8 hours from Open-Meteo. Args: location: City name or coordinates. Returns: List of hourly forecast dicts (max 8 entries). """ geo = await _geocode(location) if geo is None: logger.error("[WEATHER] Cannot fetch hourly — geocoding failed for '%s'", location) return [] try: async with httpx.AsyncClient(timeout=15) as client: resp = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": geo["latitude"], "longitude": geo["longitude"], "hourly": "temperature_2m,weather_code,precipitation_probability,wind_speed_10m", "timezone": geo["timezone"], "forecast_days": 2, }, ) resp.raise_for_status() data = resp.json() except Exception as exc: logger.error("[WEATHER] Hourly fetch failed for '%s': %s", location, exc) return [] hourly = data.get("hourly", {}) times = hourly.get("time", []) temps = hourly.get("temperature_2m", []) codes = hourly.get("weather_code", []) precips = hourly.get("precipitation_probability", []) winds = hourly.get("wind_speed_10m", []) if not times: logger.warning("[WEATHER] No hourly data for '%s'", location) return [] now = datetime.now() now_str = now.strftime("%Y-%m-%dT%H:00") upcoming: List[Dict[str, Any]] = [] started = False for i, t in enumerate(times): if not started: if t >= now_str: started = True else: continue code = int(codes[i]) if i < len(codes) else 0 upcoming.append({ "time": t.split("T")[1][:5], # "HH:MM" "temp": round(temps[i]) if i < len(temps) else 0, "icon": _wmo_icon(code), "description": _wmo_description(code), "precip_chance": round(precips[i]) if i < len(precips) else 0, "wind_kmh": round(winds[i]) if i < len(winds) else 0, }) if len(upcoming) >= 8: break logger.debug("[WEATHER] Hourly for '%s': %d slots", location, len(upcoming)) return upcoming