136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
"""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
|