ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-20 00:10:03 +02:00
Code Issues Packages Projects Releases Wiki Activity

Serializer adds full_access_path to serialized object representation

Browse Source
This commit is contained in:
Mose Müller 2024-03-26 10:52:06 +01:00
parent d9ea33abb6
commit 57e7deb552
4 changed files with 137 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/pydase/data_service/data_service_cache.py
View File

@ -45,6 +45,7 @@ class DataServiceCache:
)
except (SerializationPathError, SerializationValueError, KeyError):
return {
"full_access_path": full_access_path,
"value": None,
"type": None,
"doc": None,

65
src/pydase/utils/serializer.py
View File

@ -44,6 +44,7 @@ class SignatureDict(TypedDict):
SerializedObject = TypedDict(
"SerializedObject",
{
"full_access_path": str,
"name": NotRequired[str],
"value": "list[SerializedObject] | float | int | str | bool | dict[str, Any] | None", # noqa: E501
"type": str | None,
@ -59,28 +60,28 @@ SerializedObject = TypedDict(
class Serializer:
@staticmethod
def serialize_object(obj: Any) -> SerializedObject:
def serialize_object(obj: Any, access_path: str = "") -> SerializedObject:
result: SerializedObject
if isinstance(obj, AbstractDataService):
result = Serializer._serialize_data_service(obj)
result = Serializer._serialize_data_service(obj, access_path=access_path)
elif isinstance(obj, list):
result = Serializer._serialize_list(obj)
result = Serializer._serialize_list(obj, access_path=access_path)
elif isinstance(obj, dict):
result = Serializer._serialize_dict(obj)
result = Serializer._serialize_dict(obj, access_path=access_path)
# Special handling for u.Quantity
elif isinstance(obj, u.Quantity):
result = Serializer._serialize_quantity(obj)
result = Serializer._serialize_quantity(obj, access_path=access_path)
# Handling for Enums
elif isinstance(obj, Enum):
result = Serializer._serialize_enum(obj)
result = Serializer._serialize_enum(obj, access_path=access_path)
# Methods and coroutines
elif inspect.isfunction(obj) or inspect.ismethod(obj):
result = Serializer._serialize_method(obj)
result = Serializer._serialize_method(obj, access_path=access_path)
else:
obj_type = type(obj).__name__
@ -88,6 +89,7 @@ class Serializer:
readonly = False
doc = get_attribute_doc(obj)
result = {
"full_access_path": access_path,
"type": obj_type,
"value": value,
"readonly": readonly,
@ -97,7 +99,7 @@ class Serializer:
return result
@staticmethod
def _serialize_enum(obj: Enum) -> SerializedObject:
def _serialize_enum(obj: Enum, access_path: str = "") -> SerializedObject:
import pydase.components.coloured_enum
value = obj.name
@ -112,6 +114,7 @@ class Serializer:
obj_type = "Enum"
return {
"full_access_path": access_path,
"name": class_name,
"type": obj_type,
"value": value,
@ -123,12 +126,13 @@ class Serializer:
}
@staticmethod
def _serialize_quantity(obj: u.Quantity) -> SerializedObject:
def _serialize_quantity(obj: u.Quantity, access_path: str = "") -> SerializedObject:
obj_type = "Quantity"
readonly = False
doc = get_attribute_doc(obj)
value = {"magnitude": obj.m, "unit": str(obj.u)}
return {
"full_access_path": access_path,
"type": obj_type,
"value": value,
"readonly": readonly,
@ -136,12 +140,16 @@ class Serializer:
}
@staticmethod
def _serialize_dict(obj: dict[str, Any]) -> SerializedObject:
def _serialize_dict(obj: dict[str, Any], access_path: str = "") -> SerializedObject:
obj_type = "dict"
readonly = False
doc = get_attribute_doc(obj)
value = {key: Serializer.serialize_object(val) for key, val in obj.items()}
value = {
key: Serializer.serialize_object(val, access_path=f'{access_path}["{key}"]')
for key, val in obj.items()
}
return {
"full_access_path": access_path,
"type": obj_type,
"value": value,
"readonly": readonly,
@ -149,12 +157,16 @@ class Serializer:
}
@staticmethod
def _serialize_list(obj: list[Any]) -> SerializedObject:
def _serialize_list(obj: list[Any], access_path: str = "") -> SerializedObject:
obj_type = "list"
readonly = False
doc = get_attribute_doc(obj)
value = [Serializer.serialize_object(o) for o in obj]
value = [
Serializer.serialize_object(o, access_path=f"{access_path}[{i}]")
for i, o in enumerate(obj)
]
return {
"full_access_path": access_path,
"type": obj_type,
"value": value,
"readonly": readonly,
@ -162,7 +174,9 @@ class Serializer:
}
@staticmethod
def _serialize_method(obj: Callable[..., Any]) -> SerializedObject:
def _serialize_method(
obj: Callable[..., Any], access_path: str = ""
) -> SerializedObject:
obj_type = "method"
value = None
readonly = True
@ -176,12 +190,17 @@ class Serializer:
signature: SignatureDict = {"parameters": {}, "return_annotation": {}}
for k, v in sig.parameters.items():
default_value = cast(
dict[str, Any], {} if v.default == inspect._empty else dump(v.default)
)
default_value.pop("full_access_path", None)
signature["parameters"][k] = {
"annotation": str(v.annotation),
"default": {} if v.default == inspect._empty else dump(v.default),
"default": default_value,
}
return {
"full_access_path": access_path,
"type": obj_type,
"value": value,
"readonly": readonly,
@ -192,7 +211,9 @@ class Serializer:
}
@staticmethod
def _serialize_data_service(obj: AbstractDataService) -> SerializedObject:
def _serialize_data_service(
obj: AbstractDataService, access_path: str = ""
) -> SerializedObject:
readonly = False
doc = get_attribute_doc(obj)
obj_type = "DataService"
@ -231,7 +252,8 @@ class Serializer:
val = getattr(obj, key)
value[key] = Serializer.serialize_object(val)
path = f"{access_path}.{key}" if access_path else key
value[key] = Serializer.serialize_object(val, access_path=path)
# If there's a running task for this method
if key in obj._task_manager.tasks:
@ -244,6 +266,7 @@ class Serializer:
value[key]["doc"] = get_attribute_doc(prop) # overwrite the doc
return {
"full_access_path": access_path,
"name": obj_name,
"type": obj_type,
"value": value,
@ -303,11 +326,10 @@ def set_nested_value_by_path(
)
else:
serialized_value = dump(value)
keys_to_keep = set(serialized_value.keys())
serialized_value["full_access_path"] = path
serialized_value["readonly"] = next_level_serialized_object["readonly"]
# TODO: you might also want to pop "doc" from serialized_value if
# it is overwriting the value of the current dict
serialized_value.pop("readonly") # type: ignore
keys_to_keep = set(serialized_value.keys())
next_level_serialized_object.update(serialized_value)
@ -379,6 +401,7 @@ def get_next_level_dict_by_key(
# Appending to list
cast(list[SerializedObject], serialization_dict[attr_name]["value"]).append(
{
"full_access_path": "",
"value": None,
"type": None,
"doc": None,

8
tests/components/test_image.py
View File

@ -32,20 +32,24 @@ def test_image_serialization() -> None:
self.my_image = pydase.components.Image()
assert dump(MyService()) == {
"full_access_path": "",
"name": "MyService",
"type": "DataService",
"value": {
"my_image": {
"full_access_path": "my_image",
"name": "Image",
"type": "Image",
"value": {
"format": {
"full_access_path": "my_image.format",
"type": "str",
"value": "",
"readonly": True,
"doc": None,
},
"load_from_base64": {
"full_access_path": "my_image.load_from_base64",
"type": "method",
"value": None,
"readonly": True,
@ -72,6 +76,7 @@ def test_image_serialization() -> None:
"frontend_render": False,
},
"load_from_matplotlib_figure": {
"full_access_path": "my_image.load_from_matplotlib_figure",
"type": "method",
"value": None,
"readonly": True,
@ -95,6 +100,7 @@ def test_image_serialization() -> None:
"frontend_render": False,
},
"load_from_path": {
"full_access_path": "my_image.load_from_path",
"type": "method",
"value": None,
"readonly": True,
@ -112,6 +118,7 @@ def test_image_serialization() -> None:
"frontend_render": False,
},
"load_from_url": {
"full_access_path": "my_image.load_from_url",
"type": "method",
"value": None,
"readonly": True,
@ -126,6 +133,7 @@ def test_image_serialization() -> None:
"frontend_render": False,
},
"value": {
"full_access_path": "my_image.value",
"type": "str",
"value": "",
"readonly": True,

97
tests/utils/test_serializer.py
View File

@ -30,12 +30,40 @@ class MyEnum(enum.Enum):
@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}),
(
1,
{
"full_access_path": "",
"type": "int",
"value": 1,
"readonly": False,
"doc": None,
},
),
(
1.0,
{
"full_access_path": "",
"type": "float",
"value": 1.0,
"readonly": False,
"doc": None,
},
),
(
True,
{
"full_access_path": "",
"type": "bool",
"value": True,
"readonly": False,
"doc": None,
},
),
(
u.Quantity(10, "m"),
{
"full_access_path": "",
"type": "Quantity",
"value": {"magnitude": 10, "unit": "meter"},
"readonly": False,
@ -82,6 +110,7 @@ def test_enum_serialize() -> None:
assert dump(EnumAttribute())["value"] == {
"some_enum": {
"full_access_path": "some_enum",
"type": "Enum",
"name": "EnumClass",
"value": "FOO",
@ -92,6 +121,7 @@ def test_enum_serialize() -> None:
}
assert dump(EnumPropertyWithoutSetter())["value"] == {
"some_enum": {
"full_access_path": "some_enum",
"type": "Enum",
"name": "EnumClass",
"value": "FOO",
@ -102,6 +132,7 @@ def test_enum_serialize() -> None:
}
assert dump(EnumPropertyWithSetter())["value"] == {
"some_enum": {
"full_access_path": "some_enum",
"type": "Enum",
"name": "EnumClass",
"value": "FOO",
@ -125,6 +156,7 @@ def test_ColouredEnum_serialize() -> None:
CANCELLED = "SlateGray"
assert dump(Status.FAILED) == {
"full_access_path": "",
"type": "ColouredEnum",
"name": "Status",
"value": "FAILED",
@ -157,6 +189,7 @@ async def test_method_serialization() -> None:
assert dump(instance)["value"] == {
"some_method": {
"full_access_path": "some_method",
"type": "method",
"value": None,
"readonly": True,
@ -166,6 +199,7 @@ async def test_method_serialization() -> None:
"frontend_render": False,
},
"some_task": {
"full_access_path": "some_task",
"type": "method",
"value": TaskStatus.RUNNING.name,
"readonly": True,
@ -191,6 +225,7 @@ def test_methods_with_type_hints() -> None:
pass
assert dump(method_without_type_hint) == {
"full_access_path": "",
"async": False,
"doc": None,
"signature": {
@ -209,6 +244,7 @@ def test_methods_with_type_hints() -> None:
}
assert dump(method_with_type_hint) == {
"full_access_path": "",
"type": "method",
"value": None,
"readonly": True,
@ -223,6 +259,7 @@ def test_methods_with_type_hints() -> None:
"frontend_render": False,
}
assert dump(method_with_union_type_hint) == {
"full_access_path": "",
"type": "method",
"value": None,
"readonly": True,
@ -249,6 +286,7 @@ def test_exposed_function_serialization() -> None:
pass
assert dump(MyService().some_method) == {
"full_access_path": "",
"type": "method",
"value": None,
"readonly": True,
@ -259,6 +297,7 @@ def test_exposed_function_serialization() -> None:
}
assert dump(some_function) == {
"full_access_path": "",
"type": "method",
"value": None,
"readonly": True,
@ -286,30 +325,41 @@ def test_list_serialization() -> None:
assert dump(instance)["value"] == {
"list_attr": {
"full_access_path": "list_attr",
"doc": None,
"readonly": False,
"type": "list",
"value": [
{"doc": None, "readonly": False, "type": "int", "value": 1},
{
"full_access_path": "list_attr[0]",
"doc": None,
"readonly": False,
"type": "int",
"value": 1,
},
{
"full_access_path": "list_attr[1]",
"doc": None,
"readonly": False,
"type": "DataService",
"name": "MySubclass",
"value": {
"bool_attr": {
"full_access_path": "list_attr[1].bool_attr",
"doc": None,
"readonly": False,
"type": "bool",
"value": True,
},
"int_attr": {
"full_access_path": "list_attr[1].int_attr",
"doc": None,
"readonly": False,
"type": "int",
"value": 1,
},
"name": {
"full_access_path": "list_attr[1].name",
"doc": None,
"readonly": True,
"type": "str",
@ -335,17 +385,20 @@ def test_dict_serialization() -> None:
}
assert dump(test_dict) == {
"full_access_path": "",
"doc": None,
"readonly": False,
"type": "dict",
"value": {
"DataService_key": {
"full_access_path": '["DataService_key"]',
"name": "MyClass",
"doc": None,
"readonly": False,
"type": "DataService",
"value": {
"name": {
"full_access_path": '["DataService_key"].name',
"doc": None,
"readonly": False,
"type": "str",
@ -354,19 +407,33 @@ def test_dict_serialization() -> None:
},
},
"Quantity_key": {
"full_access_path": '["Quantity_key"]',
"doc": None,
"readonly": False,
"type": "Quantity",
"value": {"magnitude": 1.0, "unit": "s"},
},
"bool_key": {"doc": None, "readonly": False, "type": "bool", "value": True},
"bool_key": {
"full_access_path": '["bool_key"]',
"doc": None,
"readonly": False,
"type": "bool",
"value": True,
},
"float_key": {
"full_access_path": '["float_key"]',
"doc": None,
"readonly": False,
"type": "float",
"value": 1.0,
},
"int_key": {"doc": None, "readonly": False, "type": "int", "value": 1},
"int_key": {
"full_access_path": '["int_key"]',
"doc": None,
"readonly": False,
"type": "int",
"value": 1,
},
},
}
@ -387,8 +454,7 @@ def test_derived_data_service_serialization() -> None:
def name(self, value: str) -> None:
self._name = value
class DerivedService(BaseService):
...
class DerivedService(BaseService): ...
base_service_serialization = dump(BaseService())
derived_service_serialization = dump(DerivedService())
@ -415,7 +481,7 @@ def setup_dict() -> dict[str, Any]:
def my_task(self) -> None:
pass
return ServiceClass().serialize()["value"]
return ServiceClass().serialize()["value"] # type: ignore
def test_update_attribute(setup_dict: dict[str, Any]) -> None:
@ -431,6 +497,7 @@ def test_update_nested_attribute(setup_dict: dict[str, Any]) -> None:
def test_update_float_attribute_to_enum(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr2.attr3", MyEnum.RUNNING)
assert setup_dict["attr2"]["value"]["attr3"] == {
"full_access_path": "attr2.attr3",
"name": "MyEnum",
"doc": "MyEnum description",
"enum": {"FINISHED": "finished", "RUNNING": "running"},
@ -443,6 +510,7 @@ def test_update_float_attribute_to_enum(setup_dict: dict[str, Any]) -> None:
def test_update_enum_attribute_to_float(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "enum_attr", 1.01)
assert setup_dict["enum_attr"] == {
"full_access_path": "enum_attr",
"doc": None,
"readonly": False,
"type": "float",
@ -452,6 +520,7 @@ def test_update_enum_attribute_to_float(setup_dict: dict[str, Any]) -> None:
def test_update_task_state(setup_dict: dict[str, Any]) -> None:
assert setup_dict["my_task"] == {
"full_access_path": "my_task",
"async": False,
"doc": None,
"frontend_render": False,
@ -462,6 +531,7 @@ def test_update_task_state(setup_dict: dict[str, Any]) -> None:
}
set_nested_value_by_path(setup_dict, "my_task", TaskStatus.RUNNING)
assert setup_dict["my_task"] == {
"full_access_path": "my_task",
"async": False,
"doc": None,
"frontend_render": False,
@ -474,12 +544,13 @@ def test_update_task_state(setup_dict: dict[str, Any]) -> None:
def test_update_list_entry(setup_dict: dict[str, SerializedObject]) -> None:
set_nested_value_by_path(setup_dict, "attr_list[1]", 20)
assert setup_dict["attr_list"]["value"][1]["value"] == 20
assert setup_dict["attr_list"]["value"][1]["value"] == 20 # type: ignore # noqa
def test_update_list_append(setup_dict: dict[str, SerializedObject]) -> None:
set_nested_value_by_path(setup_dict, "attr_list[3]", MyEnum.RUNNING)
assert setup_dict["attr_list"]["value"][3] == {
assert setup_dict["attr_list"]["value"][3] == { # type: ignore
"full_access_path": "attr_list[3]",
"doc": "MyEnum description",
"name": "MyEnum",
"enum": {"FINISHED": "finished", "RUNNING": "running"},
@ -512,12 +583,12 @@ def test_update_invalid_path(
def test_update_list_inside_class(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr2.list_attr[1]", 40)
assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40
assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40 # noqa
def test_update_class_attribute_inside_list(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr_list[2].attr3", 50)
assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50
assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50 # noqa
def test_get_next_level_attribute_nested_dict(setup_dict: dict[str, Any]) -> None: