chore: HA-Kopie und die von der Station abgedeckten Werkzeuge wieder heraus

Der Shield-Zweig brachte zwei Verzeichnisse mit, die nichts mit dem Board zu tun haben. Nach Ruecksprache mit Michael bleibt davon nur, was es sonst nirgends gibt. ha-integration/ faellt ganz weg: die Home-Assistant-Integration hat ihr eigenes Repo (ha-ninlight, privat im Gitea) und laeuft dort in Michaels Produktivinstanz. Eine zweite Kopie hier wuerde unweigerlich auseinanderlaufen, und dann waere unklar, welche gilt. Aus qconnex-listen/ fliegen die Teile, die tools/station betriebsfertig kann: hostapd.conf, dnsmasq-qcx.conf, up.sh, down.sh und otaserve.py bauen von Hand den Zugangspunkt "qconnex-firmware" und liefern Abbilder aus - genau der Migrationsmodus der Pi-Station, dort als Dienst, mit Produkt-ID-Tabelle und Signaturpruefung. Ihre eigene hostapd.conf nennt sich im Kopf selbst "temporary experiment, do not enable as a service". Es bleibt, wofuer es bei uns nichts gibt: onboard.py spricht die ALTE Firmware ueber ihren TLS-Socket auf Port 2223 an (unser onboard.sh richtet dagegen eine bereits migrierte ninlight-Lampe ein - anderer Zweck), dazu die Analysewerkzeuge tlssniff.py, raw3333.py und logserver.py sowie bulb-map.txt. Ausserdem in diesem Commit: die erwarteten Zahlen in test_release.sh von 35/32 auf 39/36. Der Merge lief konfliktfrei durch und liess sie trotzdem falsch stehen - der Zweig zweigte vor der Reparatur der Zugangsdatenpruefung ab und konnte ihre neue Meldung nicht kennen, waehrend sie sein neuntes Ziel nicht kennen konnte. Woran die Zahlen haengen, steht jetzt im Kommentar daneben.
2026-09-26 18:01:56 +02:00
parent e10147b70f
commit 7fd279721f
16 changed files with 8 additions and 1006 deletions
@@ -1,54 +0,0 @@
 """The Ninlight integration."""
 from __future__ import annotations
 import logging
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from .const import CONF_TOKEN, DEFAULT_PORT, PLATFORMS
 from .coordinator import NinlightConfigEntry, NinlightCoordinator
 _LOGGER = logging.getLogger(__name__)
 async def async_setup_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool:
    """Set up a Ninlight bulb from a config entry."""
    coordinator = NinlightCoordinator(hass, entry)
    # Register the teardown before the first refresh: HA runs the on_unload callbacks both
    # when setup fails and when the entry unloads, and awaits the coroutine they return.
    # That covers every exit path, so no bulb client can be left with an open connection
    # slot. The bulb serves one request per connection, so this matters.
    entry.async_on_unload(coordinator.async_shutdown)
    try:
        await coordinator.async_config_entry_first_refresh()
    except Exception:
        # ConfigEntryNotReady / ConfigEntryAuthFailed propagate to HA, which retries or
        # raises reauth. Close now rather than waiting for the retry.
        await coordinator.async_shutdown()
        raise
    entry.runtime_data = coordinator
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
    return True
 async def async_unload_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool:
    """Unload a config entry."""
    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
 async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
    """Migrate an entry written by an earlier version of this integration."""
    if config_entry.version > 1:
        return False
    if "port" not in config_entry.data or CONF_TOKEN not in config_entry.data:
        new = dict(config_entry.data)
        new.setdefault("port", DEFAULT_PORT)
        new.setdefault(CONF_TOKEN, new.pop("api_key", ""))
        hass.config_entries.async_update_entry(config_entry, data=new, version=1)
        _LOGGER.debug("Migrated ninlight entry %s to v1", config_entry.entry_id)
    return True
