config_flow.py

Go to the documentation of this file.

"""Config flow for Whirlpool Appliances integration."""
 
from __future__ import annotations
 
from collections.abc import Mapping
import logging
from typing import Any
 
from aiohttp import ClientError
import voluptuous as vol
from whirlpool.appliancesmanager import AppliancesManager
from whirlpool.auth import Auth
from whirlpool.backendselector import BackendSelector
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import CONF_BRAND, CONF_BRANDS_MAP, CONF_REGIONS_MAP, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
STEP_USER_DATA_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_USERNAME): str,
        vol.Required(CONF_PASSWORD): str,
        vol.Required(CONF_REGION): vol.In(list(CONF_REGIONS_MAP)),
        vol.Required(CONF_BRAND): vol.In(list(CONF_BRANDS_MAP)),
    }
)
 
REAUTH_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_PASSWORD): str,
        vol.Required(CONF_BRAND): vol.In(list(CONF_BRANDS_MAP)),
    }
)
 
 
async def validate_input(hass: HomeAssistant, data: dict[str, str]) -> dict[str, str]:
    """Validate the user input allows us to connect.
 
    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
    """
    session = async_get_clientsession(hass)
    region = CONF_REGIONS_MAP[data[CONF_REGION]]
    brand = CONF_BRANDS_MAP[data[CONF_BRAND]]
    backend_selector = BackendSelector(brand, region)
    auth = Auth(backend_selector, data[CONF_USERNAME], data[CONF_PASSWORD], session)
    try:
        await auth.do_auth()
    except (TimeoutError, ClientError) as exc:
        raise CannotConnect from exc
 
    if not auth.is_access_token_valid():
        raise InvalidAuth
 
    appliances_manager = AppliancesManager(backend_selector, auth, session)
    await appliances_manager.fetch_appliances()
 
    if not appliances_manager.aircons and not appliances_manager.washer_dryers:
        raise NoAppliances
 
    return {"title": data[CONF_USERNAME]}
 
 
class WhirlpoolConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Whirlpool Sixth Sense."""
 
    VERSION = 1
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Handle re-authentication with Whirlpool Sixth Sense."""
        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:
        """Confirm re-authentication with Whirlpool Sixth Sense."""
        errors: dict[str, str] = {}
 
        if user_input:
            reauth_entry = self._get_reauth_entry_get_reauth_entry()
            password = user_input[CONF_PASSWORD]
            brand = user_input[CONF_BRAND]
            data = {**reauth_entry.data, CONF_PASSWORD: password, CONF_BRAND: brand}
 
            try:
                await validate_input(self.hass, data)
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            except (CannotConnect, TimeoutError):
                errors["base"] = "cannot_connect"
            else:
                return self.async_update_reload_and_abortasync_update_reload_and_abort(reauth_entry, data=data)
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="reauth_confirm",
            data_schema=REAUTH_SCHEMA,
            errors=errors,
            description_placeholders={"name": "Whirlpool"},
        )
 
    async def async_step_user(self, user_input=None) -> ConfigFlowResult:
        """Handle the initial step."""
        if user_input is None:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="user", data_schema=STEP_USER_DATA_SCHEMA
            )
 
        errors = {}
 
        try:
            info = await validate_input(self.hass, user_input)
        except CannotConnect:
            errors["base"] = "cannot_connect"
        except InvalidAuth:
            errors["base"] = "invalid_auth"
        except NoAppliances:
            errors["base"] = "no_appliances"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            errors["base"] = "unknown"
        else:
            await self.async_set_unique_idasync_set_unique_id(
                user_input[CONF_USERNAME].lower(), raise_on_progress=False
            )
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
            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=STEP_USER_DATA_SCHEMA, errors=errors
        )
 
 
class CannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect."""
 
 
class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""
 
 
class NoAppliances(HomeAssistantError):
    """Error to indicate no supported appliances in the user account."""