Flash ZBT-1 and Yellow firmwares from Core instead of using addons (home-assistant#145019)

* Make `async_flash_firmware` a public helper

* [ZBT-1] Implement flashing for Zigbee and Thread within the config flow

* WIP: Begin fixing unit tests

* WIP: Unit tests, pass 2

* WIP: pass 3

* Fix hardware unit tests

* Have the individual hardware integrations depend on the firmware flasher

* Break out firmware filter into its own helper

* Mirror to Yellow

* Simplify

* Simplify

* Revert "Have the individual hardware integrations depend on the firmware flasher"

This reverts commit 096f429.

* Move `async_flash_silabs_firmware` into `util`

* Fix existing unit tests

* Unconditionally upgrade Zigbee firmware during installation

* Fix failing error case unit tests

* Fix remaining failing unit tests

* Increase test coverage

* 100% test coverage

* Remove old translation strings

* Add new translation strings

* Do not probe OTBR firmware when completing the flow

* More translation strings

* Probe OTBR firmware info before starting the addon

Author: puddly

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -10,22 +10,6 @@
             "pick_firmware_thread": "Thread"
           }
         },
-        "install_zigbee_flasher_addon": {
-          "title": "Installing flasher",
-          "description": "Installing the Silicon Labs Flasher add-on."
-        },
-        "run_zigbee_flasher_addon": {
-          "title": "Installing Zigbee firmware",
-          "description": "Installing Zigbee firmware. This will take about a minute."
-        },
-        "uninstall_zigbee_flasher_addon": {
-          "title": "Removing flasher",
-          "description": "Removing the Silicon Labs Flasher add-on."
-        },
-        "zigbee_flasher_failed": {
-          "title": "Zigbee installation failed",
-          "description": "The Zigbee firmware installation process was unsuccessful. Ensure no other software is trying to communicate with the {model} and try again."
-        },
         "confirm_zigbee": {
           "title": "Zigbee setup complete",
           "description": "Your {model} is now a Zigbee coordinator and will be shown as discovered by the Zigbee Home Automation integration."
@@ -55,9 +39,7 @@
         "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device. If you are running Home Assistant OS in a virtual machine or in Docker, please make sure that permissions are set correctly for the device."
       },
       "progress": {
-        "install_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is installed, this may take a few minutes.",
-        "run_zigbee_flasher_addon": "Please wait while Zigbee firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes.",
-        "uninstall_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is being removed."
+        "install_firmware": "Please wait while {firmware_name} firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes."
       }
     }
   },
@@ -110,16 +92,6 @@
           "data": {
             "disable_multi_pan": "Disable multiprotocol support"
           }
