"""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-` 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}, )