config_flow.py

Go to the documentation of this file.

"""Config flow for Somfy MyLink integration."""
 
from __future__ import annotations
 
from copy import deepcopy
import logging
from typing import Any
 
from somfy_mylink_synergy import SomfyMyLinkSynergy
import voluptuous as vol
 
from homeassistant.components import dhcp
from homeassistant.config_entries import (
    ConfigEntry,
    ConfigEntryState,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import format_mac
 
from .const import (
    CONF_REVERSE,
    CONF_REVERSED_TARGET_IDS,
    CONF_SYSTEM_ID,
    CONF_TARGET_ID,
    CONF_TARGET_NAME,
    DEFAULT_PORT,
    DOMAIN,
    MYLINK_STATUS,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
async def validate_input(hass: HomeAssistant, data):
    """Validate the user input allows us to connect.
 
    Data has the keys from schema with values provided by the user.
    """
    somfy_mylink = SomfyMyLinkSynergy(
        data[CONF_SYSTEM_ID], data[CONF_HOST], data[CONF_PORT]
    )
 
    try:
        status_info = await somfy_mylink.status_info()
    except TimeoutError as ex:
        raise CannotConnect from ex
 
    if not status_info or "error" in status_info:
        _LOGGER.debug("Auth error: %s", status_info)
        raise InvalidAuth
 
    return {"title": f"MyLink {data[CONF_HOST]}"}
 
 
class SomfyConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Somfy MyLink."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize the somfy_mylink flow."""
        self.hosthost: str | None = None
        self.macmac: str | None = None
        self.ip_addressip_address: str | None = None
 
    async def async_step_dhcp(
        self, discovery_info: dhcp.DhcpServiceInfo
    ) -> ConfigFlowResult:
        """Handle dhcp discovery."""
        self._async_abort_entries_match_async_abort_entries_match({CONF_HOST: discovery_info.ip})
 
        formatted_mac = format_mac(discovery_info.macaddress)
        await self.async_set_unique_idasync_set_unique_id(format_mac(formatted_mac))
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.ip})
        self.hosthost = discovery_info.hostname
        self.macmac = formatted_mac
        self.ip_addressip_address = discovery_info.ip
        self.context["title_placeholders"] = {"ip": self.ip_addressip_address, "mac": self.macmac}
        return await self.async_step_userasync_step_userasync_step_user()
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        errors = {}
 
        if user_input is not None:
            self._async_abort_entries_match_async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
 
            try:
                info = await validate_input(self.hass, user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                return self.async_create_entryasync_create_entryasync_create_entry(title=info["title"], data=user_input)
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_HOST, default=self.ip_addressip_address): str,
                    vol.Required(CONF_SYSTEM_ID): str,
                    vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
                }
            ),
            errors=errors,
        )
 
    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> OptionsFlowHandler:
        """Get the options flow for this handler."""
        return OptionsFlowHandler(config_entry)
 
 
class OptionsFlowHandler(OptionsFlow):
    """Handle a option flow for somfy_mylink."""
 
    def __init__(self, config_entry: ConfigEntry) -> None:
        """Initialize options flow."""
        self.optionsoptions = deepcopy(dict(config_entry.options))
        self._target_id_target_id: str | None = None
 
    @callback
    def _async_callback_targets(self):
        """Return the list of targets."""
        return self.hass.data[DOMAIN][self.config_entryconfig_entryconfig_entry.entry_id][MYLINK_STATUS][
            "result"
        ]
 
    @callback
    def _async_get_target_name(self, target_id) -> str:
        """Find the name of a target in the api data."""
        mylink_targets = self._async_callback_targets_async_callback_targets()
        for cover in mylink_targets:
            if cover["targetID"] == target_id:
                return cover["name"]
        raise KeyError
 
    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle options flow."""
 
        if self.config_entryconfig_entryconfig_entry.state is not ConfigEntryState.LOADED:
            _LOGGER.error("MyLink must be connected to manage device options")
            return self.async_abortasync_abort(reason="cannot_connect")
 
        if user_input is not None:
            if target_id := user_input.get(CONF_TARGET_ID):
                return await self.async_step_target_configasync_step_target_config(None, target_id)
 
            return self.async_create_entryasync_create_entry(title="", data=self.optionsoptions)
 
        cover_dict = {None: None}
        mylink_targets = self._async_callback_targets_async_callback_targets()
        if mylink_targets:
            for cover in mylink_targets:
                cover_dict[cover["targetID"]] = cover["name"]
 
        data_schema = vol.Schema({vol.Optional(CONF_TARGET_ID): vol.In(cover_dict)})
 
        return self.async_show_formasync_show_form(step_id="init", data_schema=data_schema, errors={})
 
    async def async_step_target_config(
        self, user_input: dict[str, bool] | None = None, target_id: str | None = None
    ) -> ConfigFlowResult:
        """Handle options flow for target."""
        reversed_target_ids: dict[str | None, bool] = self.optionsoptions.setdefault(
            CONF_REVERSED_TARGET_IDS, {}
        )
 
        if user_input is not None:
            if user_input[CONF_REVERSE] != reversed_target_ids.get(self._target_id_target_id):
                reversed_target_ids[self._target_id_target_id] = user_input[CONF_REVERSE]
            return await self.async_step_initasync_step_init()
 
        self._target_id_target_id = target_id
 
        return self.async_show_formasync_show_form(
            step_id="target_config",
            data_schema=vol.Schema(
                {
                    vol.Optional(
                        CONF_REVERSE,
                        default=reversed_target_ids.get(target_id, False),
                    ): bool
                }
            ),
            description_placeholders={
                CONF_TARGET_NAME: self._async_get_target_name_async_get_target_name(target_id),
            },
            errors={},
        )
 
 
class CannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect."""
 
 
class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""