config_flow.py

Go to the documentation of this file.

"""Config flow for konnected.io integration."""
 
from __future__ import annotations
 
import asyncio
import copy
import logging
import random
import string
from typing import Any
from urllib.parse import urlparse
 
import voluptuous as vol
 
from homeassistant.components import ssdp
from homeassistant.components.binary_sensor import (
    DEVICE_CLASSES_SCHEMA,
    BinarySensorDeviceClass,
)
from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.const import (
    CONF_ACCESS_TOKEN,
    CONF_BINARY_SENSORS,
    CONF_DISCOVERY,
    CONF_HOST,
    CONF_ID,
    CONF_MODEL,
    CONF_NAME,
    CONF_PORT,
    CONF_REPEAT,
    CONF_SENSORS,
    CONF_SWITCHES,
    CONF_TYPE,
    CONF_ZONE,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
 
from .const import (
    CONF_ACTIVATION,
    CONF_API_HOST,
    CONF_BLINK,
    CONF_DEFAULT_OPTIONS,
    CONF_INVERSE,
    CONF_MOMENTARY,
    CONF_PAUSE,
    CONF_POLL_INTERVAL,
    DOMAIN,
    STATE_HIGH,
    STATE_LOW,
    ZONES,
)
from .errors import CannotConnect
from .panel import KONN_MODEL, KONN_MODEL_PRO, get_status
 
_LOGGER = logging.getLogger(__name__)
 
ATTR_KONN_UPNP_MODEL_NAME = "model_name"  # standard upnp is modelName
CONF_IO = "io"
CONF_IO_DIS = "Disabled"
CONF_IO_BIN = "Binary Sensor"
CONF_IO_DIG = "Digital Sensor"
CONF_IO_SWI = "Switchable Output"
 
CONF_MORE_STATES = "more_states"
CONF_YES = "Yes"
CONF_NO = "No"
 
CONF_OVERRIDE_API_HOST = "override_api_host"
 
KONN_MANUFACTURER = "konnected.io"
KONN_PANEL_MODEL_NAMES = {
    KONN_MODEL: "Konnected Alarm Panel",
    KONN_MODEL_PRO: "Konnected Alarm Panel Pro",
}
 
OPTIONS_IO_ANY = vol.In([CONF_IO_DIS, CONF_IO_BIN, CONF_IO_DIG, CONF_IO_SWI])
OPTIONS_IO_INPUT_ONLY = vol.In([CONF_IO_DIS, CONF_IO_BIN])
OPTIONS_IO_OUTPUT_ONLY = vol.In([CONF_IO_DIS, CONF_IO_SWI])
 
 
# Config entry schemas
IO_SCHEMA = vol.Schema(
    {
        vol.Optional("1", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("2", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("3", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("4", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("5", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("6", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("7", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("8", default=CONF_IO_DIS): OPTIONS_IO_ANY,
        vol.Optional("9", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
        vol.Optional("10", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
        vol.Optional("11", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
        vol.Optional("12", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
        vol.Optional("out", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
        vol.Optional("alarm1", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
        vol.Optional("out1", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
        vol.Optional("alarm2_out2", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
    }
)
 
BINARY_SENSOR_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_ZONE): vol.In(ZONES),
        vol.Required(
            CONF_TYPE, default=BinarySensorDeviceClass.DOOR
        ): DEVICE_CLASSES_SCHEMA,
        vol.Optional(CONF_NAME): cv.string,
        vol.Optional(CONF_INVERSE, default=False): cv.boolean,
    }
)
 
SENSOR_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_ZONE): vol.In(ZONES),
        vol.Required(CONF_TYPE, default="dht"): vol.All(
            vol.Lower, vol.In(["dht", "ds18b20"])
        ),
        vol.Optional(CONF_NAME): cv.string,
        vol.Optional(CONF_POLL_INTERVAL, default=3): vol.All(
            vol.Coerce(int), vol.Range(min=1)
        ),
    }
)
 
SWITCH_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_ZONE): vol.In(ZONES),
        vol.Optional(CONF_NAME): cv.string,
        vol.Optional(CONF_ACTIVATION, default=STATE_HIGH): vol.All(
            vol.Lower, vol.In([STATE_HIGH, STATE_LOW])
        ),
        vol.Optional(CONF_MOMENTARY): vol.All(vol.Coerce(int), vol.Range(min=10)),
        vol.Optional(CONF_PAUSE): vol.All(vol.Coerce(int), vol.Range(min=10)),
        vol.Optional(CONF_REPEAT): vol.All(vol.Coerce(int), vol.Range(min=-1)),
    }
)
 
OPTIONS_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_IO): IO_SCHEMA,
        vol.Optional(CONF_BINARY_SENSORS): vol.All(
            cv.ensure_list, [BINARY_SENSOR_SCHEMA]
        ),
        vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [SENSOR_SCHEMA]),
        vol.Optional(CONF_SWITCHES): vol.All(cv.ensure_list, [SWITCH_SCHEMA]),
        vol.Optional(CONF_BLINK, default=True): cv.boolean,
        vol.Optional(CONF_API_HOST, default=""): vol.Any("", cv.url),
        vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
    },
    extra=vol.REMOVE_EXTRA,
)
 
CONFIG_ENTRY_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_ID): cv.matches_regex("[0-9a-f]{12}"),
        vol.Required(CONF_HOST): cv.string,
        vol.Required(CONF_PORT): cv.port,
        vol.Required(CONF_MODEL): vol.Any(*KONN_PANEL_MODEL_NAMES),
        vol.Required(CONF_ACCESS_TOKEN): cv.matches_regex("[a-zA-Z0-9]+"),
        vol.Required(CONF_DEFAULT_OPTIONS): OPTIONS_SCHEMA,
    },
    extra=vol.REMOVE_EXTRA,
)
 
 
class KonnectedFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Konnected Panels."""
 
    VERSION = 1
 
    # class variable to store/share discovered host information
    DISCOVERED_HOSTS: dict[str, dict[str, Any]] = {}
 
    unique_id: str
 
    def __init__(self) -> None:
        """Initialize the Konnected flow."""
        self.data: dict[str, Any] = {}
        self.optionsoptions = OPTIONS_SCHEMA({CONF_IO: {}})
 
    async def async_gen_config(self, host, port):
        """Populate self.data based on panel status.
 
        This will raise CannotConnect if an error occurs
        """
        self.data[CONF_HOST] = host
        self.data[CONF_PORT] = port
        try:
            status = await get_status(self.hass, host, port)
            self.data[CONF_ID] = status.get("chipId", status["mac"].replace(":", ""))
        except (CannotConnect, KeyError) as err:
            raise CannotConnect from err
 
        self.data[CONF_MODEL] = status.get("model", KONN_MODEL)
        self.data[CONF_ACCESS_TOKEN] = "".join(
            random.choices(f"{string.ascii_uppercase}{string.digits}", k=20)
        )
 
    async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
        """Import a configuration.yaml config.
 
        This flow is triggered by `async_setup` for configured panels.
        """
        _LOGGER.debug(import_data)
 
        # save the data and confirm connection via user step
        await self.async_set_unique_idasync_set_unique_id(import_data["id"])
        self.optionsoptions = import_data[CONF_DEFAULT_OPTIONS]
 
        # config schema ensures we have port if we have host
        if import_data.get(CONF_HOST):
            # automatically connect if we have host info
            return await self.async_step_userasync_step_userasync_step_user(
                user_input={
                    CONF_HOST: import_data[CONF_HOST],
                    CONF_PORT: import_data[CONF_PORT],
                }
            )
 
        # if we have no host info wait for it or abort if previously configured
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
        return await self.async_step_import_confirmasync_step_import_confirm()
 
    async def async_step_import_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Confirm the user wants to import the config entry."""
        if user_input is None:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="import_confirm",
                description_placeholders={"id": self.unique_idunique_id},
            )
 
        # if we have ssdp discovered applicable host info use it
        if KonnectedFlowHandler.DISCOVERED_HOSTS.get(self.unique_idunique_id):
            return await self.async_step_userasync_step_userasync_step_user(
                user_input={
                    CONF_HOST: KonnectedFlowHandler.DISCOVERED_HOSTS[self.unique_idunique_id][
                        CONF_HOST
                    ],
                    CONF_PORT: KonnectedFlowHandler.DISCOVERED_HOSTS[self.unique_idunique_id][
                        CONF_PORT
                    ],
                }
            )
        return await self.async_step_userasync_step_userasync_step_user()
 
    async def async_step_ssdp(
        self, discovery_info: ssdp.SsdpServiceInfo
    ) -> ConfigFlowResult:
        """Handle a discovered konnected panel.
 
        This flow is triggered by the SSDP component. It will check if the
        device is already configured and attempt to finish the config if not.
        """
        _LOGGER.debug(discovery_info)
 
        try:
            if discovery_info.upnp[ssdp.ATTR_UPNP_MANUFACTURER] != KONN_MANUFACTURER:
                return self.async_abortasync_abortasync_abort(reason="not_konn_panel")
 
            if not any(
                name in discovery_info.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
                for name in KONN_PANEL_MODEL_NAMES
            ):
                _LOGGER.warning(
                    "Discovered unrecognized Konnected device %s",
                    discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME, "Unknown"),
                )
                return self.async_abortasync_abortasync_abort(reason="not_konn_panel")
 
        # If MAC is missing it is a bug in the device fw but we'll guard
        # against it since the field is so vital
        except KeyError:
            _LOGGER.error("Malformed Konnected SSDP info")
        else:
            # extract host/port from ssdp_location
            assert discovery_info.ssdp_location
            netloc = urlparse(discovery_info.ssdp_location).netloc.split(":")
            self._async_abort_entries_match_async_abort_entries_match(
                {CONF_HOST: netloc[0], CONF_PORT: int(netloc[1])}
            )
 
            try:
                status = await get_status(self.hass, netloc[0], int(netloc[1]))
            except CannotConnect:
                return self.async_abortasync_abortasync_abort(reason="cannot_connect")
 
            self.data[CONF_HOST] = netloc[0]
            self.data[CONF_PORT] = int(netloc[1])
            self.data[CONF_ID] = status.get("chipId", status["mac"].replace(":", ""))
            self.data[CONF_MODEL] = status.get("model", KONN_MODEL)
 
            KonnectedFlowHandler.DISCOVERED_HOSTS[self.data[CONF_ID]] = {
                CONF_HOST: self.data[CONF_HOST],
                CONF_PORT: self.data[CONF_PORT],
            }
            return await self.async_step_confirmasync_step_confirm()
 
        return self.async_abortasync_abortasync_abort(reason="unknown")
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Connect to panel and get config."""
        errors = {}
        if user_input:
            # build config info and wait for user confirmation
            self.data[CONF_HOST] = user_input[CONF_HOST]
            self.data[CONF_PORT] = user_input[CONF_PORT]
 
            # brief delay to allow processing of recent status req
            await asyncio.sleep(0.1)
            try:
                status = await get_status(
                    self.hass, self.data[CONF_HOST], self.data[CONF_PORT]
                )
            except CannotConnect:
                errors["base"] = "cannot_connect"
            else:
                self.data[CONF_ID] = status.get(
                    "chipId", status["mac"].replace(":", "")
                )
                self.data[CONF_MODEL] = status.get("model", KONN_MODEL)
 
                # save off our discovered host info
                KonnectedFlowHandler.DISCOVERED_HOSTS[self.data[CONF_ID]] = {
                    CONF_HOST: self.data[CONF_HOST],
                    CONF_PORT: self.data[CONF_PORT],
                }
                return await self.async_step_confirmasync_step_confirm()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            description_placeholders={
                "host": self.data.get(CONF_HOST, "Unknown"),
                "port": self.data.get(CONF_PORT, "Unknown"),
            },
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_HOST, default=self.data.get(CONF_HOST)): str,
                    vol.Required(CONF_PORT, default=self.data.get(CONF_PORT)): int,
                }
            ),
            errors=errors,
        )
 
    async def async_step_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Attempt to link with the Konnected panel.
 
        Given a configured host, will ask the user to confirm and finalize
        the connection.
        """
        if user_input is None:
            # abort and update an existing config entry if host info changes
            await self.async_set_unique_idasync_set_unique_id(self.data[CONF_ID])
            self._abort_if_unique_id_configured_abort_if_unique_id_configured(
                updates=self.data, reload_on_update=False
            )
            return self.async_show_formasync_show_formasync_show_form(
                step_id="confirm",
                description_placeholders={
                    "model": KONN_PANEL_MODEL_NAMES[self.data[CONF_MODEL]],
                    "id": self.unique_idunique_id,
                    "host": self.data[CONF_HOST],
                    "port": self.data[CONF_PORT],
                },
            )
 
        # Create access token, attach default options and create entry
        self.data[CONF_DEFAULT_OPTIONS] = self.optionsoptions
        self.data[CONF_ACCESS_TOKEN] = self.hass.data.get(DOMAIN, {}).get(
            CONF_ACCESS_TOKEN
        ) or "".join(random.choices(f"{string.ascii_uppercase}{string.digits}", k=20))
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=KONN_PANEL_MODEL_NAMES[self.data[CONF_MODEL]],
            data=self.data,
        )
 
    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> OptionsFlowHandler:
        """Return the Options Flow."""
        return OptionsFlowHandler(config_entry)
 
 
