Home_assistant/backend/app/homelab/openmeteo.py

import time
from typing import Any

import httpx

from app.config import get_settings

WEATHER_CODES: dict[int, str] = {
    0: "ясно",
    1: "преимущественно ясно",
    2: "переменная облачность",
    3: "пасмурно",
    45: "туман",
    48: "изморозь",
    51: "морось",
    53: "морось",
    55: "морось",
    61: "дождь",
    63: "дождь",
    65: "сильный дождь",
    71: "снег",
    73: "снег",
    75: "сильный снег",
    80: "ливень",
    81: "ливень",
    82: "сильный ливень",
    95: "гроза",
    96: "гроза с градом",
    99: "гроза с градом",
}

_cache: dict[str, Any] = {"data": None, "expires_at": 0.0}


class OpenMeteoClient:
    def __init__(self) -> None:
        settings = get_settings()
        self.base_url = settings.openmeteo_base_url.rstrip("/")
        self.lat = settings.weather_lat
        self.lon = settings.weather_lon
        self.location_name = settings.weather_location_name
        self.cache_ttl = settings.weather_cache_sec

    def _fetch_raw(self) -> dict[str, Any]:
        now = time.time()
        if _cache["data"] and now < _cache["expires_at"]:
            return _cache["data"]

        params = {
            "latitude": self.lat,
            "longitude": self.lon,
            "current": (
                "temperature_2m,apparent_temperature,relative_humidity_2m,"
                "precipitation,weather_code,wind_speed_10m"
            ),
            "hourly": "temperature_2m,precipitation_probability,precipitation,weather_code",
            "timezone": "auto",
            "forecast_days": 2,
        }
        with httpx.Client(timeout=15.0) as client:
            response = client.get(f"{self.base_url}/v1/forecast", params=params)
            response.raise_for_status()
            data = response.json()

        _cache["data"] = data
        _cache["expires_at"] = now + self.cache_ttl
        return data

    def fetch_current_and_hourly(self, hours_ahead: int = 12) -> dict[str, Any]:
        try:
            raw = self._fetch_raw()
        except Exception as exc:
            return {"ok": False, "error": str(exc), "location": self.location_name}

        current = raw.get("current") or {}
        hourly = raw.get("hourly") or {}
        times = hourly.get("time") or []
        limit = min(hours_ahead, len(times))
        hourly_slice = []
        for i in range(limit):
            hourly_slice.append({
                "time": times[i],
                "temperature_c": hourly.get("temperature_2m", [None])[i],
                "precipitation_mm": hourly.get("precipitation", [None])[i],
                "precipitation_probability": hourly.get("precipitation_probability", [None])[i],
                "weather_code": hourly.get("weather_code", [None])[i],
            })

        code = current.get("weather_code")
        return {
            "ok": True,
            "location": self.location_name,
            "current": {
                "time": current.get("time"),
                "temperature_c": current.get("temperature_2m"),
                "apparent_temperature_c": current.get("apparent_temperature"),
                "humidity_pct": current.get("relative_humidity_2m"),
                "precipitation_mm": current.get("precipitation"),
                "wind_speed_kmh": current.get("wind_speed_10m"),
                "weather_code": code,
                "conditions": WEATHER_CODES.get(code, "неизвестно") if code is not None else "неизвестно",
            },
            "hourly": hourly_slice,
        }

    def rain_summary(self, hours_ahead: int = 12) -> str:
        data = self.fetch_current_and_hourly(hours_ahead=hours_ahead)
        if not data.get("ok"):
            return f"Погода недоступна: {data.get('error', 'ошибка')}"

        rainy_hours = []
        for hour in data.get("hourly") or []:
            prob = hour.get("precipitation_probability")
            precip = hour.get("precipitation_mm") or 0
            if (prob is not None and prob >= 40) or precip > 0:
                time_str = (hour.get("time") or "")[11:16]
                rainy_hours.append(f"{time_str} ({prob}% вероятность, {precip} мм)")

        if rainy_hours:
            return "Ожидаются осадки: " + ", ".join(rainy_hours[:6])
        return "Существенных осадков в ближайшие часы не ожидается."


def format_weather_snapshot(data: dict[str, Any] | None = None) -> str:
    client = OpenMeteoClient()
    snapshot = data if data is not None else client.fetch_current_and_hourly(hours_ahead=6)

    lines = ["[Погода]"]
    if not snapshot.get("ok"):
        lines.append(f"Данные недоступны ({snapshot.get('error', 'ошибка')}).")
        lines.append("Для точного ответа вызови get_weather.")
        return "\n".join(lines)

    cur = snapshot.get("current") or {}
    lines.append(
        f"{snapshot.get('location')}: {cur.get('temperature_c')}°C "
        f"(ощущается {cur.get('apparent_temperature_c')}°C), "
        f"{cur.get('conditions')}, ветер {cur.get('wind_speed_kmh')} км/ч."
    )
    hourly = snapshot.get("hourly") or []
    rainy_hours = []
    for hour in hourly:
        prob = hour.get("precipitation_probability")
        precip = hour.get("precipitation_mm") or 0
        if (prob is not None and prob >= 40) or precip > 0:
            time_str = (hour.get("time") or "")[11:16]
            rainy_hours.append(f"{time_str} ({prob}% вероятность, {precip} мм)")
    if rainy_hours:
        lines.append("Ожидаются осадки: " + ", ".join(rainy_hours[:6]))
    else:
        lines.append("Существенных осадков в ближайшие часы не ожидается.")
    lines.append("Вопросы «что на улице» / «будет ли дождь» — get_weather.")
    return "\n".join(lines)