Robin von Bülow
031830454a
War eine Lampe stromlos, wurde der Verbindungsfehler als UpdateFailed gemeldet; HA schob den Config-Entry in SETUP_RETRY und die Integration fiel in den Fehlerzustand. Jetzt wird Connection-Refused/Timeout als NinlightUnavailable gefuehrt: der Coordinator liefert den letzten bekannten State mit available=False, der Entry bleibt geladen und die Entity zeigt "unavailable". Protokollfehler (400, kaputtes JSON) und Auth (401 -> Reauth) verhalten sich unveraendert. Light-Faehigkeiten (RGB/CT, Kelvin-Bereich) sind deshalb dynamische Properties — ein Entry, der mit stromloser Lampe angelegt wurde, erkennt sie, sobald die Lampe antwortet.
239 lines
9.9 KiB
Python
239 lines
9.9 KiB
Python
"""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, replace as _dc_replace
|
|
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 NinlightUnavailable(NinlightError):
|
|
"""The bulb could not be reached at all (connection refused or timeout).
|
|
|
|
A bulb that is switched off or has lost Wi-Fi is an expected state, not a fault,
|
|
so the coordinator treats this differently from protocol errors.
|
|
"""
|
|
|
|
|
|
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
|
|
# False when the last poll could not reach the bulb (powered off / offline).
|
|
# Kept separate from `on` so the entity can show "unavailable" instead of an
|
|
# error, while the light keeps its last known appearance.
|
|
available: bool = True
|
|
|
|
@classmethod
|
|
def unavailable_copy(cls, last: "NinlightState") -> "NinlightState":
|
|
"""The last known state, flagged unavailable.
|
|
|
|
Keeps the light's appearance (on/off, color) at its last real value while the
|
|
entity shows as unavailable, instead of falling back to defaults.
|
|
"""
|
|
return _dc_replace(last, available=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.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout) as err:
|
|
# No answer at all: bulb powered off, rebooting, or gone from the WLAN.
|
|
raise NinlightUnavailable(f"bulb unreachable: {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
|