config_flow.py

Go to the documentation of this file.

"""Config flow for Verisure integration."""
 
from __future__ import annotations
 
from collections.abc import Mapping
from typing import Any
 
from verisure import (
    Error as VerisureError,
    LoginError as VerisureLoginError,
    ResponseError as VerisureResponseError,
    Session as Verisure,
)
import voluptuous as vol
 
from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import callback
from homeassistant.helpers.storage import STORAGE_DIR
 
from .const import (
    CONF_GIID,
    CONF_LOCK_CODE_DIGITS,
    DEFAULT_LOCK_CODE_DIGITS,
    DOMAIN,
    LOGGER,
)
 
 
class VerisureConfigFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Verisure."""
 
    VERSION = 2
 
    email: str
    password: str
    verisure: Verisure
 
    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> VerisureOptionsFlowHandler:
        """Get the options flow for this handler."""
        return VerisureOptionsFlowHandler()
 
    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:
            self.emailemail = user_input[CONF_EMAIL]
            self.passwordpassword = user_input[CONF_PASSWORD]
            self.verisureverisure = Verisure(
                username=self.emailemail,
                password=self.passwordpassword,
                cookie_file_name=self.hass.config.path(
                    STORAGE_DIR, f"verisure_{user_input[CONF_EMAIL]}"
                ),
            )
 
            try:
                await self.hass.async_add_executor_job(self.verisureverisure.login)
            except VerisureLoginError as ex:
                if "Multifactor authentication enabled" in str(ex):
                    try:
                        await self.hass.async_add_executor_job(
                            self.verisureverisure.request_mfa
                        )
                    except (
                        VerisureLoginError,
                        VerisureError,
                        VerisureResponseError,
                    ) as mfa_ex:
                        LOGGER.debug(
                            "Unexpected response from Verisure during MFA set up, %s",
                            mfa_ex,
                        )
                        errors["base"] = "unknown_mfa"
                    else:
                        return await self.async_step_mfaasync_step_mfa()
                else:
                    LOGGER.debug("Could not log in to Verisure, %s", ex)
                    errors["base"] = "invalid_auth"
            except (VerisureError, VerisureResponseError) as ex:
                LOGGER.debug("Unexpected response from Verisure, %s", ex)
                errors["base"] = "unknown"
            else:
                return await self.async_step_installationasync_step_installation()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_EMAIL): str,
                    vol.Required(CONF_PASSWORD): str,
                }
            ),
            errors=errors,
        )
 
    async def async_step_mfa(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle multifactor authentication step."""
        errors: dict[str, str] = {}
 
        if user_input is not None:
            try:
                await self.hass.async_add_executor_job(
                    self.verisureverisure.validate_mfa, user_input[CONF_CODE]
                )
            except VerisureLoginError as ex:
                LOGGER.debug("Could not log in to Verisure, %s", ex)
                errors["base"] = "invalid_auth"
            except (VerisureError, VerisureResponseError) as ex:
                LOGGER.debug("Unexpected response from Verisure, %s", ex)
                errors["base"] = "unknown"
            else:
                return await self.async_step_installationasync_step_installation()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="mfa",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_CODE): vol.All(
                        vol.Coerce(str), vol.Length(min=6, max=6)
                    )
                }
            ),
            errors=errors,
        )
 
    async def async_step_installation(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Select Verisure installation to add."""
        installations_data = await self.hass.async_add_executor_job(
            self.verisureverisure.get_installations
        )
        installations = {
            inst["giid"]: f"{inst['alias']} ({inst['address']['street']})"
            for inst in (
                installations_data.get("data", {})
                .get("account", {})
                .get("installations", [])
            )
        }
 
        if user_input is None:
            if len(installations) != 1:
                return self.async_show_formasync_show_formasync_show_form(
                    step_id="installation",
                    data_schema=vol.Schema(
                        {vol.Required(CONF_GIID): vol.In(installations)}
                    ),
                )
            user_input = {CONF_GIID: list(installations)[0]}
 
        await self.async_set_unique_idasync_set_unique_id(user_input[CONF_GIID])
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=installations[user_input[CONF_GIID]],
            data={
                CONF_EMAIL: self.emailemail,
                CONF_PASSWORD: self.passwordpassword,
                CONF_GIID: user_input[CONF_GIID],
            },
        )
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Handle initiation of re-authentication with Verisure."""
        return await self.async_step_reauth_confirmasync_step_reauth_confirm()
 
    async def async_step_reauth_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle re-authentication with Verisure."""
        errors: dict[str, str] = {}
 
        if user_input is not None:
            self.emailemail = user_input[CONF_EMAIL]
            self.passwordpassword = user_input[CONF_PASSWORD]
 
            self.verisureverisure = Verisure(
                username=self.emailemail,
                password=self.passwordpassword,
                cookie_file_name=self.hass.config.path(
                    STORAGE_DIR, f"verisure_{user_input[CONF_EMAIL]}"
                ),
            )
 
            try:
                await self.hass.async_add_executor_job(self.verisureverisure.login)
            except VerisureLoginError as ex:
                if "Multifactor authentication enabled" in str(ex):
                    try:
                        await self.hass.async_add_executor_job(
                            self.verisureverisure.request_mfa
                        )
                    except (
                        VerisureLoginError,
                        VerisureError,
                        VerisureResponseError,
                    ) as mfa_ex:
                        LOGGER.debug(
                            "Unexpected response from Verisure during MFA set up, %s",
                            mfa_ex,
                        )
                        errors["base"] = "unknown_mfa"
                    else:
                        return await self.async_step_reauth_mfaasync_step_reauth_mfa()
                else:
                    LOGGER.debug("Could not log in to Verisure, %s", ex)
                    errors["base"] = "invalid_auth"
            except (VerisureError, VerisureResponseError) as ex:
                LOGGER.debug("Unexpected response from Verisure, %s", ex)
                errors["base"] = "unknown"
            else:
                return self.async_update_reload_and_abortasync_update_reload_and_abort(
                    self._get_reauth_entry_get_reauth_entry(),
                    data_updates={
                        CONF_EMAIL: user_input[CONF_EMAIL],
                        CONF_PASSWORD: user_input[CONF_PASSWORD],
                    },
                )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="reauth_confirm",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_EMAIL, default=self._get_reauth_entry_get_reauth_entry().data[CONF_EMAIL]
                    ): str,
                    vol.Required(CONF_PASSWORD): str,
                }
            ),
            errors=errors,
        )
 
    async def async_step_reauth_mfa(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle multifactor authentication step during re-authentication."""
        errors: dict[str, str] = {}
 
        if user_input is not None:
            try:
                await self.hass.async_add_executor_job(
                    self.verisureverisure.validate_mfa, user_input[CONF_CODE]
                )
                await self.hass.async_add_executor_job(self.verisureverisure.login)
            except VerisureLoginError as ex:
                LOGGER.debug("Could not log in to Verisure, %s", ex)
                errors["base"] = "invalid_auth"
            except (VerisureError, VerisureResponseError) as ex:
                LOGGER.debug("Unexpected response from Verisure, %s", ex)
                errors["base"] = "unknown"
            else:
                return self.async_update_reload_and_abortasync_update_reload_and_abort(
                    self._get_reauth_entry_get_reauth_entry(),
                    data_updates={
                        CONF_EMAIL: self.emailemail,
                        CONF_PASSWORD: self.passwordpassword,
                    },
                )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="reauth_mfa",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_CODE): vol.All(
                        vol.Coerce(str),
                        vol.Length(min=6, max=6),
                    )
                }
            ),
            errors=errors,
        )
 
 
class VerisureOptionsFlowHandler(OptionsFlow):
    """Handle Verisure options."""
 
    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Manage Verisure options."""
        errors: dict[str, Any] = {}
 
        if user_input is not None:
            return self.async_create_entryasync_create_entry(data=user_input)
 
        return self.async_show_formasync_show_form(
            step_id="init",
            data_schema=vol.Schema(
                {
                    vol.Optional(
                        CONF_LOCK_CODE_DIGITS,
                        description={
                            "suggested_value": self.config_entryconfig_entryconfig_entry.options.get(
                                CONF_LOCK_CODE_DIGITS, DEFAULT_LOCK_CODE_DIGITS
                            )
                        },
                    ): int,
                }
            ),
            errors=errors,
        )