config_flow.py

Go to the documentation of this file.

"""Config flow for the Open Thread Border Router integration."""
 
from __future__ import annotations
 
from contextlib import suppress
import logging
from typing import TYPE_CHECKING, cast
 
import aiohttp
import python_otbr_api
from python_otbr_api import tlv_parser
from python_otbr_api.tlv_parser import MeshcopTLVType
import voluptuous as vol
import yarl
 
from homeassistant.components.hassio import AddonError, AddonManager
from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware
from homeassistant.components.thread import async_get_preferred_dataset
from homeassistant.config_entries import SOURCE_HASSIO, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
 
from .const import DEFAULT_CHANNEL, DOMAIN
from .util import (
    compose_default_network_name,
    generate_random_pan_id,
    get_allowed_channel,
)
 
if TYPE_CHECKING:
    from . import OTBRConfigEntry
 
_LOGGER = logging.getLogger(__name__)
 
 
class AlreadyConfigured(HomeAssistantError):
    """Raised when the router is already configured."""
 
 
@callback
def get_addon_manager(hass: HomeAssistant, slug: str) -> AddonManager:
    """Get the add-on manager."""
    return AddonManager(hass, _LOGGER, "OpenThread Border Router", slug)
 
 
def _is_yellow(hass: HomeAssistant) -> bool:
    """Return True if Home Assistant is running on a Home Assistant Yellow."""
    try:
        yellow_hardware.async_info(hass)
    except HomeAssistantError:
        return False
    return True
 
 
async def _title(hass: HomeAssistant, discovery_info: HassioServiceInfo) -> str:
    """Return config entry title."""
    device: str | None = None
    addon_manager = get_addon_manager(hass, discovery_info.slug)
 
    with suppress(AddonError):
        addon_info = await addon_manager.async_get_addon_info()
        device = addon_info.options.get("device")
 
    if _is_yellow(hass) and device == "/dev/ttyAMA1":
        return f"Home Assistant Yellow ({discovery_info.name})"
 
    if device and "SkyConnect" in device:
        return f"Home Assistant SkyConnect ({discovery_info.name})"
 
    if device and "Connect_ZBT-1" in device:
        return f"Home Assistant Connect ZBT-1 ({discovery_info.name})"
 
    return discovery_info.name
 
 
