Keep track of addons and integrations when determining HA radio firmw…

…are type (home-assistant#134598)

* Replace `FirmwareGuess` with `FirmwareInfo` with owner tracking

* Fix up config flow

* Account for OTBR addon existing independent of integration

* Fix remaining unit tests

* Add some tests for ownership

* Unit test `get_zha_firmware_info`

* ZHA `homeassistant_hardware` platform

* OTBR `homeassistant_hardware` platform

* Rework imports

* Fix unit tests

* Add OTBR unit tests

* Add hassfest exemption for `homeassistant_hardware` and `otbr`

* Invert registration to decouple the hardware integration

* Revert "Add hassfest exemption for `homeassistant_hardware` and `otbr`"

This reverts commit c8c6e70.

* Fix circular imports

* Fix unit tests

* Address review comments

* Simplify API a little

* Fix `| None` mypy issues

* Remove the `unregister_firmware_info_provider` API

* 100% coverage

* Add `HardwareInfoDispatcher.register_firmware_info_callback`

* Unit test `register_firmware_info_callback` (zha)

* Unit test `register_firmware_info_callback` (otbr)

* Update existing hardware helper tests to use the new function

* Add `async_` prefix to helper function names

* Move OTBR implementation to a separate PR

* Update ZHA diagnostics snapshot

* Switch from `dict.setdefault` to `defaultdict`

* Add some error handling to `iter_firmware_info` and increase test coverage

* Oops

homeassistant/components/homeassistant_hardware/__init__.py

@@ -6,10 +6,15 @@
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.typing import ConfigType
 
-DOMAIN = "homeassistant_hardware"
+from .const import DATA_COMPONENT, DOMAIN
+from .helpers import HardwareInfoDispatcher
+
 CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
 
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the component."""
+
+    hass.data[DATA_COMPONENT] = HardwareInfoDispatcher(hass)
+
     return True

homeassistant/components/homeassistant_hardware/const.py

@@ -1,10 +1,23 @@
 """Constants for the Homeassistant Hardware integration."""
 
+from __future__ import annotations
+
 import logging
+from typing import TYPE_CHECKING
+
+from homeassistant.util.hass_dict import HassKey
+
+if TYPE_CHECKING:
+    from .helpers import HardwareInfoDispatcher
+
 
 LOGGER = logging.getLogger(__package__)
 
+DOMAIN = "homeassistant_hardware"
+DATA_COMPONENT: HassKey[HardwareInfoDispatcher] = HassKey(DOMAIN)
+
 ZHA_DOMAIN = "zha"
+OTBR_DOMAIN = "otbr"
 
 OTBR_ADDON_NAME = "OpenThread Border Router"
 OTBR_ADDON_MANAGER_DATA = "openthread_border_router"

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -25,12 +25,14 @@
 from homeassistant.helpers.hassio import is_hassio
 
 from . import silabs_multiprotocol_addon
-from .const import ZHA_DOMAIN
+from .const import OTBR_DOMAIN, ZHA_DOMAIN
 from .util import (
     ApplicationType,
+    OwningAddon,
+    OwningIntegration,
     get_otbr_addon_manager,
-    get_zha_device_path,
     get_zigbee_flasher_addon_manager,
+    guess_hardware_owners,
     probe_silabs_firmware_type,
 )
 
@@ -519,19 +521,15 @@ async def async_step_pick_firmware_zigbee(
     ) -> ConfigFlowResult:
         """Pick Zigbee firmware."""
         assert self._device is not None
+        owners = await guess_hardware_owners(self.hass, self._device)
 
-        if is_hassio(self.hass):
-            otbr_manager = get_otbr_addon_manager(self.hass)
-            otbr_addon_info = await self._async_get_addon_info(otbr_manager)
-
-            if (
-                otbr_addon_info.state != AddonState.NOT_INSTALLED
-                and otbr_addon_info.options.get("device") == self._device
-            ):
-                raise AbortFlow(
-                    "otbr_still_using_stick",
-                    description_placeholders=self._get_translation_placeholders(),
-                )
+        for info in owners:
+            for owner in info.owners:
+                if info.source == OTBR_DOMAIN and isinstance(owner, OwningAddon):
+                    raise AbortFlow(
+                        "otbr_still_using_stick",
+                        description_placeholders=self._get_translation_placeholders(),
+                    )
 
         return await super().async_step_pick_firmware_zigbee(user_input)
 
@@ -541,15 +539,14 @@ async def async_step_pick_firmware_thread(
         """Pick Thread firmware."""
         assert self._device is not None
 
-        for zha_entry in self.hass.config_entries.async_entries(
-            ZHA_DOMAIN,
-            include_ignore=False,
-            include_disabled=True,
-        ):
-            if get_zha_device_path(zha_entry) == self._device:
-                raise AbortFlow(
-                    "zha_still_using_stick",
-                    description_placeholders=self._get_translation_placeholders(),
-                )
+        owners = await guess_hardware_owners(self.hass, self._device)
+
+        for info in owners:
+            for owner in info.owners:
+                if info.source == ZHA_DOMAIN and isinstance(owner, OwningIntegration):
+                    raise AbortFlow(
+                        "zha_still_using_stick",
+                        description_placeholders=self._get_translation_placeholders(),
+                    )
 
         return await super().async_step_pick_firmware_thread(user_input)

homeassistant/components/homeassistant_hardware/helpers.py

@@ -0,0 +1,143 @@
+"""Home Assistant Hardware integration helpers."""
+
+from collections import defaultdict
+from collections.abc import AsyncIterator, Awaitable, Callable
+import logging
+from typing import Protocol
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
+
+from . import DATA_COMPONENT
+from .util import FirmwareInfo
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SyncHardwareFirmwareInfoModule(Protocol):
+    """Protocol type for Home Assistant Hardware firmware info platform modules."""
+
+    def get_firmware_info(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+    ) -> FirmwareInfo | None:
+        """Return radio firmware information for the config entry, synchronously."""
+
+
+class AsyncHardwareFirmwareInfoModule(Protocol):
+    """Protocol type for Home Assistant Hardware firmware info platform modules."""
+
+    async def async_get_firmware_info(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+    ) -> FirmwareInfo | None:
+        """Return radio firmware information for the config entry, asynchronously."""
+
+
+type HardwareFirmwareInfoModule = (
+    SyncHardwareFirmwareInfoModule | AsyncHardwareFirmwareInfoModule
+)
+
+
+class HardwareInfoDispatcher:
+    """Central dispatcher for hardware/firmware information."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the dispatcher."""
+        self.hass = hass
+        self._providers: dict[str, HardwareFirmwareInfoModule] = {}
+        self._notification_callbacks: defaultdict[
+            str, set[Callable[[FirmwareInfo], None]]
+        ] = defaultdict(set)
+
+    def register_firmware_info_provider(
+        self, domain: str, platform: HardwareFirmwareInfoModule
+    ) -> None:
+        """Register a firmware info provider."""
+        if domain in self._providers:
+            raise ValueError(
+                f"Domain {domain} is already registered as a firmware info provider"
+            )
+
+        # There is no need to handle "unregistration" because integrations cannot be
+        # wholly removed at runtime
+        self._providers[domain] = platform
+        _LOGGER.debug(
+            "Registered firmware info provider from domain %r: %s", domain, platform
+        )
+
+    def register_firmware_info_callback(
+        self, device: str, callback: Callable[[FirmwareInfo], None]
+    ) -> CALLBACK_TYPE:
+        """Register a firmware info notification callback."""
+        self._notification_callbacks[device].add(callback)
+
+        @hass_callback
+        def async_remove_callback() -> None:
+            self._notification_callbacks[device].discard(callback)
+
+        return async_remove_callback
+
+    async def notify_firmware_info(
+        self, domain: str, firmware_info: FirmwareInfo
+    ) -> None:
+        """Notify the dispatcher of new firmware information."""
+        _LOGGER.debug(
+            "Received firmware info notification from %r: %s", domain, firmware_info
+        )
+
+        for callback in self._notification_callbacks.get(firmware_info.device, []):
+            try:
+                callback(firmware_info)
+            except Exception:
+                _LOGGER.exception(
+                    "Error while notifying firmware info listener %s", callback
+                )
+
+    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():
+            for config_entry in self.hass.config_entries.async_entries(domain):
+                try:
+                    if hasattr(fw_info_module, "get_firmware_info"):
+                        fw_info = fw_info_module.get_firmware_info(
+                            self.hass, config_entry
+                        )
+                    else:
+                        fw_info = await fw_info_module.async_get_firmware_info(
+                            self.hass, config_entry
+                        )
+                except Exception:
+                    _LOGGER.exception(
+                        "Error while getting firmware info from %r", fw_info_module
+                    )
+                    continue
+
+                if fw_info is not None:
+                    yield fw_info
+
+
+@hass_callback
+def async_register_firmware_info_provider(
+    hass: HomeAssistant, domain: str, platform: HardwareFirmwareInfoModule
+) -> None:
+    """Register a firmware info provider."""
+    return hass.data[DATA_COMPONENT].register_firmware_info_provider(domain, platform)
+
+
+@hass_callback
+def async_register_firmware_info_callback(
+    hass: HomeAssistant, device: str, callback: Callable[[FirmwareInfo], None]
+) -> CALLBACK_TYPE:
+    """Register a firmware info provider."""
+    return hass.data[DATA_COMPONENT].register_firmware_info_callback(device, callback)
+
+
+@hass_callback
+def async_notify_firmware_info(
+    hass: HomeAssistant, domain: str, firmware_info: FirmwareInfo
+) -> Awaitable[None]:
+    """Notify the dispatcher of new firmware information."""
+    return hass.data[DATA_COMPONENT].notify_firmware_info(domain, firmware_info)