config_flow.py

Go to the documentation of this file.

"""Config flow for Kodi integration."""
 
from __future__ import annotations
 
import logging
from typing import Any
 
from pykodi import CannotConnectError, InvalidAuthError, Kodi, get_kodi_connection
import voluptuous as vol
 
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
    CONF_HOST,
    CONF_NAME,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_SSL,
    CONF_TIMEOUT,
    CONF_USERNAME,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import (
    CONF_WS_PORT,
    DEFAULT_PORT,
    DEFAULT_SSL,
    DEFAULT_TIMEOUT,
    DEFAULT_WS_PORT,
    DOMAIN,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
async def validate_http(hass: HomeAssistant, data):
    """Validate the user input allows us to connect over HTTP."""
 
    host = data[CONF_HOST]
    port = data[CONF_PORT]
    username = data.get(CONF_USERNAME)
    password = data.get(CONF_PASSWORD)
    ssl = data.get(CONF_SSL)
    session = async_get_clientsession(hass)
 
    _LOGGER.debug("Connecting to %s:%s over HTTP", host, port)
    khc = get_kodi_connection(
        host, port, None, username, password, ssl, session=session
    )
    kodi = Kodi(khc)
    try:
        await kodi.ping()
    except CannotConnectError as error:
        raise CannotConnect from error
    except InvalidAuthError as error:
        raise InvalidAuth from error
 
 
async def validate_ws(hass: HomeAssistant, data):
    """Validate the user input allows us to connect over WS."""
    if not (ws_port := data.get(CONF_WS_PORT)):
        return
 
    host = data[CONF_HOST]
    port = data[CONF_PORT]
    username = data.get(CONF_USERNAME)
    password = data.get(CONF_PASSWORD)
    ssl = data.get(CONF_SSL)
 
    session = async_get_clientsession(hass)
 
    _LOGGER.debug("Connecting to %s:%s over WebSocket", host, ws_port)
    kwc = get_kodi_connection(
        host, port, ws_port, username, password, ssl, session=session
    )
    try:
        await kwc.connect()
        if not kwc.connected:
            _LOGGER.warning("Cannot connect to %s:%s over WebSocket", host, ws_port)
            raise WSCannotConnect
        kodi = Kodi(kwc)
        await kodi.ping()
    except CannotConnectError as error:
        raise WSCannotConnect from error
 
 
class KodiConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Kodi."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize flow."""
        self._host_host: str | None = None
        self._port_port: int | None = DEFAULT_PORT
        self._ws_port_ws_port: int | None = DEFAULT_WS_PORT
        self._name_name: str | None = None
        self._username_username: str | None = None
        self._password_password: str | None = None
        self._ssl_ssl: bool | None = DEFAULT_SSL
        self._discovery_name_discovery_name: str | None = None
 
    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle zeroconf discovery."""
        self._host_host = discovery_info.host
        self._port_port = discovery_info.port or DEFAULT_PORT
        self._name_name = discovery_info.hostname.removesuffix(".local.")
        if not (uuid := discovery_info.properties.get("uuid")):
            return self.async_abortasync_abortasync_abort(reason="no_uuid")
 
        self._discovery_name_discovery_name = discovery_info.name
 
        await self.async_set_unique_idasync_set_unique_id(uuid)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(
            updates={
                CONF_HOST: self._host_host,
                CONF_PORT: self._port_port,
                CONF_NAME: self._name_name,
            }
        )
 
        self.context.update({"title_placeholders": {CONF_NAME: self._name_name}})
 
        try:
            await validate_http(self.hass, self._get_data_get_data())
            await validate_ws(self.hass, self._get_data_get_data())
        except InvalidAuth:
            return await self.async_step_credentialsasync_step_credentials()
        except WSCannotConnect:
            return await self.async_step_ws_portasync_step_ws_port()
        except CannotConnect:
            return self.async_abortasync_abortasync_abort(reason="cannot_connect")
        except Exception:
            _LOGGER.exception("Unexpected exception")
            return self.async_abortasync_abortasync_abort(reason="unknown")
 
        return await self.async_step_discovery_confirmasync_step_discovery_confirm()
 
    async def async_step_discovery_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle user-confirmation of discovered node."""
        if user_input is None:
            assert self._name_name is not None
            return self.async_show_formasync_show_formasync_show_form(
                step_id="discovery_confirm",
                description_placeholders={"name": self._name_name},
            )
 
        return self._create_entry_create_entry()
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        errors = {}
 
        if user_input is not None:
            self._host_host = user_input[CONF_HOST]
            self._port_port = user_input[CONF_PORT]
            self._ssl_ssl = user_input[CONF_SSL]
 
            try:
                await validate_http(self.hass, self._get_data_get_data())
                await validate_ws(self.hass, self._get_data_get_data())
            except InvalidAuth:
                return await self.async_step_credentialsasync_step_credentials()
            except WSCannotConnect:
                return await self.async_step_ws_portasync_step_ws_port()
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                return self._create_entry_create_entry()
 
        return self._show_user_form_show_user_form(errors)
 
    async def async_step_credentials(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle username and password input."""
        errors = {}
 
        if user_input is not None:
            self._username_username = user_input.get(CONF_USERNAME)
            self._password_password = user_input.get(CONF_PASSWORD)
 
            try:
                await validate_http(self.hass, self._get_data_get_data())
                await validate_ws(self.hass, self._get_data_get_data())
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            except WSCannotConnect:
                return await self.async_step_ws_portasync_step_ws_port()
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                return self._create_entry_create_entry()
 
        return self._show_credentials_form_show_credentials_form(errors)
 
    async def async_step_ws_port(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle websocket port of discovered node."""
        errors = {}
 
        if user_input is not None:
            self._ws_port_ws_port = user_input.get(CONF_WS_PORT)
 
            # optional ints return 0 rather than None when empty
            if self._ws_port_ws_port == 0:
                self._ws_port_ws_port = None
 
            try:
                await validate_ws(self.hass, self._get_data_get_data())
            except WSCannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                return self._create_entry_create_entry()
 
        return self._show_ws_port_form_show_ws_port_form(errors)
 
    async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
        """Handle import from YAML."""
        reason = None
        try:
            await validate_http(self.hass, import_data)
            await validate_ws(self.hass, import_data)
        except InvalidAuth:
            _LOGGER.exception("Invalid Kodi credentials")
            reason = "invalid_auth"
        except CannotConnect:
            _LOGGER.exception("Cannot connect to Kodi")
            reason = "cannot_connect"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            reason = "unknown"
        else:
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=import_data[CONF_NAME], data=import_data
            )
 
        return self.async_abortasync_abortasync_abort(reason=reason)
 
    @callback
    def _show_credentials_form(
        self, errors: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        schema = vol.Schema(
            {
                vol.Optional(
                    CONF_USERNAME, description={"suggested_value": self._username_username}
                ): str,
                vol.Optional(
                    CONF_PASSWORD, description={"suggested_value": self._password_password}
                ): str,
            }
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="credentials", data_schema=schema, errors=errors
        )
 
    @callback
    def _show_user_form(self, errors=None):
        default_port = self._port_port or DEFAULT_PORT
        default_ssl = self._ssl_ssl or DEFAULT_SSL
        schema = vol.Schema(
            {
                vol.Required(CONF_HOST, default=self._host_host): str,
                vol.Required(CONF_PORT, default=default_port): int,
                vol.Required(CONF_SSL, default=default_ssl): bool,
            }
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=schema, errors=errors or {}
        )
 
    @callback
    def _show_ws_port_form(self, errors=None):
        suggestion = self._ws_port_ws_port or DEFAULT_WS_PORT
        schema = vol.Schema(
            {
                vol.Optional(
                    CONF_WS_PORT, description={"suggested_value": suggestion}
                ): int
            }
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="ws_port", data_schema=schema, errors=errors or {}
        )
 
    @callback
    def _create_entry(self):
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=self._name_name or self._host_host,
            data=self._get_data_get_data(),
        )
 
    @callback
    def _get_data(self) -> dict[str, Any]:
        return {
            CONF_NAME: self._name_name,
            CONF_HOST: self._host_host,
            CONF_PORT: self._port_port,
            CONF_WS_PORT: self._ws_port_ws_port,
            CONF_USERNAME: self._username_username,
            CONF_PASSWORD: self._password_password,
            CONF_SSL: self._ssl_ssl,
            CONF_TIMEOUT: DEFAULT_TIMEOUT,
        }
 
 
class CannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect."""
 
 
class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""
 
 
class WSCannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect to websocket."""