Dual-PID/Home Assistant Erweiterung/custom_components/dual_pid/config_flow.py

from __future__ import annotations

from typing import Any
from urllib.parse import urlparse

import voluptuous as vol

from homeassistant import config_entries
from homeassistant.config_entries import OptionsFlowWithReload
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession

from .api import DualPidApiClient, DualPidApiConnectionError
from .const import (
    CONF_SCAN_INTERVAL,
    DEFAULT_NAME,
    DEFAULT_PORT,
    DEFAULT_SCAN_INTERVAL,
    DEFAULT_TIMEOUT,
    DOMAIN,
)


class CannotConnect(Exception):
    """Error to indicate we cannot connect."""


def _normalize_host(host: str) -> str:
    host = host.strip()
    if "://" in host:
        parsed = urlparse(host)
        if parsed.hostname:
            return parsed.hostname
    return host.strip("/ ")


async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
    client = DualPidApiClient(
        session=async_get_clientsession(hass),
        host=data[CONF_HOST],
        port=data[CONF_PORT],
        timeout=DEFAULT_TIMEOUT,
    )

    try:
        status = await client.async_get_status()
    except DualPidApiConnectionError as err:
        raise CannotConnect from err

    return status


def _build_schema(
    *,
    name: str = DEFAULT_NAME,
    host: str = "",
    port: int = DEFAULT_PORT,
    scan_interval: int = DEFAULT_SCAN_INTERVAL,
) -> vol.Schema:
    return vol.Schema(
        {
            vol.Required(CONF_NAME, default=name): str,
            vol.Required(CONF_HOST, default=host): str,
            vol.Required(CONF_PORT, default=port): vol.All(
                vol.Coerce(int), vol.Range(min=1, max=65535)
            ),
            vol.Required(CONF_SCAN_INTERVAL, default=scan_interval): vol.All(
                vol.Coerce(int), vol.Range(min=2, max=300)
            ),
        }
    )


def _user_schema(
    *,
    name: str = DEFAULT_NAME,
    host: str = "",
    port: int = DEFAULT_PORT,
    scan_interval: int = DEFAULT_SCAN_INTERVAL,
) -> vol.Schema:
    return _build_schema(
        name=name,
        host=host,
        port=port,
        scan_interval=scan_interval,
    )


class DualPidConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Dual PID."""

    VERSION = 1

    async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult:
        errors: dict[str, str] = {}

        if user_input is not None:
            user_input[CONF_HOST] = _normalize_host(user_input[CONF_HOST])

            for entry in self._async_current_entries():
                if (
                    entry.data.get(CONF_HOST) == user_input[CONF_HOST]
                    and entry.data.get(CONF_PORT) == user_input[CONF_PORT]
                ):
                    return self.async_abort(reason="already_configured")

            try:
                await _validate_input(self.hass, user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                errors["base"] = "unknown"
            else:
                return self.async_create_entry(
                    title=user_input[CONF_NAME],
                    data=user_input,
                )

        return self.async_show_form(
            step_id="user",
            data_schema=_user_schema(),
            errors=errors,
        )

    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: config_entries.ConfigEntry,
    ) -> config_entries.OptionsFlow:
        return DualPidOptionsFlow()

    async def async_step_reconfigure(
        self, user_input: dict[str, Any] | None = None
    ) -> FlowResult:
        errors: dict[str, str] = {}
        entry = self._get_reconfigure_entry()

        if user_input is not None:
            user_input[CONF_HOST] = _normalize_host(user_input[CONF_HOST])

            try:
                await _validate_input(self.hass, user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                errors["base"] = "unknown"
            else:
                current_entry = self._get_reconfigure_entry()
                new_data = {
                    **current_entry.data,
                    CONF_NAME: user_input[CONF_NAME],
                    CONF_HOST: user_input[CONF_HOST],
                    CONF_PORT: user_input[CONF_PORT],
                    CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL],
                }
                self.hass.config_entries.async_update_entry(
                    current_entry,
                    title=user_input[CONF_NAME],
                    options={
                        **current_entry.options,
                        CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL],
                    },
                )
                return self.async_update_reload_and_abort(
                    current_entry,
                    data_updates=new_data,
                )

        return self.async_show_form(
            step_id="reconfigure",
            data_schema=_build_schema(
                name=entry.title or entry.data.get(CONF_NAME, DEFAULT_NAME),
                host=entry.data.get(CONF_HOST, ""),
                port=entry.data.get(CONF_PORT, DEFAULT_PORT),
                scan_interval=entry.options.get(
                    CONF_SCAN_INTERVAL,
                    entry.data.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
                ),
            ),
            errors=errors,
        )


class DualPidOptionsFlow(OptionsFlowWithReload):
    """Handle options for Dual PID."""

    async def async_step_init(self, user_input: dict[str, Any] | None = None) -> FlowResult:
        if user_input is not None:
            return self.async_create_entry(data=user_input)

        return self.async_show_form(
            step_id="init",
            data_schema=self.add_suggested_values_to_schema(
                vol.Schema(
                    {
                        vol.Required(CONF_SCAN_INTERVAL): vol.All(
                            vol.Coerce(int), vol.Range(min=2, max=300)
                        )
                    }
                ),
                {
                    CONF_SCAN_INTERVAL: self.config_entry.options.get(
                        CONF_SCAN_INTERVAL,
                        self.config_entry.data.get(
                            CONF_SCAN_INTERVAL,
                            DEFAULT_SCAN_INTERVAL,
                        ),
                    )
                },
            ),
        )