config_flow.py

Go to the documentation of this file.

"""Config flow for MyPermobil integration."""
 
from __future__ import annotations
 
from collections.abc import Mapping
import logging
from typing import Any
 
from mypermobil import (
    MyPermobil,
    MyPermobilAPIException,
    MyPermobilClientException,
    MyPermobilEulaException,
)
import voluptuous as vol
 
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_REGION, CONF_TOKEN, CONF_TTL
from homeassistant.core import HomeAssistant, async_get_hass
from homeassistant.helpers import selector
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.selector import (
    TextSelector,
    TextSelectorConfig,
    TextSelectorType,
)
 
from .const import APPLICATION, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
GET_EMAIL_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_EMAIL): TextSelector(
            TextSelectorConfig(type=TextSelectorType.EMAIL)
        ),
    }
)
 
GET_TOKEN_SCHEMA = vol.Schema({vol.Required(CONF_CODE): cv.string})
 
 
class PermobilConfigFlow(ConfigFlow, domain=DOMAIN):
    """Permobil config flow."""
 
    VERSION = 1
    region_names: dict[str, str] = {}
    data: dict[str, str] = {}
 
    def __init__(self) -> None:
        """Initialize flow."""
        hass: HomeAssistant = async_get_hass()
        session = async_get_clientsession(hass)
        self.p_apip_api = MyPermobil(APPLICATION, session=session)
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Invoke when a user initiates a flow via the user interface."""
        errors: dict[str, str] = {}
 
        if user_input:
            try:
                self.p_apip_api.set_email(user_input[CONF_EMAIL])
            except MyPermobilClientException:
                _LOGGER.exception("Error validating email")
                errors["base"] = "invalid_email"
 
            self.datadata.update(user_input)
 
            await self.async_set_unique_idasync_set_unique_id(self.datadata[CONF_EMAIL])
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
        if errors or not user_input:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="user", data_schema=GET_EMAIL_SCHEMA, errors=errors
            )
        return await self.async_step_regionasync_step_region()
 
    async def async_step_region(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Invoke when a user initiates a flow via the user interface."""
        errors: dict[str, str] = {}
        if not user_input:
            # fetch the list of regions names and urls from the api
            # for the user to select from.
            try:
                self.region_namesregion_names = await self.p_apip_api.request_region_names()
                _LOGGER.debug(
                    "region names %s",
                    ",".join(list(self.region_namesregion_names.keys())),
                )
            except MyPermobilAPIException:
                _LOGGER.exception("Error requesting regions")
                errors["base"] = "region_fetch_error"
 
        else:
            region_url = self.region_namesregion_names[user_input[CONF_REGION]]
 
            self.datadata[CONF_REGION] = region_url
            self.p_apip_api.set_region(region_url)
            _LOGGER.debug("region %s", self.p_apip_api.region)
            try:
                # tell backend to send code to the users email
                await self.p_apip_api.request_application_code()
            except MyPermobilAPIException:
                _LOGGER.exception("Error requesting code")
                errors["base"] = "code_request_error"
 
        if errors or not user_input:
            # the error could either be that the fetch region did not pass
            # or that the request application code failed
            schema = vol.Schema(
                {
                    vol.Required(CONF_REGION): selector.SelectSelector(
                        selector.SelectSelectorConfig(
                            options=list(self.region_namesregion_names.keys()),
                            mode=selector.SelectSelectorMode.DROPDOWN,
                        )
                    ),
                }
            )
            return self.async_show_formasync_show_formasync_show_form(
                step_id="region", data_schema=schema, errors=errors
            )
 
        return await self.async_step_email_codeasync_step_email_code()
 
    async def async_step_email_code(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Second step in config flow to enter the email code."""
        errors: dict[str, str] = {}
 
        if user_input:
            try:
                self.p_apip_api.set_code(user_input[CONF_CODE])
                self.datadata.update(user_input)
                token, ttl = await self.p_apip_api.request_application_token()
                self.datadata[CONF_TOKEN] = token
                self.datadata[CONF_TTL] = ttl
            except (MyPermobilAPIException, MyPermobilClientException):
                # the code did not pass validation by the api client
                # or the backend returned an error when trying to validate the code
                _LOGGER.exception("Error verifying code")
                errors["base"] = "invalid_code"
            except MyPermobilEulaException:
                # The user has not accepted the EULA
                errors["base"] = "unsigned_eula"
 
        if errors or not user_input:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="email_code",
                data_schema=GET_TOKEN_SCHEMA,
                errors=errors,
                description_placeholders={"app_name": "MyPermobil"},
            )
 
        if self.sourcesourcesourcesource == SOURCE_REAUTH:
            return self.async_update_reload_and_abortasync_update_reload_and_abort(
                self._get_reauth_entry_get_reauth_entry(), title=self.datadata[CONF_EMAIL], data=self.datadata
            )
 
        return self.async_create_entryasync_create_entryasync_create_entry(title=self.datadata[CONF_EMAIL], data=self.datadata)
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Perform reauth upon an API authentication error."""
        try:
            email: str = entry_data[CONF_EMAIL]
            region: str = entry_data[CONF_REGION]
            self.p_apip_api.set_email(email)
            self.p_apip_api.set_region(region)
            self.datadata = {
                CONF_EMAIL: email,
                CONF_REGION: region,
            }
            await self.p_apip_api.request_application_code()
        except MyPermobilAPIException:
            _LOGGER.exception("Error requesting code for reauth")
            return self.async_abortasync_abortasync_abort(reason="unknown")
 
        return await self.async_step_email_codeasync_step_email_code()