Support specific version by string

Support updating to a specific software version by string. This requires
fetching all versions one by one, but is much more user friendly. The
code still checks that all restrictions are met (min/max version, etc).

Author: agners

matter_server/client/client.py

@@ -523,7 +523,11 @@ async def check_node_update(self, node_id: int) -> dict[str, Any]:
         )
         return cast(dict[str, Any], node_update)
 
-    async def update_node(self, node_id: int, software_version: int) -> None:
+    async def update_node(
+        self,
+        node_id: int,
+        software_version: int | str,
+    ) -> None:
         """Start node update to a particular version."""
         await self.send_command(
             APICommand.UPDATE_NODE, node_id=node_id, software_version=software_version

matter_server/server/device_controller.py

@@ -897,7 +897,9 @@ async def check_node_update(self, node_id: int) -> dict | None:
         return await self._check_node_update(node_id)
 
     @api_command(APICommand.UPDATE_NODE)
-    async def update_node(self, node_id: int, software_version: int) -> dict | None:
+    async def update_node(
+        self, node_id: int, software_version: int | str
+    ) -> dict | None:
         """
         Update a node to a new software version.
 
@@ -927,7 +929,7 @@ async def update_node(self, node_id: int, software_version: int) -> dict | None:
     async def _check_node_update(
         self,
         node_id: int,
-        requested_software_version: int | None = None,
+        requested_software_version: int | str | None = None,
     ) -> dict | None:
         node_logger = LOGGER.getChild(f"node_{node_id}")
         node = self._nodes[node_id]

matter_server/server/ota/__init__.py

@@ -38,15 +38,18 @@ async def check_for_update(
     vid: int,
     pid: int,
     current_software_version: int,
-    requested_software_version: int | None = None,
+    requested_software_version: int | str | None = None,
 ) -> None | dict:
     """Check for software updates."""
     if (vid, pid) in HARDCODED_UPDATES:
         update = HARDCODED_UPDATES[(vid, pid)]
         if (
             requested_software_version is None
             or update["softwareVersion"] == requested_software_version
+            or update["softwareVersionString"] == requested_software_version
         ):
             return update
 
-    return await dcl.check_for_update(vid, pid, current_software_version)
+    return await dcl.check_for_update(
+        vid, pid, current_software_version, requested_software_version
+    )

matter_server/server/ota/dcl.py

@@ -32,9 +32,15 @@ async def _get_software_version(vid: int, pid: int, software_version: int) -> An
 
 
 async def _check_update_version(
-    vid: int, pid: int, version: int, current_software_version: int
+    vid: int,
+    pid: int,
+    current_software_version: int,
+    requested_software_version: int,
+    requested_software_version_string: str | None = None,
 ) -> None | dict:
-    version_res: dict = await _get_software_version(vid, pid, version)
+    version_res: dict = await _get_software_version(
+        vid, pid, requested_software_version
+    )
     if not isinstance(version_res, dict):
         raise TypeError("Unexpected DCL response.")
 
@@ -43,6 +49,14 @@ async def _check_update_version(
 
     version_candidate: dict = cast(dict, version_res["modelVersion"])
 
+    # If we are looking for a specific version by string, check if it matches
+    if (
+        requested_software_version_string is not None
+        and version_candidate["softwareVersionString"]
+        != requested_software_version_string
+    ):
+        return None
+
     # Check minApplicableSoftwareVersion/maxApplicableSoftwareVersion
     min_sw_version = version_candidate["minApplicableSoftwareVersion"]
     max_sw_version = version_candidate["maxApplicableSoftwareVersion"]
@@ -59,13 +73,14 @@ async def check_for_update(
     vid: int,
     pid: int,
     current_software_version: int,
-    requested_software_version: int | None = None,
+    requested_software_version: int | str | None = None,
 ) -> None | dict:
     """Check if there is a software update available on the DCL."""
     try:
-        if requested_software_version is not None:
+        # If a specific version as integer is requested, just fetch it (and hope it exists)
+        if isinstance(requested_software_version, int):
             return await _check_update_version(
-                vid, pid, requested_software_version, current_software_version
+                vid, pid, current_software_version, requested_software_version
             )
 
         # Get all versions and check each one of them.
@@ -78,15 +93,15 @@ async def check_for_update(
             if version > current_software_version
         ]
 
-        # Check if there is a newer software version available
+        # Check if there is a newer software version available, no downgrade possible
         if not newer_software_versions:
             LOGGER.info("No newer software version available.")
             return None
 
         # Check if latest firmware is applicable, and backtrack from there
         for version in sorted(newer_software_versions, reverse=True):
             if version_candidate := await _check_update_version(
-                vid, pid, version, current_software_version
+                vid, pid, current_software_version, version, requested_software_version
             ):
                 return version_candidate
             LOGGER.debug("Software version %d not applicable.", version)