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

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -32,7 +32,6 @@ from .util import (
     OwningAddon,
     OwningIntegration,
     get_otbr_addon_manager,
-    get_zigbee_flasher_addon_manager,
     guess_firmware_info,
     guess_hardware_owners,
     probe_silabs_firmware_info,
@@ -167,40 +166,13 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
         ):
             return await self.async_step_confirm_zigbee()
 
-        if not is_hassio(self.hass):
+        return await self.async_step_install_zigbee_firmware()
-            return self.async_abort(
-                reason="not_hassio",
-                description_placeholders=self._get_translation_placeholders(),
-            )
 
-        # Only flash new firmware if we need to
+    async def async_step_install_zigbee_firmware(
-        fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
-        addon_info = await self._async_get_addon_info(fw_flasher_manager)
-
-        if addon_info.state == AddonState.NOT_INSTALLED:
-            return await self.async_step_install_zigbee_flasher_addon()
-
-        if addon_info.state == AddonState.NOT_RUNNING:
-            return await self.async_step_run_zigbee_flasher_addon()
-
-        # If the addon is already installed and running, fail
-        return self.async_abort(
-            reason="addon_already_running",
-            description_placeholders={
-                **self._get_translation_placeholders(),
-                "addon_name": fw_flasher_manager.addon_name,
-            },
-        )
-
-    async def async_step_install_zigbee_flasher_addon(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Show progress dialog for installing the Zigbee flasher addon."""
+        """Install Zigbee firmware."""
-        return await self._install_addon(
+        raise NotImplementedError
-            get_zigbee_flasher_addon_manager(self.hass),
-            "install_zigbee_flasher_addon",
-            "run_zigbee_flasher_addon",
-        )
 
     async def _install_addon(
         self,
@@ -254,96 +226,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
             },
         )
 
-    async def async_step_run_zigbee_flasher_addon(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Configure the flasher addon to point to the SkyConnect and run it."""
-        fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
-        addon_info = await self._async_get_addon_info(fw_flasher_manager)
-
-        assert self._device is not None
-        new_addon_config = {
-            **addon_info.options,
-            "device": self._device,
-            "baudrate": 115200,
-            "bootloader_baudrate": 115200,
-            "flow_control": True,
-        }
-
-        _LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config)
-        await self._async_set_addon_config(new_addon_config, fw_flasher_manager)
-
-        if not self.addon_start_task:
-
-            async def start_and_wait_until_done() -> None:
-                await fw_flasher_manager.async_start_addon_waiting()
-                # Now that the addon is running, wait for it to finish
-                await fw_flasher_manager.async_wait_until_addon_state(
-                    AddonState.NOT_RUNNING
-                )
-
-            self.addon_start_task = self.hass.async_create_task(
-                start_and_wait_until_done()
-            )
-
-        if not self.addon_start_task.done():
-            return self.async_show_progress(
-                step_id="run_zigbee_flasher_addon",
-                progress_action="run_zigbee_flasher_addon",
-                description_placeholders={
-                    **self._get_translation_placeholders(),
-                    "addon_name": fw_flasher_manager.addon_name,
-                },
-                progress_task=self.addon_start_task,
-            )
-
-        try:
-            await self.addon_start_task
-        except (AddonError, AbortFlow) as err:
-            _LOGGER.error(err)
-            self._failed_addon_name = fw_flasher_manager.addon_name
-            self._failed_addon_reason = "addon_start_failed"
-            return self.async_show_progress_done(next_step_id="addon_operation_failed")
-        finally:
-            self.addon_start_task = None
-
-        return self.async_show_progress_done(
-            next_step_id="uninstall_zigbee_flasher_addon"
-        )
-
-    async def async_step_uninstall_zigbee_flasher_addon(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Uninstall the flasher addon."""
-        fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
-
-        if not self.addon_uninstall_task:
-            _LOGGER.debug("Uninstalling flasher addon")
-            self.addon_uninstall_task = self.hass.async_create_task(
-                fw_flasher_manager.async_uninstall_addon_waiting()
-            )
-
-        if not self.addon_uninstall_task.done():
-            return self.async_show_progress(
-                step_id="uninstall_zigbee_flasher_addon",
-                progress_action="uninstall_zigbee_flasher_addon",
-                description_placeholders={
-                    **self._get_translation_placeholders(),
-                    "addon_name": fw_flasher_manager.addon_name,
-                },
-                progress_task=self.addon_uninstall_task,
-            )
-
-        try:
-            await self.addon_uninstall_task
-        except (AddonError, AbortFlow) as err:
-            _LOGGER.error(err)
-            # The uninstall failing isn't critical so we can just continue
-        finally:
-            self.addon_uninstall_task = None
-
-        return self.async_show_progress_done(next_step_id="confirm_zigbee")
-
     async def async_step_confirm_zigbee(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -402,24 +284,37 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
         if addon_info.state == AddonState.NOT_INSTALLED:
             return await self.async_step_install_otbr_addon()
 
-        if addon_info.state == AddonState.NOT_RUNNING:
+        if addon_info.state == AddonState.RUNNING:
-            return await self.async_step_start_otbr_addon()
+            # We only fail setup if we have an instance of OTBR running *and* it's
+            # pointing to different hardware
+            if addon_info.options["device"] != self._device:
+                return self.async_abort(
+                    reason="otbr_addon_already_running",
+                    description_placeholders={
+                        **self._get_translation_placeholders(),
+                        "addon_name": otbr_manager.addon_name,
+                    },
+                )
 
-        # If the addon is already installed and running, fail
+            # Otherwise, stop the addon before continuing to flash firmware
-        return self.async_abort(
+            await otbr_manager.async_stop_addon()
-            reason="otbr_addon_already_running",
+
-            description_placeholders={
+        return await self.async_step_install_thread_firmware()
-                **self._get_translation_placeholders(),
+
-                "addon_name": otbr_manager.addon_name,
+    async def async_step_install_thread_firmware(
-            },
+        self, user_input: dict[str, Any] | None = None
-        )
+    ) -> ConfigFlowResult:
+        """Install Thread firmware."""
+        raise NotImplementedError
 
     async def async_step_install_otbr_addon(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Show progress dialog for installing the OTBR addon."""
         return await self._install_addon(
-            get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon"
+            addon_manager=get_otbr_addon_manager(self.hass),
+            step_id="install_otbr_addon",
+            next_step_id="pick_firmware_thread",
         )
 
     async def async_step_start_otbr_addon(
@@ -435,7 +330,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
             "device": self._device,
             "baudrate": 460800,
             "flow_control": True,
-            "autoflash_firmware": True,
+            "autoflash_firmware": False,
         }
 
         _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config)

homeassistant/components/homeassistant_sky_connect/config_flow.py

@@ -2,14 +2,20 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from typing import TYPE_CHECKING, Any, Protocol
 
+from ha_silabs_firmware_client import FirmwareUpdateClient
+
 from homeassistant.components import usb
 from homeassistant.components.homeassistant_hardware import (
     firmware_config_flow,
     silabs_multiprotocol_addon,
 )
+from homeassistant.components.homeassistant_hardware.helpers import (
+    async_flash_silabs_firmware,
+)
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
     FirmwareInfo,
@@ -22,6 +28,7 @@ from homeassistant.config_entries import (
     OptionsFlow,
 )
 from homeassistant.core import callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.service_info.usb import UsbServiceInfo
 
 from .const import (
@@ -32,6 +39,7 @@ from .const import (
     FIRMWARE,
     FIRMWARE_VERSION,
     MANUFACTURER,
+    NABU_CASA_FIRMWARE_RELEASES_URL,
     PID,
     PRODUCT,
     SERIAL_NUMBER,
@@ -89,6 +97,7 @@ class HomeAssistantSkyConnectConfigFlow(
 
         self._usb_info: UsbServiceInfo | None = None
         self._hw_variant: HardwareVariant | None = None
+        self._firmware_install_task: asyncio.Task | None = None
 
     @staticmethod
     @callback
@@ -131,6 +140,78 @@ class HomeAssistantSkyConnectConfigFlow(
 
         return await self.async_step_confirm()
 
+    async def _install_firmware_step(
+        self,
+        fw_type: str,
+        firmware_name: str,
+        expected_installed_firmware_type: ApplicationType,
+        next_step_id: str,
+    ) -> ConfigFlowResult:
+        assert self._device is not None
+
+        if not self._firmware_install_task:
+            session = async_get_clientsession(self.hass)
+            client = FirmwareUpdateClient(NABU_CASA_FIRMWARE_RELEASES_URL, session)
+            manifest = await client.async_update_data()
+
+            zigbee_fw_meta = next(
+                fw
+                for fw in manifest.firmwares
+                if (
+                    fw.metadata["fw_type"] == fw_type
+                    and fw.filename.startswith(("skyconnect_", "zbt1_"))
+                )
+            )
+
+            zigbee_fw_data = await client.async_fetch_firmware(zigbee_fw_meta)
+            self._firmware_install_task = self.hass.async_create_task(
+                async_flash_silabs_firmware(
+                    hass=self.hass,
+                    device=self._device,
+                    fw_data=zigbee_fw_data,
+                    expected_installed_firmware_type=expected_installed_firmware_type,
+                    bootloader_reset_type=None,
+                    progress_callback=lambda offset, total: self.async_update_progress(
+                        offset / total
+                    ),
+                ),
+                f"Flash {firmware_name} firmware",
+            )
+
+        if not self._firmware_install_task.done():
+            return self.async_show_progress(
+                progress_action="install_firmware",
+                description_placeholders={
+                    **self._get_translation_placeholders(),
+                    "firmware_name": firmware_name,
+                },
+                progress_task=self._firmware_install_task,
+            )
+
+        return self.async_show_progress_done(next_step_id=next_step_id)
+
+    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_type="zigbee_ncp",
+            firmware_name="Zigbee",
+            expected_installed_firmware_type=ApplicationType.EZSP,
+            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_type="openthread_rcp",
+            firmware_name="OpenThread",
+            expected_installed_firmware_type=ApplicationType.SPINEL,
+            next_step_id="start_otbr_addon",
+        )
+
     def _async_flow_finished(self) -> ConfigFlowResult:
         """Create the config entry."""
         assert self._usb_info is not None