"""Config flow for ninlight lamps.""" 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 NinlightAuthError, NinlightClient, NinlightError, NinlightResponseError from .const import ( CONF_LAMP_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_lamp_base_url _LOGGER = logging.getLogger(__name__) class NinlightConfigFlow(ConfigFlow, domain=DOMAIN): """One config entry per lamp, identified by its MAC.""" VERSION = 1 def __init__(self) -> None: self._host: str = "" self._discovered_name: str = "" @staticmethod @callback def async_get_options_flow(config_entry: ConfigEntry) -> NinlightOptionsFlow: return NinlightOptionsFlow() async def _async_probe( self, host: str, token: str ) -> tuple[dict[str, Any] | None, dict[str, str]]: """Check the token (system/settings always enforces it), then read the identity.""" client = NinlightClient(async_get_clientsession(self.hass), host, token) try: await client.get_settings() info = await client.get_info() except NinlightAuthError: return None, {"base": "invalid_auth"} except NinlightResponseError: return None, {"base": "not_ninlight"} except NinlightError: return None, {"base": "cannot_connect"} except Exception: _LOGGER.exception("Unexpected error while probing %s", host) return None, {"base": "unknown"} return info, {} 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[CONF_HOST].strip() token = user_input[CONF_TOKEN] info, errors = await self._async_probe(host, token) if info is not None: await self.async_set_unique_id(format_mac(info["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} ) defaults = user_input or {} schema = vol.Schema( { vol.Required(CONF_HOST, default=defaults.get(CONF_HOST, "")): str, vol.Required(CONF_TOKEN, default=defaults.get(CONF_TOKEN, DEFAULT_TOKEN)): str, } ) return self.async_show_form(step_id="user", data_schema=schema, errors=errors) async def async_step_zeroconf(self, discovery_info: ZeroconfServiceInfo) -> ConfigFlowResult: properties = discovery_info.properties mac = properties.get("mac") if not mac: return self.async_abort(reason="not_ninlight") host = str(discovery_info.ip_address) await self.async_set_unique_id(format_mac(mac)) # Known lamp under a new address: update the entry silently and stop here. self._abort_if_unique_id_configured(updates={CONF_HOST: host}) self._host = host self._discovered_name = ( properties.get("name") or properties.get("type") or discovery_info.name ) self.context["title_placeholders"] = {"name": self._discovered_name} return await self.async_step_zeroconf_confirm() async def async_step_zeroconf_confirm( 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(self._host, user_input[CONF_TOKEN]) if info is not None: await self.async_set_unique_id(format_mac(info["mac"]), raise_on_progress=False) self._abort_if_unique_id_configured(updates={CONF_HOST: self._host}) return self.async_create_entry( title=device_name(info), data={CONF_HOST: self._host, CONF_TOKEN: user_input[CONF_TOKEN]}, ) return self.async_show_form( step_id="zeroconf_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_reauth(self, entry_data: Mapping[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._get_reauth_entry() errors: dict[str, str] = {} if user_input is not None: info, errors = await self._async_probe(entry.data[CONF_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( 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={"name": entry.title}, errors=errors, ) async def async_step_reconfigure( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: entry = 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( entry, data_updates={CONF_HOST: host, CONF_TOKEN: user_input[CONF_TOKEN]} ) schema = vol.Schema( { vol.Required(CONF_HOST, default=entry.data[CONF_HOST]): str, vol.Required(CONF_TOKEN, default=entry.data[CONF_TOKEN]): str, } ) return self.async_show_form(step_id="reconfigure", data_schema=schema, errors=errors) class NinlightOptionsFlow(OptionsFlowWithReload): """Polling interval and the address lamps use to reach Home Assistant.""" 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_LAMP_BASE_URL) or "").strip() error = validate_lamp_base_url(base) if base else None if error: errors[CONF_LAMP_BASE_URL] = error else: data: dict[str, Any] = {CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL]} if base: data[CONF_LAMP_BASE_URL] = base.rstrip("/") return self.async_create_entry(data=data) current = self.config_entry.options 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_LAMP_BASE_URL, description={"suggested_value": current.get(CONF_LAMP_BASE_URL, "")}, ): str, } ) return self.async_show_form(step_id="init", data_schema=schema, errors=errors)