Introduce hardcoded updates

Add global check for update where we can insert hardcoded updates or
updates from other sources in the future.

Author: agners

matter_server/server/device_controller.py

@@ -32,7 +32,7 @@
 from matter_server.common.models import CommissionableNodeData, CommissioningParameters
 from matter_server.server.helpers.attributes import parse_attributes_from_read_result
 from matter_server.server.helpers.utils import ping_ip
-from matter_server.server.ota.dcl import check_for_update
+from matter_server.server.ota import check_for_update
 from matter_server.server.ota.provider import ExternalOtaProvider
 from matter_server.server.sdk import ChipDeviceControllerWrapper
 

matter_server/server/ota/__init__.py

@@ -0,0 +1,28 @@
+"""OTA implementation for the Matter Server."""
+
+from matter_server.server.ota import dcl
+
+HARDCODED_UPDATES: dict[tuple[int, int], dict] = {
+    # OTA requestor example app, useful for testing
+    (0xFFF1, 0x8001): {
+        "vid": 0xFFF1,
+        "pid": 0x8001,
+        "softwareVersion": 2,
+        "softwareVersionString": "2.0",
+        "cdVersionNumber": 1,
+        "softwareVersionValid": True,
+        "minApplicableSoftwareVersion": 1,
+        "maxApplicableSoftwareVersion": 1,
+        "otaUrl": "https://github.com/agners/matter-linux-example-apps/releases/download/v1.3.0.0/chip-ota-requestor-app-x86-64.ota",
+    }
+}
+
+
+async def check_for_update(
+    vid: int, pid: int, current_software_version: int
+) -> None | dict:
+    """Check for software updates."""
+    if (vid, pid) in HARDCODED_UPDATES:
+        return HARDCODED_UPDATES[(vid, pid)]
+
+    return await dcl.check_for_update(vid, pid, current_software_version)

matter_server/server/ota/provider.py

@@ -225,7 +225,7 @@ async def download_update(self, update_desc: dict) -> None:
 
         loop = asyncio.get_running_loop()
         await loop.run_in_executor(
-            None, functools.partial(DEFAULT_UPDATES_PATH.mkdir, exists_ok=True)
+            None, functools.partial(DEFAULT_UPDATES_PATH.mkdir, exist_ok=True)
         )
 
         file_path = DEFAULT_UPDATES_PATH / file_name
@@ -236,7 +236,7 @@ async def download_update(self, update_desc: dict) -> None:
         try:
             async with ClientSession(raise_for_status=True) as session:
                 # fetch the paa certificates list
-                logging.debug("Download update from f{url}.")
+                LOGGER.debug("Download update from '%s'.", url)
                 async with session.get(url) as response:
                     with file_path.open("wb") as f:
                         while True: