109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""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 homeassistant.helpers.selector import (
|
|
SelectSelector,
|
|
SelectSelectorConfig,
|
|
SelectSelectorMode,
|
|
)
|
|
|
|
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 = "manual"
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the EnOcean config flow."""
|
|
self.dongle_path = None
|
|
self.discovery_info = None
|
|
|
|
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
|
|
"""Import a yaml configuration."""
|
|
|
|
if not await self.validate_enocean_conf(import_data):
|
|
LOGGER.warning(
|
|
"Cannot import yaml configuration: %s is not a valid dongle path",
|
|
import_data[CONF_DEVICE],
|
|
)
|
|
return self.async_abort(reason="invalid_dongle_path")
|
|
|
|
return self.create_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_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_VALUE:
|
|
return await self.async_step_manual()
|
|
if await self.validate_enocean_conf(user_input):
|
|
return self.create_enocean_entry(user_input)
|
|
errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH}
|
|
|
|
devices = await self.hass.async_add_executor_job(dongle.detect)
|
|
if len(devices) == 0:
|
|
return await self.async_step_manual(user_input)
|
|
devices.append(self.MANUAL_PATH_VALUE)
|
|
|
|
return self.async_show_form(
|
|
step_id="detect",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Required(CONF_DEVICE): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=devices,
|
|
translation_key="devices",
|
|
mode=SelectSelectorMode.LIST,
|
|
)
|
|
)
|
|
}
|
|
),
|
|
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_conf(user_input):
|
|
return self.create_enocean_entry(user_input)
|
|
default_value = user_input[CONF_DEVICE]
|
|
errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH}
|
|
|
|
return self.async_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_entry(title="EnOcean", data=user_input)
|