Use ephemeral OTA Provider instances

Instead of trying to reuse the same OTA Provider instance for multiple
OTA requests, create a new instance for each request. This is easier
to implement and also allows parallel updates.

Because the OTA Provider is now ephemeral, we need to commission it on
every update. But this is quick and reliable, so not a big deal.

To support multiple updates at once, we need to make sure the OTA
Providers use a distinct Matter port (hence passing 0) and distinct
node ids. The current implementation simply uses the target node id
plus a fixed offset. Since a single node can only run one update at a
time, this is sufficient.

Furthermore, some updates seem to have a difference in reported
versionNumberString value in the DCL vs. what is actually in the OTA
metadata. Specifically Eve updates from the Testnet DCL are such
updates (e.g. 3.2.0 vs 3.2.6705). When using the OTA Provider with the
--otaImageList option, this discrepancy is an issue and causes OTA
Provider to abort.

Using the single OTA update per OTA Provider instance allows us to use
--filepath, which doesn't check the versionNumberString.

Author: agners

matter_server/server/device_controller.py

@@ -60,7 +60,7 @@
 from .const import DATA_MODEL_SCHEMA_VERSION
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable
+    from collections.abc import Callable, Iterable
     from pathlib import Path
 
     from chip.native import PyChipError
@@ -110,11 +110,6 @@
         0, Clusters.BasicInformation.Attributes.SoftwareVersionString
     )
 )
-OTA_SOFTWARE_UPDATE_REQUESTOR_UPDATE_STATE_ATTRIBUTE_PATH = (
-    create_attribute_path_from_attribute(
-        0, Clusters.OtaSoftwareUpdateRequestor.Attributes.UpdateState
-    )
-)
 
 
 # pylint: disable=too-many-lines,too-many-instance-attributes,too-many-public-methods
@@ -131,6 +126,7 @@ def __init__(
     ):
         """Initialize the device controller."""
         self.server = server
+        self._ota_provider_dir = ota_provider_dir
 
         self._chip_device_controller = ChipDeviceControllerWrapper(
             server, paa_root_cert_dir
@@ -157,15 +153,14 @@ def __init__(
         self._polled_attributes: dict[int, set[str]] = {}
         self._custom_attribute_poller_timer: asyncio.TimerHandle | None = None
         self._custom_attribute_poller_task: asyncio.Task | None = None
-        self._ota_provider = ExternalOtaProvider(server.vendor_id, ota_provider_dir)
+        self._attribute_update_callbacks: dict[int, list[Callable]] = {}
 
     async def initialize(self) -> None:
         """Initialize the device controller."""
         self._compressed_fabric_id = (
             await self._chip_device_controller.get_compressed_fabric_id()
         )
         self._fabric_id_hex = hex(self._compressed_fabric_id)[2:]
-        await self._ota_provider.initialize()
 
     async def start(self) -> None:
         """Handle logic on controller start."""
@@ -229,9 +224,6 @@ async def stop(self) -> None:
 
         # shutdown the sdk device controller
         await self._chip_device_controller.shutdown()
-        # shutdown the OTA Provider
-        if self._ota_provider:
-            await self._ota_provider.stop()
         LOGGER.debug("Stopped.")
 
     @property
@@ -916,14 +908,29 @@ async def update_node(
             )
 
         # Add update to the OTA provider
-        await self._ota_provider.download_update(update)
+        ota_provider = ExternalOtaProvider(
+            self.server.vendor_id, self._ota_provider_dir / f"{node_id}"
+        )
 
-        # Make sure any previous instances get stopped
-        await self._ota_provider.start_update(
-            self._chip_device_controller,
-            node_id,
+        await ota_provider.initialize()
+
+        await ota_provider.download_update(update)
+
+        self._attribute_update_callbacks.setdefault(node_id, []).append(
+            ota_provider.check_update_state
         )
 
+        try:
+            # Make sure any previous instances get stopped
+            await ota_provider.start_update(
+                self._chip_device_controller,
+                node_id,
+            )
+        finally:
+            self._attribute_update_callbacks[node_id].remove(
+                ota_provider.check_update_state
+            )
+
         return update
 
     async def _check_node_update(
@@ -1017,16 +1024,6 @@ def attribute_updated(
                 # schedule a full interview of the node if the software version changed
                 loop.create_task(self.interview_node(node_id))
 
-            # work out if update state changed
-            if (
-                str(path) == OTA_SOFTWARE_UPDATE_REQUESTOR_UPDATE_STATE_ATTRIBUTE_PATH
-                and new_value != old_value
-            ):
-                if self._ota_provider:
-                    loop.create_task(
-                        self._ota_provider.check_update_state(node_id, new_value)
-                    )
-
             # store updated value in node attributes
             node.attributes[str(path)] = new_value
 
@@ -1058,6 +1055,11 @@ def attribute_updated_callback(
                 return
 
             loop.call_soon_threadsafe(attribute_updated, path, old_value, new_value)
+            if node_id in self._attribute_update_callbacks:
+                for callback in self._attribute_update_callbacks[node_id]:
+                    asyncio.run_coroutine_threadsafe(
+                        callback(path, old_value, new_value), loop
+                    )
 
         def event_callback(
             data: Attribute.EventReadResult,
@@ -1084,6 +1086,7 @@ def event_callback(
                 data=data.Data,
             )
             self.event_history.append(node_event)
+
             loop.call_soon_threadsafe(
                 self.server.signal_event, EventType.NODE_EVENT, node_event
             )