Merge pull request #56 from tiqi-group/cleanup/refactoring_serialization

Refactors DataService serialization

src/pydase/data_service/data_service.py

@@ -1,5 +1,3 @@
-import asyncio
-import inspect
 import json
 import logging
 import os
@@ -16,13 +14,13 @@ from pydase.utils.helpers import (
     convert_arguments_to_hinted_types,
     generate_paths_from_DataService_dict,
     get_class_and_instance_attributes,
-    get_component_class_names,
     get_nested_value_from_DataService_by_path_and_key,
     get_object_attr_from_path,
     is_property_attribute,
     parse_list_attr_and_index,
     update_value_if_changed,
 )
+from pydase.utils.serialization import Serializer
 from pydase.utils.warnings import (
     warn_if_instance_class_does_not_inherit_from_DataService,
 )
@@ -213,140 +211,7 @@ class DataService(rpyc.Service, AbstractDataService):
         Returns:
             dict: The serialized instance.
         """
-        result: dict[str, dict[str, Any]] = {}
-
-        # Get the dictionary of the base class
-        base_set = set(type(super()).__dict__)
-        # Get the dictionary of the derived class
-        derived_set = set(type(self).__dict__)
-        # Get the difference between the two dictionaries
-        derived_only_set = derived_set - base_set
-
-        instance_dict = set(self.__dict__)
-        # Merge the class and instance dictionaries
-        merged_set = derived_only_set | instance_dict
-
-        def get_attribute_doc(attr: Any) -> Optional[str]:
-            """This function takes an input attribute attr and returns its documentation
-            string if it's different from the documentation of its type, otherwise,
-            it returns None.
-            """
-            attr_doc = inspect.getdoc(attr)
-            attr_class_doc = inspect.getdoc(type(attr))
-            if attr_class_doc != attr_doc:
-                return attr_doc
-            else:
-                return None
-
-        # Iterate over attributes, properties, class attributes, and methods
-        for key in sorted(merged_set):
-            if key.startswith("_"):
-                continue  # Skip attributes that start with underscore
-
-            # Skip keys that start with "start_" or "stop_" and end with an async method
-            # name
-            if (key.startswith("start_") or key.startswith("stop_")) and key.split(
-                "_", 1
-            )[1] in {
-                name
-                for name, _ in inspect.getmembers(
-                    self, predicate=inspect.iscoroutinefunction
-                )
-            }:
-                continue
-
-            # Get the value of the current attribute or method
-            value = getattr(self, key)
-
-            if isinstance(value, DataService):
-                result[key] = {
-                    "type": type(value).__name__
-                    if type(value).__name__ in get_component_class_names()
-                    else "DataService",
-                    "value": value.serialize(),
-                    "readonly": False,
-                    "doc": get_attribute_doc(value),
-                }
-            elif isinstance(value, list):
-                result[key] = {
-                    "type": "list",
-                    "value": [
-                        {
-                            "type": type(item).__name__
-                            if not isinstance(item, DataService)
-                            or type(item).__name__ in get_component_class_names()
-                            else "DataService",
-                            "value": item.serialize()
-                            if isinstance(item, DataService)
-                            else item,
-                            "readonly": False,
-                            "doc": get_attribute_doc(value),
-                        }
-                        for item in value
-                    ],
-                    "readonly": False,
-                }
-            elif inspect.isfunction(value) or inspect.ismethod(value):
-                sig = inspect.signature(value)
-
-                # Store parameters and their anotations in a dictionary
-                parameters: dict[str, Optional[str]] = {}
-                for k, v in sig.parameters.items():
-                    annotation = v.annotation
-                    if annotation is not inspect._empty:
-                        if isinstance(annotation, type):
-                            # Handle regular types
-                            parameters[k] = annotation.__name__
-                        else:
-                            parameters[k] = str(annotation)
-                    else:
-                        parameters[k] = None
-                running_task_info = None
-                if (
-                    key in self._task_manager.tasks
-                ):  # If there's a running task for this method
-                    task_info = self._task_manager.tasks[key]
-                    running_task_info = task_info["kwargs"]
-
-                result[key] = {
-                    "type": "method",
-                    "async": asyncio.iscoroutinefunction(value),
-                    "parameters": parameters,
-                    "doc": get_attribute_doc(value),
-                    "readonly": True,
-                    "value": running_task_info,
-                }
-            elif isinstance(value, Enum):
-                if type(value).__base__.__name__ == "ColouredEnum":
-                    val_type = "ColouredEnum"
-                else:
-                    val_type = "Enum"
-                result[key] = {
-                    "type": val_type,
-                    "value": value.name,
-                    "enum": {
-                        name: member.value
-                        for name, member in value.__class__.__members__.items()
-                    },
-                    "readonly": False,
-                    "doc": get_attribute_doc(value),
-                }
-            else:
-                result[key] = {
-                    "type": type(value).__name__,
-                    "value": value
-                    if not isinstance(value, u.Quantity)
-                    else {"magnitude": value.m, "unit": str(value.u)},
-                    "readonly": False,
-                    "doc": get_attribute_doc(value),
-                }
-
-            if isinstance(getattr(self.__class__, key, None), property):
-                prop: property = getattr(self.__class__, key)
-                result[key]["readonly"] = prop.fset is None
-                result[key]["doc"] = get_attribute_doc(prop)
-
-        return result
+        return Serializer.serialize_object(self)["value"]
 
     def update_DataService_attribute(
         self,

src/pydase/utils/serialization.py

@@ -0,0 +1,217 @@
+import inspect
+import logging
+from collections.abc import Callable
+from enum import Enum
+from typing import Any, Optional
+
+import pydase.units as u
+from pydase.data_service.abstract_data_service import AbstractDataService
+from pydase.utils.helpers import get_component_class_names
+
+logger = logging.getLogger(__name__)
+
+
+class Serializer:
+    @staticmethod
+    def get_attribute_doc(attr: Any) -> Optional[str]:
+        """This function takes an input attribute attr and returns its documentation
+        string if it's different from the documentation of its type, otherwise,
+        it returns None.
+        """
+        attr_doc = inspect.getdoc(attr)
+        attr_class_doc = inspect.getdoc(type(attr))
+        return attr_doc if attr_class_doc != attr_doc else None
+
+    @staticmethod
+    def serialize_object(obj: Any) -> dict[str, Any]:
+        result: dict[str, Any] = {}
+        if isinstance(obj, AbstractDataService):
+            result = Serializer._serialize_DataService(obj)
+
+        elif isinstance(obj, list):
+            result = Serializer._serialize_list(obj)
+
+        elif isinstance(obj, dict):
+            result = Serializer._serialize_dict(obj)
+
+        # Special handling for u.Quantity
+        elif isinstance(obj, u.Quantity):
+            result = Serializer._serialize_Quantity(obj)
+
+        # Handling for Enums
+        elif isinstance(obj, Enum):
+            result = Serializer._serialize_enum(obj)
+
+        # Methods and coroutines
+        elif inspect.isfunction(obj) or inspect.ismethod(obj):
+            result = Serializer._serialize_method(obj)
+
+        else:
+            obj_type = type(obj).__name__
+            value = obj
+            readonly = False
+            doc = Serializer.get_attribute_doc(obj)
+            result = {
+                "type": obj_type,
+                "value": value,
+                "readonly": readonly,
+                "doc": doc,
+            }
+
+        return result
+
+    @staticmethod
+    def _serialize_enum(obj: Enum) -> dict[str, Any]:
+        value = obj.name
+        readonly = False
+        doc = Serializer.get_attribute_doc(obj)
+        if type(obj).__base__.__name__ == "ColouredEnum":
+            obj_type = "ColouredEnum"
+        else:
+            obj_type = "Enum"
+
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+            "enum": {
+                name: member.value for name, member in obj.__class__.__members__.items()
+            },
+        }
+
+    @staticmethod
+    def _serialize_Quantity(obj: u.Quantity) -> dict[str, Any]:
+        obj_type = "Quantity"
+        readonly = False
+        doc = Serializer.get_attribute_doc(obj)
+        value = {"magnitude": obj.m, "unit": str(obj.u)}
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+        }
+
+    @staticmethod
+    def _serialize_dict(obj: dict[str, Any]) -> dict[str, Any]:
+        obj_type = "dict"
+        readonly = False
+        doc = Serializer.get_attribute_doc(obj)
+        value = {key: Serializer.serialize_object(val) for key, val in obj.items()}
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+        }
+
+    @staticmethod
+    def _serialize_list(obj: list[Any]) -> dict[str, Any]:
+        obj_type = "list"
+        readonly = False
+        doc = Serializer.get_attribute_doc(obj)
+        value = [Serializer.serialize_object(o) for o in obj]
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+        }
+
+    @staticmethod
+    def _serialize_method(obj: Callable[..., Any]) -> dict[str, Any]:
+        obj_type = "method"
+        value = None
+        readonly = True
+        doc = Serializer.get_attribute_doc(obj)
+
+        # Store parameters and their anotations in a dictionary
+        sig = inspect.signature(obj)
+        parameters: dict[str, Optional[str]] = {}
+
+        for k, v in sig.parameters.items():
+            annotation = v.annotation
+            if annotation is not inspect._empty:
+                if isinstance(annotation, type):
+                    # Handle regular types
+                    parameters[k] = annotation.__name__
+                else:
+                    # Union, string annotation, Literal types, ...
+                    parameters[k] = str(annotation)
+            else:
+                parameters[k] = None
+
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+            "async": inspect.iscoroutinefunction(obj),
+            "parameters": parameters,
+        }
+
+    @staticmethod
+    def _serialize_DataService(obj: AbstractDataService) -> dict[str, Any]:
+        readonly = False
+        doc = Serializer.get_attribute_doc(obj)
+        obj_type = type(obj).__name__
+        if type(obj).__name__ not in get_component_class_names():
+            obj_type = "DataService"
+
+        # Get the dictionary of the base class
+        base_set = set(type(obj).__base__.__dict__)
+        # Get the dictionary of the derived class
+        derived_set = set(type(obj).__dict__)
+        # Get the difference between the two dictionaries
+        derived_only_set = derived_set - base_set
+
+        instance_dict = set(obj.__dict__)
+        # Merge the class and instance dictionaries
+        merged_set = derived_only_set | instance_dict
+        value = {}
+
+        # Iterate over attributes, properties, class attributes, and methods
+        for key in sorted(merged_set):
+            if key.startswith("_"):
+                continue  # Skip attributes that start with underscore
+
+            # Skip keys that start with "start_" or "stop_" and end with an async
+            # method name
+            if (key.startswith("start_") or key.startswith("stop_")) and key.split(
+                "_", 1
+            )[1] in {
+                name
+                for name, _ in inspect.getmembers(
+                    obj, predicate=inspect.iscoroutinefunction
+                )
+            }:
+                continue
+
+            val = getattr(obj, key)
+
+            value[key] = Serializer.serialize_object(val)
+
+            # If there's a running task for this method
+            if key in obj._task_manager.tasks:
+                task_info = obj._task_manager.tasks[key]
+                value[key]["value"] = task_info["kwargs"]
+
+            # If the DataService attribute is a property
+            if isinstance(getattr(obj.__class__, key, None), property):
+                prop: property = getattr(obj.__class__, key)
+                value[key]["readonly"] = prop.fset is None
+                value[key]["doc"] = Serializer.get_attribute_doc(
+                    prop
+                )  # overwrite the doc
+
+        return {
+            "type": obj_type,
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+        }
+
+
+def dump(obj: Any) -> dict[str, Any]:
+    return Serializer.serialize_object(obj)

tests/data_service/test_data_service.py

@@ -1,64 +0,0 @@
-from enum import Enum
-
-import pydase
-
-
-def test_enum_serialize() -> None:
-    class EnumClass(Enum):
-        FOO = "foo"
-        BAR = "bar"
-
-    class EnumAttribute(pydase.DataService):
-        def __init__(self) -> None:
-            self.some_enum = EnumClass.FOO
-            super().__init__()
-
-    class EnumPropertyWithoutSetter(pydase.DataService):
-        def __init__(self) -> None:
-            self._some_enum = EnumClass.FOO
-            super().__init__()
-
-        @property
-        def some_enum(self) -> EnumClass:
-            return self._some_enum
-
-    class EnumPropertyWithSetter(pydase.DataService):
-        def __init__(self) -> None:
-            self._some_enum = EnumClass.FOO
-            super().__init__()
-
-        @property
-        def some_enum(self) -> EnumClass:
-            return self._some_enum
-
-        @some_enum.setter
-        def some_enum(self, value: EnumClass) -> None:
-            self._some_enum = value
-
-    assert EnumAttribute().serialize() == {
-        "some_enum": {
-            "type": "Enum",
-            "value": "FOO",
-            "enum": {"FOO": "foo", "BAR": "bar"},
-            "readonly": False,
-            "doc": None,
-        }
-    }
-    assert EnumPropertyWithoutSetter().serialize() == {
-        "some_enum": {
-            "type": "Enum",
-            "value": "FOO",
-            "enum": {"FOO": "foo", "BAR": "bar"},
-            "readonly": True,
-            "doc": None,
-        }
-    }
-    assert EnumPropertyWithSetter().serialize() == {
-        "some_enum": {
-            "type": "Enum",
-            "value": "FOO",
-            "enum": {"FOO": "foo", "BAR": "bar"},
-            "readonly": False,
-            "doc": None,
-        }
-    }

tests/utils/test_serialization.py

@@ -0,0 +1,288 @@
+import asyncio
+from enum import Enum
+
+import pytest
+
+import pydase
+import pydase.units as u
+from pydase.components.coloured_enum import ColouredEnum
+from pydase.utils.serialization import dump
+
+
+@pytest.mark.parametrize(
+    "test_input, expected",
+    [
+        (1, {"type": "int", "value": 1, "readonly": False, "doc": None}),
+        (1.0, {"type": "float", "value": 1.0, "readonly": False, "doc": None}),
+        (True, {"type": "bool", "value": True, "readonly": False, "doc": None}),
+        (
+            u.Quantity(10, "m"),
+            {
+                "type": "Quantity",
+                "value": {"magnitude": 10, "unit": "meter"},
+                "readonly": False,
+                "doc": None,
+            },
+        ),
+    ],
+)
+def test_dump(test_input, expected):
+    assert dump(test_input) == expected
+
+
+def test_enum_serialize() -> None:
+    class EnumClass(Enum):
+        FOO = "foo"
+        BAR = "bar"
+
+    class EnumAttribute(pydase.DataService):
+        def __init__(self) -> None:
+            self.some_enum = EnumClass.FOO
+            super().__init__()
+
+    class EnumPropertyWithoutSetter(pydase.DataService):
+        def __init__(self) -> None:
+            self._some_enum = EnumClass.FOO
+            super().__init__()
+
+        @property
+        def some_enum(self) -> EnumClass:
+            return self._some_enum
+
+    class EnumPropertyWithSetter(pydase.DataService):
+        def __init__(self) -> None:
+            self._some_enum = EnumClass.FOO
+            super().__init__()
+
+        @property
+        def some_enum(self) -> EnumClass:
+            return self._some_enum
+
+        @some_enum.setter
+        def some_enum(self, value: EnumClass) -> None:
+            self._some_enum = value
+
+    assert dump(EnumAttribute())["value"] == {
+        "some_enum": {
+            "type": "Enum",
+            "value": "FOO",
+            "enum": {"FOO": "foo", "BAR": "bar"},
+            "readonly": False,
+            "doc": None,
+        }
+    }
+    assert dump(EnumPropertyWithoutSetter())["value"] == {
+        "some_enum": {
+            "type": "Enum",
+            "value": "FOO",
+            "enum": {"FOO": "foo", "BAR": "bar"},
+            "readonly": True,
+            "doc": None,
+        }
+    }
+    assert dump(EnumPropertyWithSetter())["value"] == {
+        "some_enum": {
+            "type": "Enum",
+            "value": "FOO",
+            "enum": {"FOO": "foo", "BAR": "bar"},
+            "readonly": False,
+            "doc": None,
+        }
+    }
+
+
+def test_ColouredEnum_serialize() -> None:
+    class Status(ColouredEnum):
+        PENDING = "#FFA500"
+        RUNNING = "#0000FF80"
+        PAUSED = "rgb(169, 169, 169)"
+        RETRYING = "rgba(255, 255, 0, 0.3)"
+        COMPLETED = "hsl(120, 100%, 50%)"
+        FAILED = "hsla(0, 100%, 50%, 0.7)"
+        CANCELLED = "SlateGray"
+
+    assert dump(Status.FAILED) == {
+        "type": "ColouredEnum",
+        "value": "FAILED",
+        "enum": {
+            "CANCELLED": "SlateGray",
+            "COMPLETED": "hsl(120, 100%, 50%)",
+            "FAILED": "hsla(0, 100%, 50%, 0.7)",
+            "PAUSED": "rgb(169, 169, 169)",
+            "PENDING": "#FFA500",
+            "RETRYING": "rgba(255, 255, 0, 0.3)",
+            "RUNNING": "#0000FF80",
+        },
+        "readonly": False,
+        "doc": None,
+    }
+
+
+def test_method_serialization() -> None:
+    class ClassWithMethod(pydase.DataService):
+        def some_method(self) -> str:
+            return "some method"
+
+        async def some_task(self, sleep_time: int) -> None:
+            while True:
+                await asyncio.sleep(sleep_time)
+
+    instance = ClassWithMethod()
+    instance.start_some_task(10)  # type: ignore
+
+    assert dump(instance)["value"] == {
+        "some_method": {
+            "async": False,
+            "doc": None,
+            "parameters": {},
+            "readonly": True,
+            "type": "method",
+            "value": None,
+        },
+        "some_task": {
+            "async": True,
+            "doc": None,
+            "parameters": {"sleep_time": "int"},
+            "readonly": True,
+            "type": "method",
+            "value": {"sleep_time": 10},
+        },
+    }
+
+
+def test_methods_with_type_hints() -> None:
+    def method_without_type_hint(arg_without_type_hint) -> None:
+        pass
+
+    def method_with_type_hint(some_argument: int) -> None:
+        pass
+
+    def method_with_union_type_hint(some_argument: int | float) -> None:
+        pass
+
+    assert dump(method_without_type_hint) == {
+        "async": False,
+        "doc": None,
+        "parameters": {"arg_without_type_hint": None},
+        "readonly": True,
+        "type": "method",
+        "value": None,
+    }
+
+    assert dump(method_with_type_hint) == {
+        "async": False,
+        "doc": None,
+        "parameters": {"some_argument": "int"},
+        "readonly": True,
+        "type": "method",
+        "value": None,
+    }
+
+    assert dump(method_with_union_type_hint) == {
+        "async": False,
+        "doc": None,
+        "parameters": {"some_argument": "int | float"},
+        "readonly": True,
+        "type": "method",
+        "value": None,
+    }
+
+
+def test_list_serialization() -> None:
+    class MySubclass(pydase.DataService):
+        _name = "hi"
+        bool_attr = True
+        int_attr = 1
+
+        @property
+        def name(self) -> str:
+            return self._name
+
+    class ClassWithListAttribute(pydase.DataService):
+        list_attr = [1, MySubclass()]
+
+    instance = ClassWithListAttribute()
+
+    assert dump(instance)["value"] == {
+        "list_attr": {
+            "doc": None,
+            "readonly": False,
+            "type": "list",
+            "value": [
+                {"doc": None, "readonly": False, "type": "int", "value": 1},
+                {
+                    "doc": None,
+                    "readonly": False,
+                    "type": "DataService",
+                    "value": {
+                        "bool_attr": {
+                            "doc": None,
+                            "readonly": False,
+                            "type": "bool",
+                            "value": True,
+                        },
+                        "int_attr": {
+                            "doc": None,
+                            "readonly": False,
+                            "type": "int",
+                            "value": 1,
+                        },
+                        "name": {
+                            "doc": None,
+                            "readonly": True,
+                            "type": "str",
+                            "value": "hi",
+                        },
+                    },
+                },
+            ],
+        }
+    }
+
+
+def test_dict_serialization() -> None:
+    class MyClass(pydase.DataService):
+        name = "my class"
+
+    test_dict = {
+        "int_key": 1,
+        "float_key": 1.0,
+        "bool_key": True,
+        "Quantity_key": 1.0 * u.units.s,
+        "DataService_key": MyClass(),
+    }
+
+    assert dump(test_dict) == {
+        "doc": None,
+        "readonly": False,
+        "type": "dict",
+        "value": {
+            "DataService_key": {
+                "doc": None,
+                "readonly": False,
+                "type": "DataService",
+                "value": {
+                    "name": {
+                        "doc": None,
+                        "readonly": False,
+                        "type": "str",
+                        "value": "my class",
+                    }
+                },
+            },
+            "Quantity_key": {
+                "doc": None,
+                "readonly": False,
+                "type": "Quantity",
+                "value": {"magnitude": 1.0, "unit": "s"},
+            },
+            "bool_key": {"doc": None, "readonly": False, "type": "bool", "value": True},
+            "float_key": {
+                "doc": None,
+                "readonly": False,
+                "type": "float",
+                "value": 1.0,
+            },
+            "int_key": {"doc": None, "readonly": False, "type": "int", "value": 1},
+        },
+    }