replaces ClientDeserializer with ProxyClassFactory

src/pydase/client/client.py

@@ -3,7 +3,7 @@ import time
 
 import socketio  # type: ignore
 
-from pydase.client.client_deserializer import ClientDeserializer
+from pydase.client.proxy_class_factory import ProxyClassFactory
 from pydase.utils.serializer import SerializedObject
 
 logger = logging.getLogger(__name__)
@@ -13,6 +13,7 @@ class Client:
     def __init__(self, hostname: str, port: int):
         self.sio = socketio.Client()
         self.setup_events()
+        self.proxy_class_factory = ProxyClassFactory(self.sio)
         self.proxy = None
         self.sio.connect(
             f"ws://{hostname}:{port}",
@@ -26,8 +27,7 @@ class Client:
         # TODO: subscribe to update event and update the cache of the proxy class.
         @self.sio.event
         def class_structure(data: SerializedObject) -> None:
-            ClientDeserializer._sio = self.sio
-            self.proxy = ClientDeserializer.deserialize(data)
+            self.proxy = self.proxy_class_factory.create_proxy(data)
 
     def disconnect(self) -> None:
         self.sio.disconnect()

src/pydase/client/client_deserializer.py

@@ -1,89 +0,0 @@
-import enum
-import logging
-from typing import Any, cast
-
-import socketio  # type: ignore
-
-from pydase.utils.deserializer import Deserializer, loads
-from pydase.utils.serializer import SerializedObject, dump
-
-logger = logging.getLogger(__name__)
-
-
-class ClientDeserializer(Deserializer):
-    _sio: socketio.Client
-
-    @classmethod
-    def deserialize_method(cls, serialized_object: SerializedObject) -> Any:
-        def method_proxy(self: Any, *args: Any, **kwargs: Any) -> Any:
-            serialized_response = cast(
-                dict[str, Any],
-                cls._sio.call(
-                    "trigger_method",
-                    {
-                        "access_path": serialized_object["full_access_path"],
-                        "args": dump(list(args)),
-                        "kwargs": dump(kwargs),
-                    },
-                ),
-            )
-            return loads(serialized_response)  # type: ignore
-
-        return method_proxy
-
-    @classmethod
-    def deserialize_primitive(cls, serialized_object: SerializedObject) -> Any:
-        return cls.create_attr_property(serialized_object)
-
-    @classmethod
-    def deserialize_quantity(cls, serialized_object: SerializedObject) -> Any:
-        return cls.create_attr_property(serialized_object)
-
-    @classmethod
-    def deserialize_enum(
-        cls,
-        serialized_object: SerializedObject,
-        enum_class: type[enum.Enum] = enum.Enum,
-    ) -> Any:
-        return cls.create_attr_property(serialized_object)
-
-    @classmethod
-    def deserialize_component_type(
-        cls, serialized_object: SerializedObject, base_class: type
-    ) -> Any:
-        def create_proxy_class(serialized_object: SerializedObject) -> type:
-            class_bases = (base_class,)
-            class_attrs: dict[str, Any] = {"_sio": cls._sio}
-
-            # Process and add properties based on the serialized object
-            for key, value in cast(
-                dict[str, SerializedObject], serialized_object["value"]
-            ).items():
-                class_attrs[key] = cls.deserialize(value)
-
-            # Create the dynamic class with the given name and attributes
-            return type(serialized_object["name"], class_bases, class_attrs)  # type: ignore
-
-        return create_proxy_class(serialized_object)()
-
-    @classmethod
-    def create_attr_property(cls, serialized_attr: SerializedObject) -> property:
-        def get(self) -> Any:  # type: ignore
-            return loads(
-                self._sio.call("get_value", serialized_attr["full_access_path"])
-            )
-
-        get.__doc__ = serialized_attr["doc"]
-
-        def set(self, value: Any) -> None:  # type: ignore
-            self._sio.call(
-                "update_value",
-                {
-                    "access_path": serialized_attr["full_access_path"],
-                    "value": dump(value),
-                },
-            )
-
-        if serialized_attr["readonly"]:
-            return property(get)
-        return property(get, set)

src/pydase/client/proxy_class_factory.py

@@ -0,0 +1,119 @@
+import logging
+from typing import TYPE_CHECKING, Any, cast
+
+import socketio  # type: ignore
+
+import pydase
+from pydase.utils.deserializer import Deserializer, loads
+from pydase.utils.serializer import SerializedObject, dump
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    import pydase.components
+
+    class ProxyClass(pydase.DataService):
+        _sio: socketio.Client
+
+
+logger = logging.getLogger(__name__)
+
+
+class ProxyClassFactory:
+    def __init__(self, sio_client: socketio.Client) -> None:
+        self.sio_client = sio_client
+
+    def create_proxy(self, data: SerializedObject) -> "ProxyClass":
+        proxy = self._deserialize(data)
+        proxy._sio = self.sio_client
+        return proxy
+
+    def _deserialize(self, serialized_object: SerializedObject) -> Any:
+        type_handler: dict[str | None, None | Callable[..., Any]] = {
+            None: None,
+            "int": self._create_attr_property,
+            "float": self._create_attr_property,
+            "bool": self._create_attr_property,
+            "str": self._create_attr_property,
+            "NoneType": self._create_attr_property,
+            "Quantity": self._create_attr_property,
+            "Enum": self._create_attr_property,
+            "ColouredEnum": self._create_attr_property,
+            "method": self._deserialize_method,
+            "list": loads,
+            "dict": loads,
+            "Exception": loads,
+        }
+
+        # Custom types like Components or DataService classes
+        component_class = Deserializer.get_component_class(serialized_object["type"])
+        if component_class:
+            proxy_class = self._deserialize_component_type(
+                serialized_object, component_class
+            )
+            proxy_class._sio = self.sio_client
+            return proxy_class
+
+        handler = type_handler.get(serialized_object["type"])
+        if handler:
+            return handler(serialized_object)
+        return None
+
+    def _deserialize_method(self, serialized_object: SerializedObject) -> Any:
+        def method_proxy(self: "ProxyClass", *args: Any, **kwargs: Any) -> Any:
+            serialized_response = cast(
+                dict[str, Any],
+                self._sio.call(
+                    "trigger_method",
+                    {
+                        "access_path": serialized_object["full_access_path"],
+                        "args": dump(list(args)),
+                        "kwargs": dump(kwargs),
+                    },
+                ),
+            )
+            return loads(serialized_response)  # type: ignore
+
+        return method_proxy
+
+    def _deserialize_component_type(
+        self, serialized_object: SerializedObject, base_class: type
+    ) -> Any:
+        def create_proxy_class(serialized_object: SerializedObject) -> type:
+            class_bases = (base_class,)
+            class_attrs: dict[str, Any] = {}
+
+            # Process and add properties based on the serialized object
+            for key, value in cast(
+                dict[str, SerializedObject], serialized_object["value"]
+            ).items():
+                class_attrs[key] = self._deserialize(value)
+
+            # Create the dynamic class with the given name and attributes
+            return type(serialized_object["name"], class_bases, class_attrs)  # type: ignore
+
+        return create_proxy_class(serialized_object)()
+
+    def _create_attr_property(self, serialized_attr: SerializedObject) -> property:
+        def get(self: "ProxyClass") -> Any:  # type: ignore
+            return loads(
+                cast(
+                    SerializedObject,
+                    self._sio.call("get_value", serialized_attr["full_access_path"]),
+                )
+            )
+
+        get.__doc__ = serialized_attr["doc"]
+
+        def set(self: "ProxyClass", value: Any) -> None:  # type: ignore
+            self._sio.call(
+                "update_value",
+                {
+                    "access_path": serialized_attr["full_access_path"],
+                    "value": dump(value),
+                },
+            )
+
+        if serialized_attr["readonly"]:
+            return property(get)
+        return property(get, set)