Add software update capability

Dockerfile

@@ -9,6 +9,7 @@ RUN \
     set -x \
     && apt-get update \
     && apt-get install -y --no-install-recommends \
+        curl \
         libuv1 \
         zlib1g \
         libjson-c5 \
@@ -25,6 +26,21 @@ RUN \
 
 ARG PYTHON_MATTER_SERVER
 
+ENV chip_example_url "https://github.com/home-assistant-libs/matter-linux-ota-provider/releases/download/2024.7.1"
+ARG TARGETPLATFORM
+
+RUN \
+    set -x \
+    && echo "${TARGETPLATFORM}" \
+    && if [ "${TARGETPLATFORM}" = "linux/amd64" ]; then \
+        curl -Lo /usr/local/bin/chip-ota-provider-app "${chip_example_url}/chip-ota-provider-app-x86-64"; \
+    elif [ "${TARGETPLATFORM}" = "linux/arm64" ]; then \
+        curl -Lo /usr/local/bin/chip-ota-provider-app "${chip_example_url}/chip-ota-provider-app-aarch64"; \
+    else \
+        exit 1; \
+    fi \
+    && chmod +x /usr/local/bin/chip-ota-provider-app
+
 # hadolint ignore=DL3013
 RUN \
     pip3 install --no-cache-dir "python-matter-server[server]==${PYTHON_MATTER_SERVER}"

matter_server/client/client.py

@@ -29,6 +29,7 @@
     EventType,
     MatterNodeData,
     MatterNodeEvent,
+    MatterSoftwareVersion,
     MessageType,
     NodePingResult,
     ResultMessageBase,
@@ -509,6 +510,36 @@ async def interview_node(self, node_id: int) -> None:
         """Interview a node."""
         await self.send_command(APICommand.INTERVIEW_NODE, node_id=node_id)
 
+    async def check_node_update(self, node_id: int) -> MatterSoftwareVersion | None:
+        """Check Node for updates.
+
+        Return a dict with the available update information. Most notable
+        "softwareVersion" contains the integer value of the update version which then
+        can be used for the update_node command to trigger the update.
+
+        The "softwareVersionString" is a human friendly version string.
+        """
+        data = await self.send_command(
+            APICommand.CHECK_NODE_UPDATE, node_id=node_id, require_schema=10
+        )
+        if data is None:
+            return None
+
+        return dataclass_from_dict(MatterSoftwareVersion, data)
+
+    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,
+            require_schema=10,
+        )
+
     def _prepare_message(
         self,
         command: str,

matter_server/common/const.py

@@ -2,7 +2,7 @@
 
 # schema version is used to determine compatibility between server and client
 # bump schema if we add new features and/or make other (breaking) changes
-SCHEMA_VERSION = 9
+SCHEMA_VERSION = 10
 
 
 VERBOSE_LOG_LEVEL = 5

matter_server/common/errors.py

@@ -77,6 +77,18 @@ class InvalidCommand(MatterError):
     error_code = 9
 
 
+class UpdateCheckError(MatterError):
+    """Error raised when there was an error during searching for updates."""
+
+    error_code = 10
+
+
+class UpdateError(MatterError):
+    """Error raised when there was an error during applying updates."""
+
+    error_code = 11
+
+
 def exception_from_error_code(error_code: int) -> type[MatterError]:
     """Return correct Exception class from error_code."""
     return ERROR_MAP.get(error_code, MatterError)

matter_server/common/models.py

@@ -47,6 +47,8 @@ class APICommand(str, Enum):
     PING_NODE = "ping_node"
     GET_NODE_IP_ADDRESSES = "get_node_ip_addresses"
     IMPORT_TEST_NODE = "import_test_node"
+    CHECK_NODE_UPDATE = "check_node_update"
+    UPDATE_NODE = "update_node"
 
 
 EventCallBackType = Callable[[EventType, Any], None]
@@ -209,3 +211,59 @@ class CommissioningParameters:
     setup_pin_code: int
     setup_manual_code: str
     setup_qr_code: str
+
+
+class UpdateSource(Enum):
+    """Enum with possible sources for a software update."""
+
+    MAIN_NET_DCL = "main-net-dcl"
+    TEST_NET_DCL = "test-net-dcl"
+    LOCAL = "local"
+
+
+@dataclass
+class MatterSoftwareVersion:
+    """Representation of a Matter software version. Return by the check_node_update command.
+
+    This holds Matter software version information similar to what is available from the CSA DCL.
+    https://on.dcl.csa-iot.org/#/Query/ModelVersion.
+    """
+
+    vid: int
+    pid: int
+    software_version: int
+    software_version_string: str
+    firmware_information: str | None
+    min_applicable_software_version: int
+    max_applicable_software_version: int
+    release_notes_url: str | None
+    update_source: UpdateSource
+
+    @classmethod
+    def from_dict(cls, data: dict) -> MatterSoftwareVersion:
+        """Initialize from dict."""
+        return cls(
+            vid=data["vid"],
+            pid=data["pid"],
+            software_version=data["software_version"],
+            software_version_string=data["software_version_string"],
+            firmware_information=data["firmware_information"],
+            min_applicable_software_version=data["min_applicable_software_version"],
+            max_applicable_software_version=data["max_applicable_software_version"],
+            release_notes_url=data["release_notes_url"],
+            update_source=UpdateSource(data["update_source"]),
+        )
+
+    def as_dict(self) -> dict:
+        """Return dict representation of the object."""
+        return {
+            "vid": self.vid,
+            "pid": self.pid,
+            "software_version": self.software_version,
+            "software_version_string": self.software_version_string,
+            "firmware_information": self.firmware_information,
+            "min_applicable_software_version": self.min_applicable_software_version,
+            "max_applicable_software_version": self.max_applicable_software_version,
+            "release_notes_url": self.release_notes_url,
+            "update_source": str(self.update_source),
+        }

matter_server/server/__main__.py

@@ -116,6 +116,12 @@
     nargs="+",
     help="List of node IDs to show logs from (applies only to server logs).",
 )
+parser.add_argument(
+    "--ota-provider-dir",
+    type=str,
+    default=None,
+    help="Directory where OTA Provider stores software updates and configuration.",
+)
 
 args = parser.parse_args()
 
@@ -216,6 +222,7 @@ def main() -> None:
         args.paa_root_cert_dir,
         args.enable_test_net_dcl,
         args.bluetooth_adapter,
+        args.ota_provider_dir,
     )
 
     async def handle_stop(loop: asyncio.AbstractEventLoop) -> None:

matter_server/server/const.py

@@ -20,3 +20,5 @@
     .parent.resolve()
     .joinpath("credentials/development/paa-root-certs")
 )
+
+DEFAULT_OTA_PROVIDER_DIR: Final[pathlib.Path] = pathlib.Path().cwd().joinpath("updates")