config_flow.py

Go to the documentation of this file.

"""Config flow to configure the Bluetooth integration."""
 
from __future__ import annotations
 
import platform
from typing import Any, cast
 
from bluetooth_adapters import (
    ADAPTER_ADDRESS,
    ADAPTER_MANUFACTURER,
    DEFAULT_ADDRESS,
    AdapterDetails,
    adapter_human_name,
    adapter_model,
    get_adapters,
)
from habluetooth import get_manager
import voluptuous as vol
 
from homeassistant.components import onboarding
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.core import callback
from homeassistant.helpers.schema_config_entry_flow import (
    SchemaFlowFormStep,
    SchemaOptionsFlowHandler,
)
from homeassistant.helpers.typing import DiscoveryInfoType
 
from .const import CONF_ADAPTER, CONF_DETAILS, CONF_PASSIVE, DOMAIN
from .util import adapter_title
 
OPTIONS_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_PASSIVE, default=False): bool,
    }
)
OPTIONS_FLOW = {
    "init": SchemaFlowFormStep(OPTIONS_SCHEMA),
}
 
 
def adapter_display_info(adapter: str, details: AdapterDetails) -> str:
    """Return the adapter display info."""
    name = adapter_human_name(adapter, details[ADAPTER_ADDRESS])
    model = adapter_model(details)
    manufacturer = details[ADAPTER_MANUFACTURER] or "Unknown"
    return f"{name} {manufacturer} {model}"
 
 
class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
    """Config flow for Bluetooth."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize the config flow."""
        self._adapter_adapter: str | None = None
        self._details_details: AdapterDetails | None = None
        self._adapters_adapters: dict[str, AdapterDetails] = {}
        self._placeholders_placeholders: dict[str, str] = {}
 
    async def async_step_integration_discovery(
        self, discovery_info: DiscoveryInfoType
    ) -> ConfigFlowResult:
        """Handle a flow initialized by discovery."""
        self._adapter_adapter = cast(str, discovery_info[CONF_ADAPTER])
        self._details_details = cast(AdapterDetails, discovery_info[CONF_DETAILS])
        await self.async_set_unique_idasync_set_unique_id(self._details_details[ADAPTER_ADDRESS])
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
        details = self._details_details
        self._async_set_adapter_info_async_set_adapter_info(self._adapter_adapter, details)
        return await self.async_step_single_adapterasync_step_single_adapter()
 
    @callback
    def _async_set_adapter_info(self, adapter: str, details: AdapterDetails) -> None:
        """Set the adapter info."""
        name = adapter_human_name(adapter, details[ADAPTER_ADDRESS])
        model = adapter_model(details)
        manufacturer = details[ADAPTER_MANUFACTURER]
        self._placeholders_placeholders = {
            "name": name,
            "model": model,
            "manufacturer": manufacturer or "Unknown",
        }
        self.context["title_placeholders"] = self._placeholders_placeholders
 
    async def async_step_single_adapter(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Select an adapter."""
        adapter = self._adapter_adapter
        details = self._details_details
        assert adapter is not None
        assert details is not None
        assert self._placeholders_placeholders is not None
 
        address = details[ADAPTER_ADDRESS]
 
        if user_input is not None or not onboarding.async_is_onboarded(self.hass):
            await self.async_set_unique_idasync_set_unique_id(address, raise_on_progress=False)
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=adapter_title(adapter, details), data={}
            )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="single_adapter",
            description_placeholders=self._placeholders_placeholders,
        )
 
    async def async_step_multiple_adapters(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
        if user_input is not None:
            assert self._adapters_adapters is not None
            adapter = user_input[CONF_ADAPTER]
            details = self._adapters_adapters[adapter]
            address = details[ADAPTER_ADDRESS]
            await self.async_set_unique_idasync_set_unique_id(address, raise_on_progress=False)
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=adapter_title(adapter, details), data={}
            )
 
        configured_addresses = self._async_current_ids_async_current_ids()
        bluetooth_adapters = get_adapters()
        await bluetooth_adapters.refresh()
        self._adapters_adapters = bluetooth_adapters.adapters
        system = platform.system()
        unconfigured_adapters = [
            adapter
            for adapter, details in self._adapters_adapters.items()
            if details[ADAPTER_ADDRESS] not in configured_addresses
            # DEFAULT_ADDRESS is perfectly valid on MacOS but on
            # Linux it means the adapter is not yet configured
            # or crashed
            and not (system == "Linux" and details[ADAPTER_ADDRESS] == DEFAULT_ADDRESS)
        ]
        if not unconfigured_adapters:
            ignored_adapters = len(
                self._async_current_entries_async_current_entries(include_ignore=True)
            ) - len(self._async_current_entries_async_current_entries(include_ignore=False))
            return self.async_abortasync_abortasync_abort(
                reason="no_adapters",
                description_placeholders={"ignored_adapters": str(ignored_adapters)},
            )
        if len(unconfigured_adapters) == 1:
            self._adapter_adapter = list(self._adapters_adapters)[0]
            self._details_details = self._adapters_adapters[self._adapter_adapter]
            self._async_set_adapter_info_async_set_adapter_info(self._adapter_adapter, self._details_details)
            return await self.async_step_single_adapterasync_step_single_adapter()
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="multiple_adapters",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_ADAPTER): vol.In(
                        {
                            adapter: adapter_display_info(
                                adapter, self._adapters_adapters[adapter]
                            )
                            for adapter in sorted(unconfigured_adapters)
                        }
                    ),
                }
            ),
        )
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
        return await self.async_step_multiple_adaptersasync_step_multiple_adapters()
 
    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> SchemaOptionsFlowHandler:
        """Get the options flow for this handler."""
        return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)
 
    @classmethod
    @callback
    def async_supports_options_flow(cls, config_entry: ConfigEntry) -> bool:
        """Return options flow support for this handler."""
        return bool((manager := get_manager()) and manager.supports_passive_scan)