Keep track of addons and integrations when determining HA radio firmware type (#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 c8c6e7044f005239d11fc561cca040a6d89a9b39.

* 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.core import HomeAssistant
 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.data_entry_flow import AbortFlow
 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,15 +521,11 @@ class BaseFirmwareOptionsFlow(BaseFirmwareInstallFlow, OptionsFlow):
     ) -> ConfigFlowResult:
         """Pick Zigbee firmware."""
         assert self._device is not None
+        owners = await guess_hardware_owners(self.hass, self._device)
 
-        if is_hassio(self.hass):
+        for info in owners:
-            otbr_manager = get_otbr_addon_manager(self.hass)
+            for owner in info.owners:
-            otbr_addon_info = await self._async_get_addon_info(otbr_manager)
+                if info.source == OTBR_DOMAIN and isinstance(owner, OwningAddon):
-
-            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(),
@@ -541,12 +539,11 @@ class BaseFirmwareOptionsFlow(BaseFirmwareInstallFlow, OptionsFlow):
         """Pick Thread firmware."""
         assert self._device is not None
 
-        for zha_entry in self.hass.config_entries.async_entries(
+        owners = await guess_hardware_owners(self.hass, self._device)
-            ZHA_DOMAIN,
+
-            include_ignore=False,
+        for info in owners:
-            include_disabled=True,
+            for owner in info.owners:
-        ):
+                if info.source == ZHA_DOMAIN and isinstance(owner, OwningIntegration):
-            if get_zha_device_path(zha_entry) == self._device:
                     raise AbortFlow(
                         "zha_still_using_stick",
                         description_placeholders=self._get_translation_placeholders(),

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)

homeassistant/components/homeassistant_hardware/util.py

@@ -2,27 +2,27 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections import defaultdict
 from collections.abc import Iterable
 from dataclasses import dataclass
 from enum import StrEnum
 import logging
-from typing import cast
 
 from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
 from universal_silabs_flasher.flasher import Flasher
 
 from homeassistant.components.hassio import AddonError, AddonState
-from homeassistant.config_entries import ConfigEntry, ConfigEntryState
+from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.hassio import is_hassio
 from homeassistant.helpers.singleton import singleton
 
+from . import DATA_COMPONENT
 from .const import (
     OTBR_ADDON_MANAGER_DATA,
     OTBR_ADDON_NAME,
     OTBR_ADDON_SLUG,
-    ZHA_DOMAIN,
     ZIGBEE_FLASHER_ADDON_MANAGER_DATA,
     ZIGBEE_FLASHER_ADDON_NAME,
     ZIGBEE_FLASHER_ADDON_SLUG,
@@ -55,11 +55,6 @@ class ApplicationType(StrEnum):
         return FlasherApplicationType(self.value)
 
 
-def get_zha_device_path(config_entry: ConfigEntry) -> str | None:
-    """Get the device path from a ZHA config entry."""
-    return cast(str | None, config_entry.data.get("device", {}).get("path", None))
-
-
 @singleton(OTBR_ADDON_MANAGER_DATA)
 @callback
 def get_otbr_addon_manager(hass: HomeAssistant) -> WaitingAddonManager:
@@ -84,31 +79,80 @@ def get_zigbee_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager
     )
 
 
-@dataclass(slots=True, kw_only=True)
+@dataclass(kw_only=True)
-class FirmwareGuess:
+class OwningAddon:
+    """Owning add-on."""
+
+    slug: str
+
+    def _get_addon_manager(self, hass: HomeAssistant) -> WaitingAddonManager:
+        return WaitingAddonManager(
+            hass,
+            _LOGGER,
+            f"Add-on {self.slug}",
+            self.slug,
+        )
+
+    async def is_running(self, hass: HomeAssistant) -> bool:
+        """Check if the add-on is running."""
+        addon_manager = self._get_addon_manager(hass)
+
+        try:
+            addon_info = await addon_manager.async_get_addon_info()
+        except AddonError:
+            return False
+        else:
+            return addon_info.state == AddonState.RUNNING
+
+
+@dataclass(kw_only=True)
+class OwningIntegration:
+    """Owning integration."""
+
+    config_entry_id: str
+
+    async def is_running(self, hass: HomeAssistant) -> bool:
+        """Check if the integration is running."""
+        if (entry := hass.config_entries.async_get_entry(self.config_entry_id)) is None:
+            return False
+
+        return entry.state in (
+            ConfigEntryState.LOADED,
+            ConfigEntryState.SETUP_RETRY,
+            ConfigEntryState.SETUP_IN_PROGRESS,
+        )
+
+
+@dataclass(kw_only=True)
+class FirmwareInfo:
     """Firmware guess."""
 
-    is_running: bool
+    device: str
     firmware_type: ApplicationType
+    firmware_version: str | None
+
     source: str
+    owners: list[OwningAddon | OwningIntegration]
+
+    async def is_running(self, hass: HomeAssistant) -> bool:
+        """Check if the firmware owner is running."""
+        states = await asyncio.gather(*(o.is_running(hass) for o in self.owners))
+        if not states:
+            return False
+
+        return all(states)
 
 
-async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> FirmwareGuess:
+async def guess_hardware_owners(
-    """Guess the firmware type based on installed addons and other integrations."""
+    hass: HomeAssistant, device_path: str
-    device_guesses: defaultdict[str | None, list[FirmwareGuess]] = defaultdict(list)
+) -> list[FirmwareInfo]:
+    """Guess the firmware info based on installed addons and other integrations."""
+    device_guesses: defaultdict[str, list[FirmwareInfo]] = defaultdict(list)
 
