ha-reticulum/custom_components/reticulum/config_flow.py

"""Config flow for the Reticulum integration."""

from __future__ import annotations

import asyncio
from typing import Any

import voluptuous as vol

from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.selector import (
    BooleanSelector,
    ConversationAgentSelector,
    ConversationAgentSelectorConfig,
    LanguageSelector,
    LanguageSelectorConfig,
    NumberSelector,
    NumberSelectorConfig,
    NumberSelectorMode,
    SelectSelector,
    SelectSelectorConfig,
    SelectSelectorMode,
    TextSelector,
    TextSelectorConfig,
)

from .const import (
    CONF_ALLOW_ALL,
    CONF_ALLOWED_IDENTITIES,
    CONF_ANNOUNCE_INTERVAL,
    CONF_ASSIST_AGENT,
    CONF_ASSIST_LANGUAGE,
    CONF_DEFAULT_RECIPIENT,
    CONF_DELIVERY_METHOD,
    CONF_DISPLAY_NAME,
    CONF_ENABLE_ASSIST,
    CONF_GREETING,
    CONF_INTERFACE_NAME,
    CONF_LOGLEVEL,
    CONF_PROPAGATION_NODE,
    CONF_SYNC_INTERVAL,
    CONF_TARGET_HOST,
    CONF_TARGET_PORT,
    DEFAULT_ANNOUNCE_INTERVAL,
    DEFAULT_DELIVERY_METHOD,
    DEFAULT_DISPLAY_NAME,
    DEFAULT_INTERFACE_NAME,
    DEFAULT_LOGLEVEL,
    DEFAULT_SYNC_INTERVAL,
    DEFAULT_TARGET_PORT,
    DELIVERY_METHODS,
    DOMAIN,
)


async def _test_connection(host: str, port: int) -> None:
    """Verify the Reticulum TCP server is reachable. Raises on failure."""
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), timeout=10
        )
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:  # noqa: BLE001
            pass
    except (OSError, asyncio.TimeoutError) as err:
        raise CannotConnect(str(err)) from err


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


STEP_USER_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_TARGET_HOST): cv.string,
        vol.Required(CONF_TARGET_PORT, default=DEFAULT_TARGET_PORT): vol.All(
            vol.Coerce(int), vol.Range(min=1, max=65535)
        ),
        vol.Required(CONF_DISPLAY_NAME, default=DEFAULT_DISPLAY_NAME): cv.string,
        vol.Required(
            CONF_INTERFACE_NAME, default=DEFAULT_INTERFACE_NAME
        ): cv.string,
    }
)


class ReticulumConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Reticulum."""

    VERSION = 1

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        self._async_abort_entries_match()  # single_config_entry also enforces this
        errors: dict[str, str] = {}

        if user_input is not None:
            try:
                await _test_connection(
                    user_input[CONF_TARGET_HOST], user_input[CONF_TARGET_PORT]
                )
            except CannotConnect:
                errors["base"] = "cannot_connect"
            else:
                await self.async_set_unique_id(DOMAIN)
                self._abort_if_unique_id_configured()
                return self.async_create_entry(
                    title=f"Reticulum ({user_input[CONF_DISPLAY_NAME]})",
                    data=user_input,
                    options=_default_options(),
                )

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

    @staticmethod
    @callback
    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
        """Get the options flow for this handler."""
        return ReticulumOptionsFlow()


def _default_options() -> dict[str, Any]:
    return {
        CONF_ENABLE_ASSIST: True,
        CONF_ALLOW_ALL: True,
        CONF_ALLOWED_IDENTITIES: [],
        CONF_ANNOUNCE_INTERVAL: DEFAULT_ANNOUNCE_INTERVAL,
        CONF_SYNC_INTERVAL: DEFAULT_SYNC_INTERVAL,
        CONF_DELIVERY_METHOD: DEFAULT_DELIVERY_METHOD,
        CONF_LOGLEVEL: DEFAULT_LOGLEVEL,
    }


class ReticulumOptionsFlow(OptionsFlow):
    """Handle Reticulum options."""

    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Manage the options."""
        if user_input is not None:
            # Normalise the comma/newline separated allow-list into a list.
            raw = user_input.get(CONF_ALLOWED_IDENTITIES, "")
            if isinstance(raw, str):
                user_input[CONF_ALLOWED_IDENTITIES] = [
                    item.strip()
                    for item in raw.replace("\n", ",").split(",")
                    if item.strip()
                ]
            return self.async_create_entry(data=user_input)

        opts = self.config_entry.options
        allowed = opts.get(CONF_ALLOWED_IDENTITIES, [])
        allowed_str = ", ".join(allowed) if isinstance(allowed, list) else allowed

        schema = vol.Schema(
            {
                vol.Required(
                    CONF_ENABLE_ASSIST,
                    default=opts.get(CONF_ENABLE_ASSIST, True),
                ): BooleanSelector(),
                vol.Optional(
                    CONF_ASSIST_AGENT,
                    description={"suggested_value": opts.get(CONF_ASSIST_AGENT)},
                ): ConversationAgentSelector(ConversationAgentSelectorConfig()),
                vol.Optional(
                    CONF_ASSIST_LANGUAGE,
                    description={
                        "suggested_value": opts.get(CONF_ASSIST_LANGUAGE)
                    },
                ): LanguageSelector(LanguageSelectorConfig()),
                vol.Required(
                    CONF_ALLOW_ALL, default=opts.get(CONF_ALLOW_ALL, True)
                ): BooleanSelector(),
                vol.Optional(
                    CONF_ALLOWED_IDENTITIES,
                    description={"suggested_value": allowed_str},
                ): TextSelector(TextSelectorConfig(multiline=True)),
                vol.Optional(
                    CONF_DEFAULT_RECIPIENT,
                    description={
                        "suggested_value": opts.get(CONF_DEFAULT_RECIPIENT, "")
                    },
                ): TextSelector(),
                vol.Optional(
                    CONF_GREETING,
                    description={"suggested_value": opts.get(CONF_GREETING, "")},
                ): TextSelector(TextSelectorConfig(multiline=True)),
                vol.Required(
                    CONF_DELIVERY_METHOD,
                    default=opts.get(CONF_DELIVERY_METHOD, DEFAULT_DELIVERY_METHOD),
                ): SelectSelector(
                    SelectSelectorConfig(
                        options=DELIVERY_METHODS,
                        translation_key="delivery_method",
                        mode=SelectSelectorMode.DROPDOWN,
                    )
                ),
                vol.Required(
                    CONF_ANNOUNCE_INTERVAL,
                    default=opts.get(
                        CONF_ANNOUNCE_INTERVAL, DEFAULT_ANNOUNCE_INTERVAL
                    ),
                ): NumberSelector(
                    NumberSelectorConfig(
                        min=0, max=86400, step=60, mode=NumberSelectorMode.BOX,
                        unit_of_measurement="s",
                    )
                ),
                vol.Optional(
                    CONF_PROPAGATION_NODE,
                    description={
                        "suggested_value": opts.get(CONF_PROPAGATION_NODE, "")
                    },
                ): TextSelector(),
                vol.Required(
                    CONF_SYNC_INTERVAL,
                    default=opts.get(CONF_SYNC_INTERVAL, DEFAULT_SYNC_INTERVAL),
                ): NumberSelector(
                    NumberSelectorConfig(
                        min=0, max=86400, step=60, mode=NumberSelectorMode.BOX,
                        unit_of_measurement="s",
                    )
                ),
                vol.Required(
                    CONF_LOGLEVEL, default=opts.get(CONF_LOGLEVEL, DEFAULT_LOGLEVEL)
                ): NumberSelector(
                    NumberSelectorConfig(
                        min=0, max=7, step=1, mode=NumberSelectorMode.SLIDER
                    )
                ),
            }
        )
        return self.async_show_form(step_id="init", data_schema=schema)