Support local updates with the same version string

For development it is sometimes useful to just reuse the same version
string but increment the underlying build number (which is the actual
version devices typically care about).

This change supports multiple local updates with the same version
string, and updating to an explicit version by version integer.

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