Replace FirmwareGuess with FirmwareInfo with owner tracking

homeassistant/components/homeassistant_hardware/const.py

@@ -5,6 +5,7 @@ import logging
 LOGGER = logging.getLogger(__package__)
 
 ZHA_DOMAIN = "zha"
+OTBR_DOMAIN = "otbr"
 
 OTBR_ADDON_NAME = "OpenThread Border Router"
 OTBR_ADDON_MANAGER_DATA = "openthread_border_router"

homeassistant/components/homeassistant_hardware/util.py

@@ -11,8 +11,10 @@ from typing import cast
 
 from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
 from universal_silabs_flasher.flasher import Flasher
+import voluptuous as vol
+from yarl import URL
 
-from homeassistant.components.hassio import AddonError, AddonState
+from homeassistant.components.hassio import AddonError, AddonState, valid_addon
 from homeassistant.config_entries import ConfigEntry, ConfigEntryState
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.hassio import is_hassio
@@ -22,6 +24,7 @@ from .const import (
     OTBR_ADDON_MANAGER_DATA,
     OTBR_ADDON_NAME,
     OTBR_ADDON_SLUG,
+    OTBR_DOMAIN,
     ZHA_DOMAIN,
     ZIGBEE_FLASHER_ADDON_MANAGER_DATA,
     ZIGBEE_FLASHER_ADDON_NAME,
@@ -55,11 +58,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,49 +82,140 @@ def get_zigbee_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager
     )
 
 
-@dataclass(slots=True, kw_only=True)
-class FirmwareGuess:
-    """Firmware guess."""
+@dataclass(kw_only=True)
+class OwningAddon:
+    """Owning add-on."""
 
-    is_running: bool
-    firmware_type: ApplicationType
-    source: str
+    slug: str
 
-
-async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> FirmwareGuess:
-    """Guess the firmware type based on installed addons and other integrations."""
-    device_guesses: defaultdict[str | None, list[FirmwareGuess]] = defaultdict(list)
-
-    for zha_config_entry in hass.config_entries.async_entries(ZHA_DOMAIN):
-        zha_path = get_zha_device_path(zha_config_entry)
-
-        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",
-                )
+    def _get_addon_manager(self, hass: HomeAssistant) -> WaitingAddonManager:
+        return WaitingAddonManager(
+            hass,
+            _LOGGER,
+            f"Add-on {self.slug}",
+            self.slug,
         )
 
-    if is_hassio(hass):
-        otbr_addon_manager = get_otbr_addon_manager(hass)
+    async def is_running(self, hass: HomeAssistant) -> bool:
+        """Check if the add-on is running."""
+        addon_manager = self._get_addon_manager(hass)
 
         try:
-            otbr_addon_info = await otbr_addon_manager.async_get_addon_info()
+            addon_info = await addon_manager.async_get_addon_info()
         except AddonError:
-            pass
+            return False
         else:
-            if otbr_addon_info.state != AddonState.NOT_INSTALLED:
-                otbr_path = otbr_addon_info.options.get("device")
-                device_guesses[otbr_path].append(
-                    FirmwareGuess(
-                        is_running=(otbr_addon_info.state == AddonState.RUNNING),
-                        firmware_type=ApplicationType.SPINEL,
-                        source="otbr",
-                    )
+            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 == ConfigEntryState.LOADED
+
+
+@dataclass(kw_only=True)
+class FirmwareInfo:
+    """Firmware guess."""
+
+    device: str
+    firmware_type: ApplicationType
+    firmware_version: str | None
+
+    source: str
+    owners: list[OwningAddon | OwningIntegration]
+
+
+async def get_zha_firmware_info(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> FirmwareInfo | None:
+    """Return firmware information for the ZHA instance."""
+    try:
+        # pylint: disable-next=import-outside-toplevel
+        from homeassistant.components.zha.helpers import get_zha_gateway
+    except ImportError:
+        return None
+
+    try:
+        gateway = get_zha_gateway(hass)
+    except ValueError:
+        firmware_version = None
+    else:
+        firmware_version = gateway.state.node_info.version
+
+    # We only support EZSP firmware for now
+    if config_entry.data["radio_type"] != "ezsp":
+        return None
+
+    device = config_entry.data.get("device", {}).get("device_path", None)
+    if device is None:
+        return None
+
+    return FirmwareInfo(
+        device=device,
+        firmware_type=ApplicationType.EZSP,
+        firmware_version=firmware_version,
+        source=ZHA_DOMAIN,
+        owners=[OwningIntegration(config_entry_id=config_entry.entry_id)],
     )
 
+
+async def get_otbr_firmware_info(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> FirmwareInfo | None:
+    """Return firmware information for the OpenThread Border Router."""
+    device = config_entry.data["device"]
+    if device is None:
+        return None
+
+    owners: list[OwningIntegration | OwningAddon] = [
+        OwningIntegration(config_entry_id=config_entry.entry_id)
+    ]
+
+    if is_hassio(hass) and (host := URL(config_entry.data["url"]).host) is not None:
+        try:
+            valid_addon(host)
+        except vol.Invalid:
+            pass
+        else:
+            owners.append(OwningAddon(slug=host))
+
+    return FirmwareInfo(
+        device=device,
+        firmware_type=ApplicationType.SPINEL,
+        firmware_version=cast(str | None, config_entry.data["firmware_version"]),
+        source=OTBR_DOMAIN,
+        owners=owners,
+    )
+
+
+async def guess_hardware_owners(
+    hass: HomeAssistant, device_path: str
+) -> 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):
+        firmware_info = await get_zha_firmware_info(hass, zha_config_entry)
+
+        if firmware_info is not None:
+            device_guesses[firmware_info.device].append(firmware_info)
+
+    for otbr_config_entry in hass.config_entries.async_entries(OTBR_DOMAIN):
+        firmware_info = await get_otbr_firmware_info(hass, otbr_config_entry)
+
+        if firmware_info is not None:
+            device_guesses[firmware_info.device].append(firmware_info)
+
+    if is_hassio(hass):
         multipan_addon_manager = await get_multiprotocol_addon_manager(hass)
 
         try:
@@ -136,30 +225,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(
-                        is_running=(multipan_addon_info.state == AddonState.RUNNING),
+                        FirmwareInfo(
+                            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
-    if device_path not in device_guesses:
-        return FirmwareGuess(
-            is_running=False, firmware_type=ApplicationType.EZSP, source="unknown"
-        )
-
-    # Prioritizes guesses that were pulled from a running addon or integration but keep
-    # the sort order we defined above
-    guesses = sorted(
-        device_guesses[device_path],
-        key=lambda guess: guess.is_running,
+    return device_guesses.get(device_path, [])
+
+
+async def guess_firmware_info(hass: HomeAssistant, device_path: str) -> FirmwareInfo:
+    """Guess the firmware type based on installed addons and other integrations."""
+
+    hardware_owners = await guess_hardware_owners(hass, device_path)
+
+    # Fall back to EZSP if we have no way to guess
+    if not hardware_owners:
+        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