@@ -1,214 +0,0 @@
 """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
@@ -1,166 +0,0 @@
 """Config flow for Ninlight."""
 from __future__ import annotations
 import logging
 from typing import Any
 import voluptuous as vol
 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
 from homeassistant.data_entry_flow import AbortFlow
 from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
 from .client import NinlightAuthError, NinlightClient, NinlightError
 from .const import CONF_TOKEN, DEFAULT_PORT, DEFAULT_TOKEN, DOMAIN
 _LOGGER = logging.getLogger(__name__)
 STEP_USER_DATA_SCHEMA = vol.Schema(
    {
        vol.Required("host"): str,
        vol.Optional("port", default=DEFAULT_PORT): vol.All(vol.Coerce(int), vol.Range(1, 65535)),
        vol.Optional(CONF_TOKEN, default=DEFAULT_TOKEN): vol.All(str, vol.Length(min=1, max=64)),
    }
 )
 def _normalize_mac(mac: str) -> str:
    """Keep one canonical form for the unique id (zeroconf and system/info both use `aa:bb:...`)."""
    return mac.strip().lower()
 def _is_full_mac(mac: str) -> bool:
    """True for a colon-separated 12-hex-digit MAC, the only TXT form usable as unique id.
    ninstrip devices publish the 4-byte suffix (`eb4b98`, nl_wifi.c's mac4) in their TXT
    record; two bulbs sharing a last octet would collide, so those need a probe instead.
    """
    parts = mac.strip().lower().split(":")
    return len(parts) == 6 and all(len(p) == 2 and all(c in "0123456789abcdef" for c in p) for p in parts)
 class NinlightConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Ninlight."""
    VERSION = 1
    async def _probe_mac(self, host: str, port: int) -> str:
        """Return the bulb's MAC. Needs no token: `system/info` is unauthenticated."""
        client = NinlightClient(host, DEFAULT_TOKEN, port=port)
        try:
            info = await client.get_info()
        finally:
            await client.close()
        mac = info.get("mac")
        if not isinstance(mac, str) or not mac:
            raise NinlightError("bulb did not report a MAC address")
        return _normalize_mac(mac)
    async def _probe_mac_and_token(self, host: str, port: int, token: str) -> str:
        """As above, plus verify the token against `system/settings`.
        Only that endpoint authenticates, so a probe that stops at system/info would
        happily accept a wrong token (nl_api.c's info branch never checks `authed`).
        """
        client = NinlightClient(host, token, port=port)
        try:
            info = await client.get_info()
            await client.get_settings()
        finally:
            await client.close()
        mac = info.get("mac")
        if not isinstance(mac, str) or not mac:
            raise NinlightError("bulb did not report a MAC address")
        return _normalize_mac(mac)
    async def _entry_from_data(self, data: dict[str, Any], title: str) -> ConfigFlowResult:
        await self.async_set_unique_id(data["mac"])
        # Addresses come from DHCP, so rediscovering a known MAC at a new IP must move the
        # existing entry rather than abort a re-add.
        self._abort_if_unique_id_configured(updates=data)
        return self.async_create_entry(title=title, data=data)
    async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
        errors: dict[str, str] = {}
        if user_input is not None:
            host = user_input["host"].strip()
            port = int(user_input["port"])
            token = user_input[CONF_TOKEN].strip()
            try:
                mac = await self._probe_mac_and_token(host, port, token)
            except NinlightAuthError:
                errors[CONF_TOKEN] = "invalid_auth"
            except NinlightError:
                errors["base"] = "cannot_connect"
            else:
                return await self._entry_from_data(
                    {"host": host, "port": port, CONF_TOKEN: token, "mac": mac}, title=host
                )
        return self.async_show_form(step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors)
    async def async_step_zeroconf(
        self, discovery_info: ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle discovery of a `_ninlight._tcp` service."""
        props = {str(k).lower(): str(v) for k, v in discovery_info.properties.items()}
        port = discovery_info.port or DEFAULT_PORT
        mac = props.get("mac", "")
        # A TXT record may hold no MAC at all (bulb still announcing) or only the short
        # 4-byte suffix form ninstrip publishes (`eb4b98`). Unique ids are full MACs, so
        # anything that is not one colon-form is replaced by a probe of system/info.
        if not _is_full_mac(mac):
            try:
                mac = await self._probe_mac(discovery_info.host, port)
            except NinlightError:
                raise AbortFlow("cannot_connect") from None
        mac = _normalize_mac(mac)
        await self.async_set_unique_id(mac)
        updates = {
            "host": discovery_info.host,
            "port": port,
            "mac": mac,
            "device_type": props.get("type", ""),
        }
        self._abort_if_unique_id_configured(updates=updates)
        # Instance name is `ninlight-<mac4>` until someone names the bulb; a name set on
        # the device is published in the TXT record and wins.
        title = props.get("name") or discovery_info.name.split(".", 1)[0]
        self.context["title_placeholders"] = {"name": title}
        # Create rather than prompt. A confirm step would leave the flow parked in
        # `in_progress`, where it blocks a manual add of the same bulb with
        # already_in_progress - and a bulb that is reachable but not announcing mDNS
        # (two of ours) has to be added manually, so that collision is not theoretical.
        return self.async_create_entry(title=title, data={**updates, CONF_TOKEN: DEFAULT_TOKEN})
    async def async_step_reauth(self, entry_data: dict[str, Any]) -> ConfigFlowResult:
        return await self.async_step_reauth_confirm()
    async def async_step_reauth_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
        assert entry is not None
        errors: dict[str, str] = {}
        if user_input is not None:
            token = user_input[CONF_TOKEN].strip()
            try:
                await self._probe_mac_and_token(entry.data["host"], entry.data.get("port", DEFAULT_PORT), token)
            except NinlightAuthError:
                errors[CONF_TOKEN] = "invalid_auth"
            except NinlightError:
                errors["base"] = "cannot_connect"
            else:
                self.hass.config_entries.async_update_entry(entry, data={**entry.data, CONF_TOKEN: token})
                await self.hass.config_entries.async_reload(entry.entry_id)
                return self.async_abort(reason="reauth_successful")
        return self.async_show_form(
            step_id="reauth_confirm",
            data_schema=vol.Schema({vol.Required(CONF_TOKEN, default=entry.data.get(CONF_TOKEN, "")): str}),
            errors=errors,
            description_placeholders={"name": entry.title},
        )
@@ -1,25 +0,0 @@
 """Constants for the Ninlight integration."""
 from __future__ import annotations
 from datetime import timedelta
 from typing import Final
 from homeassistant.const import Platform
 DOMAIN = "ninlight"
 PLATFORMS: Final = [Platform.LIGHT]
 CONF_TOKEN = "token"
 DEFAULT_PORT = 3333
 DEFAULT_TOKEN = "aabbccddeeff0011223344"
 # The firmware has no push channel, so state is polled. A bulb answers in well under
 # a second on the LAN; 30 s keeps the UI honest without adding noticeable REST traffic
 # for a nine bulb setup.
 SCAN_INTERVAL = timedelta(seconds=30)
 # Mirror of NL_MAX_TRANSITION_MS in src/light/nl_control.h.
 MAX_TRANSITION_MS = 60000
@@ -1,107 +0,0 @@
 """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]
@@ -1,40 +0,0 @@
 """Base entity for Ninlight devices."""
 from __future__ import annotations
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from .const import DEFAULT_PORT, DOMAIN
 from .coordinator import NinlightCoordinator, NinlightConfigEntry
 class NinlightEntity(CoordinatorEntity[NinlightCoordinator]):
    """Common device info for every entity of one bulb."""
    _attr_has_entity_name = True
    def __init__(self, coordinator: NinlightCoordinator, entry: NinlightConfigEntry) -> None:
        super().__init__(coordinator, context=entry.entry_id)
        mac = (entry.unique_id or "unknown").lower()
        info = coordinator.info or {}
        # Deliberately no `connections={("mac", mac)}`: HA treats a MAC in `connections`
        # as a network-topology link, so a bulb the FRITZ!Box tracker already knows merged
        # into that router-adjacent device and its entity id fell back to the IP. The MAC
        # still identifies the device via unique_id; identifiers are the sole identity here.
        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, mac)},
            # The bulb itself has no user-visible name field that survives rediscovery
            # cleanly, so the entry title (set from mDNS or by the user) names the device.
            name=entry.title,
            manufacturer="Klyqa",
            model=info.get("product_name") or info.get("product"),
            model_id=info.get("device_type"),
            sw_version=info.get("fw_version"),
            configuration_url=f"http://{entry.data['host']}:{entry.data.get('port', DEFAULT_PORT)}/",
        )
 def _entry_mac(entry: NinlightConfigEntry) -> str:
    mac: str = entry.unique_id or "unknown"
    return mac.lower()
