config_flow.py

Go to the documentation of this file.

"""Config flow for Linear Garage Door integration."""
 
from __future__ import annotations
 
from collections.abc import Collection, Mapping, Sequence
import logging
from typing import Any
import uuid
 
from linear_garage_door import Linear
from linear_garage_door.errors import InvalidLoginError
import voluptuous as vol
 
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
STEP_USER_DATA_SCHEMA = {
    vol.Required(CONF_EMAIL): str,
    vol.Required(CONF_PASSWORD): str,
}
 
 
async def validate_input(
    hass: HomeAssistant,
    data: dict[str, str],
) -> dict[str, Sequence[Collection[str]]]:
    """Validate the user input allows us to connect.
 
    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
    """
 
    hub = Linear()
 
    device_id = str(uuid.uuid4())
    try:
        await hub.login(
            data["email"],
            data["password"],
            device_id=device_id,
            client_session=async_get_clientsession(hass),
        )
 
        sites = await hub.get_sites()
    except InvalidLoginError as err:
        raise InvalidAuth from err
    finally:
        await hub.close()
 
    return {
        "email": data["email"],
        "password": data["password"],
        "sites": sites,
        "device_id": device_id,
    }
 
 
class LinearGarageDoorConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Linear Garage Door."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize the config flow."""
        self.datadata: dict[str, Sequence[Collection[str]]] = {}
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        data_schema = vol.Schema(STEP_USER_DATA_SCHEMA)
 
        if user_input is None:
            return self.async_show_formasync_show_formasync_show_form(step_id="user", data_schema=data_schema)
 
        errors = {}
 
        try:
            info = await validate_input(self.hass, user_input)
        except InvalidAuth:
            errors["base"] = "invalid_auth"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            errors["base"] = "unknown"
        else:
            self.datadata = info
 
            # Check if we are reauthenticating
            if self.sourcesourcesourcesource == SOURCE_REAUTH:
                return self.async_update_reload_and_abortasync_update_reload_and_abort(
                    self._get_reauth_entry_get_reauth_entry(),
                    data_updates={
                        CONF_EMAIL: self.datadata["email"],
                        CONF_PASSWORD: self.datadata["password"],
                    },
                )
 
            return await self.async_step_siteasync_step_site()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=data_schema, errors=errors
        )
 
    async def async_step_site(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle the site step."""
 
        if isinstance(self.datadata["sites"], list):
            sites: list[dict[str, str]] = self.datadata["sites"]
 
        if not user_input:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="site",
                data_schema=vol.Schema(
                    {
                        vol.Required("site"): vol.In(
                            {site["id"]: site["name"] for site in sites}
                        )
                    }
                ),
            )
 
        site_id = user_input["site"]
 
        site_name = next(site["name"] for site in sites if site["id"] == site_id)
 
        await self.async_set_unique_idasync_set_unique_id(site_id)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=site_name,
            data={
                "site_id": site_id,
                "email": self.datadata["email"],
                "password": self.datadata["password"],
                "device_id": self.datadata["device_id"],
            },
        )
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Reauth in case of a password change or other error."""
        return await self.async_step_userasync_step_userasync_step_user()
 
 
class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""
 
 
class InvalidDeviceID(HomeAssistantError):
    """Error to indicate there is invalid device ID."""