Add feature to write an attribute to a node (#341)

matter_server/client/client.py

@@ -35,6 +35,8 @@
 
 SUB_WILDCARD: Final = "*"
 
+# pylint: disable=too-many-public-methods
+
 
 class MatterClient:
     """Manage a Matter server over WebSockets."""
@@ -231,6 +233,21 @@ async def send_device_command(
             interaction_timeout_ms=interaction_timeout_ms,
         )
 
+    async def write_attribute(
+        self,
+        node_id: int,
+        attribute_path: str,
+        value: Any,
+    ) -> Any:
+        """Write an attribute(value) on a target node."""
+        return await self.send_command(
+            APICommand.WRITE_ATTRIBUTE,
+            require_schema=4,
+            node_id=node_id,
+            attribute_path=attribute_path,
+            value=value,
+        )
+
     async def remove_node(self, node_id: int) -> None:
         """Remove a Matter node/device from the fabric."""
         await self.send_command(APICommand.REMOVE_NODE, node_id=node_id)

matter_server/common/models.py

@@ -41,6 +41,7 @@ class APICommand(str, Enum):
     REMOVE_NODE = "remove_node"
     GET_VENDOR_NAMES = "get_vendor_names"
     SUBSCRIBE_ATTRIBUTE = "subscribe_attribute"
+    WRITE_ATTRIBUTE = "write_attribute"
 
 
 EventCallBackType = Callable[[EventType, Any], None]

matter_server/server/device_controller.py

@@ -383,6 +383,24 @@ async def send_device_command(
                 interactionTimeoutMs=interaction_timeout_ms,
             )
 
+    @api_command(APICommand.WRITE_ATTRIBUTE)
+    async def write_attribute(
+        self,
+        node_id: int,
+        attribute_path: str,
+        value: Any,
+    ) -> Any:
+        """Write an attribute(value) on a target node."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+        endpoint_id, cluster_id, attribute_id = parse_attribute_path(attribute_path)
+        attribute = ALL_ATTRIBUTES[cluster_id][attribute_id]()
+        attribute.value = value
+        return await self.chip_controller.WriteAttribute(
+            nodeid=node_id,
+            attributes=[(endpoint_id, attribute)],
+        )
+
     @api_command(APICommand.REMOVE_NODE)
     async def remove_node(self, node_id: int) -> None:
         """Remove a Matter node/device from the fabric."""