class OptionsFlowHandler(OptionsFlow):
    """Handle a option flow for a Konnected Panel."""
 
    def __init__(self, config_entry: ConfigEntry) -> None:
        """Initialize options flow."""
        self.modelmodel = config_entry.data[CONF_MODEL]
        self.current_optcurrent_opt = (
            config_entry.options or config_entry.data[CONF_DEFAULT_OPTIONS]
        )
 
        # as config proceeds we'll build up new options and then replace what's in the config entry
        self.new_opt: dict[str, Any] = {CONF_IO: {}}
        self.active_cfgactive_cfg: str | None = None
        self.io_cfgio_cfg: dict[str, Any] = {}
        self.current_statescurrent_states: list[dict[str, Any]] = []
        self.current_statecurrent_state = 1
 
    @callback
    def get_current_cfg(self, io_type, zone):
        """Get the current zone config."""
        return next(
            (
                cfg
                for cfg in self.current_optcurrent_opt.get(io_type, [])
                if cfg[CONF_ZONE] == zone
            ),
            {},
        )
 
    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle options flow."""
        return await self.async_step_options_ioasync_step_options_io()
 
    async def async_step_options_io(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Configure legacy panel IO or first half of pro IO."""
        errors: dict[str, str] = {}
        current_io = self.current_optcurrent_opt.get(CONF_IO, {})
 
        if user_input is not None:
            # strip out disabled io and save for options cfg
            for key, value in user_input.items():
                if value != CONF_IO_DIS:
                    self.new_opt[CONF_IO][key] = value
            return await self.async_step_options_io_extasync_step_options_io_ext()
 
        if self.modelmodel == KONN_MODEL:
            return self.async_show_formasync_show_form(
                step_id="options_io",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            "1", default=current_io.get("1", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "2", default=current_io.get("2", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "3", default=current_io.get("3", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "4", default=current_io.get("4", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "5", default=current_io.get("5", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "6", default=current_io.get("6", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "out", default=current_io.get("out", CONF_IO_DIS)
                        ): OPTIONS_IO_OUTPUT_ONLY,
                    }
                ),
                description_placeholders={
                    "model": KONN_PANEL_MODEL_NAMES[self.modelmodel],
                    "host": self.config_entryconfig_entryconfig_entry.data[CONF_HOST],
                },
                errors=errors,
            )
 
        # configure the first half of the pro board io
        if self.modelmodel == KONN_MODEL_PRO:
            return self.async_show_formasync_show_form(
                step_id="options_io",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            "1", default=current_io.get("1", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "2", default=current_io.get("2", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "3", default=current_io.get("3", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "4", default=current_io.get("4", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "5", default=current_io.get("5", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "6", default=current_io.get("6", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "7", default=current_io.get("7", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                    }
                ),
                description_placeholders={
                    "model": KONN_PANEL_MODEL_NAMES[self.modelmodel],
                    "host": self.config_entryconfig_entryconfig_entry.data[CONF_HOST],
                },
                errors=errors,
            )
 
        return self.async_abortasync_abort(reason="not_konn_panel")
 
    async def async_step_options_io_ext(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to configure the extended IO for pro."""
        errors: dict[str, str] = {}
        current_io = self.current_optcurrent_opt.get(CONF_IO, {})
 
        if user_input is not None:
            # strip out disabled io and save for options cfg
            for key, value in user_input.items():
                if value != CONF_IO_DIS:
                    self.new_opt[CONF_IO].update({key: value})
            self.io_cfgio_cfg = copy.deepcopy(self.new_opt[CONF_IO])
            return await self.async_step_options_binaryasync_step_options_binary()
 
        if self.modelmodel == KONN_MODEL:
            self.io_cfgio_cfg = copy.deepcopy(self.new_opt[CONF_IO])
            return await self.async_step_options_binaryasync_step_options_binary()
 
        if self.modelmodel == KONN_MODEL_PRO:
            return self.async_show_formasync_show_form(
                step_id="options_io_ext",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            "8", default=current_io.get("8", CONF_IO_DIS)
                        ): OPTIONS_IO_ANY,
                        vol.Required(
                            "9", default=current_io.get("9", CONF_IO_DIS)
                        ): OPTIONS_IO_INPUT_ONLY,
                        vol.Required(
                            "10", default=current_io.get("10", CONF_IO_DIS)
                        ): OPTIONS_IO_INPUT_ONLY,
                        vol.Required(
                            "11", default=current_io.get("11", CONF_IO_DIS)
                        ): OPTIONS_IO_INPUT_ONLY,
                        vol.Required(
                            "12", default=current_io.get("12", CONF_IO_DIS)
                        ): OPTIONS_IO_INPUT_ONLY,
                        vol.Required(
                            "alarm1", default=current_io.get("alarm1", CONF_IO_DIS)
                        ): OPTIONS_IO_OUTPUT_ONLY,
                        vol.Required(
                            "out1", default=current_io.get("out1", CONF_IO_DIS)
                        ): OPTIONS_IO_OUTPUT_ONLY,
                        vol.Required(
                            "alarm2_out2",
                            default=current_io.get("alarm2_out2", CONF_IO_DIS),
                        ): OPTIONS_IO_OUTPUT_ONLY,
                    }
                ),
                description_placeholders={
                    "model": KONN_PANEL_MODEL_NAMES[self.modelmodel],
                    "host": self.config_entryconfig_entryconfig_entry.data[CONF_HOST],
                },
                errors=errors,
            )
 
        return self.async_abortasync_abort(reason="not_konn_panel")
 
    async def async_step_options_binary(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to configure the IO options for binary sensors."""
        errors: dict[str, str] = {}
        if user_input is not None and self.active_cfgactive_cfg is not None:
            zone = {"zone": self.active_cfgactive_cfg}
            zone.update(user_input)
            self.new_opt[CONF_BINARY_SENSORS] = [
                *self.new_opt.get(CONF_BINARY_SENSORS, []),
                zone,
            ]
            self.io_cfgio_cfg.pop(self.active_cfgactive_cfg)
            self.active_cfgactive_cfg = None
 
        if self.active_cfgactive_cfg:
            current_cfg = self.get_current_cfgget_current_cfg(CONF_BINARY_SENSORS, self.active_cfgactive_cfg)
            return self.async_show_formasync_show_form(
                step_id="options_binary",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            CONF_TYPE,
                            default=current_cfg.get(
                                CONF_TYPE, BinarySensorDeviceClass.DOOR
                            ),
                        ): DEVICE_CLASSES_SCHEMA,
                        vol.Optional(
                            CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
                        ): str,
                        vol.Optional(
                            CONF_INVERSE, default=current_cfg.get(CONF_INVERSE, False)
                        ): bool,
                    }
                ),
                description_placeholders={
                    "zone": f"Zone {self.active_cfg}"
                    if len(self.active_cfgactive_cfg) < 3
                    else self.active_cfgactive_cfg.upper()
                },
                errors=errors,
            )
 
        # find the next unconfigured binary sensor
        for key, value in self.io_cfgio_cfg.items():
            if value == CONF_IO_BIN:
                self.active_cfgactive_cfg = key
                current_cfg = self.get_current_cfgget_current_cfg(CONF_BINARY_SENSORS, self.active_cfgactive_cfg)
                return self.async_show_formasync_show_form(
                    step_id="options_binary",
                    data_schema=vol.Schema(
                        {
                            vol.Required(
                                CONF_TYPE,
                                default=current_cfg.get(
                                    CONF_TYPE, BinarySensorDeviceClass.DOOR
                                ),
                            ): DEVICE_CLASSES_SCHEMA,
                            vol.Optional(
                                CONF_NAME,
                                default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
                            ): str,
                            vol.Optional(
                                CONF_INVERSE,
                                default=current_cfg.get(CONF_INVERSE, False),
                            ): bool,
                        }
                    ),
                    description_placeholders={
                        "zone": f"Zone {self.active_cfg}"
                        if len(self.active_cfgactive_cfg) < 3
                        else self.active_cfgactive_cfg.upper()
                    },
                    errors=errors,
                )
 
        return await self.async_step_options_digitalasync_step_options_digital()
 
    async def async_step_options_digital(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to configure the IO options for digital sensors."""
        errors: dict[str, str] = {}
        if user_input is not None and self.active_cfgactive_cfg is not None:
            zone = {"zone": self.active_cfgactive_cfg}
            zone.update(user_input)
            self.new_opt[CONF_SENSORS] = [*self.new_opt.get(CONF_SENSORS, []), zone]
            self.io_cfgio_cfg.pop(self.active_cfgactive_cfg)
            self.active_cfgactive_cfg = None
 
        if self.active_cfgactive_cfg:
            current_cfg = self.get_current_cfgget_current_cfg(CONF_SENSORS, self.active_cfgactive_cfg)
            return self.async_show_formasync_show_form(
                step_id="options_digital",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            CONF_TYPE, default=current_cfg.get(CONF_TYPE, "dht")
                        ): vol.All(vol.Lower, vol.In(["dht", "ds18b20"])),
                        vol.Optional(
                            CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
                        ): str,
                        vol.Optional(
                            CONF_POLL_INTERVAL,
                            default=current_cfg.get(CONF_POLL_INTERVAL, 3),
                        ): vol.All(vol.Coerce(int), vol.Range(min=1)),
                    }
                ),
                description_placeholders={
                    "zone": f"Zone {self.active_cfg}"
                    if len(self.active_cfgactive_cfg) < 3
                    else self.active_cfgactive_cfg.upper()
                },
                errors=errors,
            )
 
        # find the next unconfigured digital sensor
        for key, value in self.io_cfgio_cfg.items():
            if value == CONF_IO_DIG:
                self.active_cfgactive_cfg = key
                current_cfg = self.get_current_cfgget_current_cfg(CONF_SENSORS, self.active_cfgactive_cfg)
                return self.async_show_formasync_show_form(
                    step_id="options_digital",
                    data_schema=vol.Schema(
                        {
                            vol.Required(
                                CONF_TYPE, default=current_cfg.get(CONF_TYPE, "dht")
                            ): vol.All(vol.Lower, vol.In(["dht", "ds18b20"])),
                            vol.Optional(
                                CONF_NAME,
                                default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
                            ): str,
                            vol.Optional(
                                CONF_POLL_INTERVAL,
                                default=current_cfg.get(CONF_POLL_INTERVAL, 3),
                            ): vol.All(vol.Coerce(int), vol.Range(min=1)),
                        }
                    ),
                    description_placeholders={
                        "zone": f"Zone {self.active_cfg}"
                        if len(self.active_cfgactive_cfg) < 3
                        else self.active_cfgactive_cfg.upper()
                    },
                    errors=errors,
                )
 
        return await self.async_step_options_switchasync_step_options_switch()
 
    async def async_step_options_switch(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to configure the IO options for switches."""
        errors: dict[str, str] = {}
        if user_input is not None and self.active_cfgactive_cfg is not None:
            zone = {"zone": self.active_cfgactive_cfg}
            zone.update(user_input)
            del zone[CONF_MORE_STATES]
            self.new_opt[CONF_SWITCHES] = [*self.new_opt.get(CONF_SWITCHES, []), zone]
 
            # iterate through multiple switch states
            if self.current_statescurrent_states:
                self.current_statescurrent_states.pop(0)
 
            # only go to next zone if all states are entered
            self.current_statecurrent_state += 1
            if user_input[CONF_MORE_STATES] == CONF_NO:
                self.io_cfgio_cfg.pop(self.active_cfgactive_cfg)
                self.active_cfgactive_cfg = None
 
        if self.active_cfgactive_cfg:
            current_cfg = next(iter(self.current_statescurrent_states), {})
            return self.async_show_formasync_show_form(
                step_id="options_switch",
                data_schema=vol.Schema(
                    {
                        vol.Optional(
                            CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
                        ): str,
                        vol.Optional(
                            CONF_ACTIVATION,
                            default=current_cfg.get(CONF_ACTIVATION, STATE_HIGH),
                        ): vol.All(vol.Lower, vol.In([STATE_HIGH, STATE_LOW])),
                        vol.Optional(
                            CONF_MOMENTARY,
                            default=current_cfg.get(CONF_MOMENTARY, vol.UNDEFINED),
                        ): vol.All(vol.Coerce(int), vol.Range(min=10)),
                        vol.Optional(
                            CONF_PAUSE,
                            default=current_cfg.get(CONF_PAUSE, vol.UNDEFINED),
                        ): vol.All(vol.Coerce(int), vol.Range(min=10)),
                        vol.Optional(
                            CONF_REPEAT,
                            default=current_cfg.get(CONF_REPEAT, vol.UNDEFINED),
                        ): vol.All(vol.Coerce(int), vol.Range(min=-1)),
                        vol.Required(
                            CONF_MORE_STATES,
                            default=CONF_YES
                            if len(self.current_statescurrent_states) > 1
                            else CONF_NO,
                        ): vol.In([CONF_YES, CONF_NO]),
                    }
                ),
                description_placeholders={
                    "zone": f"Zone {self.active_cfg}"
                    if len(self.active_cfgactive_cfg) < 3
                    else self.active_cfgactive_cfg.upper(),
                    "state": str(self.current_statecurrent_state),
                },
                errors=errors,
            )
 
        # find the next unconfigured switch
        for key, value in self.io_cfgio_cfg.items():
            if value == CONF_IO_SWI:
                self.active_cfgactive_cfg = key
                self.current_statescurrent_states = [
                    cfg
                    for cfg in self.current_optcurrent_opt.get(CONF_SWITCHES, [])
                    if cfg[CONF_ZONE] == self.active_cfgactive_cfg
                ]
                current_cfg = next(iter(self.current_statescurrent_states), {})
                self.current_statecurrent_state = 1
                return self.async_show_formasync_show_form(
                    step_id="options_switch",
                    data_schema=vol.Schema(
                        {
                            vol.Optional(
                                CONF_NAME,
                                default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
                            ): str,
                            vol.Optional(
                                CONF_ACTIVATION,
                                default=current_cfg.get(CONF_ACTIVATION, STATE_HIGH),
                            ): vol.In(["low", "high"]),
                            vol.Optional(
                                CONF_MOMENTARY,
                                default=current_cfg.get(CONF_MOMENTARY, vol.UNDEFINED),
                            ): vol.All(vol.Coerce(int), vol.Range(min=10)),
                            vol.Optional(
                                CONF_PAUSE,
                                default=current_cfg.get(CONF_PAUSE, vol.UNDEFINED),
                            ): vol.All(vol.Coerce(int), vol.Range(min=10)),
                            vol.Optional(
                                CONF_REPEAT,
                                default=current_cfg.get(CONF_REPEAT, vol.UNDEFINED),
                            ): vol.All(vol.Coerce(int), vol.Range(min=-1)),
                            vol.Required(
                                CONF_MORE_STATES,
                                default=CONF_YES
                                if len(self.current_statescurrent_states) > 1
                                else CONF_NO,
                            ): vol.In([CONF_YES, CONF_NO]),
                        }
                    ),
                    description_placeholders={
                        "zone": f"Zone {self.active_cfg}"
                        if len(self.active_cfgactive_cfg) < 3
                        else self.active_cfgactive_cfg.upper(),
                        "state": str(self.current_statecurrent_state),
                    },
                    errors=errors,
                )
 
        return await self.async_step_options_miscasync_step_options_misc()
 
    async def async_step_options_misc(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to configure the LED behavior."""
        errors = {}
        if user_input is not None:
            # config schema only does basic schema val so check url here
            try:
                if user_input[CONF_OVERRIDE_API_HOST]:
                    cv.url(user_input.get(CONF_API_HOST, ""))
                else:
                    user_input[CONF_API_HOST] = ""
            except vol.Invalid:
                errors["base"] = "bad_host"
            else:
                # no need to store the override - can infer
                del user_input[CONF_OVERRIDE_API_HOST]
                self.new_opt.update(user_input)
                return self.async_create_entryasync_create_entry(title="", data=self.new_opt)
 
        return self.async_show_formasync_show_form(
            step_id="options_misc",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_DISCOVERY,
                        default=self.current_optcurrent_opt.get(CONF_DISCOVERY, True),
                    ): bool,
                    vol.Required(
                        CONF_BLINK, default=self.current_optcurrent_opt.get(CONF_BLINK, True)
                    ): bool,
                    vol.Required(
                        CONF_OVERRIDE_API_HOST,
                        default=bool(self.current_optcurrent_opt.get(CONF_API_HOST)),
                    ): bool,
                    vol.Optional(
                        CONF_API_HOST, default=self.current_optcurrent_opt.get(CONF_API_HOST, "")
                    ): str,
                }
            ),
            errors=errors,
        )