config_flow.py

Go to the documentation of this file.

"""Config flow to configure ecobee."""
 
from typing import Any
 
from pyecobee import (
    ECOBEE_API_KEY,
    ECOBEE_CONFIG_FILENAME,
    ECOBEE_REFRESH_TOKEN,
    Ecobee,
)
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.json import load_json_object
 
from .const import _LOGGER, CONF_REFRESH_TOKEN, DATA_ECOBEE_CONFIG, DOMAIN
 
 
class EcobeeFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle an ecobee config flow."""
 
    VERSION = 1
 
    _ecobee: Ecobee
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initiated by the user."""
        errors = {}
        stored_api_key = (
            self.hass.data[DATA_ECOBEE_CONFIG].get(CONF_API_KEY)
            if DATA_ECOBEE_CONFIG in self.hass.data
            else ""
        )
 
        if user_input is not None:
            # Use the user-supplied API key to attempt to obtain a PIN from ecobee.
            self._ecobee_ecobee = Ecobee(config={ECOBEE_API_KEY: user_input[CONF_API_KEY]})
 
            if await self.hass.async_add_executor_job(self._ecobee_ecobee.request_pin):
                # We have a PIN; move to the next step of the flow.
                return await self.async_step_authorizeasync_step_authorize()
            errors["base"] = "pin_request_failed"
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {vol.Required(CONF_API_KEY, default=stored_api_key): str}
            ),
            errors=errors,
        )
 
    async def async_step_authorize(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Present the user with the PIN so that the app can be authorized on ecobee.com."""
        errors = {}
 
        if user_input is not None:
            # Attempt to obtain tokens from ecobee and finish the flow.
            if await self.hass.async_add_executor_job(self._ecobee_ecobee.request_tokens):
                # Refresh token obtained; create the config entry.
                config = {
                    CONF_API_KEY: self._ecobee_ecobee.api_key,
                    CONF_REFRESH_TOKEN: self._ecobee_ecobee.refresh_token,
                }
                return self.async_create_entryasync_create_entryasync_create_entry(title=DOMAIN, data=config)
            errors["base"] = "token_request_failed"
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="authorize",
            errors=errors,
            description_placeholders={"pin": self._ecobee_ecobee.pin},
        )
 
    async def async_step_import(self, import_data: None) -> ConfigFlowResult:
        """Import ecobee config from configuration.yaml.
 
        Triggered by async_setup only if a config entry doesn't already exist.
        If ecobee.conf exists, we will attempt to validate the credentials
        and create an entry if valid. Otherwise, we will delegate to the user
        step so that the user can continue the config flow.
        """
        try:
            legacy_config = await self.hass.async_add_executor_job(
                load_json_object, self.hass.config.path(ECOBEE_CONFIG_FILENAME)
            )
            config = {
                ECOBEE_API_KEY: legacy_config[ECOBEE_API_KEY],
                ECOBEE_REFRESH_TOKEN: legacy_config[ECOBEE_REFRESH_TOKEN],
            }
        except (HomeAssistantError, KeyError):
            _LOGGER.debug(
                "No valid ecobee.conf configuration found for import, delegating to"
                " user step"
            )
            return await self.async_step_userasync_step_userasync_step_user(
                user_input={
                    CONF_API_KEY: self.hass.data[DATA_ECOBEE_CONFIG].get(CONF_API_KEY)
                }
            )
 
        ecobee = Ecobee(config=config)
        if await self.hass.async_add_executor_job(ecobee.refresh_tokens):
            # Credentials found and validated; create the entry.
            _LOGGER.debug(
                "Valid ecobee configuration found for import, creating configuration"
                " entry"
            )
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=DOMAIN,
                data={
                    CONF_API_KEY: ecobee.api_key,
                    CONF_REFRESH_TOKEN: ecobee.refresh_token,
                },
            )
        return await self.async_step_userasync_step_userasync_step_user(
            user_input={
                CONF_API_KEY: self.hass.data[DATA_ECOBEE_CONFIG].get(CONF_API_KEY)
            }
        )