Add read attribute command/feature (#342)

matter_server/client/client.py

@@ -233,6 +233,19 @@ async def send_device_command(
             interaction_timeout_ms=interaction_timeout_ms,
         )
 
+    async def read_attribute(
+        self,
+        node_id: int,
+        attribute_path: str,
+    ) -> Any:
+        """Read a single attribute on a node."""
+        return await self.send_command(
+            APICommand.READ_ATTRIBUTE,
+            require_schema=4,
+            node_id=node_id,
+            attribute_path=attribute_path,
+        )
+
     async def write_attribute(
         self,
         node_id: int,

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"
+    READ_ATTRIBUTE = "read_attribute"
     WRITE_ATTRIBUTE = "write_attribute"
 
 

matter_server/server/device_controller.py

@@ -383,6 +383,32 @@ async def send_device_command(
                 interactionTimeoutMs=interaction_timeout_ms,
             )
 
+    @api_command(APICommand.READ_ATTRIBUTE)
+    async def read_attribute(
+        self,
+        node_id: int,
+        attribute_path: str,
+    ) -> Any:
+        """Read a single attribute on a node."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+        node_lock = self._get_node_lock(node_id)
+        await self._resolve_node(node_id=node_id)
+        endpoint_id, cluster_id, attribute_id = parse_attribute_path(attribute_path)
+        attribute: Type[ClusterAttributeDescriptor] = ALL_ATTRIBUTES[cluster_id][
+            attribute_id
+        ]
+        async with node_lock:
+            result: Attribute.AsyncReadTransaction.ReadResponse = (
+                await self.chip_controller.Read(
+                    nodeid=node_id,
+                    attributes=[(endpoint_id, attribute)],
+                    fabricFiltered=False,
+                )
+            )
+            read_atributes = self._parse_attributes_from_read_result(result.attributes)
+            return read_atributes[attribute_path]
+
     @api_command(APICommand.WRITE_ATTRIBUTE)
     async def write_attribute(
         self,