ha-ninlight/custom_components/ninlight/api.py

"""HTTP client for the ninlight lamp REST API. Deliberately free of Home Assistant imports."""

from __future__ import annotations

import json
from typing import Any

import aiohttp

DEFAULT_PORT = 3333
DEFAULT_TIMEOUT = 10.0

_STATE_REQUIRED = ("on", "brightness", "mode", "capabilities")
_INFO_REQUIRED = ("mac", "device_type", "fw_version")


class NinlightError(Exception):
    """Base class for all lamp errors."""


class NinlightConnectionError(NinlightError):
    """The lamp could not be reached (timeout, refused, reset)."""


class NinlightAuthError(NinlightError):
    """The lamp rejected the token (HTTP 401)."""


class NinlightResponseError(NinlightError):
    """The lamp answered, but not like a ninlight lamp (no JSON, required field missing)."""


class NinlightRequestError(NinlightError):
    """The lamp rejected the request (HTTP >= 400 other than 401/409)."""

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


class NinlightBusyError(NinlightRequestError):
    """HTTP 409: an update is already running."""


class NinlightClient:
    """Talks to one lamp. Stateless apart from its address and 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

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

    async def set_state(self, **fields: Any) -> dict[str, Any]:
        """POST the fields; the lamp answers with its complete new state."""
        return _require(
            await self._request("POST", "device/state", fields), _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_settings(self) -> dict[str, Any]:
        """The one read that always requires a valid token (see spec §2)."""
        return _require(await self._request("GET", "system/settings"), (), "settings")

    async def set_sacn(self, **fields: Any) -> None:
        """Change sACN settings. Fields left out keep their value on the lamp."""
        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}})

    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 NinlightConnectionError(f"{method} {path}: {err!r}") from err

        if status == 401:
            raise NinlightAuthError(f"{method} {path}: token rejected")
        if status == 409:
            raise NinlightBusyError(status, _error_text(text))
        if status >= 400:
            raise NinlightRequestError(status, _error_text(text))
        if not text:
            return None
        try:
            return json.loads(text)
        except ValueError as err:
            raise NinlightResponseError(f"{method} {path}: no 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 NinlightResponseError(f"{what}: expected a JSON object, got {type(data).__name__}")
    missing = [name for name in fields if name not in data]
    if missing:
        raise NinlightResponseError(f"{what}: missing {', '.join(missing)}")
    return data