class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Open Thread Border Router."""
 
    VERSION = 1
 
    async def _set_dataset(self, api: python_otbr_api.OTBR, otbr_url: str) -> None:
        """Connect to the OTBR and create or apply a dataset if it doesn't have one."""
        if await api.get_active_dataset_tlvs() is None:
            allowed_channel = await get_allowed_channel(self.hass, otbr_url)
 
            thread_dataset_channel = None
            thread_dataset_tlv = await async_get_preferred_dataset(self.hass)
            if thread_dataset_tlv:
                dataset = tlv_parser.parse_tlv(thread_dataset_tlv)
                if channel := dataset.get(MeshcopTLVType.CHANNEL):
                    thread_dataset_channel = cast(tlv_parser.Channel, channel).channel
 
            if thread_dataset_tlv is not None and (
                not allowed_channel or allowed_channel == thread_dataset_channel
            ):
                await api.set_active_dataset_tlvs(bytes.fromhex(thread_dataset_tlv))
            else:
                _LOGGER.debug(
                    "not importing TLV with channel %s for %s",
                    thread_dataset_channel,
                    otbr_url,
                )
                pan_id = generate_random_pan_id()
                await api.create_active_dataset(
                    python_otbr_api.ActiveDataSet(
                        channel=allowed_channel if allowed_channel else DEFAULT_CHANNEL,
                        network_name=compose_default_network_name(pan_id),
                        pan_id=pan_id,
                    )
                )
            await api.set_enabled(True)
 
    async def _is_border_agent_id_configured(self, border_agent_id: bytes) -> bool:
        """Return True if another config entry's OTBR has the same border agent id."""
        config_entry: OTBRConfigEntry
        for config_entry in self.hass.config_entries.async_loaded_entries(DOMAIN):
            data = config_entry.runtime_data
            try:
                other_border_agent_id = await data.get_border_agent_id()
            except HomeAssistantError:
                _LOGGER.debug(
                    "Could not read border agent id from %s", data.url, exc_info=True
                )
                continue
            _LOGGER.debug(
                "border agent id for existing url %s: %s",
                data.url,
                other_border_agent_id.hex(),
            )
            if border_agent_id == other_border_agent_id:
                return True
        return False
 
    async def _connect_and_configure_router(self, otbr_url: str) -> bytes:
        """Connect to the router and configure it if needed.
 
        Will raise if the router's border agent id is in use by another config entry.
        Returns the router's border agent id.
        """
        api = python_otbr_api.OTBR(otbr_url, async_get_clientsession(self.hass), 10)
        border_agent_id = await api.get_border_agent_id()
        _LOGGER.debug("border agent id for url %s: %s", otbr_url, border_agent_id.hex())
 
        if await self._is_border_agent_id_configured_is_border_agent_id_configured(border_agent_id):
            raise AlreadyConfigured
 
        await self._set_dataset_set_dataset(api, otbr_url)
 
        return border_agent_id
 
    async def async_step_user(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Set up by user."""
        errors = {}
 
        if user_input is not None:
            url = user_input[CONF_URL].rstrip("/")
            try:
                border_agent_id = await self._connect_and_configure_router_connect_and_configure_router(url)
            except AlreadyConfigured:
                errors["base"] = "already_configured"
            except (
                python_otbr_api.OTBRError,
                aiohttp.ClientError,
                TimeoutError,
            ) as exc:
                _LOGGER.debug("Failed to communicate with OTBR@%s: %s", url, exc)
                errors["base"] = "cannot_connect"
            else:
                await self.async_set_unique_idasync_set_unique_id(border_agent_id.hex())
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title="Open Thread Border Router",
                    data={CONF_URL: url},
                )
 
        data_schema = vol.Schema({CONF_URL: str})
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=data_schema, errors=errors
        )
 
    async def async_step_hassio(
        self, discovery_info: HassioServiceInfo
    ) -> ConfigFlowResult:
        """Handle hassio discovery."""
        config = discovery_info.config
        url = f"http://{config['host']}:{config['port']}"
        config_entry_data = {"url": url}
 
        if current_entries := self._async_current_entries_async_current_entries():
            for current_entry in current_entries:
                if current_entry.source != SOURCE_HASSIO:
                    continue
                current_url = yarl.URL(current_entry.data["url"])
                if not (unique_id := current_entry.unique_id):
                    # The first version did not set a unique_id
                    # so if the entry does not have a unique_id
                    # we have to assume it's the first version
                    # This check can be removed in HA Core 2025.9
                    unique_id = discovery_info.uuid
                if (
                    unique_id != discovery_info.uuid
                    or current_url.host != config["host"]
                    or current_url.port == config["port"]
                ):
                    continue
                # Update URL with the new port
                self.hass.config_entries.async_update_entry(
                    current_entry,
                    data=config_entry_data,
                    unique_id=unique_id,  # Remove in HA Core 2025.9
                )
                return self.async_abortasync_abortasync_abort(reason="already_configured")
 
        try:
            await self._connect_and_configure_router_connect_and_configure_router(url)
        except AlreadyConfigured:
            return self.async_abortasync_abortasync_abort(reason="already_configured")
        except (
            python_otbr_api.OTBRError,
            aiohttp.ClientError,
            TimeoutError,
        ) as exc:
            _LOGGER.warning("Failed to communicate with OTBR@%s: %s", url, exc)
            return self.async_abortasync_abortasync_abort(reason="unknown")
 
        await self.async_set_unique_idasync_set_unique_id(discovery_info.uuid)
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=await _title(self.hass, discovery_info),
            data=config_entry_data,
        )