-        },
-        "install_flasher_addon": {
-          "title": "The Silicon Labs Flasher add-on installation has started"
-        },
-        "configure_flasher_addon": {
-          "title": "The Silicon Labs Flasher add-on installation has started"
-        },
-        "start_flasher_addon": {
-          "title": "Installing firmware",
-          "description": "Zigbee firmware is now being installed. This will take a few minutes."
         }
       },
       "error": {

homeassistant/components/homeassistant_hardware/strings.json

@@ -2,15 +2,12 @@
 
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Callable
-from contextlib import AsyncExitStack, asynccontextmanager
+from collections.abc import Callable
 from dataclasses import dataclass
 import logging
 from typing import Any, cast
 
 from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
-from universal_silabs_flasher.firmware import parse_firmware_image
-from universal_silabs_flasher.flasher import Flasher
 from yarl import URL
 
 from homeassistant.components.update import (
@@ -20,18 +17,12 @@
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import CALLBACK_TYPE, callback
-from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.restore_state import ExtraStoredData
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .coordinator import FirmwareUpdateCoordinator
 from .helpers import async_register_firmware_info_callback
-from .util import (
-    ApplicationType,
-    FirmwareInfo,
-    guess_firmware_info,
-    probe_silabs_firmware_info,
-)
+from .util import ApplicationType, FirmwareInfo, async_flash_silabs_firmware
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -249,19 +240,11 @@ def _update_progress(self, offset: int, total_size: int) -> None:
         self._attr_update_percentage = round((offset * 100) / total_size)
         self.async_write_ha_state()
 
-    @asynccontextmanager
-    async def _temporarily_stop_hardware_owners(
-        self, device: str
-    ) -> AsyncIterator[None]:
-        """Temporarily stop addons and integrations communicating with the device."""
-        firmware_info = await guess_firmware_info(self.hass, device)
-        _LOGGER.debug("Identified firmware info: %s", firmware_info)
-
-        async with AsyncExitStack() as stack:
-            for owner in firmware_info.owners:
-                await stack.enter_async_context(owner.temporarily_stop(self.hass))
-
-            yield
+        # Switch to an indeterminate progress bar after installation is complete, since
+        # we probe the firmware after flashing
+        if offset == total_size:
+            self._attr_update_percentage = None
+            self.async_write_ha_state()
 
     async def async_install(
         self, version: str | None, backup: bool, **kwargs: Any
@@ -278,49 +261,18 @@ async def async_install(
         fw_data = await self.coordinator.client.async_fetch_firmware(
             self._latest_firmware
         )
-        fw_image = await self.hass.async_add_executor_job(parse_firmware_image, fw_data)
-
-        device = self._current_device
-
-        flasher = Flasher(
-            device=device,
-            probe_methods=(
-                ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(),
-                ApplicationType.EZSP.as_flasher_application_type(),
-                ApplicationType.SPINEL.as_flasher_application_type(),
-                ApplicationType.CPC.as_flasher_application_type(),
-            ),
-            bootloader_reset=self.bootloader_reset_type,
-        )
 
-        async with self._temporarily_stop_hardware_owners(device):
-            try:
-                try:
-                    # Enter the bootloader with indeterminate progress
-                    await flasher.enter_bootloader()
-
-                    # Flash the firmware, with progress
-                    await flasher.flash_firmware(
-                        fw_image, progress_callback=self._update_progress
-                    )
-                except Exception as err:
-                    raise HomeAssistantError("Failed to flash firmware") from err
-
-                # Probe the running application type with indeterminate progress
-                self._attr_update_percentage = None
-                self.async_write_ha_state()
-
-                firmware_info = await probe_silabs_firmware_info(
-                    device,
-                    probe_methods=(self.entity_description.expected_firmware_type,),
-                )
-
-                if firmware_info is None:
-                    raise HomeAssistantError(
-                        "Failed to probe the firmware after flashing"
-                    )
+        try:
+            firmware_info = await async_flash_silabs_firmware(
+                hass=self.hass,
+                device=self._current_device,
+                fw_data=fw_data,
+                expected_installed_firmware_type=self.entity_description.expected_firmware_type,
+                bootloader_reset_type=self.bootloader_reset_type,
+                progress_callback=self._update_progress,
+            )
+        finally:
+            self._attr_in_progress = False
+            self.async_write_ha_state()
 
-                self._firmware_info_callback(firmware_info)
-            finally:
-                self._attr_in_progress = False
-                self.async_write_ha_state()
+        self._firmware_info_callback(firmware_info)

homeassistant/components/homeassistant_hardware/update.py

@@ -4,18 +4,20 @@
 
 import asyncio
 from collections import defaultdict
-from collections.abc import AsyncIterator, Iterable
-from contextlib import asynccontextmanager
+from collections.abc import AsyncIterator, Callable, Iterable
+from contextlib import AsyncExitStack, asynccontextmanager
 from dataclasses import dataclass
 from enum import StrEnum
 import logging
 
 from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
+from universal_silabs_flasher.firmware import parse_firmware_image
 from universal_silabs_flasher.flasher import Flasher
 
 from homeassistant.components.hassio import AddonError, AddonManager, AddonState
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.hassio import is_hassio
 from homeassistant.helpers.singleton import singleton
 
@@ -333,3 +335,52 @@ async def probe_silabs_firmware_type(
         return None
 
     return fw_info.firmware_type
+
+
+async def async_flash_silabs_firmware(
+    hass: HomeAssistant,
+    device: str,
+    fw_data: bytes,
+    expected_installed_firmware_type: ApplicationType,
+    bootloader_reset_type: str | None = None,
+    progress_callback: Callable[[int, int], None] | None = None,
+) -> FirmwareInfo:
+    """Flash firmware to the SiLabs device."""
+    firmware_info = await guess_firmware_info(hass, device)
+    _LOGGER.debug("Identified firmware info: %s", firmware_info)
+
+    fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
+
+    flasher = Flasher(
+        device=device,
+        probe_methods=(
+            ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(),
+            ApplicationType.EZSP.as_flasher_application_type(),
+            ApplicationType.SPINEL.as_flasher_application_type(),
+            ApplicationType.CPC.as_flasher_application_type(),
+        ),
+        bootloader_reset=bootloader_reset_type,
+    )
+
+    async with AsyncExitStack() as stack:
+        for owner in firmware_info.owners:
+            await stack.enter_async_context(owner.temporarily_stop(hass))
+
+        try:
+            # Enter the bootloader with indeterminate progress
+            await flasher.enter_bootloader()
+
+            # Flash the firmware, with progress
+            await flasher.flash_firmware(fw_image, progress_callback=progress_callback)
+        except Exception as err:
+            raise HomeAssistantError("Failed to flash firmware") from err
+
+        probed_firmware_info = await probe_silabs_firmware_info(
+            device,
+            probe_methods=(expected_installed_firmware_type,),
+        )
+
+    if probed_firmware_info is None:
+        raise HomeAssistantError("Failed to probe the firmware after flashing")
+
+    return probed_firmware_info

homeassistant/components/homeassistant_hardware/util.py

@@ -32,6 +32,7 @@
     FIRMWARE,
     FIRMWARE_VERSION,
     MANUFACTURER,
+    NABU_CASA_FIRMWARE_RELEASES_URL,
     PID,
     PRODUCT,
     SERIAL_NUMBER,
@@ -45,19 +46,29 @@
 
 if TYPE_CHECKING:
 
-    class TranslationPlaceholderProtocol(Protocol):
-        """Protocol describing `BaseFirmwareInstallFlow`'s translation placeholders."""
+    class FirmwareInstallFlowProtocol(Protocol):
+        """Protocol describing `BaseFirmwareInstallFlow` for a mixin."""
 
         def _get_translation_placeholders(self) -> dict[str, str]:
             return {}
 
+        async def _install_firmware_step(
+            self,
+            fw_update_url: str,
+            fw_type: str,
+            firmware_name: str,
+            expected_installed_firmware_type: ApplicationType,
+            step_id: str,
+            next_step_id: str,
+        ) -> ConfigFlowResult: ...
+
 else:
     # Multiple inheritance with `Protocol` seems to break
-    TranslationPlaceholderProtocol = object
+    FirmwareInstallFlowProtocol = object
 
 
-class SkyConnectTranslationMixin(ConfigEntryBaseFlow, TranslationPlaceholderProtocol):
-    """Translation placeholder mixin for Home Assistant SkyConnect."""
+class SkyConnectFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol):
+    """Mixin for Home Assistant SkyConnect firmware methods."""
 
     context: ConfigFlowContext
 
@@ -72,9 +83,35 @@ def _get_translation_placeholders(self) -> dict[str, str]:
 
         return placeholders
 
+    async def async_step_install_zigbee_firmware(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Install Zigbee firmware."""
+        return await self._install_firmware_step(
+            fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
+            fw_type="skyconnect_zigbee_ncp",
+            firmware_name="Zigbee",
+            expected_installed_firmware_type=ApplicationType.EZSP,
+            step_id="install_zigbee_firmware",
+            next_step_id="confirm_zigbee",
+        )
+
+    async def async_step_install_thread_firmware(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Install Thread firmware."""
+        return await self._install_firmware_step(
+            fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
+            fw_type="skyconnect_openthread_rcp",
+            firmware_name="OpenThread",
+            expected_installed_firmware_type=ApplicationType.SPINEL,
+            step_id="install_thread_firmware",
+            next_step_id="start_otbr_addon",
+        )
+
 
 class HomeAssistantSkyConnectConfigFlow(
-    SkyConnectTranslationMixin,
+    SkyConnectFirmwareMixin,
     firmware_config_flow.BaseFirmwareConfigFlow,
     domain=DOMAIN,
 ):
@@ -207,7 +244,7 @@ async def async_step_flashing_complete(
 
 
 class HomeAssistantSkyConnectOptionsFlowHandler(
-    SkyConnectTranslationMixin, firmware_config_flow.BaseFirmwareOptionsFlow
+    SkyConnectFirmwareMixin, firmware_config_flow.BaseFirmwareOptionsFlow
 ):
     """Zigbee and Thread options flow handlers."""