"""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]] = {} # Common English country names → ISO 3166-1 alpha-2 codes # (Open-Meteo returns German names, so we need this for English→code matching) _COUNTRY_CODES: Dict[str, str] = { "croatia": "hr", "kroatien": "hr", "germany": "de", "deutschland": "de", "austria": "at", "österreich": "at", "switzerland": "ch", "schweiz": "ch", "france": "fr", "frankreich": "fr", "italy": "it", "italien": "it", "spain": "es", "spanien": "es", "portugal": "pt", "greece": "gr", "griechenland": "gr", "turkey": "tr", "türkei": "tr", "netherlands": "nl", "niederlande": "nl", "belgium": "be", "belgien": "be", "poland": "pl", "polen": "pl", "czech republic": "cz", "tschechien": "cz", "hungary": "hu", "ungarn": "hu", "romania": "ro", "rumänien": "ro", "bulgaria": "bg", "bulgarien": "bg", "sweden": "se", "schweden": "se", "norway": "no", "norwegen": "no", "denmark": "dk", "dänemark": "dk", "finland": "fi", "finnland": "fi", "uk": "gb", "united kingdom": "gb", "england": "gb", "usa": "us", "united states": "us", "canada": "ca", "kanada": "ca", "australia": "au", "australien": "au", "japan": "jp", "china": "cn", "india": "in", "indien": "in", "brazil": "br", "brasilien": "br", "mexico": "mx", "mexiko": "mx", "morocco": "ma", "marokko": "ma", "egypt": "eg", "ägypten": "eg", "thailand": "th", "indonesia": "id", "indonesien": "id", "slovenia": "si", "slowenien": "si", "serbia": "rs", "serbien": "rs", "montenegro": "me", "bosnia": "ba", "bosnien": "ba", } 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: hint = country_hint.lower() # Resolve English country names to ISO codes for matching hint_code = _COUNTRY_CODES.get(hint, hint[:2]) matched = False for r in results: country = (r.get("country", "") or "").lower() country_code = (r.get("country_code", "") or "").lower() # Match: hint in DE name, DE name in hint, ISO code, or exact city name if (hint in country or country in hint or hint_code == country_code or hint == country_code): best = r matched = True break if not matched: # Prefer an exact city name match over the first result for r in results: if r.get("name", "").lower() == city_name.lower(): best = r matched = True break if not matched: 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": [], "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": 7, }, ) 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 7-day forecast daily = data.get("daily", {}) dates = daily.get("time", []) forecast: List[Dict[str, Any]] = [] for i, date_str in enumerate(dates): codes_list = daily.get("weather_code", []) code = int(codes_list[i]) if i < len(codes_list) else 0 max_temps = daily.get("temperature_2m_max", []) min_temps = daily.get("temperature_2m_min", []) sunrises = daily.get("sunrise", []) sunsets = daily.get("sunset", []) forecast.append({ "date": date_str, "max_temp": round(max_temps[i]) if i < len(max_temps) else 0, "min_temp": round(min_temps[i]) if i < len(min_temps) else 0, "icon": _wmo_icon(code), "description": _wmo_description(code), "sunrise": sunrises[i].split("T")[-1] if i < len(sunrises) and sunrises[i] else "", "sunset": sunsets[i].split("T")[-1] if i < len(sunsets) and sunsets[i] else "", }) result["forecast"] = forecast logger.info("[WEATHER] Fetched '%s': %d°C, %s", geo["name"], result["temp"], result["description"]) return result async def fetch_hourly_forecast(location: str, max_slots: int = 8) -> List[Dict[str, Any]]: """Fetch hourly forecast from Open-Meteo. Args: location: City name or coordinates. max_slots: Maximum number of hourly slots to return. Returns: List of hourly forecast dicts. """ 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) >= max_slots: break logger.debug("[WEATHER] Hourly for '%s': %d slots", location, len(upcoming)) return upcoming