Robin von Bülow
41088b13ae
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.
108 lines
4.5 KiB
Python
108 lines
4.5 KiB
Python
"""Data coordinator for a single Ninlight bulb."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextlib import suppress
|
|
from typing import Any, TypeAlias
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
|
|
from .client import (
|
|
NinlightAuthError,
|
|
NinlightBadRequest,
|
|
NinlightClient,
|
|
NinlightError,
|
|
NinlightState,
|
|
)
|
|
from .const import CONF_TOKEN, DEFAULT_PORT, DOMAIN, SCAN_INTERVAL
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class NinlightCoordinator(DataUpdateCoordinator[NinlightState]):
|
|
"""Polls one bulb. One coordinator, and one HTTP client, per config entry.
|
|
|
|
`system/info` is fetched on the first successful refresh only. It carries identity
|
|
(MAC, product name, firmware) that does not change while the bulb runs, and the bulb
|
|
accepts at most four queued TCP connections (listen(,4) in nl_http_server.c), so
|
|
skipping it on every poll keeps one of those slots free for the UI.
|
|
"""
|
|
|
|
config_entry: NinlightConfigEntry
|
|
|
|
def __init__(self, hass: HomeAssistant, entry: NinlightConfigEntry) -> None:
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
config_entry=entry,
|
|
name=f"{DOMAIN} {entry.title}",
|
|
update_interval=SCAN_INTERVAL,
|
|
)
|
|
self.client = NinlightClient(
|
|
entry.data["host"],
|
|
entry.data.get(CONF_TOKEN, ""),
|
|
port=entry.data.get("port", DEFAULT_PORT),
|
|
)
|
|
self.info: dict[str, Any] | None = None
|
|
self._identity_ok = False
|
|
self._is_strip = False
|
|
|
|
async def async_shutdown(self) -> None:
|
|
"""Close the HTTP client alongside the coordinator's own teardown."""
|
|
await self.client.close()
|
|
await super().async_shutdown()
|
|
|
|
async def _async_update_data(self) -> NinlightState:
|
|
try:
|
|
state = await self.client.get_state()
|
|
except NinlightAuthError as err:
|
|
# Only possible when the bulb has `auth_all` set; a wrong token will not fix
|
|
# itself, so surface a re-auth flow instead of retrying every interval.
|
|
raise ConfigEntryAuthFailed(str(err)) from err
|
|
except NinlightBadRequest as err:
|
|
# A poll sends no body, so this can only mean a firmware we do not match.
|
|
raise UpdateFailed(str(err)) from err
|
|
except NinlightError as err:
|
|
raise UpdateFailed(str(err)) from err
|
|
|
|
if not self._identity_ok:
|
|
# One extra pair of requests, then never again: the bulb accepts at most four
|
|
# queued TCP connections (listen(,4) in nl_http_server.c), so keeping the poll
|
|
# path to a single GET leaves slots free for the UI.
|
|
with suppress(NinlightError):
|
|
# system/info carries identity that only changes on a reboot or an OTA.
|
|
self.info = await self.client.get_info()
|
|
try:
|
|
# system/settings always demands the token, so it is the only call that
|
|
# proves ours is right. A failure here on a bulb whose state endpoint
|
|
# answered happily means the token is wrong while `auth_all` is off - the
|
|
# light would work and silently never be able to be renamed or updated.
|
|
await self.client.get_settings()
|
|
except NinlightAuthError as err:
|
|
raise ConfigEntryAuthFailed(str(err)) from err
|
|
# `product` is an ninstrip-only key in system/info (bulbs send
|
|
# product_name); once known, the heavier segment poll follows every refresh.
|
|
self._is_strip = "product" in (self.info or {})
|
|
self._identity_ok = True
|
|
|
|
segments: list[dict[str, Any]] | None = None
|
|
if self._is_strip:
|
|
# The light lives in the segment list; the plain state GET above carries
|
|
# only on/brightness for it. A firmware without the endpoint (misdetected
|
|
# strip) must not fail the whole poll for that.
|
|
try:
|
|
segments = (await self.client.get_segments()).get("segments")
|
|
except NinlightError as err:
|
|
_LOGGER.debug("segment poll failed: %s", err)
|
|
|
|
return NinlightState.from_api(state, self.info, segments)
|
|
|
|
|
|
# Declared with TypeAlias (not the 3.12 `type` statement) so the same file lints on the
|
|
# project's 3.11 tooling and runs on HA's 3.14.
|
|
NinlightConfigEntry: TypeAlias = ConfigEntry[NinlightCoordinator]
|