Reduce Connect firmware install times by removing unnecessary firmware probing (home-assistant#153012)

Author: puddly

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -155,34 +155,6 @@ async def async_step_pick_firmware(
             description_placeholders=self._get_translation_placeholders(),
         )
 
-    async def _probe_firmware_info(
-        self,
-        probe_methods: tuple[ApplicationType, ...] = (
-            # We probe in order of frequency: Zigbee, Thread, then multi-PAN
-            ApplicationType.GECKO_BOOTLOADER,
-            ApplicationType.EZSP,
-            ApplicationType.SPINEL,
-            ApplicationType.CPC,
-        ),
-    ) -> bool:
-        """Probe the firmware currently on the device."""
-        assert self._device is not None
-
-        self._probed_firmware_info = await probe_silabs_firmware_info(
-            self._device,
-            probe_methods=probe_methods,
-        )
-
-        return (
-            self._probed_firmware_info is not None
-            and self._probed_firmware_info.firmware_type
-            in (
-                ApplicationType.EZSP,
-                ApplicationType.SPINEL,
-                ApplicationType.CPC,
-            )
-        )
-
     async def _install_firmware_step(
         self,
         fw_update_url: str,
@@ -236,12 +208,6 @@ async def _install_firmware(
         expected_installed_firmware_type: ApplicationType,
     ) -> None:
         """Install firmware."""
-        if not await self._probe_firmware_info():
-            raise AbortFlow(
-                reason="unsupported_firmware",
-                description_placeholders=self._get_translation_placeholders(),
-            )
-
         assert self._device is not None
 
         # Keep track of the firmware we're working with, for error messages
@@ -250,6 +216,8 @@ async def _install_firmware(
         # Installing new firmware is only truly required if the wrong type is
         # installed: upgrading to the latest release of the current firmware type
         # isn't strictly necessary for functionality.
+        self._probed_firmware_info = await probe_silabs_firmware_info(self._device)
+
         firmware_install_required = self._probed_firmware_info is None or (
             self._probed_firmware_info.firmware_type != expected_installed_firmware_type
         )
@@ -301,7 +269,7 @@ async def _install_firmware(
             # Otherwise, fail
             raise AbortFlow(reason="firmware_download_failed") from err
 
-        await async_flash_silabs_firmware(
+        self._probed_firmware_info = await async_flash_silabs_firmware(
             hass=self.hass,
             device=self._device,
             fw_data=fw_data,
@@ -314,15 +282,6 @@ async def _install_firmware(
 
     async def _configure_and_start_otbr_addon(self) -> None:
         """Configure and start the OTBR addon."""
-
-        # Before we start the addon, confirm that the correct firmware is running
-        # and populate `self._probed_firmware_info` with the correct information
-        if not await self._probe_firmware_info(probe_methods=(ApplicationType.SPINEL,)):
-            raise AbortFlow(
-                "unsupported_firmware",
-                description_placeholders=self._get_translation_placeholders(),
-            )
-
         otbr_manager = get_otbr_addon_manager(self.hass)
         addon_info = await self._async_get_addon_info(otbr_manager)
 
@@ -444,12 +403,12 @@ async def _async_continue_picked_firmware(self) -> ConfigFlowResult:
         if self._picked_firmware_type == PickedFirmwareType.ZIGBEE:
             return await self.async_step_install_zigbee_firmware()
 
-        return await self.async_step_prepare_thread_installation()
+        return await self.async_step_install_thread_firmware()
 
-    async def async_step_prepare_thread_installation(
+    async def async_step_finish_thread_installation(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Prepare for Thread installation by stopping the OTBR addon if needed."""
+        """Finish Thread installation by starting the OTBR addon."""
         if not is_hassio(self.hass):
             return self.async_abort(
                 reason="not_hassio_thread",
@@ -459,22 +418,12 @@ async def async_step_prepare_thread_installation(
         otbr_manager = get_otbr_addon_manager(self.hass)
         addon_info = await self._async_get_addon_info(otbr_manager)
 
-        if addon_info.state == AddonState.RUNNING:
-            # 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_finish_thread_installation(
-        self, user_input: dict[str, Any] | None = None
-    ) -> ConfigFlowResult:
-        """Finish Thread installation by starting the OTBR addon."""
-        otbr_manager = get_otbr_addon_manager(self.hass)
-        addon_info = await self._async_get_addon_info(otbr_manager)
-
         if addon_info.state == AddonState.NOT_INSTALLED:
             return await self.async_step_install_otbr_addon()
 
+        if addon_info.state == AddonState.RUNNING:
+            await otbr_manager.async_stop_addon()
+
         return await self.async_step_start_otbr_addon()
 
     async def async_step_pick_firmware_zigbee(
@@ -511,12 +460,6 @@ async def async_step_continue_zigbee(
         assert self._device is not None
         assert self._hardware_name is not None
 
-        if not await self._probe_firmware_info(probe_methods=(ApplicationType.EZSP,)):
-            return self.async_abort(
-                reason="unsupported_firmware",
-                description_placeholders=self._get_translation_placeholders(),
-            )
-
         if self._zigbee_integration == ZigbeeIntegration.OTHER:
             return self._async_flow_finished()
 

homeassistant/components/homeassistant_hardware/util.py

@@ -42,9 +42,9 @@ class ApplicationType(StrEnum):
     """Application type running on a device."""
 
     GECKO_BOOTLOADER = "bootloader"
-    CPC = "cpc"
     EZSP = "ezsp"
     SPINEL = "spinel"
+    CPC = "cpc"
     ROUTER = "router"
 
     @classmethod

homeassistant/components/homeassistant_yellow/config_flow.py

@@ -27,6 +27,7 @@
 from homeassistant.components.homeassistant_hardware.util import (
     ApplicationType,
     FirmwareInfo,
+    probe_silabs_firmware_info,
 )
 from homeassistant.config_entries import (
     SOURCE_HARDWARE,
@@ -141,8 +142,10 @@ async def async_step_system(
         self, data: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Handle the initial step."""
+        assert self._device is not None
+
         # We do not actually use any portion of `BaseFirmwareConfigFlow` beyond this
-        await self._probe_firmware_info()
+        self._probed_firmware_info = await probe_silabs_firmware_info(self._device)
 
         # Kick off ZHA hardware discovery automatically if Zigbee firmware is running
         if (

tests/components/homeassistant_connect_zbt2/test_config_flow.py

@@ -72,6 +72,13 @@ async def mock_install_firmware_step(
         step_id: str,
         next_step_id: str,
     ) -> ConfigFlowResult:
+        self._probed_firmware_info = FirmwareInfo(
+            device=usb_data.device,
+            firmware_type=expected_installed_firmware_type,
+            firmware_version=fw_version,
+            owners=[],
+            source="probe",
+        )
         return await getattr(self, f"async_step_{next_step_id}")()
 
     with (
@@ -80,16 +87,6 @@ async def mock_install_firmware_step(
             autospec=True,
             side_effect=mock_install_firmware_step,
         ),
-        patch(
-            "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
-            return_value=FirmwareInfo(
-                device=usb_data.device,
-                firmware_type=fw_type,
-                firmware_version=fw_version,
-                owners=[],
-                source="probe",
-            ),
-        ),
     ):
         pick_result = await hass.config_entries.flow.async_configure(
             result["flow_id"],
@@ -157,6 +154,13 @@ async def mock_install_firmware_step(
         step_id: str,
         next_step_id: str,
     ) -> ConfigFlowResult:
+        self._probed_firmware_info = FirmwareInfo(
+            device=usb_data.device,
+            firmware_type=expected_installed_firmware_type,
+            firmware_version=fw_version,
+            owners=[],
+            source="probe",
+        )
         return await getattr(self, f"async_step_{next_step_id}")()
 
     with (
@@ -165,16 +169,6 @@ async def mock_install_firmware_step(
             autospec=True,
             side_effect=mock_install_firmware_step,
         ),
-        patch(
-            "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
-            return_value=FirmwareInfo(
-                device=usb_data.device,
-                firmware_type=fw_type,
-                firmware_version=fw_version,
-                owners=[],
-                source="probe",
-            ),
-        ),
     ):
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"],
@@ -258,6 +252,13 @@ async def mock_install_firmware_step(
         step_id: str,
         next_step_id: str,
     ) -> ConfigFlowResult:
+        self._probed_firmware_info = FirmwareInfo(
+            device=usb_data.device,
+            firmware_type=expected_installed_firmware_type,
+            firmware_version="7.4.4.0 build 0",
+            owners=[],
+            source="probe",
+        )
         return await getattr(self, f"async_step_{next_step_id}")()
 
     with (
@@ -270,16 +271,6 @@ async def mock_install_firmware_step(
             autospec=True,
             side_effect=mock_install_firmware_step,
         ),
-        patch(
-            "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
-            return_value=FirmwareInfo(
-                device=usb_data.device,
-                firmware_type=ApplicationType.EZSP,
-                firmware_version="7.4.4.0 build 0",
-                owners=[],
-                source="probe",
-            ),
-        ),
     ):
         pick_result = await hass.config_entries.options.async_configure(
             result["flow_id"],