introduce allow_sdk_types in parse value helper

Author: marcelveldt

matter_server/common/helpers/util.py

@@ -3,16 +3,17 @@
 from __future__ import annotations
 
 import base64
-from base64 import b64decode
 import binascii
+import logging
+import platform
+import socket
+from base64 import b64decode
 from dataclasses import MISSING, asdict, fields, is_dataclass
 from datetime import datetime
 from enum import Enum
 from functools import cache
-from importlib.metadata import PackageNotFoundError, version as pkg_version
-import logging
-import platform
-import socket
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as pkg_version
 from types import NoneType, UnionType
 from typing import (
     TYPE_CHECKING,
@@ -108,8 +109,13 @@ def parse_value(
     value_type: Any,
     default: Any = MISSING,
     allow_none: bool = True,
+    allow_sdk_types: bool = False,
 ) -> Any:
-    """Try to parse a value from raw (json) data and type annotations."""
+    """
+    Try to parse a value from raw (json) data and type annotations.
+
+    If allow_sdk_types is False, any SDK specific custom data types will be converted.
+    """
     # pylint: disable=too-many-return-statements,too-many-branches
 
     if isinstance(value_type, str):
@@ -131,9 +137,9 @@ def parse_value(
         return default
     if value is None and value_type is NoneType:
         return None
-    if value is None and allow_none:
-        return None
     if value is None and value_type is Nullable:
+        return Nullable() if allow_sdk_types else None
+    if value is None and allow_none:
         return None
     if is_dataclass(value_type) and isinstance(value, dict):
         return dataclass_from_dict(value_type, value)
@@ -220,12 +226,12 @@ def parse_value(
     if value_type is uint and (
         isinstance(value, int) or (isinstance(value, str) and value.isnumeric())
     ):
-        return uint(value)
+        return uint(value) if allow_sdk_types else int(value)
     if value_type is float32 and (
         isinstance(value, (float, int))
         or (isinstance(value, str) and value.isnumeric())
     ):
-        return float32(value)
+        return float32(value) if allow_sdk_types else float(value)
 
     # If we reach this point, we could not match the value with the type and we raise
     if not isinstance(value, value_type):
@@ -236,7 +242,9 @@ def parse_value(
     return value
 
 
-def dataclass_from_dict(cls: type[_T], dict_obj: dict, strict: bool = False) -> _T:
+def dataclass_from_dict(
+    cls: type[_T], dict_obj: dict, strict: bool = False, allow_sdk_types: bool = False
+) -> _T:
     """
     Create (instance of) a dataclass by providing a dict with values.
 
@@ -259,6 +267,7 @@ def dataclass_from_dict(cls: type[_T], dict_obj: dict, strict: bool = False) ->
                 type_hints[field.name],
                 field.default,
                 allow_none=not strict,
+                allow_sdk_types=allow_sdk_types,
             )
             for field in dc_fields
             if field.init

matter_server/server/device_controller.py

@@ -22,7 +22,7 @@
 
 from matter_server.common.helpers.util import convert_ip_address
 from matter_server.common.models import CommissionableNodeData, CommissioningParameters
-from matter_server.server.helpers.attributes import parse_attributes_from_read_result, parse_nullable_for_write
+from matter_server.server.helpers.attributes import parse_attributes_from_read_result
 from matter_server.server.helpers.utils import ping_ip
 
 from ..common.errors import (
@@ -619,8 +619,16 @@ async def write_attribute(
         if (node := self._nodes.get(node_id)) is None or not node.available:
             raise NodeNotReady(f"Node {node_id} is not (yet) available.")
         endpoint_id, cluster_id, attribute_id = parse_attribute_path(attribute_path)
-        attribute = ALL_ATTRIBUTES[cluster_id][attribute_id]()
-        attribute.value = parse_nullable_for_write(value)
+        attribute = cast(
+            Clusters.ClusterAttributeDescriptor,
+            ALL_ATTRIBUTES[cluster_id][attribute_id](),
+        )
+        attribute.value = parse_value(
+            name=attribute.__qualname__,
+            value=value,
+            value_type=attribute.attribute_type.Type,
+            allow_sdk_types=True,
+        )
         return await self.chip_controller.WriteAttribute(
             nodeid=node_id,
             attributes=[(endpoint_id, attribute)],
@@ -1256,4 +1264,3 @@ async def _node_offline(self, node_id: int) -> None:
         node.available = False
         self.server.signal_event(EventType.NODE_UPDATED, node)
         LOGGER.info("Marked node %s as offline", node_id)
-

matter_server/server/helpers/attributes.py

@@ -1,8 +1,6 @@
 """Helpers to manage Cluster attributes."""
 
 from typing import Any
-from copy import deepcopy
-from chip.clusters import Objects as Clusters
 
 from ...common.helpers.util import create_attribute_path
 
@@ -25,18 +23,3 @@ def parse_attributes_from_read_result(
                 )
                 result[attribute_path] = attr_value
     return result
-
-def parse_nullable_for_write(payload):
-    """Parses the Python 'None' values to CHIP 'Nullable' type for writing attribute."""
-    outload = deepcopy(payload)
-
-    if isinstance(outload, dict):
-        for k, v in outload.items():
-            outload[k] = parse_nullable_for_write(outload[k])
-    elif isinstance(outload, list):
-        for i in range(0, len(outload)):
-            outload[i] = parse_nullable_for_write(outload[i])
-    else:
-        outload = Clusters.NullValue if outload is None else outload
-
-    return outload