mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-22 09:10:01 +02:00
refactoring Serializer class
This commit is contained in:
parent
2b57df5aac
commit
6804cdf3b1
@ -1,5 +1,6 @@
|
||||
import inspect
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from enum import Enum
|
||||
from typing import Any, Optional
|
||||
|
||||
@ -19,20 +20,145 @@ 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):
|
||||
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 # You need to determine how to set this value
|
||||
readonly = False
|
||||
doc = Serializer.get_attribute_doc(obj)
|
||||
kwargs: dict[str, Any] = {}
|
||||
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"
|
||||
|
||||
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
|
||||
@ -45,9 +171,6 @@ class Serializer:
|
||||
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("_"):
|
||||
@ -82,68 +205,13 @@ class Serializer:
|
||||
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 = {
|
||||
return {
|
||||
"type": obj_type,
|
||||
"value": value,
|
||||
"readonly": readonly,
|
||||
"doc": doc,
|
||||
**kwargs,
|
||||
}
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def dump(obj: Any) -> dict[str, Any]:
|
||||
return Serializer.serialize_object(obj)
|
||||
|
Loading…
x
Reference in New Issue
Block a user