#!/usr/bin/env python3 """ WashOS — Open-Meteo Wetter-Poller v1 ======================================== Pollt Open-Meteo (kostenlos, kein API-Key) für Lüneburg. Liefert historische Tageswerte (letzte 90 Tage) + 7-Tage-Forecast für Korrelation mit Umsatzdaten. Open-Meteo Endpunkte: - /v1/archive : historisch bis 2 Tage vor heute - /v1/forecast : aktueller Tag + 7 Tage Vorhersage Starten: Wird automatisch von washos-server.py gestartet. """ import json, os, time, threading, requests from datetime import datetime, timezone, timedelta, date # ── KONFIGURATION ───────────────────────────────────────────────── # Lüneburg Koordinaten (Standort Bloomest Böcklerstr.) LAT = 53.2509 LON = 10.4115 LOCATION_NAME = "Lüneburg" TZ = "Europe/Berlin" HISTORY_DAYS = 180 # Wie weit in die Vergangenheit FORECAST_DAYS = 7 # Vorhersage-Tiefe POLL_INTERVAL = 6 * 3600 # Alle 6 Stunden (Wetter ändert sich nicht sekündlich) WEATHER_CODES = { # WMO-Codes, gekürzt auf umsatzrelevante Kategorien 0: ("Klar", "sonne"), 1: ("Überwiegend klar", "sonne"), 2: ("Teils bewölkt", "wolken"), 3: ("Bewölkt", "wolken"), 45: ("Nebel", "wolken"), 48: ("Reifnebel", "wolken"), 51: ("Leichter Nieselregen", "regen"), 53: ("Nieselregen", "regen"), 55: ("Starker Nieselregen", "regen"), 56: ("Gefrierender Nieselregen", "regen"), 57: ("Starker gefr. Niesel", "regen"), 61: ("Leichter Regen", "regen"), 63: ("Regen", "regen"), 65: ("Starker Regen", "regen"), 66: ("Gefrierender Regen", "regen"), 67: ("Starker gefr. Regen", "regen"), 71: ("Leichter Schneefall", "schnee"), 73: ("Schneefall", "schnee"), 75: ("Starker Schneefall", "schnee"), 77: ("Schneegriesel", "schnee"), 80: ("Regenschauer", "regen"), 81: ("Starke Regenschauer", "regen"), 82: ("Heftige Regenschauer", "regen"), 85: ("Leichte Schneeschauer", "schnee"), 86: ("Starke Schneeschauer", "schnee"), 95: ("Gewitter", "regen"), 96: ("Gewitter mit Hagel", "regen"), 99: ("Schweres Gewitter", "regen"), } def classify_weather(code, precip_mm, tmax_c): """Kategorie ableiten (für Umsatz-Korrelation verwendbar).""" if code is None: return "unbekannt" label, cat = WEATHER_CODES.get(int(code), ("?", "unbekannt")) # Verfeinerung: auch ohne Niederschlag zählt über Schwellwerte if cat == "sonne" and tmax_c is not None and tmax_c >= 25: return "heiss" if precip_mm is not None and precip_mm >= 5: return "regen" # mehr als 5mm = klar regnerischer Tag return cat class WeatherPoller: def __init__(self, output_dir="."): self.output_dir = output_dir self.output_file = os.path.join(output_dir, "weather.json") self.session = requests.Session() self.session.headers.update({"User-Agent": "WashOS/1.0"}) def _fetch_history(self): """Historische Daten bis vorgestern.""" end = date.today() - timedelta(days=2) start = end - timedelta(days=HISTORY_DAYS - 1) params = { "latitude": LAT, "longitude": LON, "start_date": start.isoformat(), "end_date": end.isoformat(), "daily": ",".join([ "weather_code", "temperature_2m_max", "temperature_2m_min", "temperature_2m_mean", "precipitation_sum", "rain_sum", "snowfall_sum", "precipitation_hours", "sunshine_duration", "wind_speed_10m_max", ]), "timezone": TZ, } try: r = self.session.get( "https://archive-api.open-meteo.com/v1/archive", params=params, timeout=30, ) if r.status_code == 200: return r.json() except Exception as e: print(f" [Weather] History-Fehler: {e}") return None def _fetch_forecast(self): """Heute + 7 Tage Vorhersage.""" params = { "latitude": LAT, "longitude": LON, "daily": ",".join([ "weather_code", "temperature_2m_max", "temperature_2m_min", "temperature_2m_mean", "precipitation_sum", "rain_sum", "snowfall_sum", "precipitation_hours", "sunshine_duration", "wind_speed_10m_max", ]), "timezone": TZ, "past_days": 2, # gestern + vorgestern überschneiden "forecast_days": FORECAST_DAYS, } try: r = self.session.get( "https://api.open-meteo.com/v1/forecast", params=params, timeout=30, ) if r.status_code == 200: return r.json() except Exception as e: print(f" [Weather] Forecast-Fehler: {e}") return None def _parse(self, raw, is_forecast=False): """Tageswerte parsen.""" if not raw or "daily" not in raw: return [] d = raw["daily"] out = [] today = date.today() for i, ds in enumerate(d.get("time", [])): try: dt = date.fromisoformat(ds) except Exception: continue tmax = d.get("temperature_2m_max", [None])[i] tmin = d.get("temperature_2m_min", [None])[i] tmean = d.get("temperature_2m_mean", [None])[i] precip = d.get("precipitation_sum", [None])[i] rain = d.get("rain_sum", [None])[i] snow = d.get("snowfall_sum", [None])[i] prec_h = d.get("precipitation_hours", [None])[i] sun_s = d.get("sunshine_duration", [None])[i] wind = d.get("wind_speed_10m_max", [None])[i] code = d.get("weather_code", [None])[i] label, cat_raw = WEATHER_CODES.get( int(code) if code is not None else -1, ("?", "unbekannt"), ) cat = classify_weather(code, precip, tmax) out.append({ "date": ds, "is_future": dt > today, "is_today": dt == today, "code": code, "label": label, "category": cat, "tmax": round(tmax, 1) if tmax is not None else None, "tmin": round(tmin, 1) if tmin is not None else None, "tmean": round(tmean, 1) if tmean is not None else None, "precip_mm": round(precip, 1) if precip is not None else None, "rain_mm": round(rain, 1) if rain is not None else None, "snow_cm": round(snow, 1) if snow is not None else None, "precip_h": round(prec_h, 1) if prec_h is not None else None, "sunshine_h": round(sun_s / 3600, 1) if sun_s else None, "wind_kmh": round(wind, 1) if wind is not None else None, }) return out def poll_once(self) -> dict: """Historie + Forecast holen und mergen.""" hist = self._fetch_history() fc = self._fetch_forecast() hist_days = self._parse(hist, is_forecast=False) fc_days = self._parse(fc, is_forecast=True) # Merge: Historie hat Priorität für alte Tage, # Forecast überschreibt gestern/heute/zukunft by_date = {d["date"]: d for d in hist_days} for d in fc_days: by_date[d["date"]] = d all_days = sorted(by_date.values(), key=lambda x: x["date"]) return { "meta": { "last_update": datetime.now(timezone.utc).isoformat(), "source": "open-meteo", "location": {"name": LOCATION_NAME, "lat": LAT, "lon": LON}, "history_days": len(hist_days), "forecast_days": len(fc_days), "total_days": len(all_days), }, "days": all_days, } def write_output(self, data: dict): """Schreibe weather.json (atomar).""" tmp = self.output_file + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) try: os.replace(tmp, self.output_file) except OSError: import shutil shutil.move(tmp, self.output_file) ts = datetime.now().strftime("%H:%M:%S") days = data.get("days", []) today_entry = next( (d for d in days if d.get("is_today")), None ) if today_entry: print( f"[{ts}] WEATHER | {len(days)} Tage | " f"Heute: {today_entry['label']} " f"{today_entry['tmax']}°C / " f"{today_entry['precip_mm']}mm" ) else: print(f"[{ts}] WEATHER | {len(days)} Tage geladen") def run_poller(self): """Endlos-Polling alle 6 Stunden.""" print(" [Weather] Starte Poller...") while True: try: data = self.poll_once() if data.get("days"): self.write_output(data) else: print(" [Weather] Keine Daten erhalten") except Exception as e: print(f" [Weather] Fehler: {e}") time.sleep(POLL_INTERVAL) def start_weather_thread(output_dir="."): """Startet den Weather-Poller als Daemon-Thread.""" poller = WeatherPoller(output_dir=output_dir) t = threading.Thread( target=poller.run_poller, daemon=True, name="WeatherPoller", ) t.start() return poller if __name__ == "__main__": print("Open-Meteo Weather Poller — Standalone-Test") poller = WeatherPoller() data = poller.poll_once() poller.write_output(data) print(json.dumps(data, indent=2, ensure_ascii=False)[:2000])