ha-ninstrip/custom_components/ninstrip/api.py

"""HTTP-Client fuer die REST-API von ninstrip. Frei von Home-Assistant-Importen.

Bewusst nah an ha-ninlight gehalten: gleiche Fehlerklassen, gleicher Aufbau.
Wer dort etwas korrigiert, findet es hier an derselben Stelle wieder.
"""

from __future__ import annotations

import json
from typing import Any

import aiohttp

DEFAULT_PORT = 3333
DEFAULT_TIMEOUT = 10.0

_STATE_REQUIRED = ("on", "brightness", "segment_count", "pixel_count")
_INFO_REQUIRED = ("mac", "product", "fw_version")


class NinstripError(Exception):
    """Oberklasse aller Geraetefehler."""


class NinstripConnectionError(NinstripError):
    """Das Geraet war nicht erreichbar (Zeitueberschreitung, abgewiesen, zurueckgesetzt)."""


class NinstripAuthError(NinstripError):
    """Das Geraet hat das Token abgelehnt (HTTP 401)."""


class NinstripResponseError(NinstripError):
    """Es hat geantwortet, aber nicht wie ein ninstrip (kein JSON, Pflichtfeld fehlt)."""


class NinstripRequestError(NinstripError):
    """Die Anfrage wurde abgewiesen (HTTP >= 400 ausser 401/409)."""

    def __init__(self, status: int, message: str) -> None:
        super().__init__(f"HTTP {status}: {message}")
        self.status = status
        self.message = message


class NinstripBusyError(NinstripRequestError):
    """HTTP 409 - es laeuft schon etwas, etwa eine Aktualisierung."""


class NinstripClient:
    """Spricht mit genau einem Geraet. Zustandslos bis auf Adresse und Token."""

    def __init__(
        self,
        session: aiohttp.ClientSession,
        host: str,
        token: str,
        *,
        port: int = DEFAULT_PORT,
        timeout: float = DEFAULT_TIMEOUT,
    ) -> None:
        self._session = session
        self._host = host
        self._token = token
        self._port = port
        self._timeout = timeout

    @property
    def host(self) -> str:
        return self._host

    # --- Lesen ---------------------------------------------------------

    async def get_state(self) -> dict[str, Any]:
        return _require(await self._request("GET", "device/state"), _STATE_REQUIRED, "state")

    async def get_info(self) -> dict[str, Any]:
        return _require(await self._request("GET", "system/info"), _INFO_REQUIRED, "info")

    async def get_segments(self) -> dict[str, Any]:
        return _require(await self._request("GET", "device/segments"), (), "segments")

    async def get_presets(self) -> dict[str, Any]:
        return _require(await self._request("GET", "device/presets"), ("presets",), "presets")

    async def get_effects(self) -> dict[str, Any]:
        return _require(await self._request("GET", "device/effects"), (), "effects")

    async def get_settings(self) -> dict[str, Any]:
        """Der eine Lesezugriff, der immer ein gueltiges Token verlangt."""
        return _require(await self._request("GET", "system/settings"), (), "settings")

    # --- Schreiben -----------------------------------------------------

    async def set_state(self, **fields: Any) -> dict[str, Any]:
        """POST der Felder; das Geraet antwortet mit seinem vollstaendigen Zustand."""
        return _require(
            await self._request("POST", "device/state", fields), _STATE_REQUIRED, "state"
        )

    async def put_segments(self, segments: list[dict[str, Any]]) -> Any:
        return await self._request("PUT", "device/segments", {"segments": segments})

    async def patch_segment(self, index: int, **fields: Any) -> Any:
        return await self._request("PUT", f"device/segments/{index}", fields)

    async def save_preset(self, index: int, name: str | None = None) -> Any:
        """Sichert die laufende Ansicht auf einem Szenenplatz."""
        return await self._request(
            "PUT", f"device/presets/{index}", {"name": name} if name else {}
        )

    async def apply_preset(self, index: int) -> Any:
        return await self._request("PUT", f"device/presets/{index}/apply")

    async def set_sacn(self, **fields: Any) -> None:
        """sACN-Einstellungen aendern. Weggelassene Felder behalten ihren Wert."""
        await self._request("PUT", "system/settings", {"type": "sacn", **fields})

    async def reboot(self) -> None:
        await self._request("PUT", "system/command", {"command": {"type": "reboot"}})

    async def fw_update(self, url: str) -> None:
        await self._request(
            "PUT", "system/command", {"command": {"type": "fw_update", "url": url}}
        )

    # --- Unterbau ------------------------------------------------------

    async def _request(self, method: str, path: str, payload: dict[str, Any] | None = None) -> Any:
        url = f"http://{self._host}:{self._port}/api/v1/{path}"
        try:
            async with self._session.request(
                method,
                url,
                json=payload,
                headers={"Authorization": self._token},
                timeout=aiohttp.ClientTimeout(total=self._timeout),
            ) as resp:
                status = resp.status
                text = await resp.text()
        except (TimeoutError, aiohttp.ClientError) as err:
            raise NinstripConnectionError(f"{method} {path}: {err!r}") from err

        if status == 401:
            raise NinstripAuthError(f"{method} {path}: Token abgelehnt")
        if status == 409:
            raise NinstripBusyError(status, _error_text(text))
        if status >= 400:
            raise NinstripRequestError(status, _error_text(text))
        if not text:
            return None
        try:
            return json.loads(text)
        except ValueError as err:
            raise NinstripResponseError(f"{method} {path}: kein JSON: {text[:200]!r}") from err


def _error_text(text: str) -> str:
    try:
        body = json.loads(text)
    except ValueError:
        return text[:200]
    if isinstance(body, dict) and isinstance(body.get("error"), str):
        return body["error"]
    return text[:200]


def _require(data: Any, fields: tuple[str, ...], what: str) -> dict[str, Any]:
    if not isinstance(data, dict):
        raise NinstripResponseError(f"{what}: JSON-Objekt erwartet, {type(data).__name__} bekommen")
    missing = [name for name in fields if name not in data]
    if missing:
        raise NinstripResponseError(f"{what}: es fehlt {', '.join(missing)}")
    return data