Fix support for bridged and composed devices

Author: marcelveldt

matter_server/client/models/device_type_instance.py

@@ -16,8 +16,6 @@
 class DeviceType:
     """Base class for Matter device types."""
 
-    do_not_serialize = True
-
     device_type: int
     clusters: set[type[all_clusters.Cluster]]
 
@@ -27,6 +25,10 @@ def __init_subclass__(cls, *, device_type: int, **kwargs: typing.Any) -> None:
         cls.device_type = device_type
         ALL_TYPES[device_type] = cls
 
+    def __hash__(self) -> int:
+        """Return unique hash for this object."""
+        return hash(self.device_type)
+
 
 class OrphanClusters(DeviceType, device_type=0xF001):
     """Orphan Clusters."""

matter_server/client/models/device_types.py

@@ -14,12 +14,12 @@
 )
 from matter_server.common.models import MatterNodeData
 
-from .device_type_instance import MatterDeviceTypeInstance
-from .device_types import ALL_TYPES as DEVICE_TYPES, Aggregator, BridgedDevice, RootNode
-from .node_device import (
-    AbstractMatterNodeDevice,
-    MatterBridgedNodeDevice,
-    MatterNodeDevice,
+from .device_types import (
+    ALL_TYPES as DEVICE_TYPES,
+    Aggregator,
+    BridgedDevice,
+    DeviceType,
+    RootNode,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -53,9 +53,33 @@ def __init__(
         self.node = node
         self.endpoint_id = endpoint_id
         self.clusters: dict[int, Clusters.Cluster] = {}
-        # unwrap cluster and clusterattributes from raw node data attributes
-        for attribute_path, attribute_value in attributes_data.items():
-            self.set_attribute_value(attribute_path, attribute_value)
+        self.device_types: set[DeviceType] = set()
+        self.update(attributes_data)
+
+    @property
+    def is_bridged_device(self) -> bool:
+        """Return if this endpoint represents a Bridged device."""
+        return BridgedDevice in self.device_types
+
+    @property
+    def is_composed_device(self) -> bool:
+        """Return if this endpoint belons to a composed device."""
+        return self.node.get_compose_parent(self.endpoint_id) is not None
+
+    @property
+    def device_info(self) -> Clusters.BasicInformation | Clusters.BridgedDeviceBasic:
+        """
+        Return device info.
+
+        If this endpoint represents a BridgedDevice, returns BridgedDeviceBasic.
+        If this endpoint represents a ComposedDevice, returns the info of the compose device.
+        Otherwise, returns BasicInformation from the Node itself (endpoint 0).
+        """
+        if self.is_bridged_device:
+            return self.get_cluster(Clusters.BridgedDeviceBasic)
+        if compose_parent := self.node.get_compose_parent(self.endpoint_id):
+            return compose_parent.device_info
+        return self.node.device_info
 
     def has_cluster(self, cluster: type[_CLUSTER_T] | int) -> bool:
         """Check if endpoint has a specific cluster."""
@@ -70,8 +94,8 @@ def get_cluster(self, cluster: type[_CLUSTER_T] | int) -> _CLUSTER_T | None:
         Return None if the Cluster is not present on the node.
         """
         if isinstance(cluster, type):
-            return self.clusters[cluster.id]  # type: ignore[no-any-return]
-        return self.clusters[cluster]  # type: ignore[no-any-return]
+            return self.clusters.get(cluster.id)
+        return self.clusters.get(cluster)
 
     def get_attribute_value(
         self,
@@ -155,91 +179,69 @@ def set_attribute_value(self, attribute_path: str, attribute_value: Any) -> None
         )
         setattr(cluster_instance, attribute_name, attribute_value)
 
+    def update(self, attributes_data: dict[str, Any]) -> None:
+        """Update MatterEndpoint from (endpoint-specific) raw Attributes data."""
+        # unwrap cluster and clusterattributes from raw node data attributes
+        for attribute_path, attribute_value in attributes_data.items():
+            self.set_attribute_value(attribute_path, attribute_value)
+        # extract device types from Descriptor Cluster
+        cluster = self.get_cluster(Clusters.Descriptor)
+        assert cluster is not None
+        for dev_info in cluster.deviceTypeList:  # type: ignore[unreachable]
+            device_type = DEVICE_TYPES.get(dev_info.type)
+            if device_type is None:
+                LOGGER.debug("Found unknown device type %s", dev_info)
+                continue
+            self.device_types.add(device_type)
+
+    def __repr__(self) -> str:
+        """Return the representation."""
+        return f"<MatterEndoint {self.endpoint_id} (node {self.node.node_id})>"
+
 
 class MatterNode:
     """Representation of a Matter Node."""
 
     def __init__(self, node_data: MatterNodeData) -> None:
         """Initialize MatterNode from MatterNodeData."""
         self.endpoints: dict[int, MatterEndpoint] = {}
-        self.root_device_type_instance: MatterDeviceTypeInstance[RootNode] | None = None
-        self.aggregator_device_type_instance: MatterDeviceTypeInstance[
-            Aggregator
-        ] | None = None
-        self.node_devices: list[AbstractMatterNodeDevice] = []
-        self.device_type_instances: list[MatterDeviceTypeInstance] = []
+        self._is_bridge_device: bool = False
+        # composed devices reference to other endpoints through the partsList attribute
+        # create a mapping table
+        self._composed_endpoints: dict[int, int] = {}
         self.update(node_data)
 
-    def update(self, node_data: MatterNodeData) -> None:
-        """Update MatterNode from MatterNodeData."""
-        # pylint: disable=too-many-branches
-        self.node_data = node_data
-        # collect per endpoint data
-        endpoint_data: dict[int, dict[str, Any]] = {}
-        for attribute_path, attribute_data in node_data.attributes.items():
-            endpoint_id = int(attribute_path.split("/")[0])
-            endpoint_data.setdefault(endpoint_id, {})
-            endpoint_data[endpoint_id][attribute_path] = attribute_data
-        # TODO: Should we update existing endpoints instead of overwriting them?
-        for endpoint_id, attributes_data in endpoint_data.items():
-            self.endpoints[endpoint_id] = MatterEndpoint(
-                endpoint_id=endpoint_id, attributes_data=attributes_data, node=self
-            )
-        # lookup device types from node data
-        for endpoint in self.endpoints.values():
-            # get DeviceTypeList Attribute on the Descriptor cluster
-            cluster = endpoint.get_cluster(Clusters.Descriptor)
-            if not cluster:
-                LOGGER.debug(
-                    "No Descriptor cluster found on endpoint %s, Node %s",
-                    endpoint.endpoint_id,
-                    endpoint.node.node_id,
-                )
-                continue
-
-            for dev_info in cluster.deviceTypeList:  # type: ignore[unreachable]
-                device_type = DEVICE_TYPES.get(dev_info.type)
-                if device_type is None:
-                    LOGGER.debug("Found unknown device type %s", dev_info)
-                    continue
-
-                instance: MatterDeviceTypeInstance[Any] = MatterDeviceTypeInstance(
-                    self, device_type, endpoint, dev_info.revision
-                )
-                if device_type is RootNode:
-                    self.root_device_type_instance = instance
-                elif device_type is Aggregator:
-                    self.aggregator_device_type_instance = instance
-                else:
-                    self.device_type_instances.append(instance)
-
-        if self.root_device_type_instance is None:
-            raise ValueError("No root device found")
-
-        # parse node devices
-        self.node_devices = []
-        if self.aggregator_device_type_instance:
-            for instance in self.device_type_instances:
-                if instance.device_type == BridgedDevice:
-                    self.node_devices.append(MatterBridgedNodeDevice(instance))
-        else:
-            self.node_devices.append(MatterNodeDevice(self))
-
-    def update_attribute(self, attribute_path: str, new_value: Any) -> None:
-        """Handle Attribute value update."""
-        endpoint_id = int(attribute_path.split("/")[0])
-        self.endpoints[endpoint_id].set_attribute_value(attribute_path, new_value)
-
     @property
     def node_id(self) -> int:
         """Return Node ID."""
         return self.node_data.node_id
 
+    @property
+    def name(self) -> str | None:
+        """Return friendly name for this node."""
+        if info := self.device_info:
+            return cast(str, info.nodeLabel)
+        return None
+
     @property
     def available(self) -> bool:
         """Return availability of the node."""
         return self.node_data.available
 
+    @property
+    def device_info(self) -> Clusters.BasicInformation:
+        """
+        Return device info for this Node.
+
+        Returns BasicInformation from the Node itself (endpoint 0).
+        """
+        return self.get_cluster(0, Clusters.BasicInformation)
+
+    @property
+    def is_bridge_device(self) -> bool:
+        """Return if this Node is a Bridge/Aggregator device."""
+        return self._is_bridge_device
+
     def get_attribute_value(
         self,
         endpoint: int,
@@ -270,18 +272,56 @@ def get_cluster(
         """
         return self.endpoints[endpoint].get_cluster(cluster)
 
-    @property
-    def name(self) -> str | None:
-        """Return friendly name for this node."""
-        if self.root_device_type_instance is None:
-            return None
-        return cast(
-            str,
-            self.root_device_type_instance.endpoint.get_attribute_value(
-                None,
-                Clusters.BasicInformation.Attributes.NodeLabel,
-            ),
+    def get_compose_parent(self, endpoint_id: int) -> MatterEndpoint | None:
+        """Return endpoint of parent if the endpoint belongs to a Composed device."""
+        if parent_id := self._composed_endpoints.get(endpoint_id):
+            return self.endpoints[parent_id]
+        return None
+
+    def get_compose_child_ids(self, endpoint_id: int) -> tuple[int, ...] | None:
+        """Return endpoint ID's of any childs if the endpoint represents a Composed device."""
+        return tuple(x for x, y in self._composed_endpoints.items() if y == endpoint_id)
+
+    def update(self, node_data: MatterNodeData) -> None:
+        """Update MatterNode from MatterNodeData."""
+        self.node_data = node_data
+        # collect per endpoint data
+        endpoint_data: dict[int, dict[str, Any]] = {}
+        for attribute_path, attribute_data in node_data.attributes.items():
+            endpoint_id = int(attribute_path.split("/")[0])
+            if endpoint_id not in endpoint_data:
+                endpoint_data[endpoint_id] = {}
+            endpoint_data[endpoint_id][attribute_path] = attribute_data
+        for endpoint_id, attributes_data in endpoint_data.items():
+            if endpoint_id in self.endpoints:
+                self.endpoints[endpoint_id].update(attributes_data)
+            else:
+                self.endpoints[endpoint_id] = MatterEndpoint(
+                    endpoint_id=endpoint_id, attributes_data=attributes_data, node=self
+                )
+        # lookup if this is a bridge device
+        self._is_bridge_device = any(
+            Aggregator in x.device_types for x in self.endpoints.values()
         )
+        # composed devices reference to other endpoints through the partsList attribute
+        # create a mapping table to quickly map this
+        for endpoint in self.endpoints.values():
+            if RootNode in endpoint.device_types:
+                # ignore root endoint
+                continue
+            if Aggregator in endpoint.device_types:
+                # ignore Bridge endpoint (as that will also use partsList to indicate its childs)
+                continue
+            descriptor = endpoint.get_cluster(Clusters.Descriptor)
+            assert descriptor is not None
+            if descriptor.partsList:  # type: ignore[unreachable]
+                for endpoint_id in descriptor.partsList:
+                    self._composed_endpoints[endpoint_id] = endpoint.endpoint_id
+
+    def update_attribute(self, attribute_path: str, new_value: Any) -> None:
+        """Handle Attribute value update."""
+        endpoint_id = int(attribute_path.split("/")[0])
+        self.endpoints[endpoint_id].set_attribute_value(attribute_path, new_value)
 
     def __repr__(self) -> str:
         """Return the representation."""