Automatically setup hardware integrations when firmware info is published by an integration (home-assistant#154030)

Author: puddly

homeassistant/components/homeassistant_connect_zbt2/config_flow.py

@@ -7,11 +7,18 @@
 
 from homeassistant.components import usb
 from homeassistant.components.homeassistant_hardware import firmware_config_flow
+from homeassistant.components.homeassistant_hardware.helpers import (
+    HardwareFirmwareDiscoveryInfo,
+)
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
     FirmwareInfo,
     ResetTarget,
 )
+from homeassistant.components.usb import (
+    usb_service_info_from_device,
+    usb_unique_id_from_service_info,
+)
 from homeassistant.config_entries import (
     ConfigEntry,
     ConfigEntryBaseFlow,
@@ -123,22 +130,16 @@ def async_get_options_flow(
 
     async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResult:
         """Handle usb discovery."""
-        device = discovery_info.device
-        vid = discovery_info.vid
-        pid = discovery_info.pid
-        serial_number = discovery_info.serial_number
-        manufacturer = discovery_info.manufacturer
-        description = discovery_info.description
-        unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}"
-
-        device = discovery_info.device = await self.hass.async_add_executor_job(
+        unique_id = usb_unique_id_from_service_info(discovery_info)
+
+        discovery_info.device = await self.hass.async_add_executor_job(
             usb.get_serial_by_id, discovery_info.device
         )
 
         try:
             await self.async_set_unique_id(unique_id)
         finally:
-            self._abort_if_unique_id_configured(updates={DEVICE: device})
+            self._abort_if_unique_id_configured(updates={DEVICE: discovery_info.device})
 
         self._usb_info = discovery_info
 
@@ -148,6 +149,24 @@ async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResu
 
         return await self.async_step_confirm()
 
+    async def async_step_import(
+        self, fw_discovery_info: HardwareFirmwareDiscoveryInfo
+    ) -> ConfigFlowResult:
+        """Handle import from ZHA/OTBR firmware notification."""
+        assert fw_discovery_info["usb_device"] is not None
+        usb_info = usb_service_info_from_device(fw_discovery_info["usb_device"])
+        unique_id = usb_unique_id_from_service_info(usb_info)
+
+        if await self.async_set_unique_id(unique_id, raise_on_progress=False):
+            self._abort_if_unique_id_configured(updates={DEVICE: usb_info.device})
+
+        self._usb_info = usb_info
+        self._device = usb_info.device
+        self._hardware_name = HARDWARE_NAME
+        self._probed_firmware_info = fw_discovery_info["firmware_info"]
+
+        return self._async_flow_finished()
+
     def _async_flow_finished(self) -> ConfigFlowResult:
         """Create the config entry."""
         assert self._usb_info is not None

homeassistant/components/homeassistant_hardware/const.py

@@ -19,6 +19,12 @@
 ZHA_DOMAIN = "zha"
 OTBR_DOMAIN = "otbr"
 
+HARDWARE_INTEGRATION_DOMAINS = {
+    "homeassistant_sky_connect",
+    "homeassistant_connect_zbt2",
+    "homeassistant_yellow",
+}
+
 OTBR_ADDON_NAME = "OpenThread Border Router"
 OTBR_ADDON_MANAGER_DATA = "openthread_border_router"
 OTBR_ADDON_SLUG = "core_openthread_border_router"

homeassistant/components/homeassistant_hardware/helpers.py

@@ -6,19 +6,33 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import asynccontextmanager
 import logging
-from typing import TYPE_CHECKING, Protocol
+from typing import TYPE_CHECKING, Protocol, TypedDict
 
-from homeassistant.config_entries import ConfigEntry
+from homeassistant.components.usb import (
+    USBDevice,
+    async_get_usb_matchers_for_device,
+    usb_device_from_path,
+)
+from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
 
 from . import DATA_COMPONENT
+from .const import HARDWARE_INTEGRATION_DOMAINS
 
 if TYPE_CHECKING:
     from .util import FirmwareInfo
 
+
 _LOGGER = logging.getLogger(__name__)
 
 
+class HardwareFirmwareDiscoveryInfo(TypedDict):
+    """Data for triggering hardware integration discovery via firmware notification."""
+
+    usb_device: USBDevice | None
+    firmware_info: FirmwareInfo
+
+
 class SyncHardwareFirmwareInfoModule(Protocol):
     """Protocol type for Home Assistant Hardware firmware info platform modules."""
 
@@ -46,6 +60,23 @@ async def async_get_firmware_info(
 )
 
 
+@hass_callback
+def async_get_hardware_domain_for_usb_device(
+    hass: HomeAssistant, usb_device: USBDevice
+) -> str | None:
+    """Identify which hardware domain should handle a USB device."""
+    matched = async_get_usb_matchers_for_device(hass, usb_device)
+    hw_domains = {match["domain"] for match in matched} & HARDWARE_INTEGRATION_DOMAINS
+
+    if not hw_domains:
+        return None
+
+    # We can never have two hardware integrations overlap in discovery
+    assert len(hw_domains) == 1
+
+    return list(hw_domains)[0]
+
+
 class HardwareInfoDispatcher:
     """Central dispatcher for hardware/firmware information."""
 
@@ -94,14 +125,56 @@ async def notify_firmware_info(
             "Received firmware info notification from %r: %s", domain, firmware_info
         )
 
-        for callback in self._notification_callbacks.get(firmware_info.device, []):
+        for callback in list(self._notification_callbacks[firmware_info.device]):
             try:
                 callback(firmware_info)
             except Exception:
                 _LOGGER.exception(
                     "Error while notifying firmware info listener %s", callback
                 )
 
+        await self._async_trigger_hardware_discovery(firmware_info)
+
+    async def _async_trigger_hardware_discovery(
+        self, firmware_info: FirmwareInfo
+    ) -> None:
+        """Trigger hardware integration config flows from firmware info.
+
+        Identifies which hardware integration should handle the device based on
+        USB matchers, then triggers an import flow for only that integration.
+        """
+
+        usb_device = await self.hass.async_add_executor_job(
+            usb_device_from_path, firmware_info.device
+        )
+
+        if usb_device is None:
+            _LOGGER.debug("Cannot find USB for path %s", firmware_info.device)
+            return
+
+        hardware_domain = async_get_hardware_domain_for_usb_device(
+            self.hass, usb_device
+        )
+
+        if hardware_domain is None:
+            _LOGGER.debug("No hardware integration found for device %s", usb_device)
+            return
+
+        _LOGGER.debug(
+            "Triggering %s import flow for device %s",
+            hardware_domain,
+            firmware_info.device,
+        )
+
+        await self.hass.config_entries.flow.async_init(
+            hardware_domain,
+            context={"source": SOURCE_IMPORT},
+            data=HardwareFirmwareDiscoveryInfo(
+                usb_device=usb_device,
+                firmware_info=firmware_info,
+            ),
+        )
+
     async def iter_firmware_info(self) -> AsyncIterator[FirmwareInfo]:
         """Iterate over all firmware information for all hardware."""
         for domain, fw_info_module in self._providers.items():

homeassistant/components/homeassistant_hardware/manifest.json

@@ -3,6 +3,7 @@
   "name": "Home Assistant Hardware",
   "after_dependencies": ["hassio"],
   "codeowners": ["@home-assistant/core"],
+  "dependencies": ["usb"],
   "documentation": "https://www.home-assistant.io/integrations/homeassistant_hardware",
   "integration_type": "system",
   "requirements": [

homeassistant/components/homeassistant_sky_connect/config_flow.py

@@ -10,10 +10,17 @@
     firmware_config_flow,
     silabs_multiprotocol_addon,
 )
+from homeassistant.components.homeassistant_hardware.helpers import (
+    HardwareFirmwareDiscoveryInfo,
+)
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
     FirmwareInfo,
 )
+from homeassistant.components.usb import (
+    usb_service_info_from_device,
+    usb_unique_id_from_service_info,
+)
 from homeassistant.config_entries import (
     ConfigEntry,
     ConfigEntryBaseFlow,
@@ -142,32 +149,48 @@ def async_get_options_flow(
 
     async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResult:
         """Handle usb discovery."""
-        device = discovery_info.device
-        vid = discovery_info.vid
-        pid = discovery_info.pid
-        serial_number = discovery_info.serial_number
-        manufacturer = discovery_info.manufacturer
-        description = discovery_info.description
-        unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}"
+        unique_id = usb_unique_id_from_service_info(discovery_info)
 
         if await self.async_set_unique_id(unique_id):
-            self._abort_if_unique_id_configured(updates={DEVICE: device})
+            self._abort_if_unique_id_configured(updates={DEVICE: discovery_info.device})
 
         discovery_info.device = await self.hass.async_add_executor_job(
             usb.get_serial_by_id, discovery_info.device
         )
 
         self._usb_info = discovery_info
 
-        assert description is not None
-        self._hw_variant = HardwareVariant.from_usb_product_name(description)
+        assert discovery_info.description is not None
+        self._hw_variant = HardwareVariant.from_usb_product_name(
+            discovery_info.description
+        )
 
         # Set parent class attributes
         self._device = self._usb_info.device
         self._hardware_name = self._hw_variant.full_name
 
         return await self.async_step_confirm()
 
+    async def async_step_import(
+        self, fw_discovery_info: HardwareFirmwareDiscoveryInfo
+    ) -> ConfigFlowResult:
+        """Handle import from ZHA/OTBR firmware notification."""
+        assert fw_discovery_info["usb_device"] is not None
+        usb_info = usb_service_info_from_device(fw_discovery_info["usb_device"])
+        unique_id = usb_unique_id_from_service_info(usb_info)
+
+        if await self.async_set_unique_id(unique_id, raise_on_progress=False):
+            self._abort_if_unique_id_configured(updates={DEVICE: usb_info.device})
+
+        self._usb_info = usb_info
+        assert usb_info.description is not None
+        self._hw_variant = HardwareVariant.from_usb_product_name(usb_info.description)
+        self._device = usb_info.device
+        self._hardware_name = self._hw_variant.full_name
+        self._probed_firmware_info = fw_discovery_info["firmware_info"]
+
+        return self._async_flow_finished()
+
     def _async_flow_finished(self) -> ConfigFlowResult:
         """Create the config entry."""
         assert self._usb_info is not None

homeassistant/components/homeassistant_yellow/__init__.py

@@ -41,6 +41,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     firmware = ApplicationType(entry.data[FIRMWARE])
 
+    # Auto start the multiprotocol addon if it is in use
     if firmware is ApplicationType.CPC:
         try:
             await check_multi_pan_addon(hass)