ninlight-w600/ha-integration/custom_components/ninlight/client.py

"""Asynchronous client for the Ninlight REST API.

The firmware (src/net/nl_api.c) speaks plain HTTP/1.1 on port 3333 with a raw token in
the `Authorization` header - no "Bearer" scheme prefix, no TLS, no sessions. The bulb
serves one request per connection and accepts at most four queued connections, so the
client must not keep a hot connection pool open.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

import httpx

from .const import DEFAULT_PORT


class NinlightError(Exception):
    """Base error, raised on connection problems and non-JSON replies."""


class NinlightAuthError(NinlightError):
    """The bulb rejected our token (HTTP 401)."""


class NinlightBadRequest(NinlightError):
    """The bulb rejected the payload (HTTP 400) with the reason in its body."""

    def __init__(self, message: str, reason: str | None) -> None:
        super().__init__(f"{message}: {reason}" if reason else message)
        self.reason = reason


@dataclass(slots=True)
class NinlightState:
    """A parsed `GET /api/v1/device/state` plus the identity from `system/info`."""

    on: bool
    brightness: int  # 1..100, the firmware's native unit
    mode: str  # "color" | "white"
    r: int
    g: int
    b: int
    color_temp: int  # kelvin
    ct_min: int
    ct_max: int
    has_rgb: bool
    # Settings echoed by the same reply; the coordinator exposes them as entities.
    power_on_behaviour: str
    fade_in: int
    fade_out: int
    transition_default: int
    wake_on_change: bool
    # From system/info, merged in by the coordinator.
    device_type: str | None = None
    product_name: str | None = None
    fw_version: str | None = None
    mac: str | None = None
    name: str | None = None
    rssi: int | None = None
    ip: str | None = None
    # Segment model only (ninstrip): the raw segment list; the light maps onto
    # segment 0 while a scene across the remaining segments stays untouched.
    segments: list[dict[str, Any]] = field(default_factory=list)
    # True for ninstrip firmware, whose light state lives in segments. Decided by
    # presence of `segment_count` in device/state and `product` in system/info -
    # both keys are ninstrip-only (nl_api_device.c / nl_api_system.c). Bulb
    # device_types are board ids (nl_boards[] in ninlight-w600: "evk", ...), so the
    # name is not a signal, but the keys are.
    strip: bool = False

    @classmethod
    def from_api(
        cls,
        state: dict[str, Any],
        info: dict[str, Any] | None = None,
        segments: list[dict[str, Any]] | None = None,
    ) -> NinlightState:
        color = state.get("color") or {}
        caps = state.get("capabilities") or {}
        ct = caps.get("color_temp") or [2700, 6500]
        wifi = (info or {}).get("wifi") or {}
        seg = (segments or [None])[0] or {}
        # `segment_count` and `product` exist only on ninstrip firmware; the bulb's
        # device_type is a board id and carries no signal either way.
        strip = "segment_count" in state or "product" in (info or {})
        # On a strip the light lives in segment 0; mode "off" carries no color at all,
        # so mode/color fall back to white there and is_on decides darkness anyway.
        seg_mode = seg.get("mode")
        return cls(
            on=bool(state.get("on", False)),
            brightness=int(state.get("brightness", 0)),
            mode="color" if seg_mode == "solid" else str(state.get("mode", "white")),
            r=int((seg.get("color") or color).get("r", 0)),
            g=int((seg.get("color") or color).get("g", 0)),
            b=int((seg.get("color") or color).get("b", 0)),
            color_temp=int(seg.get("color_temp", state.get("color_temp", 2700))),
            ct_min=int(ct[0]),
            ct_max=int(ct[1]),
            has_rgb=bool(caps.get("color", True)) if strip else bool(caps.get("color", False)),
            power_on_behaviour=str(state.get("power_on_behaviour", "last")),
            fade_in=int(state.get("fade_in", 0)),
            fade_out=int(state.get("fade_out", 0)),
            transition_default=int(state.get("transition_default", 0)),
            wake_on_change=bool(state.get("wake_on_change", False)),
            device_type=(info or {}).get("device_type"),
            product_name=(info or {}).get("product_name"),
            fw_version=(info or {}).get("fw_version"),
            mac=(info or {}).get("mac"),
            name=(info or {}).get("name"),
            rssi=wifi.get("rssi"),
            ip=wifi.get("sta_ip") or wifi.get("ip"),
            segments=list(segments or []),
            strip=strip,
        )


class NinlightClient:
    """One client per bulb (per config entry)."""

    def __init__(
        self,
        host: str,
        token: str,
        *,
        port: int = DEFAULT_PORT,
        session: httpx.AsyncClient | None = None,
    ) -> None:
        self._base = f"http://{host}:{port}/api/v1"
        # Raw token, not "Bearer <token>": nl_http_parse hands the header value straight to
        # token_ok(), which compares it against the stored token byte for byte.
        self._headers = {"Authorization": token}
        self._owns_session = session is None
        self._session: httpx.AsyncClient = session or httpx.AsyncClient(
            headers=self._headers,
            timeout=httpx.Timeout(8.0, connect=4.0),
            # keepalive_expiry of ~0 makes httpx close after each response instead of
            # holding one of the bulb's four listen-queue slots open between polls.
            limits=httpx.Limits(max_connections=1, max_keepalive_connections=0, keepalive_expiry=0.1),
        )

    async def close(self) -> None:
        if self._owns_session:
            await self._session.aclose()

    async def _request(self, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
        try:
            resp = await self._session.request(method, f"{self._base}{path}", json=payload)
        except httpx.InvalidURL as err:  # programming error, not a transient one
            raise NinlightError(f"invalid bulb URL: {err}") from err
        except httpx.HTTPError as err:
            raise NinlightError(f"request to bulb failed: {err}") from err

        if resp.status_code == 401:
            raise NinlightAuthError("bulb rejected the API token")
        if resp.status_code == 400:
            raise NinlightBadRequest("bulb rejected the request", _error_field(resp))
        if resp.status_code != 200:
            raise NinlightError(f"bulb answered {resp.status_code}: {_error_field(resp)}")
        try:
            data = resp.json()
        except ValueError as err:
            raise NinlightError("bulb sent a non-JSON body") from err
        if not isinstance(data, dict):
            raise NinlightError("bulb sent an unexpected JSON shape")
        return data

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

    async def get_segments(self) -> dict[str, Any]:
        """Segment model of the ninstrip firmware (bulbs have no such endpoint)."""
        return await self._request("GET", "/device/segments")

    async def set_segments(self, segments: list[dict[str, Any]]) -> dict[str, Any]:
        """Replace the whole segment list. ninstrip answers with `{"status": "ok"}`."""
        return await self._request("PUT", "/device/segments", {"segments": segments})

    async def patch_segment(self, index: int, patch: dict[str, Any]) -> dict[str, Any]:
        """Update one segment; omitted fields keep their current value."""
        return await self._request("PUT", f"/device/segments/{index}", patch)

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

    async def get_settings(self) -> dict[str, Any]:
        """`system/settings` is the one endpoint that always requires the token.

        `system/info` and a state GET answer without auth unless `auth_all` is set
        (nl_api.c: the state branch tests `s.auth_all && !authed`, the info branch tests
        nothing), so only settings proves the token is right.
        """
        return await self._request("GET", "/system/settings")

    async def set_state(self, payload: dict[str, Any]) -> dict[str, Any]:
        return await self._request("POST", "/device/state", payload)

    async def ping(self) -> dict[str, Any]:
        return await self._request("PUT", "/system/command", {"command": {"type": "ping"}})

    async def reboot(self) -> dict[str, Any]:
        return await self._request("PUT", "/system/command", {"command": {"type": "reboot"}})


def _error_field(resp: httpx.Response) -> str | None:
    try:
        body = resp.json()
    except ValueError:
        return resp.text[:120] or None
    if isinstance(body, dict):
        err = body.get("error")
        return str(err) if err is not None else None
    return None