adds Deserializer, converting SerializedObject objects back to actual objects

src/pydase/utils/deserializer.py

@@ -0,0 +1,141 @@
+import enum
+import logging
+from typing import Any, NoReturn, cast
+
+import pydase
+import pydase.components
+import pydase.server.web_server.sio_setup
+import pydase.units as u
+from pydase.utils.helpers import get_component_classes
+from pydase.utils.serializer import SerializedObject
+
+logger = logging.getLogger(__name__)
+
+
+class Deserializer:
+    @classmethod
+    def deserialize(cls, serialized_object: SerializedObject) -> Any:
+        # Main entry point for deserializing objects
+        type_handler = {
+            None: None,
+            "int": cls.deserialize_primitive,
+            "float": cls.deserialize_primitive,
+            "bool": cls.deserialize_primitive,
+            "str": cls.deserialize_primitive,
+            "NoneType": cls.deserialize_primitive,
+            "Quantity": cls.deserialize_quantity,
+            "Enum": cls.deserialize_enum,
+            "ColouredEnum": lambda serialized_object: cls.deserialize_enum(
+                serialized_object, pydase.components.ColouredEnum
+            ),
+            "list": cls.deserialize_list,
+            "dict": cls.deserialize_dict,
+            "method": cls.deserialize_method,
+            "Exception": cls.deserialize_exception,
+        }
+
+        # Custom types like Components or DataService classes
+        component_class = cls.get_component_class(serialized_object["type"])
+        if component_class:
+            return cls.deserialize_component_type(serialized_object, component_class)
+
+        handler = type_handler.get(serialized_object["type"])
+        if handler:
+            return handler(serialized_object)
+        return None
+
+    @classmethod
+    def deserialize_primitive(cls, serialized_object: SerializedObject) -> Any:
+        return serialized_object["value"]
+
+    @classmethod
+    def deserialize_quantity(cls, serialized_object: SerializedObject) -> Any:
+        return u.convert_to_quantity(serialized_object["value"])  # type: ignore
+
+    @classmethod
+    def deserialize_enum(
+        cls,
+        serialized_object: SerializedObject,
+        enum_class: type[enum.Enum] = enum.Enum,
+    ) -> Any:
+        return enum_class(serialized_object["name"], serialized_object["enum"])[  # type: ignore
+            serialized_object["value"]
+        ]
+
+    @classmethod
+    def deserialize_list(cls, serialized_object: SerializedObject) -> Any:
+        return [
+            cls.deserialize(item)
+            for item in cast(list[SerializedObject], serialized_object["value"])
+        ]
+
+    @classmethod
+    def deserialize_dict(cls, serialized_object: SerializedObject) -> Any:
+        return {
+            key: cls.deserialize(value)
+            for key, value in cast(
+                dict[str, SerializedObject], serialized_object["value"]
+            ).items()
+        }
+
+    @classmethod
+    def deserialize_method(cls, serialized_object: SerializedObject) -> Any:
+        return
+
+    @classmethod
+    def deserialize_exception(cls, serialized_object: SerializedObject) -> NoReturn:
+        exception = type(serialized_object["name"], (Exception,), {})  # type: ignore
+        raise exception(serialized_object["value"])
+
+    @staticmethod
+    def get_component_class(type_name: str | None) -> type | None:
+        for component_class in get_component_classes():
+            if type_name == component_class.__name__:
+                return component_class
+        if type_name == "DataService":
+            import pydase
+
+            return pydase.DataService
+        return None
+
+    @classmethod
+    def create_attr_property(cls, serialized_attr: SerializedObject) -> property:
+        attr_name = serialized_attr["full_access_path"].split(".")[-1]
+
+        def get(self) -> Any:  # type: ignore
+            return getattr(self, f"_{attr_name}")
+
+        get.__doc__ = serialized_attr["doc"]
+
+        def set(self, value: Any) -> None:  # type: ignore
+            return setattr(self, f"_{attr_name}", value)
+
+        if serialized_attr["readonly"]:
+            return property(get)
+        return property(get, set)
+
+    @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 = {}
+
+            # Process and add properties based on the serialized object
+            for key, value in cast(
+                dict[str, SerializedObject], serialized_object["value"]
+            ).items():
+                if value["type"] != "method":
+                    class_attrs[key] = cls.create_attr_property(value)
+                    # Initialize a placeholder for the attribute to avoid AttributeError
+                    class_attrs[f"_{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)()
+
+
+def loads(serialized_object: SerializedObject) -> Any:
+    return Deserializer.deserialize(serialized_object)