mirror of
https://github.com/tiqi-group/pydase_service_base.git
synced 2026-01-16 17:39:25 +01:00
fixes method serialization compatibility with new pydase version
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
import copy
|
||||
import inspect
|
||||
import re
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
@@ -7,11 +9,55 @@ from pydase.components import NumberSlider
|
||||
from pydase.data_service.data_service_observer import DataServiceObserver
|
||||
from pydase.units import Quantity
|
||||
from pydase.utils.helpers import get_object_attr_from_path
|
||||
from pydase.utils.serialization.serializer import dump
|
||||
from pydase.utils.serialization.types import SerializedObject
|
||||
from pydase.utils.serialization.serializer import (
|
||||
dump,
|
||||
generate_serialized_data_paths,
|
||||
get_nested_dict_by_path,
|
||||
)
|
||||
from pydase.utils.serialization.types import SerializedMethod, SerializedObject
|
||||
from pydase.version import __version__
|
||||
|
||||
|
||||
def extract_type_name(type_annotation: str) -> str | None:
|
||||
"""Extracts the type name from annotation string. For builtin types, the type string
|
||||
is enclosed like this: "<class 'type_str'>".
|
||||
|
||||
Example:
|
||||
>>> type_annotation = "<class 'int'>"
|
||||
>>> print(extract_type_name(type_annotation))
|
||||
int
|
||||
"""
|
||||
match = re.search(r"'([^']*)'", type_annotation)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return None
|
||||
|
||||
|
||||
def add_parameters_keyword_to_dict(
|
||||
serialized_method: SerializedMethod,
|
||||
) -> None:
|
||||
"""Adds "parameters" key-value pair to serialized_method dictionary."""
|
||||
parameters: dict[str, str | None] = {}
|
||||
for name, param_signature in serialized_method["signature"]["parameters"].items():
|
||||
parameters[name] = extract_type_name(param_signature["annotation"])
|
||||
|
||||
# adds the parameters keyword to the dictionary as Ionizer is expecting this
|
||||
serialized_method["parameters"] = parameters # type: ignore
|
||||
|
||||
|
||||
def update_method_serialization(
|
||||
serialized_object: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Adds a "parameter" key-value pair to each serialized method within
|
||||
serialized_object."""
|
||||
for path in generate_serialized_data_paths(serialized_object):
|
||||
nested_dict = get_nested_dict_by_path(serialized_object, path)
|
||||
if nested_dict["type"] == "method":
|
||||
add_parameters_keyword_to_dict(nested_dict)
|
||||
|
||||
return serialized_object
|
||||
|
||||
|
||||
class RPCInterface:
|
||||
"""RPC interface to be passed to tiqi_rpc.Server to interface with Ionizer."""
|
||||
|
||||
@@ -29,7 +75,9 @@ class RPCInterface:
|
||||
return self._service.__class__.__name__
|
||||
|
||||
async def get_props(self) -> SerializedObject:
|
||||
return self._service.serialize()["value"] # type: ignore
|
||||
return update_method_serialization(
|
||||
copy.deepcopy(self._service.serialize()["value"]) # type: ignore
|
||||
)
|
||||
|
||||
async def get_param(self, full_access_path: str) -> Any:
|
||||
"""Returns the value of the parameter given by the full_access_path.
|
||||
|
||||
Reference in New Issue
Block a user