ai_collections/self_created/Tools/weather_forecast.py

#!/usr/bin/env python3
"""
Weather‑7‑day forecast – German output, no API key required.
The code uses two helper classes:
  * Preparations – data acquisition & helpers.
  * Tools        – the single “print_forecast” function you asked for.
"""

from __future__ import annotations

import asyncio
from datetime import datetime
from typing import Any, Dict, List, Callable

import requests


class EventEmitter:
    def __init__(self, event_emitter: Callable[[dict], Any] = None):
        self.event_emitter = event_emitter

    async def progress_update(self, description):
        await self.emit(description)

    async def error_update(self, description):
        await self.emit(description, "error", True)

    async def success_update(self, description):
        await self.emit(description, "success", True)

    async def emit(self, description="Unknown State", status="in_progress", done=False):
        if self.event_emitter:
            await self.event_emitter(
                {
                    "type": "status",
                    "data": {
                        "status": status,
                        "description": description,
                        "done": done,
                    },
                }
            )


# --------------------------------------------------------------------------- #
# Helper mapping (German weekdays)
_WEEKDAY_GERMAN = {
    0: "Montag",
    1: "Dienstag",
    2: "Mittwoch",
    3: "Donnerstag",
    4: "Freitag",
    5: "Samstag",
    6: "Sonntag",
}


# --------------------------------------------------------------------------- #
class Preparations:
    """
    All functions that *prepare* data (fetching, calculations, formatting).
    They are kept separate from the “output” logic.
    """

    API_URL = "https://api.open-meteo.com/v1/forecast"

    @staticmethod
    def fetch_weather(lat: float, lon: float) -> Dict[str, Any]:
        """Retrieve raw forecast JSON from Open‑Meteo."""
        params = {
            "latitude": lat,
            "longitude": lon,
            "daily": (
                "temperature_2m_max,"
                "temperature_2m_min,"
                "windspeed_10m_max,"
                "precipitation_probability_mean"
            ),
            # hourly field used as a fallback for rain probability
            "hourly": "precipitation_probability",
            "timezone": "auto",
            "forecast_days": 7,
        }
        resp = requests.get(Preparations.API_URL, params=params)
        resp.raise_for_status()
        return resp.json()

    @staticmethod
    def _calc_daily_max_precip(hourly: Dict[str, List[float]]) -> List[int | None]:
        """Return the maximum hourly precipitation probability for each day."""
        if not hourly or "precipitation_probability" not in hourly:
            return []

        hourly_vals = hourly["precipitation_probability"]
        days = len(hourly_vals) // 24
        result: List[int | None] = []

        for d in range(days):
            slice_ = hourly_vals[d * 24 : (d + 1) * 24]
            result.append(int(max(slice_))) if slice_ else result.append(None)

        return result

    @staticmethod
    def _german_date(date_str: str) -> str:
        """Return German date string, e.g. 'Montag, 02.09.2025'."""
        dt = datetime.strptime(date_str, "%Y-%m-%d")
        weekday_name = _WEEKDAY_GERMAN[dt.weekday()]
        return f"{weekday_name}, {dt.strftime('%d.%m.%Y')}"

    @staticmethod
    def format_day(
        date: str,
        temp_max: float | None,
        temp_min: float | None,
        wind_kmh: float | None,
        precip_prob: int | None,
    ) -> str:
        """Return a single formatted line for one day."""
        parts = [Preparations._german_date(date)]

        if temp_max is not None and temp_min is not None:
            parts.append(f"Hoch {temp_max:.1f}°C / Niedrig {temp_min:.1f}°C")
        else:
            parts.append("Temperaturdaten fehlen")

        if wind_kmh is not None:
            parts.append(f"Winds Max. {wind_kmh:.0f} km/h")

        if precip_prob is not None:
            parts.append(f"Regenchance {precip_prob}%")
        else:
            parts.append("Regenwahrscheinlichkeit fehlt")

        return " - ".join(parts)


# --------------------------------------------------------------------------- #
class Tools:
    """
    Contains utility functions for working with weather data.
    """

    @staticmethod
    async def print_forecast(
        lat: float, lon: float, __event_emitter__: Callable[[dict], Any] = None
    ) -> None:
        """
        Fetch weather data and print a formatted forecast.

        param lat: Latitude of the location
        param lon: Longitude of the location
        """
        emitter = EventEmitter(__event_emitter__)
        try:
            await emitter.progress_update("Wetterdaten werden abgerufen...")
            raw = Preparations.fetch_weather(lat, lon)
        except requests.HTTPError as exc:
            await emitter.error_update(f"Fehler beim Abrufen der Wetterdaten: {exc}")
            return

        daily = raw["daily"]
        dates = daily.get("time", [])
        t_max_list = daily.get("temperature_2m_max")
        t_min_list = daily.get("temperature_2m_min")
        wind_list = daily.get("windspeed_10m_max")

        # Rain probability: prefer daily, otherwise fallback to hourly max
        if "precipitation_probability_mean" in daily:
            await emitter.progress_update("Regenwahrscheinlichkeit wird verarbeitet...")
            prob_list = [
                int(v) if v is not None else None
                for v in daily["precipitation_probability_mean"]
            ]
        else:
            prob_list = Preparations._calc_daily_max_precip(raw.get("hourly", {}))

        await emitter.success_update(
            "Wetterdaten erfolgreich abgerufen und verarbeitet."
        )
        # Build & print the report
        header = f"📅  Wettervorhersage für die nächsten {len(dates)} Tage\n"
        lines: List[str] = [header]

        for i, d in enumerate(dates):
            line = Preparations.format_day(
                date=d,
                temp_max=t_max_list[i] if t_max_list else None,
                temp_min=t_min_list[i] if t_min_list else None,
                wind_kmh=wind_list[i] if wind_list else None,
                precip_prob=prob_list[i] if prob_list and len(prob_list) > i else None,
            )
            lines.append(line)

        final_output = "\n".join(lines)

        return final_output


# --------------------------------------------------------------------------- #
if __name__ == "__main__":
    # Default to Berlin coordinates – replace with your own!
    LAT_DEFAULT = 52.556
    LON_DEFAULT = 13.335

    asyncio.run(Tools.print_forecast(LAT_DEFAULT, LON_DEFAULT))