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

Merge pull request #140 from tiqi-group/feat/make_serializer_methods_properly_overridable

Browse Source 
serializer: converting staticmethods to classmethods
This commit is contained in:
Mose Müller 2024-07-09 15:19:53 +02:00 committed by GitHub
commit 8afee54c51
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 32 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/pydase/utils/serialization/deserializer.py
View File

@ -108,7 +108,7 @@ class Deserializer:
return pydase.DataService return pydase.DataService
return None return None
@classmethod @classmethod # TODO: this shouldn't be a class method
def create_attr_property(cls, serialized_attr: SerializedObject) -> property: def create_attr_property(cls, serialized_attr: SerializedObject) -> property:
attr_name = serialized_attr["full_access_path"].split(".")[-1] attr_name = serialized_attr["full_access_path"].split(".")[-1]

59
src/pydase/utils/serialization/serializer.py
View File

@ -54,36 +54,36 @@ class SerializationValueError(Exception):
class Serializer: class Serializer:
@staticmethod @classmethod
def serialize_object(obj: Any, access_path: str = "") -> SerializedObject: # noqa: C901 def serialize_object(cls, obj: Any, access_path: str = "") -> SerializedObject: # noqa: C901
result: SerializedObject result: SerializedObject
if isinstance(obj, Exception): if isinstance(obj, Exception):
result = Serializer._serialize_exception(obj) result = cls._serialize_exception(obj)
elif isinstance(obj, AbstractDataService): elif isinstance(obj, AbstractDataService):
result = Serializer._serialize_data_service(obj, access_path=access_path) result = cls._serialize_data_service(obj, access_path=access_path)
elif isinstance(obj, list): elif isinstance(obj, list):
result = Serializer._serialize_list(obj, access_path=access_path) result = cls._serialize_list(obj, access_path=access_path)
elif isinstance(obj, dict): elif isinstance(obj, dict):
result = Serializer._serialize_dict(obj, access_path=access_path) result = cls._serialize_dict(obj, access_path=access_path)
# Special handling for u.Quantity # Special handling for u.Quantity
elif isinstance(obj, u.Quantity): elif isinstance(obj, u.Quantity):
result = Serializer._serialize_quantity(obj, access_path=access_path) result = cls._serialize_quantity(obj, access_path=access_path)
# Handling for Enums # Handling for Enums
elif isinstance(obj, Enum): elif isinstance(obj, Enum):
result = Serializer._serialize_enum(obj, access_path=access_path) result = cls._serialize_enum(obj, access_path=access_path)
# Methods and coroutines # Methods and coroutines
elif inspect.isfunction(obj) or inspect.ismethod(obj): elif inspect.isfunction(obj) or inspect.ismethod(obj):
result = Serializer._serialize_method(obj, access_path=access_path) result = cls._serialize_method(obj, access_path=access_path)
elif isinstance(obj, int | float | bool | str | None): elif isinstance(obj, int | float | bool | str | None):
result = Serializer._serialize_primitive(obj, access_path=access_path) result = cls._serialize_primitive(obj, access_path=access_path)
try: try:
return result return result
@ -92,8 +92,9 @@ class Serializer:
f"Could not serialized object of type {type(obj)}." f"Could not serialized object of type {type(obj)}."
) )
@staticmethod @classmethod
def _serialize_primitive( def _serialize_primitive(
cls,
obj: float | bool | str | None, obj: float | bool | str | None,
access_path: str, access_path: str,
) -> ( ) -> (
@ -112,8 +113,8 @@ class Serializer:
"value": obj, "value": obj,
} }
@staticmethod @classmethod
def _serialize_exception(obj: Exception) -> SerializedException: def _serialize_exception(cls, obj: Exception) -> SerializedException:
return { return {
"full_access_path": "", "full_access_path": "",
"doc": None, "doc": None,
@ -123,8 +124,8 @@ class Serializer:
"name": obj.__class__.__name__, "name": obj.__class__.__name__,
} }
@staticmethod @classmethod
def _serialize_enum(obj: Enum, access_path: str = "") -> SerializedEnum: def _serialize_enum(cls, obj: Enum, access_path: str = "") -> SerializedEnum:
import pydase.components.coloured_enum import pydase.components.coloured_enum
value = obj.name value = obj.name
@ -149,9 +150,9 @@ class Serializer:
}, },
} }
@staticmethod @classmethod
def _serialize_quantity( def _serialize_quantity(
obj: u.Quantity, access_path: str = "" cls, obj: u.Quantity, access_path: str = ""
) -> SerializedQuantity: ) -> SerializedQuantity:
doc = get_attribute_doc(obj) doc = get_attribute_doc(obj)
value: u.QuantityDict = {"magnitude": obj.m, "unit": str(obj.u)} value: u.QuantityDict = {"magnitude": obj.m, "unit": str(obj.u)}
@ -163,13 +164,15 @@ class Serializer:
"doc": doc, "doc": doc,
} }
@staticmethod @classmethod
def _serialize_dict(obj: dict[str, Any], access_path: str = "") -> SerializedDict: def _serialize_dict(
cls, obj: dict[str, Any], access_path: str = ""
) -> SerializedDict:
readonly = False readonly = False
doc = get_attribute_doc(obj) doc = get_attribute_doc(obj)
value = {} value = {}
for key, val in obj.items(): for key, val in obj.items():
value[key] = Serializer.serialize_object( value[key] = cls.serialize_object(
val, access_path=f'{access_path}["{key}"]' val, access_path=f'{access_path}["{key}"]'
) )
return { return {
@ -180,12 +183,12 @@ class Serializer:
"doc": doc, "doc": doc,
} }
@staticmethod @classmethod
def _serialize_list(obj: list[Any], access_path: str = "") -> SerializedList: def _serialize_list(cls, obj: list[Any], access_path: str = "") -> SerializedList:
readonly = False readonly = False
doc = get_attribute_doc(obj) doc = get_attribute_doc(obj)
value = [ value = [
Serializer.serialize_object(o, access_path=f"{access_path}[{i}]") cls.serialize_object(o, access_path=f"{access_path}[{i}]")
for i, o in enumerate(obj) for i, o in enumerate(obj)
] ]
return { return {
@ -196,9 +199,9 @@ class Serializer:
"doc": doc, "doc": doc,
} }
@staticmethod @classmethod
def _serialize_method( def _serialize_method(
obj: Callable[..., Any], access_path: str = "" cls, obj: Callable[..., Any], access_path: str = ""
) -> SerializedMethod: ) -> SerializedMethod:
readonly = True readonly = True
doc = get_attribute_doc(obj) doc = get_attribute_doc(obj)
@ -231,9 +234,9 @@ class Serializer:
"frontend_render": frontend_render, "frontend_render": frontend_render,
} }
@staticmethod @classmethod
def _serialize_data_service( def _serialize_data_service(
obj: AbstractDataService, access_path: str = "" cls, obj: AbstractDataService, access_path: str = ""
) -> SerializedDataService: ) -> SerializedDataService:
readonly = False readonly = False
doc = get_attribute_doc(obj) doc = get_attribute_doc(obj)
@ -274,7 +277,7 @@ class Serializer:
val = getattr(obj, key) val = getattr(obj, key)
path = f"{access_path}.{key}" if access_path else key path = f"{access_path}.{key}" if access_path else key
serialized_object = Serializer.serialize_object(val, access_path=path) serialized_object = cls.serialize_object(val, access_path=path)
# If there's a running task for this method # If there's a running task for this method
if serialized_object["type"] == "method" and key in obj._task_manager.tasks: if serialized_object["type"] == "method" and key in obj._task_manager.tasks: