mass-hacs-gate-button/custom_components/main_gate_controller/config_flow.py

"""Config flow for the Main Gate Controller integration."""

from __future__ import annotations

from typing import Any

import voluptuous as vol
from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    OptionsFlow,
    OptionsFlowWithConfigEntry,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.selector import (
    BooleanSelector,
    EntitySelector,
    EntitySelectorConfig,
    NumberSelector,
    NumberSelectorConfig,
    TextSelector,
)

from .const import (
    CONF_NAME,
    CONF_NOTIFY_CLOSING_TEXT,
    CONF_NOTIFY_ENABLED,
    CONF_NOTIFY_ENTITIES,
    CONF_NOTIFY_OPENING_TEXT,
    CONF_OPEN_DURATION,
    CONF_PULSE_DURATION,
    CONF_SWITCH_ENTITY_ID,
    DEFAULT_NAME,
    DEFAULT_NOTIFY_CLOSING_TEXT,
    DEFAULT_NOTIFY_ENABLED,
    DEFAULT_NOTIFY_OPENING_TEXT,
    DEFAULT_OPEN_DURATION,
    DEFAULT_PULSE_DURATION,
    DOMAIN,
    MAX_OPEN_DURATION,
    MAX_PULSE_DURATION,
    MIN_OPEN_DURATION,
    MIN_PULSE_DURATION,
    NOTIFY_DOMAIN,
    SWITCH_DOMAIN,
)
from .coordinator import get_setting

_USER_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_NAME, default=DEFAULT_NAME): TextSelector(),
        vol.Required(CONF_SWITCH_ENTITY_ID): EntitySelector(
            EntitySelectorConfig(domain=SWITCH_DOMAIN)
        ),
        vol.Required(CONF_OPEN_DURATION, default=DEFAULT_OPEN_DURATION): NumberSelector(
            NumberSelectorConfig(
                min=MIN_OPEN_DURATION,
                max=MAX_OPEN_DURATION,
                step=1,
                unit_of_measurement="s",
                mode="box",
            )
        ),
        vol.Required(CONF_PULSE_DURATION, default=DEFAULT_PULSE_DURATION): NumberSelector(
            NumberSelectorConfig(
                min=MIN_PULSE_DURATION,
                max=MAX_PULSE_DURATION,
                step=0.1,
                unit_of_measurement="s",
                mode="box",
            )
        ),
        vol.Required(CONF_NOTIFY_ENABLED, default=DEFAULT_NOTIFY_ENABLED): BooleanSelector(),
        vol.Optional(CONF_NOTIFY_ENTITIES, default=[]): EntitySelector(
            EntitySelectorConfig(domain=NOTIFY_DOMAIN, multiple=True)
        ),
        vol.Optional(
            CONF_NOTIFY_OPENING_TEXT, default=DEFAULT_NOTIFY_OPENING_TEXT
        ): TextSelector(),
        vol.Optional(
            CONF_NOTIFY_CLOSING_TEXT, default=DEFAULT_NOTIFY_CLOSING_TEXT
        ): TextSelector(),
    }
)


def _options_schema(entry: ConfigEntry) -> vol.Schema:
    return vol.Schema(
        {
            vol.Required(
                CONF_OPEN_DURATION,
                default=float(get_setting(entry, CONF_OPEN_DURATION, DEFAULT_OPEN_DURATION)),
            ): NumberSelector(
                NumberSelectorConfig(
                    min=MIN_OPEN_DURATION,
                    max=MAX_OPEN_DURATION,
                    step=1,
                    unit_of_measurement="s",
                    mode="box",
                )
            ),
            vol.Required(
                CONF_PULSE_DURATION,
                default=float(get_setting(entry, CONF_PULSE_DURATION, DEFAULT_PULSE_DURATION)),
            ): NumberSelector(
                NumberSelectorConfig(
                    min=MIN_PULSE_DURATION,
                    max=MAX_PULSE_DURATION,
                    step=0.1,
                    unit_of_measurement="s",
                    mode="box",
                )
            ),
            vol.Required(
                CONF_NOTIFY_ENABLED,
                default=bool(get_setting(entry, CONF_NOTIFY_ENABLED, DEFAULT_NOTIFY_ENABLED)),
            ): BooleanSelector(),
            vol.Optional(
                CONF_NOTIFY_ENTITIES,
                default=list(get_setting(entry, CONF_NOTIFY_ENTITIES, []) or []),
            ): EntitySelector(
                EntitySelectorConfig(domain=NOTIFY_DOMAIN, multiple=True)
            ),
            vol.Optional(
                CONF_NOTIFY_OPENING_TEXT,
                default=get_setting(
                    entry, CONF_NOTIFY_OPENING_TEXT, DEFAULT_NOTIFY_OPENING_TEXT
                ),
            ): TextSelector(),
            vol.Optional(
                CONF_NOTIFY_CLOSING_TEXT,
                default=get_setting(
                    entry, CONF_NOTIFY_CLOSING_TEXT, DEFAULT_NOTIFY_CLOSING_TEXT
                ),
            ): TextSelector(),
        }
    )


@callback
def _validate_input(hass: HomeAssistant, user_input: dict[str, Any]) -> dict[str, str]:
    """Return a dict of validation errors keyed by field name."""
    errors: dict[str, str] = {}

    name = user_input.get(CONF_NAME)
    if not isinstance(name, str) or not name.strip():
        errors[CONF_NAME] = "invalid_name"

    switch = user_input.get(CONF_SWITCH_ENTITY_ID)
    if not isinstance(switch, str) or not switch.startswith(f"{SWITCH_DOMAIN}."):
        errors[CONF_SWITCH_ENTITY_ID] = "invalid_switch_domain"
    elif hass.states.get(switch) is None:
        errors[CONF_SWITCH_ENTITY_ID] = "switch_not_found"

    if user_input.get(CONF_NOTIFY_ENABLED) and not user_input.get(CONF_NOTIFY_ENTITIES):
        errors[CONF_NOTIFY_ENTITIES] = "notify_no_targets"

    return errors


@callback
def _validate_options(user_input: dict[str, Any]) -> dict[str, str]:
    errors: dict[str, str] = {}
    if user_input.get(CONF_NOTIFY_ENABLED) and not user_input.get(CONF_NOTIFY_ENTITIES):
        errors[CONF_NOTIFY_ENTITIES] = "notify_no_targets"
    return errors


class MainGateConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle the user-initiated config flow for a single gate."""

    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:
            switch = user_input.get(CONF_SWITCH_ENTITY_ID)
            if isinstance(switch, str):
                await self.async_set_unique_id(switch)
                self._abort_if_unique_id_configured()

            errors = _validate_input(self.hass, user_input)
            if not errors:
                title = str(user_input[CONF_NAME]).strip() or DEFAULT_NAME
                return self.async_create_entry(title=title, 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: ConfigEntry) -> OptionsFlow:
        return MainGateOptionsFlow(config_entry)


class MainGateOptionsFlow(OptionsFlowWithConfigEntry):
    """Handle the options flow for an existing gate."""

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

            return self.async_show_form(
                step_id="init",
                data_schema=self.add_suggested_values_to_schema(
                    _options_schema(self.config_entry), user_input
                ),
                errors=errors,
            )

        return self.async_show_form(
            step_id="init",
            data_schema=_options_schema(self.config_entry),
        )