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] 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)