config_flow.py

Go to the documentation of this file.

"""Config flow to configure SmartThings."""
 
from http import HTTPStatus
import logging
from typing import Any
 
from aiohttp import ClientResponseError
from pysmartthings import APIResponseError, AppOAuth, SmartThings
from pysmartthings.installedapp import format_install_url
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import (
    APP_OAUTH_CLIENT_NAME,
    APP_OAUTH_SCOPES,
    CONF_APP_ID,
    CONF_INSTALLED_APP_ID,
    CONF_LOCATION_ID,
    CONF_REFRESH_TOKEN,
    DOMAIN,
    VAL_UID_MATCHER,
)
from .smartapp import (
    create_app,
    find_app,
    format_unique_id,
    get_webhook_url,
    setup_smartapp,
    setup_smartapp_endpoint,
    update_app,
    validate_webhook_requirements,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle configuration of SmartThings integrations."""
 
    VERSION = 2
 
    api: SmartThings
    app_id: str
    location_id: str
 
    def __init__(self) -> None:
        """Create a new instance of the flow handler."""
        self.access_tokenaccess_token: str | None = None
        self.oauth_client_secretoauth_client_secret = None
        self.oauth_client_idoauth_client_id = None
        self.installed_app_idinstalled_app_id = None
        self.refresh_tokenrefresh_token = None
        self.endpoints_initializedendpoints_initialized = False
 
    async def async_step_import(self, import_data: None) -> ConfigFlowResult:
        """Occurs when a previously entry setup fails and is re-initiated."""
        return await self.async_step_userasync_step_userasync_step_user(import_data)
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Validate and confirm webhook setup."""
        if not self.endpoints_initializedendpoints_initialized:
            self.endpoints_initializedendpoints_initialized = True
            await setup_smartapp_endpoint(
                self.hass, len(self._async_current_entries_async_current_entries()) == 0
            )
        webhook_url = get_webhook_url(self.hass)
 
        # Abort if the webhook is invalid
        if not validate_webhook_requirements(self.hass):
            return self.async_abortasync_abortasync_abort(
                reason="invalid_webhook_url",
                description_placeholders={
                    "webhook_url": webhook_url,
                    "component_url": (
                        "https://www.home-assistant.io/integrations/smartthings/"
                    ),
                },
            )
 
        # Show the confirmation
        if user_input is None:
            return self.async_show_formasync_show_formasync_show_form(
                step_id="user",
                description_placeholders={"webhook_url": webhook_url},
            )
 
        # Show the next screen
        return await self.async_step_patasync_step_pat()
 
    async def async_step_pat(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Get the Personal Access Token and validate it."""
        errors: dict[str, str] = {}
        if user_input is None or CONF_ACCESS_TOKEN not in user_input:
            return self._show_step_pat_show_step_pat(errors)
 
        self.access_tokenaccess_token = user_input[CONF_ACCESS_TOKEN]
 
        # Ensure token is a UUID
        if not VAL_UID_MATCHER.match(self.access_tokenaccess_token):
            errors[CONF_ACCESS_TOKEN] = "token_invalid_format"
            return self._show_step_pat_show_step_pat(errors)
 
        # Setup end-point
        self.apiapi = SmartThings(async_get_clientsession(self.hass), self.access_tokenaccess_token)
        try:
            app = await find_app(self.hass, self.apiapi)
            if app:
                await app.refresh()  # load all attributes
                await update_app(self.hass, app)
                # Find an existing entry to copy the oauth client
                existing = next(
                    (
                        entry
                        for entry in self._async_current_entries_async_current_entries()
                        if entry.data[CONF_APP_ID] == app.app_id
                    ),
                    None,
                )
                if existing:
                    self.oauth_client_idoauth_client_id = existing.data[CONF_CLIENT_ID]
                    self.oauth_client_secretoauth_client_secret = existing.data[CONF_CLIENT_SECRET]
                else:
                    # Get oauth client id/secret by regenerating it
                    app_oauth = AppOAuth(app.app_id)
                    app_oauth.client_name = APP_OAUTH_CLIENT_NAME
                    app_oauth.scope.extend(APP_OAUTH_SCOPES)
                    client = await self.apiapi.generate_app_oauth(app_oauth)
                    self.oauth_client_secretoauth_client_secret = client.client_secret
                    self.oauth_client_idoauth_client_id = client.client_id
            else:
                app, client = await create_app(self.hass, self.apiapi)
                self.oauth_client_secretoauth_client_secret = client.client_secret
                self.oauth_client_idoauth_client_id = client.client_id
            setup_smartapp(self.hass, app)
            self.app_idapp_id = app.app_id
 
        except APIResponseError as ex:
            if ex.is_target_error():
                errors["base"] = "webhook_error"
            else:
                errors["base"] = "app_setup_error"
            _LOGGER.exception(
                "API error setting up the SmartApp: %s", ex.raw_error_response
            )
            return self._show_step_pat_show_step_pat(errors)
        except ClientResponseError as ex:
            if ex.status == HTTPStatus.UNAUTHORIZED:
                errors[CONF_ACCESS_TOKEN] = "token_unauthorized"
                _LOGGER.debug(
                    "Unauthorized error received setting up SmartApp", exc_info=True
                )
            elif ex.status == HTTPStatus.FORBIDDEN:
                errors[CONF_ACCESS_TOKEN] = "token_forbidden"
                _LOGGER.debug(
                    "Forbidden error received setting up SmartApp", exc_info=True
                )
            else:
                errors["base"] = "app_setup_error"
                _LOGGER.exception("Unexpected error setting up the SmartApp")
            return self._show_step_pat_show_step_pat(errors)
        except Exception:
            errors["base"] = "app_setup_error"
            _LOGGER.exception("Unexpected error setting up the SmartApp")
            return self._show_step_pat_show_step_pat(errors)
 
        return await self.async_step_select_locationasync_step_select_location()
 
    async def async_step_select_location(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Ask user to select the location to setup."""
        if user_input is None or CONF_LOCATION_ID not in user_input:
            # Get available locations
            existing_locations = [
                entry.data[CONF_LOCATION_ID] for entry in self._async_current_entries_async_current_entries()
            ]
            locations = await self.apiapi.locations()
            locations_options = {
                location.location_id: location.name
                for location in locations
                if location.location_id not in existing_locations
            }
            if not locations_options:
                return self.async_abortasync_abortasync_abort(reason="no_available_locations")
 
            return self.async_show_formasync_show_formasync_show_form(
                step_id="select_location",
                data_schema=vol.Schema(
                    {vol.Required(CONF_LOCATION_ID): vol.In(locations_options)}
                ),
            )
 
        self.location_idlocation_id = user_input[CONF_LOCATION_ID]
        await self.async_set_unique_idasync_set_unique_id(format_unique_id(self.app_idapp_id, self.location_idlocation_id))
        return await self.async_step_authorizeasync_step_authorize()
 
    async def async_step_authorize(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Wait for the user to authorize the app installation."""
        user_input = {} if user_input is None else user_input
        self.installed_app_idinstalled_app_id = user_input.get(CONF_INSTALLED_APP_ID)
        self.refresh_tokenrefresh_token = user_input.get(CONF_REFRESH_TOKEN)
        if self.installed_app_idinstalled_app_id is None:
            # Launch the external setup URL
            url = format_install_url(self.app_idapp_id, self.location_idlocation_id)
            return self.async_external_stepasync_external_step(step_id="authorize", url=url)
 
        return self.async_external_step_doneasync_external_step_done(next_step_id="install")
 
    def _show_step_pat(self, errors):
        if self.access_tokenaccess_token is None:
            # Get the token from an existing entry to make it easier to setup multiple locations.
            self.access_tokenaccess_token = next(
                (
                    entry.data.get(CONF_ACCESS_TOKEN)
                    for entry in self._async_current_entries_async_current_entries()
                ),
                None,
            )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="pat",
            data_schema=vol.Schema(
                {vol.Required(CONF_ACCESS_TOKEN, default=self.access_tokenaccess_token): str}
            ),
            errors=errors,
            description_placeholders={
                "token_url": "https://account.smartthings.com/tokens",
                "component_url": (
                    "https://www.home-assistant.io/integrations/smartthings/"
                ),
            },
        )
 
    async def async_step_install(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Create a config entry at completion of a flow and authorization of the app."""
        data = {
            CONF_ACCESS_TOKEN: self.access_tokenaccess_token,
            CONF_REFRESH_TOKEN: self.refresh_tokenrefresh_token,
            CONF_CLIENT_ID: self.oauth_client_idoauth_client_id,
            CONF_CLIENT_SECRET: self.oauth_client_secretoauth_client_secret,
            CONF_LOCATION_ID: self.location_idlocation_id,
            CONF_APP_ID: self.app_idapp_id,
            CONF_INSTALLED_APP_ID: self.installed_app_idinstalled_app_id,
        }
 
        location = await self.apiapi.location(data[CONF_LOCATION_ID])
 
        return self.async_create_entryasync_create_entryasync_create_entry(title=location.name, data=data)