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

adds start and stop methods for tasks in socketio client

Browse Source
This commit is contained in:
Mose Müller 2024-03-27 15:18:40 +01:00
parent 9d3264de1f
commit 454b0fb7d1
1 changed files with 33 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
src/pydase/client/proxy_class_factory.py
View File

@ -1,11 +1,12 @@
import logging
from copy import copy
from typing import TYPE_CHECKING, Any, cast
import socketio # type: ignore
import pydase
from pydase.utils.deserializer import Deserializer, loads
from pydase.utils.serializer import SerializedObject, dump
from pydase.utils.serializer import SerializedMethod, SerializedObject, dump
if TYPE_CHECKING:
from collections.abc import Callable
@ -24,7 +25,7 @@ class ProxyClassFactory:
self.sio_client = sio_client
def create_proxy(self, data: SerializedObject) -> "ProxyClass":
proxy = self._deserialize(data)
proxy: "ProxyClass" = self._deserialize(data)
proxy._sio = self.sio_client
return proxy
@ -39,7 +40,6 @@ class ProxyClassFactory:
"Quantity": self._create_attr_property,
"Enum": self._create_attr_property,
"ColouredEnum": self._create_attr_property,
"method": self._deserialize_method,
"list": loads,
"dict": loads,
"Exception": loads,
@ -60,7 +60,7 @@ class ProxyClassFactory:
return proxy_class
return None
def _deserialize_method(self, serialized_object: SerializedObject) -> Any:
def _deserialize_method(self, serialized_object: SerializedMethod) -> Any:
def method_proxy(self: "ProxyClass", *args: Any, **kwargs: Any) -> Any:
serialized_response = cast(
dict[str, Any],
@ -80,6 +80,11 @@ class ProxyClassFactory:
def _deserialize_component_type(
self, serialized_object: SerializedObject, base_class: type
) -> Any:
def add_prefix_to_last_path_element(s: str, prefix: str) -> str:
parts = s.split(".")
parts[-1] = f"{prefix}_{parts[-1]}"
return ".".join(parts)
def create_proxy_class(serialized_object: SerializedObject) -> type:
class_bases = (base_class,)
class_attrs: dict[str, Any] = {}
@ -88,6 +93,29 @@ class ProxyClassFactory:
for key, value in cast(
dict[str, SerializedObject], serialized_object["value"]
).items():
if value["type"] == "method":
if value["async"] is True:
start_method = copy(value)
start_method["full_access_path"] = (
add_prefix_to_last_path_element(
start_method["full_access_path"], "start"
)
)
stop_method = copy(value)
stop_method["full_access_path"] = (
add_prefix_to_last_path_element(
stop_method["full_access_path"], "stop"
)
)
class_attrs[f"start_{key}"] = self._deserialize_method(
start_method
)
class_attrs[f"stop_{key}"] = self._deserialize_method(
stop_method
)
else:
class_attrs[key] = self._deserialize_method(value)
else:
class_attrs[key] = self._deserialize(value)
# Create the dynamic class with the given name and attributes