""" BAN YARO — Wetter via Open-Meteo - get_day_alert(): standortbezogener Wetter-Alarm-Push (Hitze/Gewitter) - get_weather_for_location(): API-Endpoint, beliebiger Standort mit TTL-Cache """ import asyncio import time import logging import httpx from datetime import datetime logger = logging.getLogger(__name__) # WMO-Wettercodes → (Beschreibung, Phosphor-Icon-Name) _WMO = { 0: ('Klar', 'sun'), 1: ('Überwiegend klar', 'cloud-sun'), 2: ('Teilweise bewölkt', 'cloud-sun'), 3: ('Bedeckt', 'cloud'), 45: ('Nebel', 'cloud-fog'), 48: ('Gefrierender Nebel', 'cloud-fog'), 51: ('Leichter Nieselregen', 'cloud-rain'), 53: ('Nieselregen', 'cloud-rain'), 55: ('Starker Nieselregen', 'cloud-rain'), 61: ('Leichter Regen', 'cloud-rain'), 63: ('Regen', 'cloud-rain'), 65: ('Starker Regen', 'cloud-rain'), 71: ('Leichter Schnee', 'snowflake'), 73: ('Schnee', 'snowflake'), 75: ('Starker Schnee', 'snowflake'), 77: ('Schneekörner', 'snowflake'), 80: ('Leichte Schauer', 'cloud-rain'), 81: ('Schauer', 'cloud-rain'), 82: ('Starke Schauer', 'cloud-rain'), 85: ('Schneeschauer', 'snowflake'), 86: ('Starke Schneeschauer', 'snowflake'), 95: ('Gewitter', 'cloud-lightning'), 96: ('Gewitter mit Hagel', 'cloud-lightning'), 99: ('Schweres Gewitter', 'cloud-lightning'), } # TTL-Cache: (round(lat,1), round(lon,1)) → (timestamp, data) _location_cache: dict = {} _CACHE_TTL = 1800 # 30 Minuten async def get_weather_for_location(lat: float, lon: float) -> dict: """Holt aktuelles Wetter für einen Standort. 30-min TTL-Cache.""" key = (round(lat, 1), round(lon, 1)) now = time.time() if key in _location_cache: ts, cached = _location_cache[key] if now - ts < _CACHE_TTL: return cached url = ( "https://api.open-meteo.com/v1/forecast" f"?latitude={lat}&longitude={lon}" "¤t=temperature_2m,apparent_temperature,weathercode,windspeed_10m,is_day" "&daily=precipitation_probability_max,uv_index_max" "&hourly=precipitation_probability" "&timezone=Europe%2FBerlin&forecast_days=1" ) async with httpx.AsyncClient(timeout=8.0) as client: resp, geo_resp = await asyncio.gather( client.get(url), client.get( f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json&zoom=10", headers={"User-Agent": "BanYaro/1.0 support@banyaro.app"} ), return_exceptions=True, ) resp.raise_for_status() raw = resp.json() # Ortsname aus Reverse-Geocoding location_name = None try: if not isinstance(geo_resp, Exception) and geo_resp.status_code == 200: geo = geo_resp.json() addr = geo.get("address", {}) location_name = (addr.get("city") or addr.get("town") or addr.get("village") or addr.get("municipality") or addr.get("county") or geo.get("name")) except Exception: pass cur = raw.get('current', {}) daily = raw.get('daily', {}) hourly = raw.get('hourly', {}) temp = cur.get('temperature_2m') feels_like = cur.get('apparent_temperature') wcode = cur.get('weathercode', 0) wind = cur.get('windspeed_10m') is_day = cur.get('is_day', 1) _daily_precip_max = (daily.get('precipitation_probability_max') or [None])[0] # Fallback uv = (daily.get('uv_index_max') or [None])[0] desc, icon = _WMO.get(wcode, ('Unbekannt', 'cloud')) if wcode == 0 and not is_day: icon = 'moon' month = datetime.now().month zecken = None if temp is not None and temp > 7.0 and 3 <= month <= 10: zecken = 'hoch' if temp > 20 else ('mittel' if temp > 12 else 'niedrig') # Nächste Regenstunde + Umschwung-Warnung next_rain_time = None rain_warning_time = None # Stunde mit ≥40%-Sprung gegenüber Vorststunde already_raining = wcode >= 51 now_h = datetime.now().hour h_times = hourly.get('time', []) h_precip = hourly.get('precipitation_probability', []) # Niederschlag fürs Pill: Höchstwert der NÄCHSTEN 3 STUNDEN (ab aktueller Stunde) statt Tages-Max # — relevanter fürs "jetzt/gleich Gassi?". h_precip ist nach Stunde indiziert (0 = heute 00:00); # der Slice ist am Tagesende automatisch kürzer (forecast_days=1). _next3 = [p for p in h_precip[now_h:now_h + 3] if p is not None] precip = max(_next3) if _next3 else _daily_precip_max # Index-Liste nur für Stunden im Fenster now_h+1 … now_h+12 window = [] for t, p in zip(h_times, h_precip): try: entry_h = int(t[11:13]) except Exception: continue if entry_h <= now_h or entry_h > now_h + 12: continue window.append((entry_h, p if p is not None else 0)) if not already_raining: for entry_h, p in window: if next_rain_time is None and p >= 20: next_rain_time = f"{entry_h:02d}:00" break # Umschwung: Sprung ≥40% von einer Stunde zur nächsten prev_p = wcode >= 51 and 100 or (h_precip[now_h] if now_h < len(h_precip) else 0) or 0 for entry_h, p in window: if p - prev_p >= 40: rain_warning_time = f"{entry_h:02d}:00" break prev_p = p data = { 'temp_c': temp, 'feels_like_c': feels_like, 'weathercode': wcode, 'desc': desc, 'icon': icon, 'wind_kmh': wind, 'precip_prob': precip, 'uv_index': uv, 'is_day': bool(is_day), 'zecken_warnung': zecken, 'next_rain_time': next_rain_time, 'rain_warning_time': rain_warning_time, 'location_name': location_name, } _location_cache[key] = (now, data) return data async def get_weather_for_date(lat: float, lon: float, datum: str, hour: int | None = None) -> dict: """Wetter für ein bestimmtes Datum (YYYY-MM-DD) an einem Standort. Für nachträgliches Positionieren alter Tagebucheinträge: heutiges Datum → aktuelles Wetter; vergangene Tage → stündliche Historie. Open-Meteo Forecast-API deckt die jüngsten ~90 Tage Vergangenheit ab, ältere Tage holt die Archive-API. `hour` (0–23, Default 12) wählt die Tagesstunde. Rückgabe hat dieselbe Struktur wie get_weather_for_location (die nach vorn gerichteten Felder next_rain_time/rain_warning_time sind None).""" from datetime import date as _date try: target = _date.fromisoformat(datum[:10]) except (ValueError, TypeError): return await get_weather_for_location(lat, lon) today = datetime.now().date() if target >= today: # Heute/Zukunft → aktuelles Wetter (inkl. Vorhersage-Felder + Ortsname). return await get_weather_for_location(lat, lon) hour = 12 if hour is None else max(0, min(23, hour)) age_days = (today - target).days base = ("https://archive-api.open-meteo.com/v1/archive" if age_days > 90 else "https://api.open-meteo.com/v1/forecast") url = ( f"{base}?latitude={lat}&longitude={lon}" f"&start_date={target.isoformat()}&end_date={target.isoformat()}" "&hourly=temperature_2m,apparent_temperature,weathercode,windspeed_10m,is_day" "&daily=precipitation_probability_max,uv_index_max" "&timezone=Europe%2FBerlin" ) async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(url) resp.raise_for_status() raw = resp.json() hourly = raw.get('hourly', {}) daily = raw.get('daily', {}) def _at(arr): try: return arr[hour] except (IndexError, TypeError, KeyError): return None temp = _at(hourly.get('temperature_2m', [])) feels_like = _at(hourly.get('apparent_temperature', [])) wcode = _at(hourly.get('weathercode', [])) or 0 wind = _at(hourly.get('windspeed_10m', [])) is_day_v = _at(hourly.get('is_day', [])) is_day = 1 if is_day_v is None else is_day_v precip = (daily.get('precipitation_probability_max') or [None])[0] uv = (daily.get('uv_index_max') or [None])[0] desc, icon = _WMO.get(wcode, ('Unbekannt', 'cloud')) if wcode == 0 and not is_day: icon = 'moon' zecken = None if temp is not None and temp > 7.0 and 3 <= target.month <= 10: zecken = 'hoch' if temp > 20 else ('mittel' if temp > 12 else 'niedrig') return { 'temp_c': temp, 'feels_like_c': feels_like, 'weathercode': wcode, 'desc': desc, 'icon': icon, 'wind_kmh': wind, 'precip_prob': precip, 'uv_index': uv, 'is_day': bool(is_day), 'zecken_warnung': zecken, 'next_rain_time': None, 'rain_warning_time': None, 'location_name': None, 'historical': True, } # WMO-Wettercodes 95–99 = Gewitter THUNDERSTORM_CODES = {95, 96, 99} async def get_day_alert(lat: float, lon: float) -> dict: """ Tagesprognose (heute) für EINEN Standort — Grundlage für den standortbezogenen Wetter-Alarm-Push (Hitze / Gewitter). Gibt zurück: {"max_temp_c": float|None, "thunderstorm": bool} """ url = ( "https://api.open-meteo.com/v1/forecast" f"?latitude={lat}&longitude={lon}" "&daily=temperature_2m_max,precipitation_probability_max,weathercode" "&timezone=Europe%2FBerlin&forecast_days=1" ) async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(url) resp.raise_for_status() data = resp.json() daily = data.get("daily", {}) temp = (daily.get("temperature_2m_max") or [None])[0] precip_prob = (daily.get("precipitation_probability_max") or [None])[0] weathercode = (daily.get("weathercode") or [None])[0] thunderstorm = ( precip_prob is not None and precip_prob > 50 and weathercode is not None and int(weathercode) in THUNDERSTORM_CODES ) return {"max_temp_c": temp, "thunderstorm": thunderstorm} # --------------------------------------------------------------------------- # 7-Tage-Vorhersage # --------------------------------------------------------------------------- import asyncio # noqa: E402 — appended section _forecast_cache: dict = {} _FORECAST_TTL = 3600 # 1 Stunde _WDAY_DE = ['Mo', 'Di', 'Mi', 'Do', 'Fr', 'Sa', 'So'] def _wind_dir(deg: float) -> str: dirs = ['N', 'NO', 'O', 'SO', 'S', 'SW', 'W', 'NW'] idx = round(deg / 45) % 8 return dirs[idx] def _asphalt_temp(air_max: float, uv_max: float) -> tuple[float, str]: # UV-Bonus skaliert mit Temperatur: unter 10°C kaum Aufheizung, ab 30°C voll t_factor = max(0.0, min(1.0, (air_max - 5) / 25)) bonus = min(uv_max * 3.0 * t_factor, 30.0) asphalt = air_max + bonus if asphalt <= 30: warn = 'safe' elif asphalt <= 40: warn = 'warm' elif asphalt <= 55: warn = 'hot' else: warn = 'danger' return round(asphalt, 1), warn def _pollen_lvl(val: float | None) -> dict: if val is None: return {'level': 0, 'label': 'keine'} if val < 5: return {'level': 1, 'label': 'niedrig'} if val < 25: return {'level': 2, 'label': 'mittel'} if val < 100: return {'level': 3, 'label': 'hoch'} return {'level': 4, 'label': 'sehr hoch'} async def get_forecast(lat: float, lon: float) -> dict: """7-Tage-Wettervorhersage inkl. Pollen, Asphalttemperatur, Zecken. 1h TTL-Cache.""" key = (round(lat, 2), round(lon, 2)) now = time.time() if key in _forecast_cache: ts, cached = _forecast_cache[key] if now - ts < _FORECAST_TTL: return cached forecast_url = ( "https://api.open-meteo.com/v1/forecast" f"?latitude={lat}&longitude={lon}" "&daily=temperature_2m_max,temperature_2m_min,apparent_temperature_max," "apparent_temperature_min,precipitation_probability_max,precipitation_sum," "weathercode,windspeed_10m_max,winddirection_10m_dominant,uv_index_max," "sunrise,sunset" "&hourly=precipitation_probability,precipitation,weathercode" "&timezone=auto&forecast_days=7" ) pollen_url = ( "https://air-quality-api.open-meteo.com/v1/air-quality" f"?latitude={lat}&longitude={lon}" "&hourly=alder_pollen,birch_pollen,grass_pollen,mugwort_pollen,ragweed_pollen" "&timezone=auto&forecast_days=7" ) async with httpx.AsyncClient(timeout=10.0) as client: forecast_task = client.get(forecast_url) pollen_task = client.get(pollen_url) geo_task = client.get( f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json&zoom=10", headers={"User-Agent": "BanYaro/1.0 support@banyaro.app"} ) forecast_resp, pollen_resp, geo_resp_fc = await asyncio.gather( forecast_task, pollen_task, geo_task, return_exceptions=True ) location_name_fc = None try: if not isinstance(geo_resp_fc, Exception) and geo_resp_fc.status_code == 200: addr = geo_resp_fc.json().get("address", {}) location_name_fc = (addr.get("city") or addr.get("town") or addr.get("village") or addr.get("municipality") or addr.get("county")) except Exception: pass # --- Forecast (required) --- if isinstance(forecast_resp, Exception): raise forecast_resp forecast_resp.raise_for_status() raw = forecast_resp.json() daily = raw.get('daily', {}) hourly_fc = raw.get('hourly', {}) timezone = raw.get('timezone', 'auto') dates = daily.get('time', []) temp_max = daily.get('temperature_2m_max', []) temp_min = daily.get('temperature_2m_min', []) feels_max = daily.get('apparent_temperature_max', []) feels_min = daily.get('apparent_temperature_min', []) precip_prob = daily.get('precipitation_probability_max', []) precip_sum = daily.get('precipitation_sum', []) wcodes = daily.get('weathercode', []) wind_kmh = daily.get('windspeed_10m_max', []) wind_deg = daily.get('winddirection_10m_dominant', []) uv_index = daily.get('uv_index_max', []) sunrises = daily.get('sunrise', []) sunsets = daily.get('sunset', []) # --- Hourly precipitation data grouped by day --- hourly_times = hourly_fc.get('time', []) hourly_pp = hourly_fc.get('precipitation_probability', []) hourly_precip = hourly_fc.get('precipitation', []) hourly_wcode = hourly_fc.get('weathercode', []) # Build: date_str → list of {hour, precip_prob, precip, weathercode} _hourly_by_day: dict = {} for idx, ts_str in enumerate(hourly_times): day_str = ts_str[:10] # 'YYYY-MM-DD' hour_str = ts_str[11:16] # 'HH:MM' entry = { 'hour': hour_str, 'precip_prob': hourly_pp[idx] if idx < len(hourly_pp) else None, 'precip': hourly_precip[idx] if idx < len(hourly_precip) else None, 'weathercode': int(hourly_wcode[idx]) if idx < len(hourly_wcode) and hourly_wcode[idx] is not None else None, } _hourly_by_day.setdefault(day_str, []).append(entry) # --- Pollen (optional) --- pollen_daily: dict | None = None if not isinstance(pollen_resp, Exception): try: pollen_resp.raise_for_status() praw = pollen_resp.json() hourly = praw.get('hourly', {}) htimes = hourly.get('time', []) # aggregate hourly → daily max per type pollen_types = { 'erle': hourly.get('alder_pollen', []), 'birke': hourly.get('birch_pollen', []), 'graeser': hourly.get('grass_pollen', []), 'beifuss': hourly.get('mugwort_pollen', []), 'ambrosia': hourly.get('ragweed_pollen', []), } # build date → max mapping per type pollen_daily = {ptype: {} for ptype in pollen_types} for i, ts_str in enumerate(htimes): day_str = ts_str[:10] # 'YYYY-MM-DD' for ptype, vals in pollen_types.items(): v = vals[i] if i < len(vals) else None if v is not None: prev = pollen_daily[ptype].get(day_str) pollen_daily[ptype][day_str] = max(prev, v) if prev is not None else v except Exception as e: logger.warning(f"Pollen-Abruf fehlgeschlagen: {e}") pollen_daily = None # --- Assemble days --- days = [] for i, date_str in enumerate(dates): wcode = int(wcodes[i]) if i < len(wcodes) and wcodes[i] is not None else 0 desc, icon = _WMO.get(wcode, ('Unbekannt', 'cloud')) t_max = temp_max[i] if i < len(temp_max) else None t_min = temp_min[i] if i < len(temp_min) else None f_max = feels_max[i] if i < len(feels_max) else None f_min = feels_min[i] if i < len(feels_min) else None pp = precip_prob[i] if i < len(precip_prob) else None ps = precip_sum[i] if i < len(precip_sum) else None wk = wind_kmh[i] if i < len(wind_kmh) else None wd_deg = wind_deg[i] if i < len(wind_deg) else None uv = uv_index[i] if i < len(uv_index) else None # Sunrise / Sunset → HH:MM only (format: "2025-05-02T06:12") sunrise_raw = sunrises[i] if i < len(sunrises) else None sunset_raw = sunsets[i] if i < len(sunsets) else None sunrise_hm = sunrise_raw[11:16] if sunrise_raw and len(sunrise_raw) >= 16 else sunrise_raw sunset_hm = sunset_raw[11:16] if sunset_raw and len(sunset_raw) >= 16 else sunset_raw # Weekday try: dt_obj = datetime.strptime(date_str, '%Y-%m-%d') wday = _WDAY_DE[dt_obj.weekday()] except Exception: wday = '' # Asphalt asphalt_t, asphalt_w = _asphalt_temp(t_max or 0.0, uv or 0.0) # Zecken month = datetime.strptime(date_str, '%Y-%m-%d').month zecken = None if t_max is not None and t_max > 7.0 and 3 <= month <= 10: zecken = 'hoch' if t_max > 20 else ('mittel' if t_max > 12 else 'niedrig') # Pollen if pollen_daily is not None: pollen_out = { pt: _pollen_lvl(pollen_daily[pt].get(date_str)) for pt in ('erle', 'birke', 'graeser', 'beifuss', 'ambrosia') } else: pollen_out = None days.append({ 'date': date_str, 'wday': wday, 'weathercode': wcode, 'desc': desc, 'icon': icon, 'temp_max': t_max, 'temp_min': t_min, 'feels_max': f_max, 'feels_min': f_min, 'precip_prob': pp, 'precip_sum': ps, 'wind_kmh': wk, 'wind_dir': _wind_dir(wd_deg) if wd_deg is not None else None, 'wind_dir_deg': wd_deg, 'uv_index': uv, 'sunrise': sunrise_hm, 'sunset': sunset_hm, 'asphalt_temp': asphalt_t, 'asphalt_warn': asphalt_w, 'pollen': pollen_out, 'zecken': zecken, 'thunderstorm': wcode in {95, 96, 99}, 'paw_cold': wcode in {71, 73, 75, 77, 85, 86} or (t_min is not None and t_min < 0), 'hourly': _hourly_by_day.get(date_str, []), }) result = {'timezone': timezone, 'days': days, 'location_name': location_name_fc} _forecast_cache[key] = (now, result) _log_forecast(round(lat, 1), round(lon, 1), days) return result def _log_forecast(lat_r: float, lon_r: float, days: list) -> None: """Speichert jeden Forecast-Tag in weather_log (INSERT OR IGNORE — kein Überschreiben).""" if not days: return try: import json from database import db with db() as conn: for d in days: pollen = d.get('pollen') or {} conn.execute(""" INSERT OR IGNORE INTO weather_log (date, lat_r, lon_r, temp_max, temp_min, feels_max, precip_prob, precip_sum, wind_kmh, wind_dir, uv_index, weathercode, weatherdesc, sunrise, sunset, asphalt_temp, asphalt_warn, zecken, pollen_erle, pollen_birke, pollen_graeser, pollen_beifuss, pollen_ambrosia, forecast_json) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( d['date'], lat_r, lon_r, d.get('temp_max'), d.get('temp_min'), d.get('feels_max'), d.get('precip_prob'), d.get('precip_sum'), d.get('wind_kmh'), d.get('wind_dir'), d.get('uv_index'), d.get('weathercode'), d.get('desc'), d.get('sunrise'), d.get('sunset'), d.get('asphalt_temp'), d.get('asphalt_warn'), d.get('zecken'), pollen.get('erle', {}).get('level'), pollen.get('birke', {}).get('level'), pollen.get('graeser', {}).get('level'), pollen.get('beifuss', {}).get('level'), pollen.get('ambrosia',{}).get('level'), json.dumps(d, ensure_ascii=False), )) except Exception as e: logger.warning(f"weather_log insert fehlgeschlagen: {e}")