root/daily-briefing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
daily-briefing/server/services/weather_service.py
Sam 2f56be835e Add: München als 3. Wetter-Location + Wetter-Detail-Modal 
- München als tertiärer Standort (iris-Akzent) hinzugefügt
- Klick auf WeatherCard öffnet Detail-Modal mit:
  - 24h stündliche Prognose (horizontal scrollbar)
  - 7-Tage-Vorhersage mit Temperaturbalken
  - Wind, Feuchte, Sonnenauf/-untergang
- Backend: 7-Tage statt 3-Tage Forecast, 24 Hourly-Slots pro Standort
- Backend: forecast_3day → forecast Feldname-Konsistenz
- Dashboard: 3-Spalten Wetter-Grid statt 4 (HourlyForecast → Modal)
- Admin: Tertiärer Standort konfigurierbar
- THERMAL Design: iris glow, modal animation, Portal-basiertes Modal

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 01:13:49 +01:00

367 lines
13 KiB
Python
Raw Blame History

"""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
Reference in a new issue View git blame Copy permalink