-    for zha_config_entry in hass.config_entries.async_entries(ZHA_DOMAIN):
+    async for firmware_info in hass.data[DATA_COMPONENT].iter_firmware_info():
-        zha_path = get_zha_device_path(zha_config_entry)
+        device_guesses[firmware_info.device].append(firmware_info)
-
-        if zha_path is not None:
-            device_guesses[zha_path].append(
-                FirmwareGuess(
-                    is_running=(zha_config_entry.state == ConfigEntryState.LOADED),
-                    firmware_type=ApplicationType.EZSP,
-                    source="zha",
-                )
-            )
 
+    # It may be possible for the OTBR addon to be present without the integration
     if is_hassio(hass):
         otbr_addon_manager = get_otbr_addon_manager(hass)
 
@@ -119,14 +163,22 @@ async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> Firmware
         else:
             if otbr_addon_info.state != AddonState.NOT_INSTALLED:
                 otbr_path = otbr_addon_info.options.get("device")
+
+                # Only create a new entry if there are no existing OTBR ones
+                if otbr_path is not None and not any(
+                    info.source == "otbr" for info in device_guesses[otbr_path]
+                ):
                     device_guesses[otbr_path].append(
-                    FirmwareGuess(
+                        FirmwareInfo(
-                        is_running=(otbr_addon_info.state == AddonState.RUNNING),
+                            device=otbr_path,
                             firmware_type=ApplicationType.SPINEL,
+                            firmware_version=None,
                             source="otbr",
+                            owners=[OwningAddon(slug=otbr_addon_manager.addon_slug)],
                         )
                     )
 
+    if is_hassio(hass):
         multipan_addon_manager = await get_multiprotocol_addon_manager(hass)
 
         try:
@@ -136,30 +188,48 @@ async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> Firmware
         else:
             if multipan_addon_info.state != AddonState.NOT_INSTALLED:
                 multipan_path = multipan_addon_info.options.get("device")
+
+                if multipan_path is not None:
                     device_guesses[multipan_path].append(
-                    FirmwareGuess(
+                        FirmwareInfo(
-                        is_running=(multipan_addon_info.state == AddonState.RUNNING),
+                            device=multipan_path,
                             firmware_type=ApplicationType.CPC,
+                            firmware_version=None,
                             source="multiprotocol",
+                            owners=[
+                                OwningAddon(slug=multipan_addon_manager.addon_slug)
+                            ],
                         )
                     )
 
-    # Fall back to EZSP if we can't guess the firmware type
+    return device_guesses.get(device_path, [])
-    if device_path not in device_guesses:
+
-        return FirmwareGuess(
+
-            is_running=False, firmware_type=ApplicationType.EZSP, source="unknown"
+async def guess_firmware_info(hass: HomeAssistant, device_path: str) -> FirmwareInfo:
-        )
+    """Guess the firmware type based on installed addons and other integrations."""
-
+
-    # Prioritizes guesses that were pulled from a running addon or integration but keep
+    hardware_owners = await guess_hardware_owners(hass, device_path)
-    # the sort order we defined above
+
-    guesses = sorted(
+    # Fall back to EZSP if we have no way to guess
-        device_guesses[device_path],
+    if not hardware_owners:
-        key=lambda guess: guess.is_running,
+        return FirmwareInfo(
+            device=device_path,
+            firmware_type=ApplicationType.EZSP,
+            firmware_version=None,
+            source="unknown",
+            owners=[],
         )
 
+    # Prioritize guesses that are pulled from a real source
+    guesses = [
+        (guess, sum([await owner.is_running(hass) for owner in guess.owners]))
+        for guess in hardware_owners
+    ]
+    guesses.sort(key=lambda p: p[1])
     assert guesses
 
-    return guesses[-1]
+    # Pick the best one. We use a stable sort so ZHA < OTBR < multi-PAN
+    return guesses[-1][0]
 
 
 async def probe_silabs_firmware_type(

homeassistant/components/homeassistant_sky_connect/__init__.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import logging
 
-from homeassistant.components.homeassistant_hardware.util import guess_firmware_type
+from homeassistant.components.homeassistant_hardware.util import guess_firmware_info
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 
@@ -33,7 +33,7 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
             # Add-on startup with type service get started before Core, always (e.g. the
             # Multi-Protocol add-on). Probing the firmware would interfere with the add-on,
             # so we can't safely probe here. Instead, we must make an educated guess!
-            firmware_guess = await guess_firmware_type(
+            firmware_guess = await guess_firmware_info(
                 hass, config_entry.data["device"]
             )
 

homeassistant/components/homeassistant_yellow/__init__.py

@@ -10,7 +10,7 @@ from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon
 )
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
-    guess_firmware_type,
+    guess_firmware_info,
 )
 from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -75,7 +75,7 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
             # Add-on startup with type service get started before Core, always (e.g. the
             # Multi-Protocol add-on). Probing the firmware would interfere with the add-on,
             # so we can't safely probe here. Instead, we must make an educated guess!
-            firmware_guess = await guess_firmware_type(hass, RADIO_DEVICE)
+            firmware_guess = await guess_firmware_info(hass, RADIO_DEVICE)
 
             new_data = {**config_entry.data}
             new_data[FIRMWARE] = firmware_guess.firmware_type.value

homeassistant/components/zha/__init__.py

@@ -12,6 +12,10 @@ from zha.zigbee.device import get_device_automation_triggers
 from zigpy.config import CONF_DATABASE, CONF_DEVICE, CONF_DEVICE_PATH
 from zigpy.exceptions import NetworkSettingsInconsistent, TransientConnectionError
 
+from homeassistant.components.homeassistant_hardware.helpers import (
+    async_notify_firmware_info,
+    async_register_firmware_info_provider,
+)
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     CONF_TYPE,
@@ -25,7 +29,7 @@ from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.typing import ConfigType
 
-from . import repairs, websocket_api
+from . import homeassistant_hardware, repairs, websocket_api
 from .const import (
     CONF_BAUDRATE,
     CONF_CUSTOM_QUIRKS_PATH,
@@ -110,6 +114,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     ha_zha_data = HAZHAData(yaml_config=config.get(DOMAIN, {}))
     hass.data[DATA_ZHA] = ha_zha_data
 
+    async_register_firmware_info_provider(hass, DOMAIN, homeassistant_hardware)
+
     return True
 
 
@@ -218,6 +224,13 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
         hass.bus.async_listen(EVENT_CORE_CONFIG_UPDATE, update_config)
     )
 
+    if fw_info := homeassistant_hardware.get_firmware_info(hass, config_entry):
+        await async_notify_firmware_info(
+            hass,
+            DOMAIN,
+            firmware_info=fw_info,
+        )
+
     await ha_zha_data.gateway_proxy.async_initialize_devices_and_entities()
     await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
     async_dispatcher_send(hass, SIGNAL_ADD_ENTITIES)

homeassistant/components/zha/homeassistant_hardware.py

@@ -0,0 +1,43 @@
+"""Home Assistant Hardware firmware utilities."""
+
+from __future__ import annotations
+
+from homeassistant.components.homeassistant_hardware.util import (
+    ApplicationType,
+    FirmwareInfo,
+    OwningIntegration,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+
+from .const import DOMAIN
+from .helpers import get_zha_gateway
+
+
+@callback
+def get_firmware_info(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> FirmwareInfo | None:
+    """Return firmware information for the ZHA instance, synchronously."""
+
+    # We only support EZSP firmware for now
+    if config_entry.data.get("radio_type", None) != "ezsp":
+        return None
+
+    if (device := config_entry.data.get("device", {}).get("path")) is None:
+        return None
+
+    try:
+        gateway = get_zha_gateway(hass)
+    except ValueError:
+        firmware_version = None
+    else:
+        firmware_version = gateway.state.node_info.version
+
+    return FirmwareInfo(
+        device=device,
+        firmware_type=ApplicationType.EZSP,
+        firmware_version=firmware_version,
+        source=DOMAIN,
+        owners=[OwningIntegration(config_entry_id=config_entry.entry_id)],
+    )

tests/components/homeassistant_hardware/test_config_flow.py

@@ -23,6 +23,7 @@ from homeassistant.components.homeassistant_hardware.util import (
 from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.setup import async_setup_component
 
 from tests.common import (
     MockConfigEntry,
@@ -106,7 +107,7 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow):
 
 
 @pytest.fixture(autouse=True)
-def mock_test_firmware_platform(
+async def mock_test_firmware_platform(
     hass: HomeAssistant,
 ) -> Generator[None]:
     """Fixture for a test config flow."""
@@ -116,6 +117,8 @@ def mock_test_firmware_platform(
     mock_integration(hass, mock_module)
     mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
 
+    await async_setup_component(hass, "homeassistant_hardware", {})
+
     with mock_config_flow(TEST_DOMAIN, FakeFirmwareConfigFlow):
         yield
 
@@ -189,6 +192,10 @@ def mock_addon_info(
             "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager",
             return_value=mock_otbr_manager,
         ),
+        patch(
+            "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager",
+            return_value=mock_otbr_manager,
+        ),
         patch(
             "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_zigbee_flasher_addon_manager",
             return_value=mock_flasher_manager,
@@ -197,6 +204,10 @@ def mock_addon_info(
             "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio",
             return_value=is_hassio,
         ),
+        patch(
+            "homeassistant.components.homeassistant_hardware.util.is_hassio",
+            return_value=is_hassio,
+        ),
         patch(
             "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_type",
             return_value=app_type,

tests/components/homeassistant_hardware/test_config_flow_failures.py

@@ -1,6 +1,6 @@
 """Test the Home Assistant hardware firmware config flow failure cases."""
 
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -9,7 +9,11 @@ from homeassistant.components.homeassistant_hardware.firmware_config_flow import
     STEP_PICK_FIRMWARE_THREAD,
     STEP_PICK_FIRMWARE_ZIGBEE,
 )
-from homeassistant.components.homeassistant_hardware.util import ApplicationType
+from homeassistant.components.homeassistant_hardware.util import (
+    ApplicationType,
+    FirmwareInfo,
+    OwningIntegration,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
 
@@ -548,13 +552,19 @@ async def test_options_flow_zigbee_to_thread_zha_configured(
 
     assert await hass.config_entries.async_setup(config_entry.entry_id)
 
-    # Set up ZHA as well
+    # Pretend ZHA is using the stick
-    zha_config_entry = MockConfigEntry(
+    with patch(
-        domain="zha",
+        "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners",
-        data={"device": {"path": TEST_DEVICE}},
+        return_value=[
+            FirmwareInfo(
+                device=TEST_DEVICE,
+                firmware_type=ApplicationType.EZSP,
+                firmware_version="1.2.3.4",
+                source="zha",
+                owners=[OwningIntegration(config_entry_id="some_config_entry_id")],
             )
-    zha_config_entry.add_to_hass(hass)
+        ],
-
+    ):
         # Confirm options flow
         result = await hass.config_entries.options.async_init(config_entry.entry_id)
 
@@ -563,6 +573,7 @@ async def test_options_flow_zigbee_to_thread_zha_configured(
             result["flow_id"],
             user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
         )
+
     assert result["type"] == FlowResultType.ABORT
     assert result["reason"] == "zha_still_using_stick"
 

tests/components/homeassistant_hardware/test_helpers.py

@@ -0,0 +1,185 @@
+"""Test hardware helpers."""
+
+import logging
+from unittest.mock import AsyncMock, MagicMock, Mock, call
+
+import pytest
+
+from homeassistant.components.homeassistant_hardware.const import DATA_COMPONENT
+from homeassistant.components.homeassistant_hardware.helpers import (
+    async_notify_firmware_info,
+    async_register_firmware_info_callback,
+    async_register_firmware_info_provider,
+)
+from homeassistant.components.homeassistant_hardware.util import (
+    ApplicationType,
+    FirmwareInfo,
+)
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+
+FIRMWARE_INFO_EZSP = FirmwareInfo(
+    device="/dev/serial/by-id/device1",
+    firmware_type=ApplicationType.EZSP,
+    firmware_version=None,
+    source="zha",
+    owners=[AsyncMock(is_running=AsyncMock(return_value=True))],
+)
+
+FIRMWARE_INFO_SPINEL = FirmwareInfo(
+    device="/dev/serial/by-id/device2",
+    firmware_type=ApplicationType.SPINEL,
+    firmware_version=None,
+    source="otbr",
+    owners=[AsyncMock(is_running=AsyncMock(return_value=True))],
+)
+
+
+async def test_dispatcher_registration(hass: HomeAssistant) -> None:
+    """Test HardwareInfoDispatcher registration."""
+
+    await async_setup_component(hass, "homeassistant_hardware", {})
+
+    # Mock provider 1 with a synchronous method to pull firmware info
+    provider1_config_entry = MockConfigEntry(
+        domain="zha",
+        unique_id="some_unique_id1",
+        data={},
+    )
+    provider1_config_entry.add_to_hass(hass)
+    provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED)
+
+    provider1_firmware = MagicMock(spec=["get_firmware_info"])
+    provider1_firmware.get_firmware_info = MagicMock(return_value=FIRMWARE_INFO_EZSP)
+    async_register_firmware_info_provider(hass, "zha", provider1_firmware)
+
+    # Mock provider 2 with an asynchronous method to pull firmware info
+    provider2_config_entry = MockConfigEntry(
+        domain="otbr",
+        unique_id="some_unique_id2",
+        data={},
+    )
+    provider2_config_entry.add_to_hass(hass)
+    provider2_config_entry.mock_state(hass, ConfigEntryState.LOADED)
+
+    provider2_firmware = MagicMock(spec=["async_get_firmware_info"])
+    provider2_firmware.async_get_firmware_info = AsyncMock(
+        return_value=FIRMWARE_INFO_SPINEL
+    )
+    async_register_firmware_info_provider(hass, "otbr", provider2_firmware)
+
+    # Double registration won't work
+    with pytest.raises(ValueError, match="Domain zha is already registered"):
+        async_register_firmware_info_provider(hass, "zha", provider1_firmware)
+
+    # We can iterate over the results
+    info = [i async for i in hass.data[DATA_COMPONENT].iter_firmware_info()]
+    assert info == [
+        FIRMWARE_INFO_EZSP,
+        FIRMWARE_INFO_SPINEL,
+    ]
+
+    callback1 = Mock()
+    cancel1 = async_register_firmware_info_callback(
+        hass, "/dev/serial/by-id/device1", callback1
+    )
+
+    callback2 = Mock()
+    cancel2 = async_register_firmware_info_callback(
+        hass, "/dev/serial/by-id/device2", callback2
+    )
+
+    # And receive notification callbacks
+    await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP)
+    await async_notify_firmware_info(hass, "otbr", firmware_info=FIRMWARE_INFO_SPINEL)
+    await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP)
+    cancel1()
+    await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP)
+    await async_notify_firmware_info(hass, "otbr", firmware_info=FIRMWARE_INFO_SPINEL)
+    cancel2()
+
+    assert callback1.mock_calls == [
+        call(FIRMWARE_INFO_EZSP),
+        call(FIRMWARE_INFO_EZSP),
+    ]
+
+    assert callback2.mock_calls == [
+        call(FIRMWARE_INFO_SPINEL),
+        call(FIRMWARE_INFO_SPINEL),
+    ]
+
+
+async def test_dispatcher_iter_error_handling(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test HardwareInfoDispatcher ignoring errors from firmware info providers."""
+
+    await async_setup_component(hass, "homeassistant_hardware", {})
+
+    provider1_config_entry = MockConfigEntry(
+        domain="zha",
+        unique_id="some_unique_id1",
+        data={},
+    )
+    provider1_config_entry.add_to_hass(hass)
+    provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED)
+
+    provider1_firmware = MagicMock(spec=["get_firmware_info"])
+    provider1_firmware.get_firmware_info = MagicMock(side_effect=Exception("Boom!"))
+    async_register_firmware_info_provider(hass, "zha", provider1_firmware)
+
+    provider2_config_entry = MockConfigEntry(
+        domain="otbr",
+        unique_id="some_unique_id2",
+        data={},
+    )
+    provider2_config_entry.add_to_hass(hass)
+    provider2_config_entry.mock_state(hass, ConfigEntryState.LOADED)
+
+    provider2_firmware = MagicMock(spec=["async_get_firmware_info"])
+    provider2_firmware.async_get_firmware_info = AsyncMock(
+        return_value=FIRMWARE_INFO_SPINEL
+    )
+    async_register_firmware_info_provider(hass, "otbr", provider2_firmware)
+
+    with caplog.at_level(logging.ERROR):
+        info = [i async for i in hass.data[DATA_COMPONENT].iter_firmware_info()]
+
+    assert info == [FIRMWARE_INFO_SPINEL]
+    assert "Error while getting firmware info from" in caplog.text
+
+
+async def test_dispatcher_callback_error_handling(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test HardwareInfoDispatcher ignoring errors from firmware info callbacks."""
+
+    await async_setup_component(hass, "homeassistant_hardware", {})
+    provider1_config_entry = MockConfigEntry(
+        domain="zha",
+        unique_id="some_unique_id1",
+        data={},
+    )
+    provider1_config_entry.add_to_hass(hass)
+    provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED)
+
+    provider1_firmware = MagicMock(spec=["get_firmware_info"])
+    provider1_firmware.get_firmware_info = MagicMock(return_value=FIRMWARE_INFO_EZSP)
+    async_register_firmware_info_provider(hass, "zha", provider1_firmware)
+
+    callback1 = Mock(side_effect=Exception("Some error"))
+    async_register_firmware_info_callback(hass, "/dev/serial/by-id/device1", callback1)
+
+    callback2 = Mock()
+    async_register_firmware_info_callback(hass, "/dev/serial/by-id/device1", callback2)
+
+    with caplog.at_level(logging.ERROR):
+        await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP)
+
+    assert "Error while notifying firmware info listener" in caplog.text
+
+    assert callback1.mock_calls == [call(FIRMWARE_INFO_EZSP)]
+    assert callback2.mock_calls == [call(FIRMWARE_INFO_EZSP)]

tests/components/homeassistant_hardware/test_util.py

@@ -1,18 +1,21 @@
 """Test hardware utilities."""
 
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from homeassistant.components.hassio import AddonError, AddonInfo, AddonState
+from homeassistant.components.homeassistant_hardware.helpers import (
+    async_register_firmware_info_provider,
+)
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
-    FirmwareGuess,
+    FirmwareInfo,
-    FlasherApplicationType,
+    OwningAddon,
-    get_zha_device_path,
+    OwningIntegration,
-    guess_firmware_type,
+    guess_firmware_info,
-    probe_silabs_firmware_type,
 )
-from homeassistant.config_entries import ConfigEntryState
+from homeassistant.config_entries import ConfigEntry, ConfigEntryState
 from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
 
 from tests.common import MockConfigEntry
 
@@ -21,7 +24,21 @@ ZHA_CONFIG_ENTRY = MockConfigEntry(
     unique_id="some_unique_id",
     data={
         "device": {
-            "path": "socket://1.2.3.4:5678",
+            "path": "/dev/ttyUSB1",
+            "baudrate": 115200,
+            "flow_control": None,
+        },
+        "radio_type": "ezsp",
+    },
+    version=4,
+)
+
+ZHA_CONFIG_ENTRY2 = MockConfigEntry(
+    domain="zha",
+    unique_id="some_other_unique_id",
+    data={
+        "device": {
+            "path": "/dev/ttyUSB2",
             "baudrate": 115200,
             "flow_control": None,
         },
@@ -31,153 +48,202 @@ ZHA_CONFIG_ENTRY = MockConfigEntry(
 )
 
 
-def test_get_zha_device_path() -> None:
+async def test_guess_firmware_info_unknown(hass: HomeAssistant) -> None:
-    """Test extracting the ZHA device path from its config entry."""
-    assert (
-        get_zha_device_path(ZHA_CONFIG_ENTRY) == ZHA_CONFIG_ENTRY.data["device"]["path"]
-    )
-
-
-def test_get_zha_device_path_ignored_discovery() -> None:
-    """Test extracting the ZHA device path from an ignored ZHA discovery."""
-    config_entry = MockConfigEntry(
-        domain="zha",
-        unique_id="some_unique_id",
-        data={},
-        version=4,
-    )
-
-    assert get_zha_device_path(config_entry) is None
-
-
-async def test_guess_firmware_type_unknown(hass: HomeAssistant) -> None:
     """Test guessing the firmware type."""
 
-    assert (await guess_firmware_type(hass, "/dev/missing")) == FirmwareGuess(
+    await async_setup_component(hass, "homeassistant_hardware", {})
-        is_running=False, firmware_type=ApplicationType.EZSP, source="unknown"
+
+    assert (await guess_firmware_info(hass, "/dev/missing")) == FirmwareInfo(
+        device="/dev/missing",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version=None,
+        source="unknown",
+        owners=[],
     )
 
 
-async def test_guess_firmware_type(hass: HomeAssistant) -> None:
+async def test_guess_firmware_info_integrations(hass: HomeAssistant) -> None:
-    """Test guessing the firmware."""
+    """Test guessing the firmware via OTBR and ZHA."""
-    path = ZHA_CONFIG_ENTRY.data["device"]["path"]
 
-    ZHA_CONFIG_ENTRY.add_to_hass(hass)
+    await async_setup_component(hass, "homeassistant_hardware", {})
 
-    ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.NOT_LOADED)
+    # One instance of ZHA and two OTBRs
-    assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
+    zha = MockConfigEntry(domain="zha", unique_id="some_unique_id_1")
-        is_running=False, firmware_type=ApplicationType.EZSP, source="zha"
+    zha.add_to_hass(hass)
+
+    otbr1 = MockConfigEntry(domain="otbr", unique_id="some_unique_id_2")
+    otbr1.add_to_hass(hass)
+
+    otbr2 = MockConfigEntry(domain="otbr", unique_id="some_unique_id_3")
+    otbr2.add_to_hass(hass)
+
+    # First ZHA is running with the stick
+    zha_firmware_info = FirmwareInfo(
+        device="/dev/serial/by-id/device1",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version=None,
+        source="zha",
+        owners=[AsyncMock(is_running=AsyncMock(return_value=True))],
     )
 
-    # When ZHA is running, we indicate as such when guessing
+    # First OTBR: neither the addon or the integration are loaded
-    ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.LOADED)
+    otbr_firmware_info1 = FirmwareInfo(
-    assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
+        device="/dev/serial/by-id/device1",
-        is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
+        firmware_type=ApplicationType.SPINEL,
+        firmware_version=None,
+        source="otbr",
+        owners=[
+            AsyncMock(is_running=AsyncMock(return_value=False)),
+            AsyncMock(is_running=AsyncMock(return_value=False)),
+        ],
     )
 
-    mock_otbr_addon_manager = AsyncMock()
+    # Second OTBR: fully running but is with an unrelated device
-    mock_multipan_addon_manager = AsyncMock()
+    otbr_firmware_info2 = FirmwareInfo(
-
+        device="/dev/serial/by-id/device2",  # An unrelated device
-    with (
+        firmware_type=ApplicationType.SPINEL,
-        patch(
+        firmware_version=None,
-            "homeassistant.components.homeassistant_hardware.util.is_hassio",
+        source="otbr",
-            return_value=True,
+        owners=[
-        ),
+            AsyncMock(is_running=AsyncMock(return_value=True)),
-        patch(
+            AsyncMock(is_running=AsyncMock(return_value=True)),
-            "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager",
+        ],
-            return_value=mock_otbr_addon_manager,
-        ),
-        patch(
-            "homeassistant.components.homeassistant_hardware.util.get_multiprotocol_addon_manager",
-            return_value=mock_multipan_addon_manager,
-        ),
-    ):
-        mock_otbr_addon_manager.async_get_addon_info.side_effect = AddonError()
-        mock_multipan_addon_manager.async_get_addon_info.side_effect = AddonError()
-
-        # Hassio errors are ignored and we still go with ZHA
-        assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
-            is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
     )
 
-        mock_otbr_addon_manager.async_get_addon_info.side_effect = None
+    mock_zha_hardware_info = MagicMock(spec=["get_firmware_info"])
-        mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
+    mock_zha_hardware_info.get_firmware_info = MagicMock(return_value=zha_firmware_info)
+    async_register_firmware_info_provider(hass, "zha", mock_zha_hardware_info)
+
+    async def mock_otbr_async_get_firmware_info(
+        hass: HomeAssistant, config_entry: ConfigEntry
+    ) -> FirmwareInfo | None:
+        return {
+            otbr1.entry_id: otbr_firmware_info1,
+            otbr2.entry_id: otbr_firmware_info2,
+        }.get(config_entry.entry_id)
+
+    mock_otbr_hardware_info = MagicMock(spec=["async_get_firmware_info"])
+    mock_otbr_hardware_info.async_get_firmware_info = AsyncMock(
+        side_effect=mock_otbr_async_get_firmware_info
+    )
+    async_register_firmware_info_provider(hass, "otbr", mock_otbr_hardware_info)
+
+    # ZHA wins for the first stick, since it's actually running
+    assert (
+        await guess_firmware_info(hass, "/dev/serial/by-id/device1")
+    ) == zha_firmware_info
+
+    # Second stick is communicating exclusively with the second OTBR
+    assert (
+        await guess_firmware_info(hass, "/dev/serial/by-id/device2")
+    ) == otbr_firmware_info2
+
+    # If we stop ZHA, OTBR will take priority
+    zha_firmware_info.owners[0].is_running.return_value = False
+    otbr_firmware_info1.owners[0].is_running.return_value = True
+    assert (
+        await guess_firmware_info(hass, "/dev/serial/by-id/device1")
+    ) == otbr_firmware_info1
+
+
+async def test_owning_addon(hass: HomeAssistant) -> None:
+    """Test `OwningAddon`."""
+
+    owning_addon = OwningAddon(slug="some-addon-slug")
+
+    # Explicitly running
+    with patch(
+        "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager"
+    ) as mock_manager:
+        mock_manager.return_value.async_get_addon_info = AsyncMock(
+            return_value=AddonInfo(
                 available=True,
-            hostname=None,
+                hostname="core_some_addon_slug",
-            options={"device": "/some/other/device"},
+                options={},
                 state=AddonState.RUNNING,
                 update_available=False,
                 version="1.0.0",
             )
-
-        # We will prefer ZHA, as it is running (and actually pointing to the device)
-        assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
-            is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
         )
+        assert (await owning_addon.is_running(hass)) is True
 
-        mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
+    # Explicitly not running
+    with patch(
+        "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager"
+    ) as mock_manager:
+        mock_manager.return_value.async_get_addon_info = AsyncMock(
+            return_value=AddonInfo(
                 available=True,
-            hostname=None,
+                hostname="core_some_addon_slug",
-            options={"device": path},
+                options={},
                 state=AddonState.NOT_RUNNING,
                 update_available=False,
                 version="1.0.0",
             )
-
-        # We will still prefer ZHA, as it is the one actually running
-        assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
-            is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
         )
+        assert (await owning_addon.is_running(hass)) is False
 
-        mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
+    # Failed to get status
-            available=True,
-            hostname=None,
-            options={"device": path},
-            state=AddonState.RUNNING,
-            update_available=False,
-            version="1.0.0",
-        )
-
-        # Finally, ZHA loses out to OTBR
-        assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
-            is_running=True, firmware_type=ApplicationType.SPINEL, source="otbr"
-        )
-
-        mock_multipan_addon_manager.async_get_addon_info.side_effect = None
-        mock_multipan_addon_manager.async_get_addon_info.return_value = AddonInfo(
-            available=True,
-            hostname=None,
-            options={"device": path},
-            state=AddonState.RUNNING,
-            update_available=False,
-            version="1.0.0",
-        )
-
-        # Which will lose out to multi-PAN
-        assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
-            is_running=True, firmware_type=ApplicationType.CPC, source="multiprotocol"
-        )
-
-
-async def test_probe_silabs_firmware_type() -> None:
-    """Test probing Silabs firmware type."""
-
     with patch(
-        "homeassistant.components.homeassistant_hardware.util.Flasher.probe_app_type",
+        "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager"
-        side_effect=RuntimeError,
+    ) as mock_manager:
-    ):
+        mock_manager.return_value.async_get_addon_info = AsyncMock(
-        assert (await probe_silabs_firmware_type("/dev/ttyUSB0")) is None
+            side_effect=AddonError()
-
-    with patch(
-        "homeassistant.components.homeassistant_hardware.util.Flasher.probe_app_type",
-        side_effect=lambda self: setattr(self, "app_type", FlasherApplicationType.EZSP),
-        autospec=True,
-    ) as mock_probe_app_type:
-        # The application type constant is converted back and forth transparently
-        result = await probe_silabs_firmware_type(
-            "/dev/ttyUSB0", probe_methods=[ApplicationType.EZSP]
         )
-        assert result is ApplicationType.EZSP
+        assert (await owning_addon.is_running(hass)) is False
 
-        flasher = mock_probe_app_type.mock_calls[0].args[0]
+
-        assert flasher._probe_methods == [FlasherApplicationType.EZSP]
+async def test_owning_integration(hass: HomeAssistant) -> None:
+    """Test `OwningIntegration`."""
+    config_entry = MockConfigEntry(domain="mock_domain", unique_id="some_unique_id")
+    config_entry.add_to_hass(hass)
+
+    owning_integration = OwningIntegration(config_entry_id=config_entry.entry_id)
+
+    # Explicitly running
+    config_entry.mock_state(hass, ConfigEntryState.LOADED)
+    assert (await owning_integration.is_running(hass)) is True
+
+    # Explicitly not running
+    config_entry.mock_state(hass, ConfigEntryState.NOT_LOADED)
+    assert (await owning_integration.is_running(hass)) is False
+
+    # Missing config entry
+    owning_integration2 = OwningIntegration(config_entry_id="some_nonexistenct_id")
+    assert (await owning_integration2.is_running(hass)) is False
+
+
+async def test_firmware_info(hass: HomeAssistant) -> None:
+    """Test `FirmwareInfo`."""
+
+    owner1 = AsyncMock()
+    owner2 = AsyncMock()
+
+    firmware_info = FirmwareInfo(
+        device="/dev/ttyUSB1",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version="1.0.0",
+        source="zha",
+        owners=[owner1, owner2],
+    )
+
+    # Both running
+    owner1.is_running.return_value = True
+    owner2.is_running.return_value = True
+    assert (await firmware_info.is_running(hass)) is True
+
+    # Only one running
+    owner1.is_running.return_value = True
+    owner2.is_running.return_value = False
+    assert (await firmware_info.is_running(hass)) is False
+
+    # No owners
+    firmware_info2 = FirmwareInfo(
+        device="/dev/ttyUSB1",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version="1.0.0",
+        source="zha",
+        owners=[],
+    )
+
+    assert (await firmware_info2.is_running(hass)) is False

tests/components/homeassistant_sky_connect/test_init.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
-    FirmwareGuess,
+    FirmwareInfo,
 )
 from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
 from homeassistant.core import HomeAssistant
@@ -32,11 +32,13 @@ async def test_config_entry_migration_v2(hass: HomeAssistant) -> None:
     config_entry.add_to_hass(hass)
 
     with patch(
-        "homeassistant.components.homeassistant_sky_connect.guess_firmware_type",
+        "homeassistant.components.homeassistant_sky_connect.guess_firmware_info",
-        return_value=FirmwareGuess(
+        return_value=FirmwareInfo(
-            is_running=True,
+            device="/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
+            firmware_version=None,
             firmware_type=ApplicationType.SPINEL,
             source="otbr",
+            owners=[],
         ),
     ):
         await hass.config_entries.async_setup(config_entry.entry_id)

tests/components/homeassistant_yellow/test_init.py

@@ -8,7 +8,7 @@ from homeassistant.components import zha
 from homeassistant.components.hassio import DOMAIN as HASSIO_DOMAIN
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
-    FirmwareGuess,
+    FirmwareInfo,
 )
 from homeassistant.components.homeassistant_yellow.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
@@ -49,11 +49,13 @@ async def test_setup_entry(
             return_value=onboarded,
         ),
         patch(
-            "homeassistant.components.homeassistant_yellow.guess_firmware_type",
+            "homeassistant.components.homeassistant_yellow.guess_firmware_info",
-            return_value=FirmwareGuess(  # Nothing is setup
+            return_value=FirmwareInfo(  # Nothing is setup
-                is_running=False,
+                device="/dev/ttyAMA1",
+                firmware_version=None,
                 firmware_type=ApplicationType.EZSP,
                 source="unknown",
+                owners=[],
             ),
         ),
     ):

tests/components/zha/conftest.py

@@ -155,6 +155,7 @@ async def zigpy_app_controller():
     app.state.node_info.ieee = zigpy.types.EUI64.convert("00:15:8d:00:02:32:4f:32")
     app.state.node_info.manufacturer = "Coordinator Manufacturer"
     app.state.node_info.model = "Coordinator Model"
+    app.state.node_info.version = "7.1.4.0 build 389"
     app.state.network_info.pan_id = 0x1234
     app.state.network_info.extended_pan_id = app.state.node_info.ieee
     app.state.network_info.channel = 15

tests/components/zha/snapshots/test_diagnostics.ambr

@@ -75,7 +75,7 @@
         'manufacturer': 'Coordinator Manufacturer',
         'model': 'Coordinator Model',
         'nwk': 0,
-        'version': None,
+        'version': '7.1.4.0 build 389',
       }),
     }),
     'config': dict({

tests/components/zha/test_homeassistant_hardware.py

@@ -0,0 +1,120 @@
+"""Test Home Assistant Hardware platform for ZHA."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from zigpy.application import ControllerApplication
+
+from homeassistant.components.homeassistant_hardware.helpers import (
+    async_register_firmware_info_callback,
+)
+from homeassistant.components.homeassistant_hardware.util import (
+    ApplicationType,
+    FirmwareInfo,
+    OwningIntegration,
+)
+from homeassistant.components.zha.homeassistant_hardware import get_firmware_info
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+
+
+async def test_get_firmware_info_normal(hass: HomeAssistant) -> None:
+    """Test `get_firmware_info`."""
+
+    zha = MockConfigEntry(
+        domain="zha",
+        unique_id="some_unique_id",
+        data={
+            "device": {
+                "path": "/dev/ttyUSB1",
+                "baudrate": 115200,
+                "flow_control": None,
+            },
+            "radio_type": "ezsp",
+        },
+        version=4,
+    )
+    zha.add_to_hass(hass)
+    zha.mock_state(hass, ConfigEntryState.LOADED)
+
+    # With ZHA running
+    with patch(
+        "homeassistant.components.zha.homeassistant_hardware.get_zha_gateway"
+    ) as mock_get_zha_gateway:
+        mock_get_zha_gateway.return_value.state.node_info.version = "1.2.3.4"
+        fw_info_running = get_firmware_info(hass, zha)
+
+    assert fw_info_running == FirmwareInfo(
+        device="/dev/ttyUSB1",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version="1.2.3.4",
+        source="zha",
+        owners=[OwningIntegration(config_entry_id=zha.entry_id)],
+    )
+    assert await fw_info_running.is_running(hass) is True
+
+    # With ZHA not running
+    zha.mock_state(hass, ConfigEntryState.NOT_LOADED)
+    fw_info_not_running = get_firmware_info(hass, zha)
+
+    assert fw_info_not_running == FirmwareInfo(
+        device="/dev/ttyUSB1",
+        firmware_type=ApplicationType.EZSP,
+        firmware_version=None,
+        source="zha",
+        owners=[OwningIntegration(config_entry_id=zha.entry_id)],
+    )
+    assert await fw_info_not_running.is_running(hass) is False
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        # Missing data
+        {},
+        # Bad radio type
+        {"device": {"path": "/dev/ttyUSB1"}, "radio_type": "znp"},
+    ],
+)
+async def test_get_firmware_info_errors(
+    hass: HomeAssistant, data: dict[str, str | int | None]
+) -> None:
+    """Test `get_firmware_info` with config entry data format errors."""
+    zha = MockConfigEntry(
+        domain="zha",
+        unique_id="some_unique_id",
+        data=data,
+        version=4,
+    )
+    zha.add_to_hass(hass)
+
+    assert (get_firmware_info(hass, zha)) is None
+
+
+async def test_hardware_firmware_info_provider_notification(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_zigpy_connect: ControllerApplication,
+) -> None:
+    """Test that the ZHA gateway provides hardware and firmware information."""
+    config_entry.add_to_hass(hass)
+
+    await async_setup_component(hass, "homeassistant_hardware", {})
+
+    callback = MagicMock()
+    async_register_firmware_info_callback(hass, "/dev/ttyUSB0", callback)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+
+    callback.assert_called_once_with(
+        FirmwareInfo(
+            device="/dev/ttyUSB0",
+            firmware_type=ApplicationType.EZSP,
+            firmware_version="7.1.4.0 build 389",
+            source="zha",
+            owners=[OwningIntegration(config_entry_id=config_entry.entry_id)],
+        )
+    )