From 055c5e2797a6921af71344516221f7456b2365fd Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Thu, 23 May 2024 18:20:14 +0200 Subject: [PATCH] [Python] Make AttributePath more pythonic Use dataclass default initializer to initialize AttributePath. Use static method to initialize from Cluster or Attribute. Also hash the integer fields directly, this is more efficient than formatting a string first. --- src/controller/python/chip/ChipDeviceCtrl.py | 21 +++++-------- .../python/chip/clusters/Attribute.py | 31 +++++++------------ .../test/test_scripts/cluster_objects.py | 2 +- 3 files changed, 20 insertions(+), 34 deletions(-) diff --git a/src/controller/python/chip/ChipDeviceCtrl.py b/src/controller/python/chip/ChipDeviceCtrl.py index d63a3772e62dc6..33d5efec46543c 100644 --- a/src/controller/python/chip/ChipDeviceCtrl.py +++ b/src/controller/python/chip/ChipDeviceCtrl.py @@ -1170,33 +1170,26 @@ def _parseAttributePathTuple(self, pathTuple: typing.Union[ # Concrete path typing.Tuple[int, typing.Type[ClusterObjects.ClusterAttributeDescriptor]] ]): - endpoint = None - cluster = None - attribute = None - if pathTuple == ('*') or pathTuple == (): # Wildcard - pass + return ClusterAttribute.AttributePath() elif not isinstance(pathTuple, tuple): if isinstance(pathTuple, int): - endpoint = pathTuple + return ClusterAttribute.AttributePath(EndpointId=pathTuple) elif issubclass(pathTuple, ClusterObjects.Cluster): - cluster = pathTuple + return ClusterAttribute.AttributePath.from_cluster(EndpointId=None, Cluster=pathTuple) elif issubclass(pathTuple, ClusterObjects.ClusterAttributeDescriptor): - attribute = pathTuple + return ClusterAttribute.AttributePath.from_attribute(EndpointId=None, Attribute=pathTuple) else: raise ValueError("Unsupported Attribute Path") else: # endpoint + (cluster) attribute / endpoint + cluster - endpoint = pathTuple[0] if issubclass(pathTuple[1], ClusterObjects.Cluster): - cluster = pathTuple[1] + return ClusterAttribute.AttributePath.from_cluster(EndpointId=pathTuple[0], Cluster=pathTuple[1]) elif issubclass(pathTuple[1], ClusterAttribute.ClusterAttributeDescriptor): - attribute = pathTuple[1] + return ClusterAttribute.AttributePath.from_attribute(EndpointId=pathTuple[0], Attribute=pathTuple[1]) else: raise ValueError("Unsupported Attribute Path") - return ClusterAttribute.AttributePath( - EndpointId=endpoint, Cluster=cluster, Attribute=attribute) def _parseDataVersionFilterTuple(self, pathTuple: typing.List[typing.Tuple[int, typing.Type[ClusterObjects.Cluster], int]]): endpoint = None @@ -1530,7 +1523,7 @@ def ZCLReadAttribute(self, cluster, attribute, nodeid, endpoint, groupid, blocki result = asyncio.run(self.ReadAttribute( nodeid, [(endpoint, attributeType)])) - path = ClusterAttribute.AttributePath( + path = ClusterAttribute.AttributePath.from_attribute( EndpointId=endpoint, Attribute=attributeType) return im.AttributeReadResult(path=im.AttributePath(nodeId=nodeid, endpointId=path.EndpointId, clusterId=path.ClusterId, attributeId=path.AttributeId), status=0, value=result[endpoint][clusterType][attributeType], dataVersion=result[endpoint][clusterType][ClusterAttribute.DataVersion]) diff --git a/src/controller/python/chip/clusters/Attribute.py b/src/controller/python/chip/clusters/Attribute.py index 9e46eed469d39f..818246be80b3d8 100644 --- a/src/controller/python/chip/clusters/Attribute.py +++ b/src/controller/python/chip/clusters/Attribute.py @@ -60,30 +60,23 @@ class AttributePath: ClusterId: int = None AttributeId: int = None - def __init__(self, EndpointId: int = None, Cluster=None, Attribute=None, ClusterId=None, AttributeId=None): - self.EndpointId = EndpointId - if Cluster is not None: - # Wildcard read for a specific cluster - if (Attribute is not None) or (ClusterId is not None) or (AttributeId is not None): - raise Warning( - "Attribute, ClusterId and AttributeId is ignored when Cluster is specified") - self.ClusterId = Cluster.id - return - if Attribute is not None: - if (ClusterId is not None) or (AttributeId is not None): - raise Warning( - "ClusterId and AttributeId is ignored when Attribute is specified") - self.ClusterId = Attribute.cluster_id - self.AttributeId = Attribute.attribute_id - return - self.ClusterId = ClusterId - self.AttributeId = AttributeId + @staticmethod + def from_cluster(EndpointId: int, Cluster: Cluster) -> AttributePath: + if Cluster is None: + raise ValueError("Cluster cannot be None") + return AttributePath(EndpointId=EndpointId, ClusterId=Cluster.id) + + @staticmethod + def from_attribute(EndpointId: int, Attribute: ClusterAttributeDescriptor) -> AttributePath: + if Attribute is None: + raise ValueError("Attribute cannot be None") + return AttributePath(EndpointId=EndpointId, ClusterId=Attribute.cluster_id, AttributeId=Attribute.attribute_id) def __str__(self) -> str: return f"{self.EndpointId}/{self.ClusterId}/{self.AttributeId}" def __hash__(self): - return str(self).__hash__() + return hash((self.EndpointId, self.ClusterId, self.AttributeId)) @dataclass diff --git a/src/controller/python/test/test_scripts/cluster_objects.py b/src/controller/python/test/test_scripts/cluster_objects.py index 53516c1dc75e9b..37f6819cbe66a4 100644 --- a/src/controller/python/test/test_scripts/cluster_objects.py +++ b/src/controller/python/test/test_scripts/cluster_objects.py @@ -164,7 +164,7 @@ async def TestWriteRequest(cls, devCtrl): ] ) expectedRes = [ - AttributeStatus(Path=AttributePath( + AttributeStatus(Path=AttributePath.from_attribute( EndpointId=1, Attribute=Clusters.UnitTesting.Attributes.ListLongOctetString), Status=chip.interaction_model.Status.Success), ]