config_flow.py

Go to the documentation of this file.

"""Config flow for Hunter Douglas PowerView integration."""
 
from __future__ import annotations
 
import logging
from typing import TYPE_CHECKING, Any, Self
 
import voluptuous as vol
 
from homeassistant.components import dhcp, zeroconf
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_VERSION, CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
 
from .const import DOMAIN, HUB_EXCEPTIONS
from .util import async_connect_hub
 
_LOGGER = logging.getLogger(__name__)
 
HAP_SUFFIX = "._hap._tcp.local."
POWERVIEW_G2_SUFFIX = "._powerview._tcp.local."
POWERVIEW_G3_SUFFIX = "._PowerView-G3._tcp.local."
 
 
async def validate_input(hass: HomeAssistant, hub_address: str) -> dict[str, str]:
    """Validate the user input allows us to connect.
 
    Data has the keys from DATA_SCHEMA with values provided by the user.
    """
    api = await async_connect_hub(hass, hub_address)
    hub = api.hub
    device_info = api.device_info
    if hub.role != "Primary":
        raise UnsupportedDevice(
            f"{hub.name} ({hub.hub_address}) is the {hub.role} Hub. "
            "Only the Primary can manage shades"
        )
 
    _LOGGER.debug("Connection made using api version: %s", hub.api_version)
 
    # Return info that you want to store in the config entry.
    return {
        "title": device_info.name,
        "unique_id": device_info.serial_number,
        CONF_API_VERSION: hub.api_version,
    }
 
 
class PowerviewConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Hunter Douglas PowerView."""
 
    VERSION = 1
    MINOR_VERSION = 2
 
    def __init__(self) -> None:
        """Initialize the powerview config flow."""
        self.powerview_configpowerview_config: dict = {}
        self.discovered_ipdiscovered_ip: str | None = None
        self.discovered_namediscovered_name: str | None = None
        self.data_schema: dict = {vol.Required(CONF_HOST): str}
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        errors: dict[str, str] = {}
 
        if user_input is not None:
            info, error = await self._async_validate_or_error_async_validate_or_error(user_input[CONF_HOST])
 
            if info and not error:
                self.powerview_configpowerview_config = {
                    CONF_HOST: user_input[CONF_HOST],
                    CONF_NAME: info["title"],
                    CONF_API_VERSION: info[CONF_API_VERSION],
                }
                await self.async_set_unique_idasync_set_unique_id(info["unique_id"])
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title=self.powerview_configpowerview_config[CONF_NAME],
                    data={
                        CONF_HOST: self.powerview_configpowerview_config[CONF_HOST],
                        CONF_API_VERSION: self.powerview_configpowerview_config[CONF_API_VERSION],
                    },
                )
 
            if TYPE_CHECKING:
                assert error is not None
            errors["base"] = error
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=vol.Schema(self.data_schema), errors=errors
        )
 
    async def _async_validate_or_error(
        self, host: str
    ) -> tuple[dict[str, str], None] | tuple[None, str]:
        self._async_abort_entries_match_async_abort_entries_match({CONF_HOST: host})
 
        try:
            info = await validate_input(self.hass, host)
        except HUB_EXCEPTIONS:
            return None, "cannot_connect"
        except UnsupportedDevice:
            return None, "unsupported_device"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            return None, "unknown"
 
        return info, None
 
    async def async_step_dhcp(
        self, discovery_info: dhcp.DhcpServiceInfo
    ) -> ConfigFlowResult:
        """Handle DHCP discovery."""
        self.discovered_ipdiscovered_ip = discovery_info.ip
        self.discovered_namediscovered_name = discovery_info.hostname
        return await self.async_step_discovery_confirmasync_step_discovery_confirm()
 
    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle zeroconf discovery."""
        self.discovered_ipdiscovered_ip = discovery_info.host
        name = discovery_info.name.removesuffix(POWERVIEW_G2_SUFFIX)
        name = name.removesuffix(POWERVIEW_G3_SUFFIX)
        self.discovered_namediscovered_name = name
        return await self.async_step_discovery_confirmasync_step_discovery_confirm()
 
    async def async_step_homekit(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle HomeKit discovery."""
        self.discovered_ipdiscovered_ip = discovery_info.host
        name = discovery_info.name.removesuffix(HAP_SUFFIX)
        self.discovered_namediscovered_name = name
        return await self.async_step_discovery_confirmasync_step_discovery_confirm()
 
    async def async_step_discovery_confirm(self) -> ConfigFlowResult:
        """Confirm dhcp or homekit discovery."""
        # If we already have the host configured do
        # not open connections to it if we can avoid it.
        assert self.discovered_ipdiscovered_ip and self.discovered_namediscovered_name is not None
        if self.hass.config_entries.flow.async_has_matching_flow(self):
            return self.async_abortasync_abortasync_abort(reason="already_in_progress")
 
        self._async_abort_entries_match_async_abort_entries_match({CONF_HOST: self.discovered_ipdiscovered_ip})
        info, error = await self._async_validate_or_error_async_validate_or_error(self.discovered_ipdiscovered_ip)
        if error:
            return self.async_abortasync_abortasync_abort(reason=error)
        assert info is not None
 
        api_version = info[CONF_API_VERSION]
        if not self.discovered_namediscovered_name:
            self.discovered_namediscovered_name = f"Powerview Generation {api_version}"
 
        await self.async_set_unique_idasync_set_unique_id(info["unique_id"], raise_on_progress=False)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured({CONF_HOST: self.discovered_ipdiscovered_ip})
 
        self.powerview_configpowerview_config = {
            CONF_HOST: self.discovered_ipdiscovered_ip,
            CONF_NAME: self.discovered_namediscovered_name,
            CONF_API_VERSION: api_version,
        }
        return await self.async_step_linkasync_step_link()
 
    def is_matching(self, other_flow: Self) -> bool:
        """Return True if other_flow is matching this flow."""
        return other_flow.discovered_ip == self.discovered_ipdiscovered_ip
 
    async def async_step_link(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Attempt to link with Powerview."""
        if user_input is not None:
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=self.powerview_configpowerview_config[CONF_NAME],
                data={
                    CONF_HOST: self.powerview_configpowerview_config[CONF_HOST],
                    CONF_API_VERSION: self.powerview_configpowerview_config[CONF_API_VERSION],
                },
            )
 
        self._set_confirm_only_set_confirm_only()
        self.context["title_placeholders"] = self.powerview_configpowerview_config
        return self.async_show_formasync_show_formasync_show_form(
            step_id="link", description_placeholders=self.powerview_configpowerview_config
        )
 
 
class UnsupportedDevice(HomeAssistantError):
    """Error to indicate the device is not supported."""