Refactor subscription logic

matter_server/client/client.py

@@ -67,6 +67,10 @@ def subscribe(
         Optionally filter by specific events or node attributes.
         Returns:
             function to unsubscribe.
+
+        NOTE: To receive attribute changed events,
+        you must also register the attributes to subscribe to
+        with the `subscribe_attributes` method.
         """
         # for fast lookups we create a key based on the filters, allowing
         # a "catch all" with a wildcard (*).
@@ -94,11 +98,11 @@ def unsubscribe() -> None:
 
         return unsubscribe
 
-    async def get_nodes(self) -> list[MatterNode]:
+    def get_nodes(self) -> list[MatterNode]:
         """Return all Matter nodes."""
         return list(self._nodes.values())
 
-    async def get_node(self, node_id: int) -> MatterNode:
+    def get_node(self, node_id: int) -> MatterNode:
         """Return Matter node by id."""
         return self._nodes[node_id]
 
@@ -229,6 +233,22 @@ async def remove_node(self, node_id: int) -> None:
         """Remove a Matter node/device from the fabric."""
         await self.send_command(APICommand.REMOVE_NODE, node_id=node_id)
 
+    async def subscribe_attribute(
+        self, node_id: int, attribute_path: str | list[str]
+    ) -> None:
+        """
+        Subscribe to given AttributePath(s).
+
+        Either supply a single attribute path or a list of paths.
+        The given attribute path(s) will be added to the list of attributes that
+        are watched for the given node. This is persistent over restarts.
+        """
+        await self.send_command(
+            APICommand.SUBSCRIBE_ATTRIBUTE,
+            node_id=node_id,
+            attribute_path=attribute_path,
+        )
+
     async def send_command(
         self,
         command: str,
@@ -366,7 +386,7 @@ def _handle_incoming_message(self, msg: MessageType) -> None:
 
         # handle EventMessage
         if isinstance(msg, EventMessage):
-            self.logger.debug("Received event: %s", msg)
+            self.logger.debug("Received event: %s", msg.event)
             self._handle_event_message(msg)
             return
 
@@ -392,10 +412,20 @@ def _handle_event_message(self, msg: EventMessage) -> None:
                 node.update(node_data)
             self._signal_event(event, data=node, node_id=node.node_id)
             return
-        if msg.event == EventType.NODE_DELETED:
+        if msg.event == EventType.NODE_REMOVED:
             node_id = msg.data
+            self._signal_event(EventType.NODE_REMOVED, data=node_id, node_id=node_id)
+            # cleanup node only after signalling subscribers
             self._nodes.pop(node_id, None)
-            self._signal_event(EventType.NODE_DELETED, data=node_id, node_id=node_id)
+            return
+        if msg.event == EventType.ENDPOINT_REMOVED:
+            node_id = msg.data["node_id"]
+            self._signal_event(
+                EventType.ENDPOINT_REMOVED, data=msg.data, node_id=node_id
+            )
+            # cleanup endpoint only after signalling subscribers
+            if node := self._nodes.get(node_id):
+                node.endpoints.pop(msg.data["endpoint_id"], None)
             return
         if msg.event == EventType.ATTRIBUTE_UPDATED:
             # data is tuple[node_id, attribute_path, new_value]

matter_server/client/connection.py

@@ -3,6 +3,7 @@
 
 import asyncio
 import logging
+import os
 import pprint
 from typing import Any, Callable, Dict, Final, cast
 
@@ -33,6 +34,7 @@
 from .models.node import MatterNode
 
 LOGGER = logging.getLogger(f"{__package__}.connection")
+VERBOSE_LOGGER = os.environ.get("MATTER_VERBOSE_LOGGING")
 SUB_WILDCARD: Final = "*"
 
 
@@ -129,7 +131,10 @@ async def receive_message_or_raise(self) -> MessageType:
             raise InvalidMessage("Received invalid JSON.") from err
 
         if LOGGER.isEnabledFor(logging.DEBUG):
-            LOGGER.debug("Received message:\n%s\n", pprint.pformat(ws_msg))
+            if VERBOSE_LOGGER:
+                LOGGER.debug("Received message:\n%s\n", pprint.pformat(ws_msg))
+            else:
+                LOGGER.debug("Received message: %s ...", ws_msg.data[:50])
 
         return msg
 
@@ -143,7 +148,10 @@ async def send_message(self, message: CommandMessage) -> None:
             raise NotConnected
 
         if LOGGER.isEnabledFor(logging.DEBUG):
-            LOGGER.debug("Publishing message:\n%s\n", pprint.pformat(message))
+            if VERBOSE_LOGGER:
+                LOGGER.debug("Publishing message:\n%s\n", pprint.pformat(message))
+            else:
+                LOGGER.debug("Publishing message: %s", message)
 
         assert self._ws_client
         assert isinstance(message, CommandMessage)

matter_server/client/models/node.py

@@ -216,7 +216,6 @@ class MatterNode:
     def __init__(self, node_data: MatterNodeData) -> None:
         """Initialize MatterNode from MatterNodeData."""
         self.endpoints: dict[int, MatterEndpoint] = {}
-        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] = {}
@@ -251,7 +250,7 @@ def device_info(self) -> Clusters.BasicInformation:
     @property
     def is_bridge_device(self) -> bool:
         """Return if this Node is a Bridge/Aggregator device."""
-        return self._is_bridge_device
+        return self.node_data.is_bridge
 
     def get_attribute_value(
         self,
@@ -310,10 +309,6 @@ def update(self, node_data: MatterNodeData) -> None:
                 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():
@@ -339,6 +334,9 @@ def update(self, node_data: MatterNodeData) -> None:
     def update_attribute(self, attribute_path: str, new_value: Any) -> None:
         """Handle Attribute value update."""
         endpoint_id = int(attribute_path.split("/")[0])
+        if endpoint_id not in self.endpoints:
+            # race condition when a bridge is in the process of adding a new endpoint
+            return
         self.endpoints[endpoint_id].set_attribute_value(attribute_path, new_value)
 
     def __repr__(self) -> str:

matter_server/common/const.py

@@ -2,4 +2,4 @@
 
 # schema version is used to determine compatibility between server and client
 # bump schema if we add new features and/or make other (breaking) changes
-SCHEMA_VERSION = 3
+SCHEMA_VERSION = 4

matter_server/common/helpers/json.py

@@ -4,7 +4,6 @@
 from dataclasses import is_dataclass
 from typing import Any
 
-from chip.clusters.Attribute import ValueDecodeFailure
 from chip.clusters.Types import Nullable
 from chip.tlv import float32, uint
 import orjson
@@ -22,8 +21,6 @@ def json_encoder_default(obj: Any) -> Any:
     """
     if getattr(obj, "do_not_serialize", None):
         return None
-    if isinstance(obj, ValueDecodeFailure):
-        return None
     if isinstance(obj, (set, tuple)):
         return list(obj)
     if isinstance(obj, float32):

matter_server/common/helpers/util.py

@@ -105,7 +105,7 @@ def parse_value(name: str, value: Any, value_type: Any, default: Any = MISSING)
         return dataclass_from_dict(value_type, value)
     # get origin value type and inspect one-by-one
     origin: Any = get_origin(value_type)
-    if origin in (list, tuple) and isinstance(value, list | tuple):
+    if origin in (list, tuple, set) and isinstance(value, (list, tuple, set)):
         return origin(
             parse_value(name, subvalue, get_args(value_type)[0])
             for subvalue in value

matter_server/common/models.py

@@ -14,10 +14,12 @@ class EventType(Enum):
 
     NODE_ADDED = "node_added"
     NODE_UPDATED = "node_updated"
-    NODE_DELETED = "node_deleted"
+    NODE_REMOVED = "node_removed"
     NODE_EVENT = "node_event"
     ATTRIBUTE_UPDATED = "attribute_updated"
     SERVER_SHUTDOWN = "server_shutdown"
+    ENDPOINT_ADDED = "endpoint_added"
+    ENDPOINT_REMOVED = "endpoint_removed"
 
 
 class APICommand(str, Enum):
@@ -38,6 +40,7 @@ class APICommand(str, Enum):
     DEVICE_COMMAND = "device_command"
     REMOVE_NODE = "remove_node"
     GET_VENDOR_NAMES = "get_vendor_names"
+    SUBSCRIBE_ATTRIBUTE = "subscribe_attribute"
 
 
 EventCallBackType = Callable[[EventType, Any], None]
@@ -66,8 +69,15 @@ class MatterNodeData:
     last_interview: datetime
     interview_version: int
     available: bool = False
+    is_bridge: bool = False
     # attributes are stored in form of AttributePath: ENDPOINT/CLUSTER_ID/ATTRIBUTE_ID
     attributes: dict[str, Any] = field(default_factory=dict)
+    # all attribute subscriptions we need to persist for this node,
+    # a set of tuples in format (endpoint_id, cluster_id, attribute_id)
+    # where each value can also be a '*' for wildcard
+    attribute_subscriptions: set[tuple[int | str, int | str, int | str]] = field(
+        default_factory=set
+    )
 
 
 @dataclass

matter_server/server/__main__.py

@@ -70,6 +70,8 @@ def main() -> None:
         handlers = [logging.FileHandler(args.log_file)]
     logging.basicConfig(handlers=handlers, level=args.log_level.upper())
     coloredlogs.install(level=args.log_level.upper())
+    if not logging.getLogger().isEnabledFor(logging.DEBUG):
+        logging.getLogger("chip").setLevel(logging.WARNING)
 
     # make sure storage path exists
     if not os.path.isdir(args.storage_path):