chore(tools): qconnex-Migrationswerkzeug und HA-Integration ins Repo
qconnex-listen/ (OTA-Falle: AP + DHCP + Range-Server, Onboarding-Client) und ha-integration/ (Home-Assistant-custom_component ninlight) waren seit der Erfindung nur auf diesem Rechner. Bilder, Request-Log, hostapd-Log und PIDs ueber .gitignore draussen - die Images sind generierbar, sobald die Signierschluessel erreichbar sind. otaserve.py bildet jetzt auch die Montana-Black-Produkt-ID auf ledshield_rgb ab.
This commit is contained in:
Robin von Bülow
@@ -0,0 +1,214 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user