config_flow.py

Go to the documentation of this file.

"""Config flow to configure the LG Netcast TV integration."""
 
from __future__ import annotations
 
import contextlib
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
 
from pylgnetcast import AccessTokenError, LgNetCastClient, SessionIdError
import voluptuous as vol
 
from homeassistant import config_entries
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import (
    CONF_ACCESS_TOKEN,
    CONF_HOST,
    CONF_ID,
    CONF_MODEL,
    CONF_NAME,
)
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.data_entry_flow import AbortFlow
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util.network import is_host_valid
 
from .const import DEFAULT_NAME, DOMAIN
from .helpers import LGNetCastDetailDiscoveryError, async_discover_netcast_details
 
DISPLAY_ACCESS_TOKEN_INTERVAL = timedelta(seconds=1)
 
 
class LGNetCast(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for LG Netcast TV integration."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize config flow."""
        self.clientclient: LgNetCastClient | None = None
        self.device_config: dict[str, Any] = {}
        self._discovered_devices: dict[str, Any] = {}
        self._track_interval_track_interval: CALLBACK_TYPE | None = None
 
    def create_client(self) -> None:
        """Create LG Netcast client from config."""
        host = self.device_config[CONF_HOST]
        access_token = self.device_config.get(CONF_ACCESS_TOKEN)
        self.clientclient = LgNetCastClient(host, access_token)
 
    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:
            host = user_input[CONF_HOST]
            if is_host_valid(host):
                self.device_config[CONF_HOST] = host
                return await self.async_step_authorizeasync_step_authorize()
 
            errors[CONF_HOST] = "invalid_host"
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
            errors=errors,
        )
 
    async def async_discover_client(self):
        """Handle Discovery step."""
        self.create_clientcreate_client()
 
        if TYPE_CHECKING:
            assert self.clientclient is not None
 
        if self.device_config.get(CONF_ID):
            return
 
        try:
            details = await async_discover_netcast_details(self.hass, self.clientclient)
        except LGNetCastDetailDiscoveryError as err:
            raise AbortFlow("cannot_connect") from err
 
        if (unique_id := details["uuid"]) is None:
            raise AbortFlow("invalid_host")
 
        self.device_config[CONF_ID] = unique_id
        self.device_config[CONF_MODEL] = details["model_name"]
 
        if CONF_NAME not in self.device_config:
            self.device_config[CONF_NAME] = details["friendly_name"] or DEFAULT_NAME
 
    async def async_step_authorize(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle Authorize step."""
        errors: dict[str, str] = {}
        self.async_stop_display_access_tokenasync_stop_display_access_token()
 
        if user_input is not None and user_input.get(CONF_ACCESS_TOKEN) is not None:
            self.device_config[CONF_ACCESS_TOKEN] = user_input[CONF_ACCESS_TOKEN]
 
        await self.async_discover_clientasync_discover_client()
        assert self.clientclient is not None
 
        await self.async_set_unique_idasync_set_unique_id(self.device_config[CONF_ID])
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(
            updates={CONF_HOST: self.device_config[CONF_HOST]}
        )
 
        try:
            await self.hass.async_add_executor_job(
                self.clientclient._get_session_id  # noqa: SLF001
            )
        except AccessTokenError:
            if user_input is not None:
                errors[CONF_ACCESS_TOKEN] = "invalid_access_token"
        except SessionIdError:
            errors["base"] = "cannot_connect"
        else:
            return await self.async_create_deviceasync_create_device()
 
        self._track_interval_track_interval = async_track_time_interval(
            self.hass,
            self.async_display_access_tokenasync_display_access_token,
            DISPLAY_ACCESS_TOKEN_INTERVAL,
            cancel_on_shutdown=True,
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="authorize",
            data_schema=vol.Schema(
                {
                    vol.Optional(CONF_ACCESS_TOKEN): vol.All(str, vol.Length(max=6)),
                }
            ),
            errors=errors,
        )
 
    async def async_display_access_token(self, _: datetime | None = None):
        """Display access token on screen."""
        assert self.clientclient is not None
        with contextlib.suppress(AccessTokenError, SessionIdError):
            await self.hass.async_add_executor_job(
                self.clientclient._get_session_id  # noqa: SLF001
            )
 
    @callback
    def async_remove(self):
        """Terminate Access token display if flow is removed."""
        self.async_stop_display_access_tokenasync_stop_display_access_token()
 
    def async_stop_display_access_token(self):
        """Stop Access token request if running."""
        if self._track_interval_track_interval is not None:
            self._track_interval_track_interval()
            self._track_interval_track_interval = None
 
    async def async_create_device(self) -> ConfigFlowResult:
        """Create LG Netcast TV Device from config."""
        assert self.clientclient
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=self.device_config[CONF_NAME], data=self.device_config
        )