config_flow.py

Go to the documentation of this file.

"""Config flow for the Bang & Olufsen integration."""
 
from __future__ import annotations
 
from ipaddress import AddressValueError, IPv4Address
from typing import Any, TypedDict
 
from aiohttp.client_exceptions import ClientConnectorError
from mozart_api.exceptions import ApiException
from mozart_api.mozart_client import MozartClient
import voluptuous as vol
 
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_MODEL
from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig
from homeassistant.util.ssl import get_default_context
 
from .const import (
    ATTR_FRIENDLY_NAME,
    ATTR_ITEM_NUMBER,
    ATTR_SERIAL_NUMBER,
    ATTR_TYPE_NUMBER,
    COMPATIBLE_MODELS,
    CONF_SERIAL_NUMBER,
    DEFAULT_MODEL,
    DOMAIN,
)
from .util import get_serial_number_from_jid
 
 
class EntryData(TypedDict, total=False):
    """TypedDict for config_entry data."""
 
    host: str
    jid: str
    model: str
    name: str
 
 
# Map exception types to strings
_exception_map = {
    ApiException: "api_exception",
    ClientConnectorError: "client_connector_error",
    TimeoutError: "timeout_error",
    AddressValueError: "invalid_ip",
}
 
 
class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a config flow."""
 
    _beolink_jid = ""
    _client: MozartClient
    _host = ""
    _model = ""
    _name = ""
    _serial_number = ""
 
    def __init__(self) -> None:
        """Init the config flow."""
 
    VERSION = 1
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        data_schema = vol.Schema(
            {
                vol.Required(CONF_HOST): str,
                vol.Required(CONF_MODEL, default=DEFAULT_MODEL): SelectSelector(
                    SelectSelectorConfig(options=COMPATIBLE_MODELS)
                ),
            }
        )
 
        if user_input is not None:
            self._host_host_host = user_input[CONF_HOST]
            self._model_model_model = user_input[CONF_MODEL]
 
            # Check if the IP address is a valid IPv4 address.
            try:
                IPv4Address(self._host_host_host)
            except AddressValueError as error:
                return self.async_show_formasync_show_formasync_show_form(
                    step_id="user",
                    data_schema=data_schema,
                    errors={"base": _exception_map[type(error)]},
                )
 
            self._client_client = MozartClient(
                host=self._host_host_host, ssl_context=get_default_context()
            )
 
            # Try to get information from Beolink self method.
            async with self._client_client:
                try:
                    beolink_self = await self._client_client.get_beolink_self(
                        _request_timeout=3
                    )
                except (
                    ApiException,
                    ClientConnectorError,
                    TimeoutError,
                ) as error:
                    return self.async_show_formasync_show_formasync_show_form(
                        step_id="user",
                        data_schema=data_schema,
                        errors={"base": _exception_map[type(error)]},
                    )
 
            self._beolink_jid_beolink_jid_beolink_jid = beolink_self.jid
            self._serial_number_serial_number_serial_number = get_serial_number_from_jid(beolink_self.jid)
 
            await self.async_set_unique_idasync_set_unique_id(self._serial_number_serial_number_serial_number)
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
            return await self._create_entry_create_entry()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=data_schema,
        )
 
    async def async_step_zeroconf(
        self, discovery_info: ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle discovery using Zeroconf."""
 
        # Check if the discovered device is a Mozart device
        if ATTR_FRIENDLY_NAME not in discovery_info.properties:
            return self.async_abortasync_abortasync_abort(reason="not_mozart_device")
 
        # Ensure that an IPv4 address is received
        self._host_host_host = discovery_info.host
        try:
            IPv4Address(self._host_host_host)
        except AddressValueError:
            return self.async_abortasync_abortasync_abort(reason="ipv6_address")
 
        # Check connection to ensure valid address is received
        self._client_client = MozartClient(self._host_host_host, ssl_context=get_default_context())
 
        async with self._client_client:
            try:
                await self._client_client.get_beolink_self(_request_timeout=3)
            except (ClientConnectorError, TimeoutError):
                return self.async_abortasync_abortasync_abort(reason="invalid_address")
 
        self._model_model_model = discovery_info.hostname[:-16].replace("-", " ")
        self._serial_number_serial_number_serial_number = discovery_info.properties[ATTR_SERIAL_NUMBER]
        self._beolink_jid_beolink_jid_beolink_jid = f"{discovery_info.properties[ATTR_TYPE_NUMBER]}.{discovery_info.properties[ATTR_ITEM_NUMBER]}.{self._serial_number}@products.bang-olufsen.com"
 
        await self.async_set_unique_idasync_set_unique_id(self._serial_number_serial_number_serial_number)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(updates={CONF_HOST: self._host_host_host})
 
        # Set the discovered device title
        self.context["title_placeholders"] = {
            "name": discovery_info.properties[ATTR_FRIENDLY_NAME]
        }
 
        return await self.async_step_zeroconf_confirmasync_step_zeroconf_confirm()
 
    async def _create_entry(self) -> ConfigFlowResult:
        """Create the config entry for a discovered or manually configured Bang & Olufsen device."""
        # Ensure that created entities have a unique and easily identifiable id and not a "friendly name"
        self._name_name_name = f"{self._model}-{self._serial_number}"
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=self._name_name_name,
            data=EntryData(
                host=self._host_host_host,
                jid=self._beolink_jid_beolink_jid_beolink_jid,
                model=self._model_model_model,
                name=self._name_name_name,
            ),
        )
 
    async def async_step_zeroconf_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Confirm the configuration of the device."""
        if user_input is not None:
            return await self._create_entry_create_entry()
 
        self._set_confirm_only_set_confirm_only()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="zeroconf_confirm",
            description_placeholders={
                CONF_HOST: self._host_host_host,
                CONF_MODEL: self._model_model_model,
                CONF_SERIAL_NUMBER: self._serial_number_serial_number_serial_number,
            },
            last_step=True,
        )