Simon/wetter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: __pycache__/app.cpython-310.pyc

Browse Source 
modified:   app.py
	modified:   requirements.txt
	modified:   static/css/style.css
	modified:   templates/base.html
	modified:   templates/index.html
	modified:   templates/weather.html
This commit is contained in:
SimolZimol
2026-04-21 11:34:19 +02:00
parent 55bda8ea5e
commit 164d0eb6a8
7 changed files with 364 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

381
app.py
View File

@@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
from flask import Flask, render_template, request, jsonify
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut
@@ -8,10 +8,26 @@ import pandas as pd
import datetime as _dt
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__)
# ── Parameter (wetterdienst ≥ 0.100, Format: resolution/dataset/parameter) ───
# ── 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)
# ── MOSMIX Parameter ─────────────────────────────────────────────────────────
MOSMIX_PARAMS = [
"hourly/large/temperature_air_mean_2m",
"hourly/large/wind_speed",
@@ -22,12 +38,19 @@ MOSMIX_PARAMS = [
"hourly/large/precipitation_height_last_1h",
"hourly/large/sunshine_duration",
"hourly/large/probability_precipitation_height_gt_0_1mm_last_1h",
"hourly/large/uv_index",
]
# ── Hilfsfunktionen ──────────────────────────────────────────────────────────
def _get_berlin():
try:
import zoneinfo
return zoneinfo.ZoneInfo("Europe/Berlin")
except ImportError:
import pytz
return pytz.timezone("Europe/Berlin")
def geocode_location(query: str):
geolocator = Nominatim(user_agent="dwd-wetter-app/1.0")
def geocode_location(query):
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
try:
loc = geolocator.geocode(query, language="de", timeout=10)
if loc:
@@ -36,16 +59,12 @@ def geocode_location(query: str):
pass
return 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))
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:
@@ -53,82 +72,144 @@ def _isnan(v):
except (TypeError, ValueError):
return True
def _round_temp(k):
"""Wetterdienst gives temperature already in degC."""
if k is None or _isnan(k):
return None
return round(float(k), 1)
def feels_like(temp_c, wind_kmh, cloud_pct):
if temp_c is None:
return None
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)
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)
return temp_c
def get_sun_times(lat, lon, date=None):
try:
loc = LocationInfo(latitude=lat, longitude=lon, timezone="Europe/Berlin")
d = date or _dt.date.today()
s = astral_sun(loc.observer, date=d, tzinfo=_get_berlin())
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 get_dwd_warnings(lat, lon):
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
data = json.loads(text)
warnings = []
for region_warns in data.get("warnings", {}).values():
for w in region_warns:
level = w.get("level", 0)
if level >= 1:
warnings.append({
"level": level,
"type": w.get("event", ""),
"headline": w.get("headline", ""),
"description": w.get("description", ""),
})
warnings.sort(key=lambda x: x["level"], reverse=True)
result = warnings[:3]
_warn_cache[key] = result
return result
except Exception:
_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 weather_icon(cloud_pct, precip_mm, rain_prob, temp_c):
if temp_c is not None and temp_c <= 0 and (precip_mm and precip_mm > 0 or rain_prob and rain_prob >= 40):
return "❄️"
if precip_mm and precip_mm > 0.2:
return "🌧️"
if rain_prob is not None and rain_prob >= 60:
return "🌦️"
if rain_prob is not None and rain_prob >= 30 and cloud_pct is not None and cloud_pct > 50:
return "🌦️"
if cloud_pct is not None:
if cloud_pct > 80: return "☁️"
if cloud_pct > 35: return "⛅"
return "☀️"
def get_mosmix_forecast(lat, lon, hours=72):
"""Holt MOSMIX-Vorhersage für die nächsten Stunden."""
cache_key = (round(lat,2), round(lon,2), hours)
if cache_key in _forecast_cache:
return _forecast_cache[cache_key]
try:
# Zeitzone Berlin für alle Anzeige-Timestamps
try:
import zoneinfo
_berlin = zoneinfo.ZoneInfo("Europe/Berlin")
except ImportError:
import pytz
_berlin = pytz.timezone("Europe/Berlin")
berlin = _get_berlin()
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):
if df is None or (hasattr(df,"__len__") and len(df)==0):
return [], {}
# Polars → Pandas
if hasattr(df, "to_pandas"):
df = df.to_pandas()
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()
if hasattr(sdf,"to_pandas"): sdf = sdf.to_pandas()
station_info = sdf.iloc[0].to_dict()
df = df.sort_values("date").copy()
# Auf die nächsten `hours` Stunden begrenzen
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"):
params = {row["parameter"]: row["value"] for _, row in group.iterrows()}
temp_c = _round_temp(params.get("temperature_air_mean_2m"))
ff = params.get("wind_speed")
wind_kmh = round(float(ff) * 3.6, 1) if not _isnan(ff) else None
fx1 = params.get("wind_gust_max_last_1h")
gust_kmh = round(float(fx1) * 3.6, 1) if not _isnan(fx1) else None
pppp = params.get("pressure_air_site_reduced")
pressure = round(float(pppp), 1) if not _isnan(pppp) else None
# Niederschlag: rr1 (gesamt)
rr1c = params.get("precipitation_height_significant_weather_last_1h")
rr1 = params.get("precipitation_height_last_1h")
precip_raw = rr1c if not _isnan(rr1c) else (rr1 if not _isnan(rr1) else None)
precip = round(float(precip_raw), 1) if precip_raw is not None else 0.0
# Regenwahrscheinlichkeit (Wert kommt als Bruchteil 0.0–1.0)
rprob_raw = params.get("probability_precipitation_height_gt_0_1mm_last_1h")
rain_prob = round(float(rprob_raw) * 100) if not _isnan(rprob_raw) else None
n = params.get("cloud_cover_total")
clouds = round(float(n)) if not _isnan(n) else None
sun = params.get("sunshine_duration")
sun_min = round(float(sun) / 60) if not _isnan(sun) else 0
wind_dir_v = params.get("wind_direction")
wind_dir = float(wind_dir_v) if not _isnan(wind_dir_v) else None
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")
clouds = round(float(n)) if not _isnan(n) else 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
uv_raw = p.get("uv_index")
uv = round(float(uv_raw),1) if not _isnan(uv_raw) else None
feels = feels_like(temp_c, wind_kmh, clouds)
dt_local = pd.Timestamp(date_val).tz_convert(berlin).tz_localize(None)
forecast.append({
"datetime": pd.Timestamp(date_val).tz_convert(_berlin).tz_localize(None),
"datetime": dt_local,
"temp_c": temp_c,
"feels_like": feels,
"wind_kmh": wind_kmh,
"gust_kmh": gust_kmh,
"pressure_hpa": pressure,
@@ -137,182 +218,126 @@ def get_mosmix_forecast(lat, lon, hours=72):
"cloud_pct": clouds,
"sun_min": sun_min,
"wind_dir": wind_dir,
"uv_index": uv,
"icon": weather_icon(clouds, precip, rain_prob, temp_c),
})
return forecast, station_info
result_data = (forecast, station_info)
_forecast_cache[cache_key] = result_data
return result_data
except Exception:
traceback.print_exc()
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 weather_icon(cloud_pct, precip_mm, rain_prob, temp_c):
# Schnee
if temp_c is not None and temp_c <= 0 and (precip_mm and precip_mm > 0 or rain_prob and rain_prob >= 40):
return "❄️"
# Regen (tatsächlicher Niederschlag)
if precip_mm and precip_mm > 0.2:
return "🌧️"
# Hohe Regenwahrscheinlichkeit
if rain_prob is not None and rain_prob >= 60:
return "🌦️"
if rain_prob is not None and rain_prob >= 30:
if cloud_pct is not None and cloud_pct > 50:
return "🌦️"
if cloud_pct is not None:
if cloud_pct > 80:
return "☁️"
if cloud_pct > 35:
return "⛅"
return "☀️"
# ── Routen ───────────────────────────────────────────────────────────────────
@app.route("/")
def index():
return render_template("index.html")
@app.route("/wetter", methods=["GET"])
def wetter():
ort = request.args.get("ort", "").strip()
if not ort:
return render_template("index.html", error="Bitte einen Ort eingeben.")
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 = None, None, None
if lat_param and lon_param:
try:
lat = float(lat_param)
lon = float(lon_param)
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
loc = geolocator.reverse((lat, lon), language="de", timeout=10)
display_name = loc.address if loc else f"{lat:.2f}, {lon:.2f}"
if not ort or ort == "Mein Standort":
ort = display_name.split(",")[0]
except Exception:
lat, lon, display_name = None, None, None
lat, lon, display_name = geocode_location(ort)
if lat is None:
return render_template("index.html",
error=f'Ort "{ort}" konnte nicht gefunden werden.')
if not ort:
return render_template("index.html", error="Bitte einen Ort eingeben.")
lat, lon, display_name = geocode_location(ort)
if lat is None:
return render_template("index.html", error=f'Ort "{ort}" konnte nicht gefunden werden.')
forecast, mosmix_station = get_mosmix_forecast(lat, lon, hours=72)
if not forecast:
return render_template(
"index.html",
error="Keine Wetterdaten von DWD verfügbar. Bitte später erneut versuchen.",
)
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)
# "Aktuell" = erste Stunde >= jetzt (Berliner Lokalzeit)
now_utc = _dt.datetime.now(_dt.timezone.utc).replace(tzinfo=None)
try:
import zoneinfo
berlin = zoneinfo.ZoneInfo("Europe/Berlin")
except ImportError:
import pytz
berlin = pytz.timezone("Europe/Berlin")
now_local_dt = _dt.datetime.now(berlin)
now_local = now_local_dt.strftime("%H:%M")
# Auf volle Stunde abrunden → aktuelle Stunde als Startpunkt
berlin = _get_berlin()
now_local_dt = _dt.datetime.now(berlin)
now_local = now_local_dt.strftime("%H:%M")
now_berlin_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
dt_naive = dt.replace(tzinfo=None) if hasattr(dt,"tzinfo") and dt.tzinfo is not None else dt
if dt_naive >= now_berlin_naive:
current_idx = i
break
current = forecast[current_idx]
# Stundenliste ab jetzt kürzen
current = forecast[current_idx]
forecast = forecast[current_idx:]
# Tageszusammenfassung
sunrise, sunset, dawn, dusk = get_sun_times(lat, lon)
warnings = get_dwd_warnings(lat, lon)
daily = {}
for h in forecast:
dt = h["datetime"]
day = dt.date() if hasattr(dt, "date") else str(dt)[:10]
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": []}
if h["temp_c"] is not None:
daily[day]["temps"].append(h["temp_c"])
daily[day] = {"temps":[], "precip":0.0, "cloud":[], "wind":[], "icons":[], "rain_prob":[], "uv":[]}
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["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),
"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,
"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": max(set(d["icons"]), key=d["icons"].count),
})
# Chart-Daten (erste 48 h)
chart_labels, chart_temps, chart_precip, chart_rain_prob = [], [], [], []
for h in forecast[:48]:
dt = h["datetime"]
label = (dt.strftime("%d.%m %H:%M") if hasattr(dt, "strftime")
else str(dt)[5:16])
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_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,
forecast=forecast[:48],
daily=daily_summary,
chart_labels=chart_labels,
chart_temps=chart_temps,
chart_precip=chart_precip,
chart_rain_prob=chart_rain_prob,
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,
sunrise=sunrise, sunset=sunset,
warnings=warnings,
forecast=forecast[:48], daily=daily_summary,
chart_labels=chart_labels, chart_temps=chart_temps,
chart_precip=chart_precip, chart_rain_prob=chart_rain_prob,
wind_dir_name=wind_direction_name,
)
@app.route("/api/suggest")
def suggest():
q = request.args.get("q", "").strip()
if len(q) < 2:
return jsonify([])
geolocator = Nominatim(user_agent="dwd-wetter-app/1.0")
q = request.args.get("q","").strip()
if len(q) < 2: return jsonify([])
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
try:
results = geolocator.geocode(
q, exactly_one=False, limit=5, language="de",
addressdetails=True, timeout=5,
)
return jsonify(
[{"name": r.address, "lat": r.latitude, "lon": r.longitude}
for r in results]
if results else []
)
results = geolocator.geocode(q, exactly_one=False, limit=5, language="de", addressdetails=True, timeout=5)
return jsonify([{"name":r.address,"lat":r.latitude,"lon":r.longitude} for r in results] if results else [])
except Exception:
return jsonify([])
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)
app.run(debug=True, host="0.0.0.0", port=5000)