feat: erste Fassung der ninstrip-Integration fuer Home Assistant

Auf dem Geruest von ha-ninlight aufgebaut und bewusst parallel gehalten: gleiche Fehlerklassen in api.py, gleicher Coordinator mit Sperre und Fehlertoleranz, gleicher Einrichtungsablauf samt Zeroconf und Reauth. Eigenes Repo statt Erweiterung von ha-ninlight: die Geraete melden sich unter verschiedenen mDNS-Diensten, das Datenmodell unterscheidet sich deutlich (Segmente und Szenen gegen ein einzelnes Licht), und ha-ninlight laeuft produktiv mit vier Lampen - Verzweigungen dort haetten beide Produkte gefaehrdet. Entitaeten: light (Hauptschalter und Helligkeit), select (Szene), button (Neustart), vier Diagnosesensoren. In ha-dev eingebunden und am echten Geraet geprueft: Aus- und Einschalten, Helligkeit 40 Prozent, Szenenwechsel auf Regenbogen - das Geraet meldet jeweils genau das zurueck.
2026-09-20 03:42:42 +02:00
commit adec66a6a2
16 changed files with 1140 additions and 0 deletions
@@ -0,0 +1,177 @@
+"""Einrichtungsablauf fuer ninstrip-Geraete."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from homeassistant.config_entries import (
+    ConfigEntry,
+    ConfigFlow,
+    ConfigFlowResult,
+    OptionsFlowWithReload,
+)
+from homeassistant.const import CONF_HOST, CONF_TOKEN
+from homeassistant.core import callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.device_registry import format_mac
+from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
+import voluptuous as vol
+
+from .api import NinstripAuthError, NinstripClient, NinstripError, NinstripResponseError
+from .const import (
+    CONF_SCAN_INTERVAL,
+    DEFAULT_SCAN_INTERVAL,
+    DOMAIN,
+    MAX_SCAN_INTERVAL,
+    MIN_SCAN_INTERVAL,
+)
+from .entity import device_name
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class NinstripConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Ein Eintrag je Geraet, erkannt an seiner MAC."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        self._host: str = ""
+        self._discovered_name: str = ""
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry: ConfigEntry) -> NinstripOptionsFlow:
+        return NinstripOptionsFlow()
+
+    async def _async_probe(
+        self, host: str, token: str
+    ) -> tuple[dict[str, Any] | None, dict[str, str]]:
+        """Erst das Token pruefen - system/settings verlangt es immer - dann die Kennung lesen.
+
+        Wichtig: system/info allein taugt nicht als Pruefung, es antwortet auch
+        ohne Token. Genau wie bei ninlight.
+        """
+        client = NinstripClient(async_get_clientsession(self.hass), host, token)
+        try:
+            await client.get_settings()
+            info = await client.get_info()
+        except NinstripAuthError:
+            return None, {"base": "invalid_auth"}
+        except NinstripResponseError:
+            return None, {"base": "not_ninstrip"}
+        except NinstripError:
+            return None, {"base": "cannot_connect"}
+        except Exception:
+            _LOGGER.exception("Unerwarteter Fehler beim Pruefen von %s", host)
+            return None, {"base": "unknown"}
+        return info, {}
+
+    async def _async_finish(self, host: str, token: str, info: dict[str, Any]) -> ConfigFlowResult:
+        mac = format_mac(info["mac"])
+        await self.async_set_unique_id(mac)
+        self._abort_if_unique_id_configured(updates={CONF_HOST: host})
+        return self.async_create_entry(
+            title=device_name(info), data={CONF_HOST: host, CONF_TOKEN: token}
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            info, errors = await self._async_probe(
+                user_input[CONF_HOST], user_input[CONF_TOKEN]
+            )
+            if info is not None:
+                return await self._async_finish(
+                    user_input[CONF_HOST], user_input[CONF_TOKEN], info
+                )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST, default=self._host): str,
+                    vol.Required(CONF_TOKEN): str,
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: ZeroconfServiceInfo
+    ) -> ConfigFlowResult:
+        """Gefunden ueber _ninstrip._tcp."""
+        self._host = discovery_info.host
+        props = discovery_info.properties or {}
+        mac = props.get("mac")
+        if mac:
+            await self.async_set_unique_id(format_mac(str(mac)))
+            self._abort_if_unique_id_configured(updates={CONF_HOST: self._host})
+        self._discovered_name = str(props.get("name") or discovery_info.name.split(".")[0])
+        self.context["title_placeholders"] = {"name": self._discovered_name}
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Das Token steht einmalig auf der seriellen Konsole - es muss von Hand herein."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            info, errors = await self._async_probe(self._host, user_input[CONF_TOKEN])
+            if info is not None:
+                return await self._async_finish(self._host, user_input[CONF_TOKEN], info)
+        return self.async_show_form(
+            step_id="discovery_confirm",
+            data_schema=vol.Schema({vol.Required(CONF_TOKEN): str}),
+            description_placeholders={"name": self._discovered_name, "host": self._host},
+            errors=errors,
+        )
+
+    async def async_step_reauth(
+        self, entry_data: Mapping[str, Any]
+    ) -> ConfigFlowResult:
+        self._host = entry_data[CONF_HOST]
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Nach einem Werksreset hat das Geraet ein neues Token."""
+        errors: dict[str, str] = {}
+        entry = self._get_reauth_entry()
+        if user_input is not None:
+            info, errors = await self._async_probe(self._host, user_input[CONF_TOKEN])
+            if info is not None:
+                return self.async_update_reload_and_abort(
+                    entry, data_updates={CONF_TOKEN: user_input[CONF_TOKEN]}
+                )
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema({vol.Required(CONF_TOKEN): str}),
+            description_placeholders={"host": self._host},
+            errors=errors,
+        )
+
+
+class NinstripOptionsFlow(OptionsFlowWithReload):
+    """Bisher nur die Abfragehaeufigkeit."""
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        if user_input is not None:
+            return self.async_create_entry(data=user_input)
+        current = self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_SCAN_INTERVAL, default=current): vol.All(
+                        vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL, max=MAX_SCAN_INTERVAL)
+                    )
+                }
+            ),
+        )