ha-ninstrip/custom_components/ninstrip/config_flow.py

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

    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:
        """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,
        )