Use our own global models for CommissionableNodeData and Commissionin… (#520)

Author: marcelveldt

matter_server/client/client.py

@@ -22,6 +22,7 @@
 from ..common.models import (
     APICommand,
     CommandMessage,
+    CommissionableNodeData,
     CommissioningParameters,
     ErrorResultMessage,
     EventMessage,
@@ -198,6 +199,15 @@ async def open_commissioning_window(
             ),
         )
 
+    async def discover_commissionable_nodes(
+        self,
+    ) -> list[CommissionableNodeData]:
+        """Discover Commissionable Nodes (discovered on BLE or mDNS)."""
+        return [
+            dataclass_from_dict(CommissionableNodeData, x)
+            for x in await self.send_command(APICommand.DISCOVER, require_schema=7)
+        ]
+
     async def get_matter_fabrics(self, node_id: int) -> list[MatterFabricData]:
         """
         Get Matter fabrics from a device.

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 = 6
+SCHEMA_VERSION = 7

matter_server/common/models.py

@@ -175,13 +175,31 @@ class ServerInfoMessage:
 
 
 @dataclass
-class CommissioningParameters:
-    """
-    Object that is returned on the 'open_commisisoning_window' command.
+class CommissionableNodeData:
+    """Object that is returned on the 'discover_commissionable_nodes' command."""
+
+    instance_name: str | None = None
+    host_name: str | None = None
+    port: int | None = None
+    long_discriminator: int | None = None
+    vendor_id: int | None = None
+    product_id: int | None = None
+    commissioning_mode: int | None = None
+    device_type: int | None = None
+    device_name: str | None = None
+    pairing_instruction: str | None = None
+    pairing_hint: int | None = None
+    mrp_retry_interval_idle: int | None = None
+    mrp_retry_interval_active: int | None = None
+    supports_tcp: bool | None = None
+    addresses: list[str] | None = None
+    rotating_id: str | None = None
+
 
-    NOTE: This is just a copy of the dataclass specified in chip.ChipDeviceCtrl
-    """
+@dataclass
+class CommissioningParameters:
+    """Object that is returned on the 'open_commisisoning_window' command."""
 
-    setupPinCode: int  # pylint: disable=invalid-name
-    setupManualCode: str  # pylint: disable=invalid-name
-    setupQRCode: str  # pylint: disable=invalid-name
+    setup_pin_code: int
+    setup_manual_code: str
+    setup_qr_code: str

matter_server/server/device_controller.py

@@ -11,15 +11,14 @@
 import random
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterable, TypeVar, cast
 
-from chip.ChipDeviceCtrl import CommissionableNode, DeviceProxyWrapper
+from chip.ChipDeviceCtrl import DeviceProxyWrapper
 from chip.clusters import Attribute, Objects as Clusters
 from chip.clusters.Attribute import ValueDecodeFailure
 from chip.clusters.ClusterObjects import ALL_ATTRIBUTES, ALL_CLUSTERS, Cluster
-from chip.discovery import CommissionableNode as CommissionableNodeData
 from chip.exceptions import ChipStackError
 
 from matter_server.common.helpers.util import convert_ip_address
-from matter_server.common.models import CommissioningParameters
+from matter_server.common.models import CommissionableNodeData, CommissioningParameters
 from matter_server.server.helpers.attributes import parse_attributes_from_read_result
 from matter_server.server.helpers.utils import ping_ip
 
@@ -391,37 +390,57 @@ async def open_commissioning_window(
         if discriminator is None:
             discriminator = 3840  # TODO generate random one
 
-        return await self._call_sdk(
+        sdk_result = await self._call_sdk(
             self.chip_controller.OpenCommissioningWindow,
             nodeid=node_id,
             timeout=timeout,
             iteration=iteration,
             discriminator=discriminator,
             option=option,
         )
+        return CommissioningParameters(
+            setup_pin_code=sdk_result.setupPinCode,
+            setup_manual_code=sdk_result.setupManualCode,
+            setup_qr_code=sdk_result.setupQRCode,
+        )
 
     @api_command(APICommand.DISCOVER)
     async def discover_commissionable_nodes(
         self,
-    ) -> CommissionableNodeData | list[CommissionableNodeData] | None:
+    ) -> list[CommissionableNodeData]:
         """Discover Commissionable Nodes (discovered on BLE or mDNS)."""
         if self.chip_controller is None:
             raise RuntimeError("Device Controller not initialized.")
 
-        result = await self._call_sdk(
+        sdk_result = await self._call_sdk(
             self.chip_controller.DiscoverCommissionableNodes,
         )
-
-        def convert(cn: CommissionableNode) -> CommissionableNodeData:
-            cnd = CommissionableNodeData()
-            # pylint: disable=no-member
-            for field in CommissionableNodeData.__dataclass_fields__:
-                setattr(cnd, field, getattr(cn, field))
-            return cnd
-
-        if isinstance(result, list):
-            return [convert(c) for c in result]
-        return convert(result)
+        if sdk_result is None:
+            return []
+        # ensure list
+        if not isinstance(sdk_result, list):
+            sdk_result = [sdk_result]
+        return [
+            CommissionableNodeData(
+                instance_name=x.instanceName,
+                host_name=x.hostName,
+                port=x.port,
+                long_discriminator=x.longDiscriminator,
+                vendor_id=x.vendorId,
+                product_id=x.productId,
+                commissioning_mode=x.commissioningMode,
+                device_type=x.deviceType,
+                device_name=x.deviceName,
+                pairing_instruction=x.pairingInstruction,
+                pairing_hint=x.pairingHint,
+                mrp_retry_interval_idle=x.mrpRetryIntervalIdle,
+                mrp_retry_interval_active=x.mrpRetryIntervalActive,
+                supports_tcp=x.supportsTcp,
+                addresses=x.addresses,
+                rotating_id=x.rotatingId,
+            )
+            for x in sdk_result
+        ]
 
     @api_command(APICommand.INTERVIEW_NODE)
     async def interview_node(self, node_id: int) -> None: