From 2eb0eb84cf4c5b290c2258ca4c0cff8f3f563d41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mose=20M=C3=BCller?= Date: Thu, 2 Nov 2023 14:10:33 +0100 Subject: [PATCH 1/3] moves serialization into separate class in the utils module --- src/pydase/data_service/data_service.py | 139 +--------------------- src/pydase/utils/serialization.py | 149 ++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 137 deletions(-) create mode 100644 src/pydase/utils/serialization.py diff --git a/src/pydase/data_service/data_service.py b/src/pydase/data_service/data_service.py index 97c4678..5014bd2 100644 --- a/src/pydase/data_service/data_service.py +++ b/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, diff --git a/src/pydase/utils/serialization.py b/src/pydase/utils/serialization.py new file mode 100644 index 0000000..6f81b6c --- /dev/null +++ b/src/pydase/utils/serialization.py @@ -0,0 +1,149 @@ +import inspect +import logging +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)) + if attr_class_doc != attr_doc: + return attr_doc + else: + return None + + @staticmethod + def serialize_object(obj: Any): + obj_type = type(obj).__name__ + value = obj + readonly = False # You need to determine how to set this value + doc = Serializer.get_attribute_doc(obj) + kwargs: dict[str, Any] = {} + + if isinstance(obj, AbstractDataService): + # 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 = {} + + if type(value).__name__ not in get_component_class_names(): + obj_type = "DataService" + + # 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 + + elif isinstance(value, list): + obj_type = "list" + value = [Serializer.serialize_object(o) for o in value] + + elif isinstance(value, dict): + obj_type = "dict" + value = {key: Serializer.serialize_object(val) for key, val in obj.items()} + + # Special handling for u.Quantity + elif isinstance(obj, u.Quantity): + value = {"magnitude": obj.m, "unit": str(obj.u)} + + # Handling for Enums + elif isinstance(obj, Enum): + value = obj.name + if type(obj).__base__.__name__ == "ColouredEnum": + obj_type = "ColouredEnum" + else: + obj_type = "Enum" + kwargs = { + "enum": { + name: member.value + for name, member in obj.__class__.__members__.items() + }, + } + + # Methods and coroutines + elif inspect.isfunction(obj) or inspect.ismethod(obj): + 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: + # Union, string annotation, Literal types, ... + parameters[k] = str(annotation) + else: + parameters[k] = None + value = None + obj_type = "method" + readonly = True + kwargs = { + "async": inspect.iscoroutinefunction(obj), + "parameters": parameters, + } + + # Construct the result dictionary + result = { + "type": obj_type, + "value": value, + "readonly": readonly, + "doc": doc, + **kwargs, + } + + return result + + +def dump(obj: Any) -> dict[str, Any]: + return Serializer.serialize_object(obj) From 2b57df5aac2cde3c1af07546c1d8fec00cbceb8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mose=20M=C3=BCller?= Date: Thu, 2 Nov 2023 14:11:08 +0100 Subject: [PATCH 2/3] adds tests for serialization (and moves tests from test_data_service) --- tests/data_service/test_data_service.py | 64 ------ tests/utils/test_serialization.py | 288 ++++++++++++++++++++++++ 2 files changed, 288 insertions(+), 64 deletions(-) create mode 100644 tests/utils/test_serialization.py diff --git a/tests/data_service/test_data_service.py b/tests/data_service/test_data_service.py index 8d08a3c..e69de29 100644 --- a/tests/data_service/test_data_service.py +++ b/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, - } - } diff --git a/tests/utils/test_serialization.py b/tests/utils/test_serialization.py new file mode 100644 index 0000000..465128e --- /dev/null +++ b/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}, + }, + } From 6804cdf3b1afe1fd1ba3923b846b04426f9fb6c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mose=20M=C3=BCller?= Date: Thu, 2 Nov 2023 14:33:16 +0100 Subject: [PATCH 3/3] refactoring Serializer class --- src/pydase/utils/serialization.py | 270 +++++++++++++++++++----------- 1 file changed, 169 insertions(+), 101 deletions(-) diff --git a/src/pydase/utils/serialization.py b/src/pydase/utils/serialization.py index 6f81b6c..674d530 100644 --- a/src/pydase/utils/serialization.py +++ b/src/pydase/utils/serialization.py @@ -1,5 +1,6 @@ import inspect import logging +from collections.abc import Callable from enum import Enum from typing import Any, Optional @@ -19,130 +20,197 @@ class Serializer: """ attr_doc = inspect.getdoc(attr) attr_class_doc = inspect.getdoc(type(attr)) - if attr_class_doc != attr_doc: - return attr_doc - else: - return None + return attr_doc if attr_class_doc != attr_doc else None @staticmethod - def serialize_object(obj: Any): - obj_type = type(obj).__name__ - value = obj - readonly = False # You need to determine how to set this value - doc = Serializer.get_attribute_doc(obj) - kwargs: dict[str, Any] = {} - + def serialize_object(obj: Any) -> dict[str, Any]: + result: dict[str, Any] = {} if isinstance(obj, AbstractDataService): - # 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 + result = Serializer._serialize_DataService(obj) - instance_dict = set(obj.__dict__) - # Merge the class and instance dictionaries - merged_set = derived_only_set | instance_dict - value = {} + elif isinstance(obj, list): + result = Serializer._serialize_list(obj) - if type(value).__name__ not in get_component_class_names(): - obj_type = "DataService" - - # 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 - - elif isinstance(value, list): - obj_type = "list" - value = [Serializer.serialize_object(o) for o in value] - - elif isinstance(value, dict): - obj_type = "dict" - value = {key: Serializer.serialize_object(val) for key, val in obj.items()} + elif isinstance(obj, dict): + result = Serializer._serialize_dict(obj) # Special handling for u.Quantity elif isinstance(obj, u.Quantity): - value = {"magnitude": obj.m, "unit": str(obj.u)} + result = Serializer._serialize_Quantity(obj) # Handling for Enums elif isinstance(obj, Enum): - value = obj.name - if type(obj).__base__.__name__ == "ColouredEnum": - obj_type = "ColouredEnum" - else: - obj_type = "Enum" - kwargs = { - "enum": { - name: member.value - for name, member in obj.__class__.__members__.items() - }, - } + result = Serializer._serialize_enum(obj) # Methods and coroutines elif inspect.isfunction(obj) or inspect.ismethod(obj): - sig = inspect.signature(value) + result = Serializer._serialize_method(obj) - # 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: - # Union, string annotation, Literal types, ... - parameters[k] = str(annotation) - else: - parameters[k] = None - value = None - obj_type = "method" - readonly = True - kwargs = { - "async": inspect.iscoroutinefunction(obj), - "parameters": parameters, + 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, } - # Construct the result dictionary - result = { + 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, - **kwargs, + "enum": { + name: member.value for name, member in obj.__class__.__members__.items() + }, } - return result + @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]: