Fix issues with parsing models on Python 3.9

.gitignore

@@ -6,4 +6,5 @@ venv
 dist
 .DS_Store
 credentials/
-build/
+build/
+venv39/

matter_server/client/client.py

@@ -32,7 +32,7 @@
     dataclass_from_dict,
     dataclass_to_dict,
     parse_value,
-    parse_message
+    parse_message,
 )
 from ..common.models.api_command import APICommand
 from ..common.models.events import EventType
@@ -197,7 +197,7 @@ async def send_device_command(
             APICommand.DEVICE_COMMAND,
             node_id=node_id,
             endpoint=endpoint,
-            payload=payload
+            payload=payload,
         )
 
     async def remove_node(self, node_id: int) -> None:

matter_server/common/helpers/json.py

@@ -42,7 +42,9 @@ def json_dumps(data: Any) -> str:
     """Dump json string."""
     return orjson.dumps(
         data,
-        option=orjson.OPT_NON_STR_KEYS | orjson.OPT_INDENT_2 | orjson.OPT_PASSTHROUGH_DATACLASS,
+        option=orjson.OPT_NON_STR_KEYS
+        | orjson.OPT_INDENT_2
+        | orjson.OPT_PASSTHROUGH_DATACLASS,
         default=json_encoder_default,
     ).decode("utf-8")
 

matter_server/common/helpers/util.py

@@ -35,7 +35,7 @@
 except:  # noqa
     # older python version
     NoneType = type(None)
-    UnionType = type(int | str)
+    UnionType = type(Union)
 
 CHIP_CLUSTERS_PKG_NAME = "home-assistant-chip-clusters"
 CHIP_CORE_PKG_NAME = "home-assistant-chip-core"
@@ -87,15 +87,20 @@ def parse_utc_timestamp(datetimestr: str):
     return datetime.fromisoformat(datetimestr.replace("Z", "+00:00"))
 
 
-def parse_value(name: str, value: Any, value_type: Type | str, default: Any = MISSING):
+def parse_value(
+    name: str, value: Any, value_type: Union[Type, str], default: Any = MISSING
+):
     """Try to parse a value from raw (json) data and type definitions."""
     if isinstance(value, dict) and "_type" in value:
         return dataclass_from_dict(None, value)
     if isinstance(value_type, str):
         # type is provided as string
         if value_type == "type":
             return locate(value) or eval(value)
-        value_type = eval(value_type)
+        try:
+            value_type = eval(value_type)
+        except TypeError:
+            pass
 
     elif isinstance(value, dict):
         if hasattr(value_type, "from_dict"):
