diff --git a/src/pydase/utils/deserializer.py b/src/pydase/utils/deserializer.py new file mode 100644 index 0000000..76bbef5 --- /dev/null +++ b/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)