From 24bc6d357379eca5c85aff0d007a19f701b68876 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 12 May 2025 16:31:10 -0400 Subject: [PATCH] [ZBT-1] Implement flashing for Zigbee and Thread within the config flow --- .../firmware_config_flow.py | 163 ++++-------------- .../homeassistant_sky_connect/config_flow.py | 81 +++++++++ 2 files changed, 110 insertions(+), 134 deletions(-) diff --git a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py index 1b4840e5a98..86cc2ed6794 100644 --- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py +++ b/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 self.async_abort( - reason="not_hassio", - description_placeholders=self._get_translation_placeholders(), - ) + return await self.async_step_install_zigbee_firmware() - # Only flash new firmware if we need to - 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( + async def async_step_install_zigbee_firmware( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Show progress dialog for installing the Zigbee flasher addon.""" - return await self._install_addon( - get_zigbee_flasher_addon_manager(self.hass), - "install_zigbee_flasher_addon", - "run_zigbee_flasher_addon", - ) + """Install Zigbee firmware.""" + raise NotImplementedError 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: - return await self.async_step_start_otbr_addon() + if addon_info.state == AddonState.RUNNING: + # 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 - return self.async_abort( - reason="otbr_addon_already_running", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": otbr_manager.addon_name, - }, - ) + # Otherwise, stop the addon before continuing to flash firmware + await otbr_manager.async_stop_addon() + + return await self.async_step_install_thread_firmware() + + 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) diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index eb5ea214b3e..e69e845feb6 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/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