From b3a7d0d761c919e03e09f97b6448920f6f5801d1 Mon Sep 17 00:00:00 2001 From: Marcel van der Veldt Date: Wed, 30 Nov 2022 11:21:47 +0100 Subject: [PATCH] Improve helper functions (#101) * Improve the helper methods and fix some typos --- .../common/models/device_type_instance.py | 32 ++++++------ matter_server/common/models/node.py | 50 ++++++++++++------- matter_server/common/models/node_device.py | 6 +-- 3 files changed, 53 insertions(+), 35 deletions(-) diff --git a/matter_server/common/models/device_type_instance.py b/matter_server/common/models/device_type_instance.py index 8e55a392..3da33bba 100644 --- a/matter_server/common/models/device_type_instance.py +++ b/matter_server/common/models/device_type_instance.py @@ -25,34 +25,38 @@ def __init__( self, node: MatterNode, device_type: _DEVICE_TYPE_T, - endpoint_id: int, + endpoint: int, device_revision: int, ) -> None: self.node = node self.device_type = device_type self.device_revision = device_revision - self.endpoint_id = endpoint_id + self.endpoint = endpoint @property def attributes(self) -> list[MatterAttribute]: - return self.node.get_endpoint_attributes(self.endpoint_id) + """Return all Attributes belonging to this DeviceTypeInstance.""" + return [ + attr + for attr in self.node.get_endpoint_attributes(self.endpoint) + if attr.cluster_type in self.device_type.clusters + ] def has_cluster(self, cluster: type[all_clusters.Cluster]) -> bool: """Check if device has a specific cluster.""" - return any(x for x in self.attributes if x.cluster_type == cluster) + # only return True if the cluster belongs to this device type + # and is actually present in the atributes. + return cluster in self.device_type.clusters and any( + x for x in self.attributes if x.cluster_type == cluster + ) def get_cluster(self, cluster: type[_CLUSTER_T]) -> _CLUSTER_T | None: """Get the cluster object.""" - if not self.has_cluster(cluster): + # only return Cluster if the cluster belongs to this device type + # and is actually present in the atributes. + if cluster not in self.device_type.clusters: return None - - # instantiate a Cluster object from the properties - return cluster( - **{ - x.attribute_name: x.value - for x in self.node.get_cluster_attributes(cluster, self.endpoint_id) - } - ) + return self.node.get_cluster(self.endpoint, cluster) def __repr__(self): - return f"" + return f"" diff --git a/matter_server/common/models/node.py b/matter_server/common/models/node.py index 2286ca17..03275fb1 100644 --- a/matter_server/common/models/node.py +++ b/matter_server/common/models/node.py @@ -26,13 +26,15 @@ LOGGER = logging.getLogger(__name__) +_CLUSTER_T = TypeVar("_CLUSTER_T", bound=Clusters.Cluster) + def create_attribute_path(endpoint: int, cluster_id: int, attribute_id: int) -> str: """ Create path/identifier for an Attribute. Returns same output as `Attribute.AttributePath` - endpoint_id/cluster_id/attribute_id + endpoint/cluster_id/attribute_id """ return f"{endpoint}/{cluster_id}/{attribute_id}" @@ -133,30 +135,18 @@ def __post_init__(self): else: self.node_devices.append(MatterNodeDevice(self)) - def has_cluster( - self, cluster: type[Cluster] | int, endpoint: int | None = None - ) -> bool: - """Check if node has a specific cluster.""" - return any( - x - for x in self.attributes.values() - if cluster in (x.cluster_type, x.cluster_id) - and (endpoint is None or x.endpoint == endpoint) - ) - def get_endpoint_attributes(self, endpoint: int) -> list[MatterAttribute]: """Return Matter Attributes for given endpoint.""" return [x for x in self.attributes.values() if x.endpoint == endpoint] def get_cluster_attributes( - self, cluster: type[Clusters.Cluster], endpoint: int | None = None + self, endpoint: int, cluster: type[Clusters.Cluster] | int ) -> list[MatterAttribute]: - """Return Matter Attributes for given cluster.""" + """Return all Attributes for given cluster.""" return [ x for x in self.attributes.values() - if x.cluster_type == cluster - and (endpoint is None or x.endpoint == endpoint) + if cluster in (x.cluster_type, x.cluster_id) and x.endpoint == endpoint ] def get_attribute( @@ -174,18 +164,42 @@ def get_attribute( and attribute in (x.attribute_id, x.attribute_name, x.attribute_type) ) + def has_cluster(self, endpoint: int, cluster: type[Cluster] | int) -> bool: + """Check if node has a specific cluster.""" + return any( + x + for x in self.attributes.values() + if cluster in (x.cluster_type, x.cluster_id) + and (endpoint is None or x.endpoint == endpoint) + ) + + def get_cluster( + self, endpoint: int, cluster: type[_CLUSTER_T] + ) -> _CLUSTER_T | None: + """ + Get a full Cluster object containing all attributes. + + Returns None is the Cluster is not present on the node. + """ + atrributes = self.get_cluster_attributes(endpoint, cluster) + if len(atrributes) == 0: + return None + + # instantiate a Cluster object from the properties + return cluster(**{x.attribute_name: x.value for x in atrributes}) + @property def name(self) -> str: """Return friendly name for this node.""" return self.get_attribute( - self.root_device_type_instance.endpoint_id, Clusters.Basic, "nodeLabel" + self.root_device_type_instance.endpoint, Clusters.Basic, "nodeLabel" ) @property def unique_id(self) -> str: """Return uniqueID for this node.""" return self.get_attribute( - self.root_device_type_instance.endpoint_id, Clusters.Basic, "uniqueID" + self.root_device_type_instance.endpoint, Clusters.Basic, "uniqueID" ) def __repr__(self): diff --git a/matter_server/common/models/node_device.py b/matter_server/common/models/node_device.py index 9d6a4d19..1b8d190d 100644 --- a/matter_server/common/models/node_device.py +++ b/matter_server/common/models/node_device.py @@ -68,14 +68,14 @@ def device_info(self) -> Clusters.BridgedDeviceBasic: ) def device_type_instances(self) -> list[MatterDeviceTypeInstance]: - endpoint_id = self.bridged_device_type_instance.endpoint_id + endpoint = self.bridged_device_type_instance.endpoint return [ inst for inst in self.bridged_device_type_instance.node.device_type_instances - if inst.endpoint_id == endpoint_id + if inst.endpoint == endpoint and inst != self.bridged_device_type_instance ] def __repr__(self) -> str: bridged = self.bridged_device_type_instance - return f"" + return f""