# -*- coding: utf-8 -*- from flask import Flask, render_template, request, jsonify from geopy.geocoders import Nominatim from geopy.exc import GeocoderTimedOut import math import pandas as pd import datetime as _dt import logging from collections import deque from threading import Lock from time import time from wetterdienst.provider.dwd.mosmix import DwdMosmixRequest from cachetools import TTLCache from astral import LocationInfo from astral.sun import sun as astral_sun import requests as _requests app = Flask(__name__) app.logger.setLevel(logging.INFO) _GEOLOCATOR = Nominatim(user_agent="skywatcher-app/1.0") # ── Zeitzone ──────────────────────────────────────────────────────────── try: import zoneinfo BERLIN = zoneinfo.ZoneInfo("Europe/Berlin") except ImportError: import pytz BERLIN = pytz.timezone("Europe/Berlin") # ── Cache: Forecasts 45 Min, DWD-Warnungen 15 Min ──────────────────────────── _forecast_cache = TTLCache(maxsize=64, ttl=2700) _warn_cache = TTLCache(maxsize=32, ttl=900) _suggest_cache = TTLCache(maxsize=256, ttl=900) _sun_cache: dict = {} # (lat_r, lon_r, date) -> (naive_sunrise_ts, naive_sunset_ts) # API protection: simple in-memory rate limiting by client IP. _suggest_rate_lock = Lock() _suggest_rate_hits = {} _SUGGEST_RATE_WINDOW_SECONDS = 60 _SUGGEST_RATE_MAX_REQUESTS = 30 STATE_ALIASES = { "baden-wurttemberg": {"baden-wurttemberg", "baden wuerttemberg", "bw"}, "bayern": {"bayern", "by"}, "berlin": {"berlin", "be"}, "brandenburg": {"brandenburg", "bb"}, "bremen": {"bremen", "hb"}, "hamburg": {"hamburg", "hh"}, "hessen": {"hessen", "he"}, "mecklenburg-vorpommern": {"mecklenburg-vorpommern", "mecklenburg vorpommern", "mv"}, "niedersachsen": {"niedersachsen", "ni"}, "nordrhein-westfalen": {"nordrhein-westfalen", "nordrhein westfalen", "nrw", "nw"}, "rheinland-pfalz": {"rheinland-pfalz", "rheinland pfalz", "rp"}, "saarland": {"saarland", "sl"}, "sachsen": {"sachsen", "sn"}, "sachsen-anhalt": {"sachsen-anhalt", "sachsen anhalt", "st"}, "schleswig-holstein": {"schleswig-holstein", "schleswig holstein", "sh"}, "thuringen": {"thuringen", "thueringen", "th"}, } # ── MOSMIX Parameter ───────────────────────────────────────────────────────── MOSMIX_PARAMS = [ "hourly/large/temperature_air_mean_2m", "hourly/large/wind_speed", "hourly/large/wind_direction", "hourly/large/wind_gust_max_last_1h", "hourly/large/cloud_cover_total", "hourly/large/pressure_air_site_reduced", "hourly/large/precipitation_height_last_1h", "hourly/large/sunshine_duration", "hourly/large/probability_precipitation_height_gt_0_1mm_last_1h", "hourly/large/uv_index", "hourly/large/visibility", "hourly/large/weather_significant", ] def _get_berlin(): try: import zoneinfo return zoneinfo.ZoneInfo("Europe/Berlin") except ImportError: import pytz return pytz.timezone("Europe/Berlin") _TF = None # TimezoneFinder singleton def _get_location_tz(lat, lon): """Return (tzinfo, tz_name) for the given coordinates using timezonefinder.""" global _TF try: if _TF is None: from timezonefinder import TimezoneFinder _TF = TimezoneFinder() tz_name = _TF.timezone_at(lat=lat, lng=lon) if tz_name: try: import zoneinfo return zoneinfo.ZoneInfo(tz_name), tz_name except ImportError: import pytz return pytz.timezone(tz_name), tz_name except Exception: pass return _get_berlin(), "Europe/Berlin" def _normalize_text(value): if not value: return "" text = str(value).strip().lower() replacements = { "ä": "a", "ö": "o", "ü": "u", "ß": "ss", } for src, dst in replacements.items(): text = text.replace(src, dst) return " ".join(text.split()) def _state_tokens(state_name): normalized = _normalize_text(state_name) if not normalized: return set() for _, aliases in STATE_ALIASES.items(): if normalized in aliases: return aliases return {normalized} def _warning_matches_state(state_hint, warning_state, warning_state_short): hint_tokens = _state_tokens(state_hint) warning_tokens = _state_tokens(warning_state) warning_short = _normalize_text(warning_state_short) if warning_short: warning_tokens.add(warning_short) return bool(hint_tokens and warning_tokens and hint_tokens.intersection(warning_tokens)) def _extract_state_from_location(loc): try: return (loc.raw or {}).get("address", {}).get("state") except Exception: return None def _extract_location_names(loc): """Return a list of normalised name strings (city, county, …) from a Nominatim result for matching against DWD warning regionName fields.""" names = [] try: raw = (loc.raw or {}).get("address", {}) for key in ("city", "town", "village", "hamlet", "municipality", "city_district", "county", "district", "suburb", "borough"): val = raw.get(key) if val: n = _normalize_text(val) if n and len(n) >= 3 and n not in names: names.append(n) except Exception: pass return names def _client_ip(req): forwarded = req.headers.get("X-Forwarded-For", "").strip() if forwarded: return forwarded.split(",")[0].strip() return req.remote_addr or "unknown" def _is_suggest_rate_limited(client_ip): now = time() with _suggest_rate_lock: hits = _suggest_rate_hits.get(client_ip) if hits is None: hits = deque() _suggest_rate_hits[client_ip] = hits cutoff = now - _SUGGEST_RATE_WINDOW_SECONDS while hits and hits[0] < cutoff: hits.popleft() if len(hits) >= _SUGGEST_RATE_MAX_REQUESTS: return True hits.append(now) if len(hits) == 1: stale_ips = [ip for ip, values in _suggest_rate_hits.items() if not values or values[-1] < cutoff] for stale_ip in stale_ips: _suggest_rate_hits.pop(stale_ip, None) return False def geocode_location(query): try: loc = _GEOLOCATOR.geocode(query, language="de", addressdetails=True, timeout=10) if loc: return (loc.latitude, loc.longitude, loc.address, _extract_state_from_location(loc), _extract_location_names(loc)) except GeocoderTimedOut: pass except Exception: app.logger.exception("Geocoding failed for query '%s'", query) return None, None, None, None, [] def haversine(lat1, lon1, lat2, lon2): R = 6371 dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = (math.sin(dlat/2)**2 + math.cos(math.radians(lat1))*math.cos(math.radians(lat2))*math.sin(dlon/2)**2) return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1-a)) def _isnan(v): try: return math.isnan(float(v)) except (TypeError, ValueError): return True def _round_temp(k): if k is None or _isnan(k): return None return round(float(k), 1) def _clamp(value, min_value, max_value): return max(min_value, min(max_value, value)) def _estimate_uv(dt_local, lat, cloud_pct): """Rough UV index estimate for Central Europe when the MOSMIX station doesn't provide the UVI parameter. Uses time-of-day, season, latitude, and cloud cover.""" hour = dt_local.hour + dt_local.minute / 60.0 if hour < 5.5 or hour > 20.5: return 0.0 noon = 13.0 # approximate solar noon in Germany (CET) hour_factor = max(0.0, math.cos(math.pi * (hour - noon) / 14.0)) # Typical clear-sky peak UV at solar noon for ~51°N, Jan–Dec monthly_peak = [1.0, 1.8, 3.5, 5.0, 6.5, 7.5, 7.2, 6.2, 4.5, 2.5, 1.2, 0.8] seasonal = monthly_peak[dt_local.month - 1] lat_factor = max(0.5, 1.0 - (float(lat) - 51.0) * 0.015) cloud_factor = 1.0 - (cloud_pct or 0) / 100.0 * 0.75 uv = seasonal * hour_factor * lat_factor * cloud_factor return round(max(0.0, uv), 1) def uv_risk_info(uv_index): if uv_index is None: return "–", "na" uv = float(uv_index) if uv < 3: return "niedrig", "low" if uv < 6: return "moderat", "moderate" if uv < 8: return "hoch", "high" if uv < 11: return "sehr hoch", "very-high" return "extrem", "extreme" def hour_confidence_score(temp_c, precip_mm, rain_prob, wind_kmh, gust_kmh, cloud_pct): score = 100 if temp_c is None: score -= 18 if precip_mm is not None: score -= _clamp(precip_mm * 20, 0, 40) if rain_prob is not None: score -= _clamp(rain_prob * 0.35, 0, 30) if wind_kmh is not None: score -= _clamp((wind_kmh - 25) * 0.6, 0, 18) if gust_kmh is not None: score -= _clamp((gust_kmh - 45) * 0.45, 0, 16) if cloud_pct is not None: score -= _clamp((cloud_pct - 85) * 0.4, 0, 8) score = int(round(_clamp(score, 5, 99))) if score >= 80: return score, "hoch" if score >= 60: return score, "mittel" return score, "niedrig" def activity_score(temp_c, precip_mm, rain_prob, wind_kmh, gust_kmh, uv_index): score = 100.0 if temp_c is not None: score -= abs(temp_c - 20) * 3.5 if precip_mm is not None: score -= _clamp(precip_mm * 35, 0, 45) if rain_prob is not None: score -= _clamp(rain_prob * 0.45, 0, 35) if wind_kmh is not None: score -= _clamp((wind_kmh - 18) * 0.7, 0, 16) if gust_kmh is not None: score -= _clamp((gust_kmh - 35) * 0.55, 0, 12) if uv_index is not None and uv_index > 6: score -= _clamp((uv_index - 6) * 6, 0, 16) return int(round(_clamp(score, 0, 100))) def best_activity_window(forecast, horizon_hours=24, window_size=2): hours = forecast[:horizon_hours] if len(hours) < window_size: return None best = None for i in range(0, len(hours) - window_size + 1): segment = hours[i:i + window_size] scores = [h.get("activity_score") for h in segment if h.get("activity_score") is not None] if not scores: continue avg_score = round(sum(scores) / len(scores)) if best is None or avg_score > best["score"]: best = { "start": segment[0]["datetime"], "end": segment[-1]["datetime"], "score": int(avg_score), } return best def pressure_trend_info(forecast, step_hours=6): if len(forecast) <= step_hours: return None, None p0 = forecast[0].get("pressure_hpa") p1 = forecast[step_hours].get("pressure_hpa") if p0 is None or p1 is None: return None, None delta = round(p1 - p0, 1) if delta >= 1.5: return delta, "steigend" if delta <= -1.5: return delta, "fallend" return delta, "stabil" def temp_trend_info(forecast, step_hours=6): if len(forecast) <= step_hours: return None, None t0 = forecast[0].get("temp_c") t1 = forecast[step_hours].get("temp_c") if t0 is None or t1 is None: return None, None delta = round(t1 - t0, 1) if delta >= 1.0: return delta, "wärmer" if delta <= -1.0: return delta, "kälter" return delta, "konstant" def _parse_warning_datetime(value): if value in (None, ""): return None try: if isinstance(value, (int, float)): ts = pd.Timestamp(value, unit="ms", tz="UTC") else: v = str(value).strip() if v.isdigit(): ts = pd.Timestamp(int(v), unit="ms", tz="UTC") else: ts = pd.Timestamp(v) if ts.tzinfo is None: ts = ts.tz_localize("UTC") return ts.tz_convert(_get_berlin()).tz_localize(None).to_pydatetime() except Exception: return None # ── Icon key → static/icons/{key}.png ────────────────────────────────────── # sonne clear day # wolkig(2) sun + one cloud (partly cloudy) # wolke single cloud (mostly cloudy) # wolkig two clouds (overcast) # wolkig(1) sun+cloud+rain (showers) # regen heavy rain # schnee snow # blitz dry thunderstorm (WW 17, 91-92) # sturm thunderstorm with rain (WW 95, 97-98) # hagel hail (WW 27, 89-90, 93-94) # schneebedeckt hail + snow (hagel WW + temp ≤ 2°C) # sturmundhagel thunderstorm + rain + hail (WW 96, 99) # schneebedecktsonne snow + sunshine (temp ≤ 2°C, cloud < 60 %) # nebel dense fog (visibility < 1000 m, day) # nebel_nacht dense fog (visibility < 1000 m, night) # nebel_wolkig foggy + overcast (1000-5000 m, cloud > 60 %) # wolkig_nebel_sonne patchy fog / haze (1000-5000 m, less cloud) # nacht clear night (moon) # nacht(1) moon behind cloud # nacht(2) moon+cloud+rain # nacht(3) moon+cloud+snow _WW_HAIL = {27, 89, 90, 93, 94} # hail showers (no thunder) _WW_GRAUPEL = {87, 88} # graupel → schneebedeckt _WW_THUNDER_HAIL = {96, 99} # thunderstorm + hail → sturmundhagel _WW_THUNDER_RAIN = {91, 92, 95, 97, 98} # thunderstorm + precipitation _WW_THUNDER_DRY = {17} # dry thunderstorm def weather_icon(cloud_pct, precip_mm, rain_prob, temp_c, is_night=False, visibility_m=None, weather_code=None): """Return the icon key for static/icons/{key}.png.""" # Fog (takes priority; fog unlikely when precipitating heavily) if visibility_m is not None and visibility_m < 5000 and not (precip_mm and precip_mm > 0.5): if visibility_m < 1000: return "nebel_nacht" if is_night else "nebel" if cloud_pct is not None and cloud_pct > 60: return "nebel_wolkig" return "wolkig_nebel_sonne" # ── WW-code based icons (thunderstorm / hail / graupel) ────────── if weather_code is not None: ww = int(weather_code) if ww in _WW_THUNDER_HAIL: return "sturmundhagel" if ww in _WW_HAIL: return "schneebedeckt" if (temp_c is not None and temp_c <= 2) else "hagel" if ww in _WW_GRAUPEL: return "schneebedeckt" if ww in _WW_THUNDER_RAIN: return "sturm" if ww in _WW_THUNDER_DRY: return "blitz" # Snow / sleet (day and night) if temp_c is not None and temp_c <= 2 and ( (precip_mm and precip_mm > 0.1) or (rain_prob and rain_prob >= 40) ): if is_night: return "nacht(3)" return "schneebedecktsonne" if (cloud_pct is not None and cloud_pct < 60) else "schnee" # ── Night icons ─────────────────────────────────────────────────── if is_night: if (precip_mm and precip_mm > 0.1) or (rain_prob is not None and rain_prob >= 50): return "nacht(2)" if rain_prob is not None and rain_prob >= 30 and cloud_pct is not None and cloud_pct > 50: return "nacht(2)" if cloud_pct is not None: if cloud_pct > 80: return "wolkig" # too cloudy to see moon if cloud_pct > 30: return "nacht(1)" return "nacht" # ── Day icons ───────────────────────────────────────────────────── if (precip_mm and precip_mm > 0.1) or (rain_prob is not None and rain_prob >= 50): return "wolkig(1)" if (cloud_pct is None or cloud_pct < 75) else "regen" if rain_prob is not None and rain_prob >= 30 and cloud_pct is not None and cloud_pct > 50: return "wolkig(1)" if cloud_pct is not None: if cloud_pct > 80: return "wolkig" if cloud_pct > 50: return "wolke" if cloud_pct > 20: return "wolkig(2)" return "sonne" def pick_daily_icon(hours): """Choose the most representative icon key for a whole day.""" if not hours: return "sonne" # Thunderstorm / hail: any hour with matching WW code takes priority ww_codes = [h.get("weather_code") for h in hours if h.get("weather_code") is not None] if ww_codes: has_t_hail = any(w in _WW_THUNDER_HAIL for w in ww_codes) has_hail = any(w in _WW_HAIL for w in ww_codes) has_graupel = any(w in _WW_GRAUPEL for w in ww_codes) has_t_rain = any(w in _WW_THUNDER_RAIN for w in ww_codes) has_t_dry = any(w in _WW_THUNDER_DRY for w in ww_codes) avg_temp = sum(h["temp_c"] for h in hours if h.get("temp_c") is not None) n_temp = sum(1 for h in hours if h.get("temp_c") is not None) mean_temp = avg_temp / n_temp if n_temp else 10 if has_t_hail: return "sturmundhagel" if has_hail: return "schneebedeckt" if mean_temp <= 2 else "hagel" if has_graupel: return "schneebedeckt" if has_t_rain: return "sturm" if has_t_dry: return "blitz" # Fog: majority of hours with reduced visibility and no heavy precipitation fog_hours = [ h for h in hours if h.get("visibility_m") is not None and h["visibility_m"] < 5000 and (h.get("precip_mm") or 0) <= 0.5 ] if len(fog_hours) >= len(hours) // 2 and fog_hours: avg_vis = sum(h["visibility_m"] for h in fog_hours) / len(fog_hours) avg_cloud_fog = sum(h.get("cloud_pct") or 0 for h in fog_hours) / len(fog_hours) if avg_vis < 1000: return "nebel" if avg_cloud_fog > 60: return "nebel_wolkig" return "wolkig_nebel_sonne" # Snow / sleet if any( h.get("temp_c") is not None and h["temp_c"] <= 2 and ((h.get("precip_mm") or 0) > 0.1 or (h.get("rain_prob") or 0) >= 40) for h in hours ): return "schnee" clouds = [h.get("cloud_pct") for h in hours if h.get("cloud_pct") is not None] avg_cloud = sum(clouds) / len(clouds) if clouds else 0 # Significant rain/showers if any((h.get("precip_mm") or 0) >= 0.8 or (h.get("rain_prob") or 0) >= 65 for h in hours): return "regen" if avg_cloud >= 75 else "wolkig(1)" # Light showers if any((h.get("precip_mm") or 0) > 0.1 or (h.get("rain_prob") or 0) >= 35 for h in hours): return "wolkig(1)" # Cloud cover if avg_cloud > 80: return "wolkig" if avg_cloud > 50: return "wolke" if avg_cloud > 20: return "wolkig(2)" return "sonne" def feels_like(temp_c, wind_kmh, cloud_pct): """Apparent / perceived temperature. * ≤10 °C + wind >4.8 km/h → Windchill (JAG/TI formula) * ≥27 °C → Heat index (Rothfusz, RH 60 %) * 10–27 °C → Small sun/wind corrections: - Very clear sky (<20 % clouds): up to +2 °C - Notable wind (>20 km/h): up to -2 °C Only reported when result differs by ≥1 °C from actual temp. """ if temp_c is None: return None # ── Wind chill (cold range) ─────────────────────────────────────── if temp_c <= 10 and wind_kmh is not None and wind_kmh > 4.8: v = wind_kmh wc = 13.12 + 0.6215*temp_c - 11.37*(v**0.16) + 0.3965*temp_c*(v**0.16) return round(wc, 1) # ── Heat index (hot range) ──────────────────────────────────────── if temp_c >= 27: rh = 60 hi = (-8.78469475556 + 1.61139411*temp_c + 2.33854883889*rh - 0.14611605*temp_c*rh - 0.012308094*temp_c**2 - 0.016424828*rh**2 + 0.002211732*temp_c**2*rh + 0.00072546*temp_c*rh**2 - 0.000003582*temp_c**2*rh**2) return round(hi, 1) # ── Mild range: conservative corrections only ───────────────────── adjusted = float(temp_c) # Sunshine: very clear sky (<20 % clouds) adds max +2 °C if cloud_pct is not None and cloud_pct < 20: adjusted += (20 - cloud_pct) / 20 * 2.0 # Wind: only meaningful wind (>20 km/h) cools, max -2 °C if wind_kmh is not None and wind_kmh > 20: adjusted -= _clamp((wind_kmh - 20) / 60 * 2.0, 0, 2.0) result = round(adjusted, 1) return result if abs(result - temp_c) >= 1.0 else temp_c def get_sun_times(lat, lon, date=None): try: loc_tz, tz_name = _get_location_tz(lat, lon) loc = LocationInfo(latitude=lat, longitude=lon, timezone=tz_name) d = date or _dt.date.today() s = astral_sun(loc.observer, date=d, tzinfo=loc_tz) return (s["sunrise"].strftime("%H:%M"), s["sunset"].strftime("%H:%M"), s["dawn"].strftime("%H:%M"), s["dusk"].strftime("%H:%M")) except Exception: return None, None, None, None def _sunrise_sunset(lat, lon, d): """Return (naive_sunrise, naive_sunset) in location local time for date d. Results are cached in _sun_cache to avoid recomputing for every forecast hour.""" key = (round(lat, 1), round(lon, 1), d) if key in _sun_cache: return _sun_cache[key] try: loc_tz, tz_name = _get_location_tz(lat, lon) loc = LocationInfo(latitude=lat, longitude=lon, timezone=tz_name) s = astral_sun(loc.observer, date=d, tzinfo=loc_tz) sr = pd.Timestamp(s["sunrise"]).tz_convert(loc_tz).tz_localize(None) ss = pd.Timestamp(s["sunset"]).tz_convert(loc_tz).tz_localize(None) result = (sr, ss) except Exception: result = (None, None) _sun_cache[key] = result return result def get_dwd_warnings(lat, lon, state_hint=None, location_names=None): """Fetch DWD warnings and filter by regionName matching the user's municipality/county. The DWD JSON uses warncell-IDs as keys and each warning carries a ``regionName`` field (e.g. "Krefeld", "Kreis Kleve - Niederrhein", "Sauerland") that identifies the affected area far more precisely than the federal-state field. We only include a warning when its regionName contains at least one of the place names extracted from the geocoded location (city, county, …). Warnings for neighbouring NRW regions (Eifel, Sauerland, …) are therefore never shown for Krefeld. Duplicates (same event type + onset + expires) are removed before returning. """ loc_names = location_names or [] key = (round(lat, 1), round(lon, 1)) if key in _warn_cache: return _warn_cache[key] try: url = "https://www.dwd.de/DWD/warnungen/warnapp/json/warnings.json" resp = _requests.get(url, timeout=8) if resp.status_code != 200: _warn_cache[key] = [] return [] text = resp.text if text.startswith("warnWetter.loadWarnings("): text = text[len("warnWetter.loadWarnings("):-2] import json as _json data = _json.loads(text) matched = [] for region_warns in data.get("warnings", {}).values(): for w in (region_warns or []): level = w.get("level", 0) if level < 1: continue region_name = w.get("regionName", "") or "" norm_region = _normalize_text(region_name) # Require the user's city / county to appear in regionName region_match = False for name in loc_names: if name and len(name) >= 3 and name in norm_region: region_match = True break if not region_match: continue matched.append({ "level": level, "type": w.get("event", ""), "headline": w.get("headline", ""), "description": w.get("description", ""), "onset": w.get("onset", ""), "expires": w.get("expires", ""), }) # Deduplicate: same event type + onset time + expires time seen = set() deduped = [] for w in matched: dk = (w["type"], w["onset"], w["expires"]) if dk not in seen: seen.add(dk) deduped.append(w) deduped.sort(key=lambda x: x["level"], reverse=True) result = deduped[:5] _warn_cache[key] = result return result except Exception: app.logger.exception("Could not load DWD warnings") _warn_cache[key] = [] return [] def wind_direction_name(degrees): if degrees is None or _isnan(degrees): return "–" dirs = ["N","NNO","NO","ONO","O","OSO","SO","SSO","S","SSW","SW","WSW","W","WNW","NW","NNW"] idx = round(float(degrees)/22.5) % 16 return dirs[idx] def get_mosmix_forecast(lat, lon, hours=72, loc_tz=None): cache_key = (round(lat,2), round(lon,2), hours) if cache_key in _forecast_cache: return _forecast_cache[cache_key] try: if loc_tz is None: loc_tz, _ = _get_location_tz(lat, lon) req = DwdMosmixRequest(parameters=MOSMIX_PARAMS) nearest = req.filter_by_rank(latlon=(lat, lon), rank=1) result = nearest.values.all() df = result.df if df is None or (hasattr(df,"__len__") and len(df)==0): return [], {} if hasattr(df,"to_pandas"): df = df.to_pandas() station_info = {} sdf = nearest.df if sdf is not None and len(sdf) > 0: if hasattr(sdf,"to_pandas"): sdf = sdf.to_pandas() station_info = sdf.iloc[0].to_dict() df = df.sort_values("date").copy() min_date = df["date"].min() cutoff = min_date + pd.Timedelta(hours=hours) df = df[df["date"] <= cutoff] forecast = [] for date_val, group in df.groupby("date"): p = {row["parameter"]: row["value"] for _, row in group.iterrows()} temp_c = _round_temp(p.get("temperature_air_mean_2m")) ff = p.get("wind_speed") wind_kmh = round(float(ff)*3.6,1) if not _isnan(ff) else None fx1 = p.get("wind_gust_max_last_1h") gust_kmh = round(float(fx1)*3.6,1) if not _isnan(fx1) else None pppp = p.get("pressure_air_site_reduced") pressure = round(float(pppp),1) if not _isnan(pppp) else None rr1c = p.get("precipitation_height_significant_weather_last_1h") rr1 = p.get("precipitation_height_last_1h") prec_raw = rr1c if not _isnan(rr1c) else (rr1 if not _isnan(rr1) else None) precip = round(float(prec_raw),1) if prec_raw is not None else 0.0 rprob = p.get("probability_precipitation_height_gt_0_1mm_last_1h") rain_prob = round(float(rprob)*100) if not _isnan(rprob) else None n = p.get("cloud_cover_total") if not _isnan(n): nf = float(n) # wetterdienst returns cloud cover as a 0-1 fraction (like probabilities); # guard against stations that already return 0-100 percent. clouds = _clamp(round(nf * 100 if nf <= 1.0 else nf), 0, 100) else: clouds = None sun = p.get("sunshine_duration") sun_min = round(float(sun)/60) if not _isnan(sun) else 0 wd = p.get("wind_direction") wind_dir = float(wd) if not _isnan(wd) else None vis_raw = p.get("visibility") visibility_m = round(float(vis_raw)) if not _isnan(vis_raw) else None ww_raw = p.get("weather_significant") weather_code = int(float(ww_raw)) if not _isnan(ww_raw) else None uv_raw = p.get("uv_index") dt_local = pd.Timestamp(date_val).tz_convert(loc_tz).tz_localize(None) # Determine day/night for icon selection _sr, _ss = _sunrise_sunset(lat, lon, dt_local.date()) is_night = bool( _sr is not None and _ss is not None and (pd.Timestamp(dt_local) < _sr or pd.Timestamp(dt_local) > _ss) ) if not _isnan(uv_raw): uv = round(float(uv_raw), 1) else: uv = _estimate_uv(dt_local, lat, clouds) uv_label, uv_level = uv_risk_info(uv) feels = feels_like(temp_c, wind_kmh, clouds) confidence_score, confidence_label = hour_confidence_score(temp_c, precip, rain_prob, wind_kmh, gust_kmh, clouds) a_score = activity_score(temp_c, precip, rain_prob, wind_kmh, gust_kmh, uv) forecast.append({ "datetime": dt_local, "temp_c": temp_c, "feels_like": feels, "wind_kmh": wind_kmh, "gust_kmh": gust_kmh, "pressure_hpa": pressure, "precip_mm": precip, "rain_prob": rain_prob, "cloud_pct": clouds, "sun_min": sun_min, "wind_dir": wind_dir, "uv_index": uv, "uv_label": uv_label, "uv_level": uv_level, "confidence": confidence_score, "confidence_label": confidence_label, "activity_score": a_score, "visibility_m": visibility_m, "weather_code": weather_code, "icon": weather_icon(clouds, precip, rain_prob, temp_c, is_night=is_night, visibility_m=visibility_m, weather_code=weather_code), }) result_data = (forecast, station_info) _forecast_cache[cache_key] = result_data return result_data except Exception: app.logger.exception("MOSMIX forecast loading failed") return [], {} def filter_unrealistic_warnings(warnings, forecast, now_local=None): """Remove warnings that contradict the MOSMIX forecast. Frost warnings are suppressed when the minimum forecast temperature within the warning window (or next 48 h if the window is unknown) stays above 3 °C. Rain / heavy-rain warnings are suppressed when neither significant precipitation nor a meaningful rain probability is forecast. Both onset/expires timestamps from DWD and forecast datetimes are compared as naive Berlin-local datetimes so no tz-mismatch can cause an empty overlap window. """ if not warnings or not forecast: return warnings frost_keywords = {"frost", "glatte", "glatt", "eis", "schnee"} rain_keywords = {"regen", "starkregen", "dauerregen"} result = [] for w in warnings: warn_type = _normalize_text(w.get("type", "")) headline = _normalize_text(w.get("headline", "")) onset_dt = _parse_warning_datetime(w.get("onset")) expires_dt = _parse_warning_datetime(w.get("expires")) # Discard already-expired warnings if now_local is not None and expires_dt is not None and expires_dt < now_local: app.logger.info("Discarded expired warning: %s (expires %s)", w.get("headline"), expires_dt) continue # Select forecast hours that fall within the warning window. # Convert everything to pandas Timestamps (naive) so comparisons are type-safe. if onset_dt is not None and expires_dt is not None: ts_onset = pd.Timestamp(onset_dt) ts_expires = pd.Timestamp(expires_dt) relevant = [ h for h in forecast[:48] if h.get("datetime") is not None and ts_onset <= pd.Timestamp(h["datetime"]) <= ts_expires ] # If the window is entirely in the future and forecast doesn't reach it, # fall back to the full 48-h slice to be conservative. if not relevant: relevant = forecast[:48] else: relevant = forecast[:48] temps = [h["temp_c"] for h in relevant if h.get("temp_c") is not None] precips = [h["precip_mm"] for h in relevant if h.get("precip_mm") is not None] rainprobs = [h["rain_prob"] for h in relevant if h.get("rain_prob") is not None] min_temp = min(temps) if temps else None max_precip = max(precips) if precips else 0.0 max_rain_prob = max(rainprobs) if rainprobs else 0 is_frost = any(kw in warn_type or kw in headline for kw in frost_keywords) is_rain = any(kw in warn_type or kw in headline for kw in rain_keywords) skip = False if is_frost and min_temp is not None and min_temp > 3: app.logger.info("Suppressed frost warning (min %.1f °C): %s", min_temp, w.get("headline")) skip = True elif is_rain and max_precip < 0.2 and max_rain_prob < 35: app.logger.info("Suppressed rain warning (max_precip %.1f mm): %s", max_precip, w.get("headline")) skip = True if not skip: w["onset_dt"] = onset_dt w["expires_dt"] = expires_dt result.append(w) return result @app.route("/") def index(): return render_template("index.html") @app.route("/wetter", methods=["GET"]) def wetter(): ort = request.args.get("ort","").strip() lat_param = request.args.get("lat") lon_param = request.args.get("lon") # Geolocation via Browser-Koordinaten lat, lon, display_name, state_hint, location_names = None, None, None, None, [] if lat_param and lon_param: try: lat = float(lat_param) lon = float(lon_param) if not (-90 <= lat <= 90 and -180 <= lon <= 180): raise ValueError("Invalid coordinate range") loc = _GEOLOCATOR.reverse((lat, lon), language="de", addressdetails=True, timeout=10) display_name = loc.address if loc else f"{lat:.2f}, {lon:.2f}" state_hint = _extract_state_from_location(loc) if loc else None location_names = _extract_location_names(loc) if loc else [] if not ort or ort == "Mein Standort": ort = display_name.split(",")[0] except Exception: app.logger.info("Invalid or unusable browser coordinates for /wetter") lat, lon, display_name, state_hint, location_names = None, None, None, None, [] if lat is None: if not ort: return render_template("index.html", error="Bitte einen Ort eingeben.") lat, lon, display_name, state_hint, location_names = geocode_location(ort) if lat is None: return render_template("index.html", error=f'Ort "{ort}" konnte nicht gefunden werden.') loc_tz, location_tz = _get_location_tz(lat, lon) forecast, mosmix_station = get_mosmix_forecast(lat, lon, hours=240, loc_tz=loc_tz) if not forecast: return render_template("index.html", error="Keine Wetterdaten verfügbar. Bitte später erneut versuchen.") station_name = mosmix_station.get("name", ort) station_id = mosmix_station.get("station_id", "–") station_lat = float(mosmix_station.get("latitude", lat)) station_lon = float(mosmix_station.get("longitude", lon)) station_dist = round(haversine(lat, lon, station_lat, station_lon), 1) now_local_dt = _dt.datetime.now(loc_tz) now_local = now_local_dt.strftime("%H:%M") now_loc_naive = now_local_dt.replace(minute=0, second=0, microsecond=0, tzinfo=None) current_idx = 0 for i, h in enumerate(forecast): dt = h["datetime"] dt_naive = dt.replace(tzinfo=None) if hasattr(dt,"tzinfo") and dt.tzinfo is not None else dt if dt_naive >= now_loc_naive: current_idx = i break current = forecast[current_idx] forecast = forecast[current_idx:] sunrise, sunset, dawn, dusk = get_sun_times(lat, lon) warnings = get_dwd_warnings(lat, lon, state_hint=state_hint, location_names=location_names) warnings = filter_unrealistic_warnings(warnings, forecast, now_local=now_local_dt.replace(tzinfo=None)) pressure_delta, pressure_trend = pressure_trend_info(forecast) temp_delta_6h, temp_trend_6h = temp_trend_info(forecast) best_window = best_activity_window(forecast, horizon_hours=24, window_size=2) daily = {} for h in forecast: dt = h["datetime"] day = dt.date() if hasattr(dt,"date") else str(dt)[:10] if day not in daily: daily[day] = {"temps":[], "precip":0.0, "cloud":[], "wind":[], "icons":[], "rain_prob":[], "uv":[], "hours":[]} daily[day]["hours"].append(h) if h["temp_c"] is not None: daily[day]["temps"].append(h["temp_c"]) daily[day]["precip"] += h.get("precip_mm") or 0 if h["cloud_pct"] is not None: daily[day]["cloud"].append(h["cloud_pct"]) if h["wind_kmh"] is not None: daily[day]["wind"].append(h["wind_kmh"]) if h.get("rain_prob") is not None: daily[day]["rain_prob"].append(h["rain_prob"]) if h.get("uv_index") is not None: daily[day]["uv"].append(h["uv_index"]) daily[day]["icons"].append(h["icon"]) daily_summary = [] for day, d in daily.items(): daily_summary.append({ "date": day, "temp_min": min(d["temps"]) if d["temps"] else None, "temp_max": max(d["temps"]) if d["temps"] else None, "precip": round(d["precip"],1), "rain_prob": max(d["rain_prob"]) if d["rain_prob"] else None, "cloud": round(sum(d["cloud"])/len(d["cloud"])) if d["cloud"] else None, "wind_max": max(d["wind"]) if d["wind"] else None, "uv_max": max(d["uv"]) if d["uv"] else None, "icon": pick_daily_icon(d["hours"]), }) chart_labels, chart_temps, chart_feels, chart_precip, chart_rain_prob = [], [], [], [], [] for h in forecast[:72]: dt = h["datetime"] label = dt.strftime("%d.%m %H:%M") if hasattr(dt,"strftime") else str(dt)[5:16] chart_labels.append(label) chart_temps.append(h["temp_c"]) chart_feels.append(h.get("feels_like")) chart_precip.append(h.get("precip_mm") or 0) chart_rain_prob.append(h.get("rain_prob") or 0) return render_template( "weather.html", ort=ort, display_name=display_name, lat=lat, lon=lon, station_name=station_name, station_id=station_id, station_dist=station_dist, current=current, now_local=now_local, location_tz=location_tz, pressure_delta=pressure_delta, pressure_trend=pressure_trend, temp_delta_6h=temp_delta_6h, temp_trend_6h=temp_trend_6h, best_window=best_window, sunrise=sunrise, sunset=sunset, warnings=warnings, forecast=forecast[:48], daily=daily_summary, chart_labels=chart_labels, chart_temps=chart_temps, chart_feels=chart_feels, chart_precip=chart_precip, chart_rain_prob=chart_rain_prob, wind_dir_name=wind_direction_name, uv_risk_info=uv_risk_info, ) @app.route("/api/suggest") def suggest(): q = request.args.get("q","").strip() if len(q) < 2: return jsonify([]) client_ip = _client_ip(request) if _is_suggest_rate_limited(client_ip): return jsonify({"error": "Zu viele Anfragen. Bitte kurz warten."}), 429 cache_key = _normalize_text(q) if cache_key in _suggest_cache: return jsonify(_suggest_cache[cache_key]) try: results = _GEOLOCATOR.geocode(q, exactly_one=False, limit=5, language="de", addressdetails=True, timeout=5) payload = [{"name": r.address, "lat": r.latitude, "lon": r.longitude} for r in results] if results else [] _suggest_cache[cache_key] = payload return jsonify(payload) except Exception: app.logger.exception("Suggest lookup failed for query '%s'", q) return jsonify([]) if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=5000)