@@ -184,7 +189,7 @@ def parse_value(name: str, value: Any, value_type: Type | str, default: Any = MI
     return value
 
 
-def dataclass_from_dict(cls: dataclass | None, dict_obj: dict, strict=False):
+def dataclass_from_dict(cls: Optional[dataclass], dict_obj: dict, strict=False):
     """
     Create (instance of) a dataclass by providing a dict with values.
 

matter_server/common/models/device_type_instance.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Callable, Coroutine, Generic, TypeVar
+from typing import TYPE_CHECKING, Callable, Coroutine, Generic, Optional, TypeVar
 
 from .device_types import DeviceType
 from chip.clusters import Objects as all_clusters
@@ -50,7 +50,7 @@ def has_cluster(self, cluster: type[all_clusters.Cluster]) -> bool:
             x for x in self.attributes if x.cluster_type == cluster
         )
 
-    def get_cluster(self, cluster: type[_CLUSTER_T]) -> _CLUSTER_T | None:
+    def get_cluster(self, cluster: type[_CLUSTER_T]) -> Optional[_CLUSTER_T]:
         """Get the cluster object."""
         # only return Cluster if the cluster belongs to this device type
         # and is actually present in the atributes.

matter_server/common/models/message.py

@@ -3,7 +3,7 @@
 
 from dataclasses import dataclass
 from enum import Enum, IntEnum
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 from .events import EventType
 from .server_information import ServerInfo
@@ -15,7 +15,7 @@ class CommandMessage:
 
     message_id: str
     command: str
-    args: dict[str, Any] | None = None
+    args: Union[dict[str, Any], None] = None
 
 
 @dataclass
@@ -46,7 +46,7 @@ class ErrorResultMessage(ResultMessageBase):
     """Message sent when a command did not execute succesfully"""
 
     error_code: ErrorCode
-    details: str | None = None
+    details: Optional[str] = None
 
 
 @dataclass
@@ -69,5 +69,3 @@ class ServerInfoMessage(ServerInfo):
     ErrorResultMessage,
     ServerInfoMessage,
 ]
-
-

matter_server/common/models/node.py

@@ -5,7 +5,7 @@
 from datetime import datetime
 import inspect
 import logging
-from typing import TYPE_CHECKING, Any, Dict, TypeVar
+from typing import TYPE_CHECKING, Any, Dict, Optional, TypeVar, Union
 
 from chip.clusters import (
     Objects as Clusters,
@@ -73,12 +73,12 @@ class MatterNode:
     attributes: Dict[str, MatterAttribute]
     # below attributes are derrived from the attributes in post init.
     endpoints: list[int] = field(default_factory=list, init=False)
-    root_device_type_instance: MatterDeviceTypeInstance[RootNode] | None = field(
+    root_device_type_instance: Optional[MatterDeviceTypeInstance[RootNode]] = field(
         default=None, init=False
     )
-    aggregator_device_type_instance: MatterDeviceTypeInstance[
-        Aggregator
-    ] | None = field(default=None, init=False)
+    aggregator_device_type_instance: Optional[
+        MatterDeviceTypeInstance[Aggregator]
+    ] = field(default=None, init=False)
     device_type_instances: list[MatterDeviceTypeInstance] = field(
         default_factory=list, init=False
     )
@@ -134,7 +134,7 @@ def get_endpoint_attributes(self, endpoint: int) -> list[MatterAttribute]:
         return [x for x in self.attributes.values() if x.endpoint == endpoint]
 
     def get_cluster_attributes(
-        self, endpoint: int, cluster: type[Clusters.Cluster] | int
+        self, endpoint: int, cluster: Union[type[Clusters.Cluster], int]
     ) -> list[MatterAttribute]:
         """Return all Attributes for given cluster."""
         return [
@@ -147,7 +147,7 @@ def get_attribute(
         self,
         endpoint: int,
         cluster: type[Clusters.Cluster],
-        attribute: str | int | type,
+        attribute: Union[str, int, type],
     ) -> MatterAttribute:
         """Return Matter Attribute for given parameters."""
         return next(
@@ -158,7 +158,9 @@ def get_attribute(
             and attribute in (x.attribute_id, x.attribute_name, x.attribute_type)
         )
 
-    def has_cluster(self, endpoint: int, cluster: type[Clusters.Cluster] | int) -> bool:
+    def has_cluster(
+        self, endpoint: int, cluster: Union[type[Clusters.Cluster], int]
+    ) -> bool:
         """Check if node has a specific cluster."""
         return any(
             x
@@ -169,7 +171,7 @@ def has_cluster(self, endpoint: int, cluster: type[Clusters.Cluster] | int) -> b
 
     def get_cluster(
         self, endpoint: int, cluster: type[_CLUSTER_T]
-    ) -> _CLUSTER_T | None:
+    ) -> Optional[_CLUSTER_T]:
         """
         Get a full Cluster object containing all attributes.
 

matter_server/common/models/node_device.py

@@ -72,8 +72,7 @@ def device_type_instances(self) -> list[MatterDeviceTypeInstance]:
         return [
             inst
             for inst in self.bridged_device_type_instance.node.device_type_instances
-            if inst.endpoint == endpoint
-            and inst != self.bridged_device_type_instance
+            if inst.endpoint == endpoint and inst != self.bridged_device_type_instance
         ]
 
     def __repr__(self) -> str:

matter_server/server/client_handler.py

@@ -101,7 +101,9 @@ async def handle_client(self) -> web.WebSocketResponse:
                 self._logger.debug("Received: %s", msg.data)
 
                 try:
-                    command_msg = dataclass_from_dict(CommandMessage, json_loads(msg.data))
+                    command_msg = dataclass_from_dict(
+                        CommandMessage, json_loads(msg.data)
+                    )
                 except ValueError:
                     disconnect_warn = f"Received invalid JSON: {msg.data}"
                     break