Support local updates with the same version string (#1027)

Author: agners

matter_server/server/device_controller.py

@@ -1099,10 +1099,12 @@ async def _check_node_update(
             raise UpdateCheckError("Update found, but no OTA URL provided.")
 
         node_logger.info(
-            "New software update found: %s on %s (current %s).",
+            "Software update found: %s (%s) from %s, current %s (%s)).",
             update["softwareVersionString"],
+            update["softwareVersion"],
             update_source,
             software_version_string,
+            software_version,
         )
         return update_source, update
 

matter_server/server/ota/__init__.py

@@ -8,7 +8,9 @@
 from matter_server.common.models import UpdateSource
 from matter_server.server.ota import dcl
 
-_local_updates: dict[tuple[int, int], dict] = {}
+MatterProduct = tuple[int, int]
+
+_local_updates: dict[MatterProduct, dict[int | str, dict]] = {}
 
 
 async def load_local_updates(ota_provider_dir: Path) -> None:
@@ -21,9 +23,12 @@ def _load_update(ota_provider_dir: Path) -> None:
             with open(update_file) as f:
                 update = json.load(f)
                 model_version = update["modelVersion"]
-                _local_updates[(model_version["vid"], model_version["pid"])] = (
-                    model_version
-                )
+                model_key = (model_version["vid"], model_version["pid"])
+                update_dict = _local_updates.get(model_key, {})
+                # Store by string or integer, this allows update by both
+                update_dict[model_version["softwareVersion"]] = model_version
+                update_dict[model_version["softwareVersionString"]] = model_version
+                _local_updates[model_key] = update_dict
 
     await asyncio.get_running_loop().run_in_executor(
         None, _load_update, ota_provider_dir
@@ -38,14 +43,16 @@ async def check_for_update(
     requested_software_version: int | str | None = None,
 ) -> tuple[UpdateSource, dict] | tuple[None, None]:
     """Check for software updates."""
-    if (vid, pid) in _local_updates:
-        local_update = _local_updates[(vid, pid)]
-        if (
-            requested_software_version is None
-            or local_update["softwareVersion"] == requested_software_version
-            or local_update["softwareVersionString"] == requested_software_version
-        ):
-            return UpdateSource.LOCAL, local_update
+    if local_updates := _local_updates.get((vid, pid)):
+        logger.info("Local updates found for this device")
+        if requested_software_version is None:
+            # Use integer version to reliably determine absolute latest version
+            versions = filter(
+                lambda version: isinstance(version, int), local_updates.keys()
+            )
+            return UpdateSource.LOCAL, local_updates[max(versions)]
+        if requested_software_version in local_updates:
+            return UpdateSource.LOCAL, local_updates[requested_software_version]
 
     if dcl_update := await dcl.check_for_update(
         logger, vid, pid, current_software_version, requested_software_version