Add feature to write an attribute to a node

matter_server/client/client.py

@@ -31,7 +31,7 @@
 from .models.node import MatterFabricData, MatterNode
 
 if TYPE_CHECKING:
-    from chip.clusters.Objects import ClusterCommand
+    from chip.clusters.Objects import ClusterCommand, ClusterAttributeDescriptor
 
 SUB_WILDCARD: Final = "*"
 
@@ -231,6 +231,28 @@ async def send_device_command(
             interaction_timeout_ms=interaction_timeout_ms,
         )
 
+    async def write_attribute(
+        self,
+        node_id: int,
+        endpoint_id: int,
+        attribute: ClusterAttributeDescriptor,
+        response_type: Any | None = None,
+        timed_request_timeout_ms: int | None = None,
+        interaction_timeout_ms: int | None = None,
+    ) -> Any:
+        """Write an attribute(value) on a target node."""
+        return await self.send_command(
+            APICommand.WRITE_ATTRIBUTE,
+            node_id=node_id,
+            endpoint_id=endpoint_id,
+            cluster_id=attribute.cluster_id,
+            attribute_id=attribute.attribute_id,
+            value=attribute.value,
+            response_type=response_type,
+            timed_request_timeout_ms=timed_request_timeout_ms,
+            interaction_timeout_ms=interaction_timeout_ms,
+        )
+
     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,29 @@ async def send_device_command(
                 interactionTimeoutMs=interaction_timeout_ms,
             )
 
+    @api_command(APICommand.WRITE_ATTRIBUTE)
+    async def write_attribute(
+        self,
+        node_id: int,
+        endpoint_id: int,
+        cluster_id: int,
+        attribute_id: int,
+        value: Any,
+        timed_request_timeout_ms: int | None = None,
+        interaction_timeout_ms: int | None = None,
+    ) -> Any:
+        """Write an attribute(value) on a target node."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+        attribute = ALL_ATTRIBUTES[cluster_id][attribute_id]()
+        attribute.value = value
+        return await self.chip_controller.WriteAttribute(
+            nodeid=node_id,
+            attributes=[(endpoint_id, attribute)],
+            timedRequestTimeoutMs=timed_request_timeout_ms,
+            interactionTimeoutMs=interaction_timeout_ms,
+        )
+
     @api_command(APICommand.REMOVE_NODE)
     async def remove_node(self, node_id: int) -> None:
         """Remove a Matter node/device from the fabric."""