config_flow.py

Go to the documentation of this file.

"""Config flows for the ENOcean integration."""
 
from typing import Any
 
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_DEVICE
 
from . import dongle
from .const import DOMAIN, ERROR_INVALID_DONGLE_PATH, LOGGER
 
 
class EnOceanFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle the enOcean config flows."""
 
    VERSION = 1
    MANUAL_PATH_VALUE = "Custom path"
 
    def __init__(self) -> None:
        """Initialize the EnOcean config flow."""
        self.dongle_pathdongle_path = None
        self.discovery_infodiscovery_info = None
 
    async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
        """Import a yaml configuration."""
 
        if not await self.validate_enocean_confvalidate_enocean_conf(import_data):
            LOGGER.warning(
                "Cannot import yaml configuration: %s is not a valid dongle path",
                import_data[CONF_DEVICE],
            )
            return self.async_abortasync_abortasync_abort(reason="invalid_dongle_path")
 
        return self.create_enocean_entrycreate_enocean_entry(import_data)
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle an EnOcean config flow start."""
        return await self.async_step_detectasync_step_detect()
 
    async def async_step_detect(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Propose a list of detected dongles."""
        errors = {}
        if user_input is not None:
            if user_input[CONF_DEVICE] == self.MANUAL_PATH_VALUEMANUAL_PATH_VALUE:
                return await self.async_step_manualasync_step_manual()
            if await self.validate_enocean_confvalidate_enocean_conf(user_input):
                return self.create_enocean_entrycreate_enocean_entry(user_input)
            errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH}
 
        bridges = await self.hass.async_add_executor_job(dongle.detect)
        if len(bridges) == 0:
            return await self.async_step_manualasync_step_manual(user_input)
 
        bridges.append(self.MANUAL_PATH_VALUEMANUAL_PATH_VALUE)
        return self.async_show_formasync_show_formasync_show_form(
            step_id="detect",
            data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(bridges)}),
            errors=errors,
        )
 
    async def async_step_manual(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Request manual USB dongle path."""
        default_value = None
        errors = {}
        if user_input is not None:
            if await self.validate_enocean_confvalidate_enocean_conf(user_input):
                return self.create_enocean_entrycreate_enocean_entry(user_input)
            default_value = user_input[CONF_DEVICE]
            errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH}
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="manual",
            data_schema=vol.Schema(
                {vol.Required(CONF_DEVICE, default=default_value): str}
            ),
            errors=errors,
        )
 
    async def validate_enocean_conf(self, user_input) -> bool:
        """Return True if the user_input contains a valid dongle path."""
        dongle_path = user_input[CONF_DEVICE]
        return await self.hass.async_add_executor_job(dongle.validate_path, dongle_path)
 
    def create_enocean_entry(self, user_input):
        """Create an entry for the provided configuration."""
        return self.async_create_entryasync_create_entryasync_create_entry(title="EnOcean", data=user_input)