config_flow.py

Go to the documentation of this file.

"""Config flow for Sense integration."""
 
from collections.abc import Mapping
from functools import partial
import logging
from typing import Any
 
from sense_energy import (
    ASyncSenseable,
    SenseAuthenticationException,
    SenseMFARequiredException,
)
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_PASSWORD, CONF_TIMEOUT
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import ACTIVE_UPDATE_RATE, DEFAULT_TIMEOUT, DOMAIN, SENSE_CONNECT_EXCEPTIONS
 
_LOGGER = logging.getLogger(__name__)
 
DATA_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_EMAIL): str,
        vol.Required(CONF_PASSWORD): str,
        vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): vol.Coerce(int),
    }
)
 
 
class SenseConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Sense."""
 
    VERSION = 1
 
    _gateway: ASyncSenseable
 
    def __init__(self) -> None:
        """Init Config ."""
        self._auth_data_auth_data: dict[str, Any] = {}
 
    async def validate_input(self, data: Mapping[str, Any]) -> None:
        """Validate the user input allows us to connect.
 
        Data has the keys from DATA_SCHEMA with values provided by the user.
        """
        self._auth_data_auth_data.update(dict(data))
        timeout = self._auth_data_auth_data[CONF_TIMEOUT]
        client_session = async_get_clientsession(self.hass)
 
        # Creating the AsyncSenseable object loads
        # ssl certificates which does blocking IO
        self._gateway_gateway = await self.hass.async_add_executor_job(
            partial(
                ASyncSenseable,
                api_timeout=timeout,
                wss_timeout=timeout,
                client_session=client_session,
            )
        )
        self._gateway_gateway.rate_limit = ACTIVE_UPDATE_RATE
        await self._gateway_gateway.authenticate(
            self._auth_data_auth_data[CONF_EMAIL], self._auth_data_auth_data[CONF_PASSWORD]
        )
 
    async def create_entry_from_data(self) -> ConfigFlowResult:
        """Create the entry from the config data."""
        self._auth_data_auth_data["access_token"] = self._gateway_gateway.sense_access_token
        self._auth_data_auth_data["user_id"] = self._gateway_gateway.sense_user_id
        self._auth_data_auth_data["device_id"] = self._gateway_gateway.device_id
        self._auth_data_auth_data["refresh_token"] = self._gateway_gateway.refresh_token
        self._auth_data_auth_data["monitor_id"] = self._gateway_gateway.sense_monitor_id
        existing_entry = await self.async_set_unique_idasync_set_unique_id(self._auth_data_auth_data[CONF_EMAIL])
        if not existing_entry:
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=self._auth_data_auth_data[CONF_EMAIL], data=self._auth_data_auth_data
            )
 
        return self.async_update_reload_and_abortasync_update_reload_and_abort(existing_entry, data=self._auth_data_auth_data)
 
    async def validate_input_and_create_entry(
        self, user_input: Mapping[str, Any], errors: dict[str, str]
    ) -> ConfigFlowResult | None:
        """Validate the input and create the entry from the data."""
        try:
            await self.validate_inputvalidate_input(user_input)
        except SenseMFARequiredException:
            return await self.async_step_validationasync_step_validation()
        except SENSE_CONNECT_EXCEPTIONS:
            errors["base"] = "cannot_connect"
        except SenseAuthenticationException:
            errors["base"] = "invalid_auth"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            errors["base"] = "unknown"
        else:
            return await self.create_entry_from_datacreate_entry_from_data()
        return None
 
    async def async_step_validation(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle validation (2fa) step."""
        errors = {}
        if user_input:
            try:
                await self._gateway_gateway.validate_mfa(user_input[CONF_CODE])
            except SENSE_CONNECT_EXCEPTIONS:
                errors["base"] = "cannot_connect"
            except SenseAuthenticationException:
                errors["base"] = "invalid_auth"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                return await self.create_entry_from_datacreate_entry_from_data()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="validation",
            data_schema=vol.Schema({vol.Required(CONF_CODE): vol.All(str, vol.Strip)}),
            errors=errors,
        )
 
    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:
            if result := await self.validate_input_and_create_entryvalidate_input_and_create_entry(user_input, errors):
                return result
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=DATA_SCHEMA, errors=errors
        )
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Handle configuration by re-auth."""
        self._auth_data_auth_data = dict(entry_data)
        return await self.async_step_reauth_validateasync_step_reauth_validate(entry_data)
 
    async def async_step_reauth_validate(
        self, user_input: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Handle reauth and validation."""
        errors: dict[str, str] = {}
        if user_input is not None:
            if result := await self.validate_input_and_create_entryvalidate_input_and_create_entry(user_input, errors):
                return result
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="reauth_validate",
            data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
            errors=errors,
            description_placeholders={
                CONF_EMAIL: self._auth_data_auth_data[CONF_EMAIL],
            },
        )