From a60ad8b6a342d60de5b6f726a4a304bc1c4e8ad2 Mon Sep 17 00:00:00 2001 From: Cecille Freeman Date: Tue, 6 Aug 2024 21:42:41 -0400 Subject: [PATCH] python: Add direct attribute paths to Read Supports one particular use case: read one or all endpoints, all clusters, specific (global) attribute. See spec 8.9.2.4. This is an allowed wildcard construct that is not currently expressable in the API. Test: Used on wildcard read for matter_testing_support. This is therefore tested on any test using that decorator - switch and timesync. --- src/controller/python/chip/ChipDeviceCtrl.py | 18 +++++++++++++++--- src/python_testing/matter_testing_support.py | 3 ++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/controller/python/chip/ChipDeviceCtrl.py b/src/controller/python/chip/ChipDeviceCtrl.py index 5989a42977aafc..24fe59486f78d5 100644 --- a/src/controller/python/chip/ChipDeviceCtrl.py +++ b/src/controller/python/chip/ChipDeviceCtrl.py @@ -1351,8 +1351,12 @@ def _parseAttributePathTuple(self, pathTuple: typing.Union[ # Wildcard attribute id typing.Tuple[int, typing.Type[ClusterObjects.Cluster]], # Concrete path - typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]] + typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]], + # Directly specified attribute path + ClusterAttribute.AttributePath ]): + if isinstance(pathTuple, ClusterAttribute.AttributePath): + return pathTuple if pathTuple == ('*') or pathTuple == (): # Wildcard return ClusterAttribute.AttributePath() @@ -1437,7 +1441,9 @@ async def Read(self, nodeid: int, attributes: typing.Optional[typing.List[typing # Wildcard attribute id typing.Tuple[int, typing.Type[ClusterObjects.Cluster]], # Concrete path - typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]] + typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]], + # Directly specified attribute path + ClusterAttribute.AttributePath ]]] = None, dataVersionFilters: typing.Optional[typing.List[typing.Tuple[int, typing.Type[ClusterObjects.Cluster], int]]] = None, events: typing.Optional[typing.List[ typing.Union[ @@ -1476,6 +1482,8 @@ async def Read(self, nodeid: int, attributes: typing.Optional[typing.List[typing ReadAttribute(1, [ Clusters.BasicInformation ] ) -- case 5 above. ReadAttribute(1, [ (1, Clusters.BasicInformation.Attributes.Location ] ) -- case 1 above. + An AttributePath can also be specified directly by [chip.cluster.Attribute.AttributePath(...)] + dataVersionFilters: A list of tuples of (endpoint, cluster, data version). events: A list of tuples of varying types depending on the type of read being requested: @@ -1543,7 +1551,9 @@ async def ReadAttribute(self, nodeid: int, attributes: typing.Optional[typing.Li # Wildcard attribute id typing.Tuple[int, typing.Type[ClusterObjects.Cluster]], # Concrete path - typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]] + typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]], + # Directly specified attribute path + ClusterAttribute.AttributePath ]]], dataVersionFilters: typing.Optional[typing.List[typing.Tuple[int, typing.Type[ClusterObjects.Cluster], int]]] = None, returnClusterObject: bool = False, reportInterval: typing.Optional[typing.Tuple[int, int]] = None, @@ -1568,6 +1578,8 @@ async def ReadAttribute(self, nodeid: int, attributes: typing.Optional[typing.Li ReadAttribute(1, [ Clusters.BasicInformation ] ) -- case 5 above. ReadAttribute(1, [ (1, Clusters.BasicInformation.Attributes.Location ] ) -- case 1 above. + An AttributePath can also be specified directly by [chip.cluster.Attribute.AttributePath(...)] + returnClusterObject: This returns the data as consolidated cluster objects, with all attributes for a cluster inside a single cluster-wide cluster object. diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index 8275b8f0f83326..69f9e04ba57904 100644 --- a/src/python_testing/matter_testing_support.py +++ b/src/python_testing/matter_testing_support.py @@ -49,6 +49,7 @@ # isort: on import chip.clusters as Clusters +from chip.clusters import Attribute import chip.logging import chip.native from chip import discovery @@ -1863,7 +1864,7 @@ async def get_accepted_endpoints_for_test(self: MatterBaseTest, accept_function: Returns a list of endpoints on which the test should be run given the accept_function for the test. """ - wildcard = await self.default_controller.Read(self.dut_node_id, [()]) + wildcard = await self.default_controller.Read(self.dut_node_id, [(Clusters.Descriptor), Attribute.AttributePath(None, None, GlobalAttributeIds.ATTRIBUTE_LIST_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.FEATURE_MAP_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID)]) return [e for e in wildcard.attributes.keys() if accept_function(wildcard, e)]