"""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_DEVICE_BASE_URL, CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL, DEFAULT_TOKEN, DOMAIN, MAX_SCAN_INTERVAL, MIN_SCAN_INTERVAL, ) from .entity import device_name from .urls import validate_device_base_url _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, default=DEFAULT_TOKEN): str, } ), errors=errors, ) def _kennung_aus_mdns(self, roh: str) -> str | None: """Die Kennung zu dem, was im TXT-Satz unter "mac" steht. Eine volle MAC ist die Kennung. Die Firmware wirbt aber nur mit den letzten vier Hex-Stellen (`nl_wifi_mac4()`, siehe nl_mdns.c) - daraus laesst sich keine bauen. Als *Hinweis* taugen sie trotzdem: endet genau ein eingerichteter Eintrag so, ist er gemeint. Bei mehreren Treffern bleibt es unentschieden. Dann lieber nachfragen als den falschen Eintrag stillschweigend auf eine fremde Adresse umzuschreiben - vier Stellen sind nicht eindeutig. """ kennung = format_mac(roh) if len(kennung) == 17: return kennung kurz = roh.replace(":", "").replace("-", "").lower() if len(kurz) != 4: return None treffer = [ eintrag.unique_id for eintrag in self._async_current_entries(include_ignore=False) if eintrag.unique_id and eintrag.unique_id.replace(":", "").endswith(kurz) ] return treffer[0] if len(treffer) == 1 else None 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 and (kennung := self._kennung_aus_mdns(str(mac))): await self.async_set_unique_id(kennung) 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: """Vorbelegt mit dem Werks-Token - ein frisch onboardetes Geraet antwortet damit.""" 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, default=DEFAULT_TOKEN): str}), description_placeholders={"name": self._discovered_name, "host": self._host}, errors=errors, ) async def async_step_reconfigure( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Adresse und Token eines bestehenden Geraets aendern. Gegengeprueft wird die MAC: wer aus Versehen die Adresse eines anderen Geraets eintraegt, soll nicht stillschweigend den falschen Eintrag umbiegen. """ eintrag = self._get_reconfigure_entry() errors: dict[str, str] = {} if user_input is not None: host = user_input[CONF_HOST].strip() info, errors = await self._async_probe(host, user_input[CONF_TOKEN]) if info is not None: await self.async_set_unique_id(format_mac(info["mac"])) self._abort_if_unique_id_mismatch(reason="wrong_device") return self.async_update_reload_and_abort( eintrag, data_updates={CONF_HOST: host, CONF_TOKEN: user_input[CONF_TOKEN]} ) return self.async_show_form( step_id="reconfigure", data_schema=vol.Schema( { vol.Required(CONF_HOST, default=eintrag.data[CONF_HOST]): str, vol.Required(CONF_TOKEN, default=eintrag.data[CONF_TOKEN]): str, } ), 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): """Abfragehaeufigkeit und die Adresse, unter der Geraete Home Assistant erreichen.""" async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: errors: dict[str, str] = {} if user_input is not None: base = str(user_input.get(CONF_DEVICE_BASE_URL) or "").strip() fehler = validate_device_base_url(base) if base else None if fehler: errors[CONF_DEVICE_BASE_URL] = fehler else: daten: dict[str, Any] = {CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL]} if base: daten[CONF_DEVICE_BASE_URL] = base.rstrip("/") return self.async_create_entry(data=daten) current = self.config_entry.options return self.async_show_form( step_id="init", data_schema=vol.Schema( { vol.Required( CONF_SCAN_INTERVAL, default=current.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL), ): vol.All( vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL, max=MAX_SCAN_INTERVAL) ), vol.Optional( CONF_DEVICE_BASE_URL, description={"suggested_value": current.get(CONF_DEVICE_BASE_URL, "")}, ): str, } ), errors=errors, )