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