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_SCAN_INTERVAL,
    DEFAULT_SCAN_INTERVAL,
    DEFAULT_TOKEN,
    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, 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_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)
                    )
                }
            ),
        )