feat(api): HTTP-Client fuer die ninlight-Lampen

custom_components/ninlight/api.py

@@ -0,0 +1,135 @@
+"""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 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