@@ -1,182 +0,0 @@
 """Light platform for Ninlight bulbs."""
 from __future__ import annotations
 import logging
 from typing import Any
 from homeassistant.components.light import (
    ATTR_BRIGHTNESS,
    ATTR_COLOR_TEMP_KELVIN,
    ATTR_RGB_COLOR,
    ATTR_TRANSITION,
    ColorMode,
    LightEntity,
    LightEntityFeature,
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from .client import NinlightBadRequest, NinlightError, NinlightState
 from .const import MAX_TRANSITION_MS
 from .coordinator import NinlightConfigEntry, NinlightCoordinator
 from .entity import NinlightEntity
 _LOGGER = logging.getLogger(__name__)
 async def async_setup_entry(
    hass: HomeAssistant,
    entry: NinlightConfigEntry,
    async_add_entities: AddEntitiesCallback,
 ) -> None:
    """Set up the light for a Ninlight bulb config entry."""
    async_add_entities([NinlightLight(entry.runtime_data, entry)])
 def _to_ha_brightness(pct: int) -> int:
    """Bulb brightness is 1..100; HA brightness is 0..255."""
    return max(1, min(255, round(pct * 255 / 100)))
 def _to_bulb_brightness(level: int) -> int:
    """The bulb rejects brightness 0 and anything above 100 (invalid brightness)."""
    return max(1, min(100, round(level * 100 / 255)))
 class NinlightLight(NinlightEntity, LightEntity):
    """A Klyqa bulb running the Ninlight firmware."""
    # No translation_key: with has_entity_name and a single entity per device, HA names the
    # light after the device, giving light.schreibtisch_links rather than ..._bulb.
    _attr_name = None
    def __init__(self, coordinator: NinlightCoordinator, entry: NinlightConfigEntry) -> None:
        super().__init__(coordinator, entry)
        state = coordinator.data
        self._attr_unique_id = f"{(entry.unique_id or '').lower()}-light"
        # Both modes are advertised for RGB bulbs: the firmware keeps the last color and
        # the last color temperature and picks between them with `mode`, which is exactly
        # how HA models a color_temp + rgb light.
        self._attr_supported_color_modes = {ColorMode.COLOR_TEMP, ColorMode.RGB} if state.has_rgb else {
            ColorMode.COLOR_TEMP
        }
        self._attr_supported_features = LightEntityFeature.TRANSITION
        self._attr_min_color_temp_kelvin = state.ct_min
        self._attr_max_color_temp_kelvin = state.ct_max
    @property
    def is_on(self) -> bool:
        return self.coordinator.data.on
    @property
    def brightness(self) -> int | None:
        return _to_ha_brightness(self.coordinator.data.brightness)
    @property
    def color_mode(self) -> ColorMode:
        state = self.coordinator.data
        if not state.on:
            # Nothing is lit; reporting a mode would contradict is_on=False.
            return ColorMode.UNKNOWN
        if state.has_rgb and state.mode == "color":
            return ColorMode.RGB
        return ColorMode.COLOR_TEMP
    @property
    def color_temp_kelvin(self) -> int | None:
        state = self.coordinator.data
        return state.color_temp if self.color_mode == ColorMode.COLOR_TEMP else None
    @property
    def rgb_color(self) -> tuple[int, int, int] | None:
        state = self.coordinator.data
        if self.color_mode == ColorMode.RGB:
            return (state.r, state.g, state.b)
        return None
    async def async_turn_on(self, **kwargs: Any) -> None:
        state = self.coordinator.data
        payload: dict[str, Any] = {"on": True}
        if (level := kwargs.get(ATTR_BRIGHTNESS)) is not None:
            payload["brightness"] = _to_bulb_brightness(int(level))
        elif not state.on:
            # HA expects turn_on without a brightness to restore the previous level. The
            # firmware already does that (it keeps `brightness` while off), so sending it
            # would only re-assert a value the bulb may have clamped itself - but a bulb
            # that has never been on reports brightness 0, which its API rejects. Mirror
            # the last known value only when the bulb actually has one.
            if state.brightness < 1:
                payload["brightness"] = 100
        # The firmware rejects color and color_temp in one request with
        # "color and color_temp are exclusive", so pick one. HA only sends both when a
        # caller passes them together; color is the more specific intent and wins.
        kelvin = kwargs.get(ATTR_COLOR_TEMP_KELVIN)
        rgb = kwargs.get(ATTR_RGB_COLOR)
        seg: dict[str, Any] | None = None
        if rgb is not None and state.has_rgb:
            r, g, b = (int(max(0, min(255, v))) for v in rgb)
            if state.strip:
                seg = {"mode": "solid", "color": {"r": r, "g": g, "b": b}}
            else:
                payload["color"] = {"r": r, "g": g, "b": b}
        elif kelvin is not None:
            ct = max(state.ct_min, min(state.ct_max, int(kelvin)))
            if state.strip:
                seg = {"mode": "white", "color_temp": ct}
            else:
                payload["color_temp"] = ct
        if (transition := kwargs.get(ATTR_TRANSITION)) is not None:
            payload["transition"] = max(0, min(MAX_TRANSITION_MS, int(float(transition) * 1000)))
        await self._apply(payload, seg)
    async def async_turn_off(self, **kwargs: Any) -> None:
        payload: dict[str, Any] = {"on": False}
        if (transition := kwargs.get(ATTR_TRANSITION)) is not None:
            payload["transition"] = max(0, min(MAX_TRANSITION_MS, int(float(transition) * 1000)))
        await self._apply(payload, None)
    async def _apply(self, payload: dict[str, Any], seg_patch: dict[str, Any] | None) -> None:
        """Send one command and update immediately.
        POST /device/state answers with the full new state, so the coordinator does not
        need a refresh round-trip and the UI settles on the real bulb state at once.
        On the segment-model firmware the color goes into segment 0 as a separate PATCH
        (`PUT /device/segments/0`, fields omitted keep their value); the master on and
        brightness stay in the state POST, because the firmware's fade rides on them.
        Segment brightness stays untouched so master scaling remains the single knob.
        """
        try:
            data = await self.coordinator.client.set_state(payload)
            if seg_patch is not None:
                segs = self.coordinator.data.segments
                start = segs[0]["start"] if segs else 0
                end = segs[0]["end"] if segs else max(0, int(data.get("pixel_count", 1)) - 1)
                await self.coordinator.client.patch_segment(
                    0, {**seg_patch, "start": start, "end": end}
                )
        except NinlightBadRequest as err:
            # A rejected command is a bug or a firmware mismatch, not a network fault:
            # surface it to the caller and leave the entity's state alone.
            raise HomeAssistantError(str(err)) from err
        except NinlightError as err:
            self.coordinator.async_set_update_error(err)
            raise HomeAssistantError(str(err)) from err
        # Push the device's own reply through the coordinator so every entity of this
        # device (light, sensors) settles on the real state at once.
        segments = None
        if self.coordinator.data.strip:
            try:
                segments = (await self.coordinator.client.get_segments()).get("segments")
            except NinlightError:
                pass  # the next poll recovers it; the state reply above is still right
        self.coordinator.async_set_updated_data(
            NinlightState.from_api(data, self.coordinator.info, segments)
        )
@@ -1,12 +0,0 @@
 {
  "domain": "ninlight",
  "name": "Ninlight",
  "codeowners": [],
  "config_flow": true,
  "documentation": "https://github.com/ninharp/ninlight-w600",
  "integration_type": "device",
  "iot_class": "local_polling",
  "requirements": [],
  "version": "0.1.0",
  "zeroconf": ["_ninlight._tcp.local.", "_ninstrip._tcp.local."]
 }
@@ -1,35 +0,0 @@
 {
  "config": {
    "steps": {
      "user": {
        "title": "Ninlight bulb",
        "description": "Connect to a bulb running the Ninlight firmware. Use the address shown in your router, or `ninlight-<mac>.local`.",
        "data": {
          "host": "Host",
          "port": "Port",
          "token": "API token"
        },
        "data_description": {
          "token": "The bulb's api.token setting. Defaults to the firmware's built-in token."
        }
      },
      "reauth_confirm": {
        "title": "Reauthenticate {name}",
        "description": "The bulb rejected the stored API token. Enter its current token.",
        "data": {
          "token": "API token"
        }
      }
    },
    "error": {
      "cannot_connect": "Could not reach the bulb on port 3333. Check the address and that the bulb is powered.",
      "invalid_auth": "Wrong API token."
    },
    "abort": {
      "already_configured": "This bulb is already configured.",
      "already_in_progress": "A setup flow for this bulb is already running.",
      "cannot_connect": "Could not reach the bulb.",
      "reauth_successful": "The API token was updated."
    }
  }
 }
@@ -1,35 +0,0 @@
 {
  "config": {
    "steps": {
      "user": {
        "title": "Ninlight bulb",
        "description": "Connect to a bulb running the Ninlight firmware. Use the address shown in your router, or `ninlight-<mac>.local`.",
        "data": {
          "host": "Host",
          "port": "Port",
          "token": "API token"
        },
        "data_description": {
          "token": "The bulb's api.token setting. Defaults to the firmware's built-in token."
        }
      },
      "reauth_confirm": {
        "title": "Reauthenticate {name}",
        "description": "The bulb rejected the stored API token. Enter its current token.",
        "data": {
          "token": "API token"
        }
      }
    },
    "error": {
      "cannot_connect": "Could not reach the bulb on port 3333. Check the address and that the bulb is powered.",
      "invalid_auth": "Wrong API token."
    },
    "abort": {
      "already_configured": "This bulb is already configured.",
      "already_in_progress": "A setup flow for this bulb is already running.",
      "cannot_connect": "Could not reach the bulb.",
      "reauth_successful": "The API token was updated."
    }
  }
 }
@@ -1,13 +0,0 @@
 # dnsmasq for the qconnex-firmware listener (no service integration)
 interface=wlan0
 bind-interfaces
 listen-address=192.168.8.1
 no-resolv
 server=192.168.178.1
 address=/#/192.168.8.1
 dhcp-range=192.168.8.10,192.168.8.200,255.255.255.0,10m
 dhcp-option=3,192.168.8.1
 dhcp-option=6,192.168.8.1
 log-dhcp
 log-queries
 no-ping
@@ -1,11 +0,0 @@
 #!/usr/bin/env bash
 # Tear down the qconnex-firmware migration listener.
 set -u
 HERE=$(cd "$(dirname "$0")" && pwd)
 SUDO=${SUDO:-sudo -n}
 [ -f "$HERE/otaserve.pid" ] && $SUDO pkill -f "python3 $HERE/otaserve.py" 2>/dev/null; rm -f "$HERE/otaserve.pid"
 [ -f "$HERE/dnsmasq.pid" ] && $SUDO kill "$(cat "$HERE/dnsmasq.pid")" 2>/dev/null && rm -f "$HERE/dnsmasq.pid"
 $SUDO pkill -f "hostapd -B -f $HERE/hostapd.log" 2>/dev/null
 $SUDO ip addr flush dev wlan0 2>/dev/null
 $SUDO nmcli dev set wlan0 managed yes 2>/dev/null || true
 echo "listener down"
@@ -1,15 +0,0 @@
 # qconnex-firmware listener AP — temporary experiment, do not enable as a service
 interface=wlan0
 driver=nl80211
 ssid=qconnex-firmware
 hw_mode=g
 channel=6
 ieee80211n=0
 auth_algs=1
 # WPA2 with the password the firmware's bootup-OTA STA config expects.
 # (Open AP worked for the W600 bulbs; ESP32-C3 STA threshold is WPA_PSK.)
 wpa=2
 wpa_passphrase=qconnex-firmware
 wpa_pairwise=CCMP
 rsn_pairwise=CCMP
 wmm_enabled=0
@@ -1,72 +0,0 @@
 #!/usr/bin/env python3
 """Serve qcx OTA images from ./images with HTTP Range support, plus request logging."""
 import http.server, os, re, sys, datetime
 ROOT = os.path.join(sys.path[0], "images")
 LOG = os.path.join(sys.path[0], "qcx-requests.log")
 class H(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    def log_message(self, *a):
        pass
    def do_GET(self):
        path = self.path.lstrip("/")
        with open(LOG, "a") as f:
            f.write("=== GET %s  (%s) %s\nPATH: %s\n" % (datetime.datetime.now(), self.client_address[0],
                        dict(self.headers), path))
        fp = os.path.join(ROOT, path)
        # any bootup-OTA request for a qcx image that doesn't exist on purpose
        # (unknown product/version combo) -> serve the ninlight migrate image
        if not os.path.isfile(fp) and path.startswith("qcx_firmware_"):
            if "ledshield" in path:
                guess = "ninlight_ledshield_rgb_migrate_dev.img"
            elif "rgbc3" in path:
                guess = "ninlight_ninstrip-rgbc3_migrate_dev.img"
            elif ".gu10" in path:
                guess = "ninlight_gu10_rgb_migrate_dev.img"
            else:
                guess = "ninlight_e27_rgb_migrate_dev.img"
            fp = os.path.join(ROOT, guess)
        if not os.path.isfile(fp):
            p = b"not found\n"
            self.send_response(404)
            self.send_header("Content-Length", str(len(p)))
            self.end_headers()
            self.wfile.write(p)
            return
        size = os.path.getsize(fp)
        start, rng = 0, self.headers.get("Range")
        if rng and (m := re.match(r"bytes=(\d+)-(\d*)$", rng)):
            start = int(m.group(1))
            end = int(m.group(2)) if m.group(2) else size - 1
        else:
            end = size - 1
        end = min(end, size - 1)
        length = max(0, end - start + 1)
        self.send_response(206 if rng else 200)
        self.send_header("Content-Length", str(length))
        self.send_header("Content-Type", "application/octet-stream")
        if rng:
            self.send_header("Content-Range", "bytes %d-%d/%d" % (start, end, size))
        self.end_headers()
        with open(fp, "rb") as f:
            f.seek(start)
            left = length
            while left > 0:
                chunk = f.read(min(8192, left))
                if not chunk:
                    break
                self.wfile.write(chunk)
                left -= len(chunk)
        print("served %s %d-%d to %s" % (path, start, end, self.client_address[0]), flush=True)
    do_POST = do_HEAD = do_PUT = do_GET
 if __name__ == "__main__":
    os.makedirs(ROOT, exist_ok=True)
    print("qcx ota server on 0.0.0.0:80 root=%s" % ROOT, flush=True)
    http.server.ThreadingHTTPServer(("0.0.0.0", 80), H).serve_forever()
@@ -1,22 +0,0 @@
 #!/usr/bin/env bash
 # Bring up the qconnex-firmware migration listener (AP + DHCP/DNS + OTA server).
 # Takes wlan0 as an open AP "qconnex-firmware" on 192.168.8.1. Machine stays
 # online via ethernet. Teardown: ./down.sh
 set -euo pipefail
 HERE=$(cd "$(dirname "$0")" && pwd)
 SUDO=${SUDO:-sudo -n}
 # wlan0 must be free of any NetworkManager claim
 $SUDO nmcli dev set wlan0 managed no 2>/dev/null || true
 $SUDO ip link set wlan0 up
 $SUDO ip addr replace 192.168.8.1/24 dev wlan0
 $SUDO dnsmasq --conf-file="$HERE/dnsmasq-qcx.conf" --pid-file="$HERE/dnsmasq.pid"
 $SUDO hostapd -B -f "$HERE/hostapd.log" "$HERE/hostapd.conf"
 sleep 1
 # serve images on :80 (needs root to bind <1024)
 nohup $SUDO python3 "$HERE/otaserve.py" >/dev/null 2>&1 &
 echo $! > "$HERE/otaserve.pid"
 echo "AP qconnex-firmware up on 192.168.8.1; serving $HERE/images (log: $HERE/qcx-requests.log)"
@@ -410,9 +410,14 @@ else
 fi
 echo "--- Meldung der Zugangsdatenpruefung nennt Dateien und entpackte Abbilder ---"
-# 8 ota + 8 flash + 16 migrate + bruno/ninlight.yml + README.md + onboard.sh = 35,
+# 9 ota + 9 flash + 18 migrate + bruno/ninlight.yml + README.md + onboard.sh = 39,
-# davon 32 Abbilder. SHA256SUMS entsteht erst nach der Pruefung.
+# davon 36 Abbilder. SHA256SUMS entsteht erst nach der Pruefung.
-if echo "$OUT" | grep -qF "sauber (35 Dateien geprueft, davon 32 Abbilder entpackt)"; then
+#
 # Die Zahlen haengen an der Laenge von TARGETS: mit ledshield_rgb kamen vier
 # Dateien dazu (je ein _ota, ein _gz und zwei signierte Migrationsabzuege), alle
 # vier sind Abbilder. Wer ein Ziel ergaenzt, passt sie hier mit an - der Merge
 # des Shield-Zweigs lief konfliktfrei durch und liess trotzdem 35/32 stehen.
 if echo "$OUT" | grep -qF "sauber (39 Dateien geprueft, davon 36 Abbilder entpackt)"; then
  pass "Meldung nennt Dateizahl und Zahl der entpackten Abbilder"
 else
  fail "Meldung der Zugangsdatenpruefung unerwartet: $(echo "$OUT" | grep -i sauber || echo '(keine sauber-Zeile)')"