bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

fix(rpc_server): enhance serialization logic for BECConnector objects and fix return types

Browse Source
This commit is contained in:
wakonig_k
2025-04-11 13:34:05 +02:00
parent 4dc59aa5e9
commit 125afc8907
1 changed files with 20 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
bec_widgets/utils/rpc_server.py
View File

@ -4,7 +4,7 @@ import functools
import traceback import traceback
import types import types
from contextlib import contextmanager from contextlib import contextmanager
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, TypeVar
from bec_lib.client import BECClient from bec_lib.client import BECClient
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
@ -28,6 +28,9 @@ else:
logger = bec_logger.logger logger = bec_logger.logger
T = TypeVar("T")
@contextmanager @contextmanager
def rpc_exception_hook(err_func): def rpc_exception_hook(err_func):
"""This context replaces the popup message box for error display with a specific hook""" """This context replaces the popup message box for error display with a specific hook"""
@ -143,13 +146,22 @@ class RPCServer:
res = self.serialize_object(res) res = self.serialize_object(res)
return res return res
def serialize_object(self, obj): def serialize_object(self, obj: T) -> None | dict | T:
if isinstance(obj, BECConnector): """
Serialize all BECConnector objects.
Args:
obj: The object to be serialized.
Returns:
None | dict | T: The serialized object or None if the object is not a BECConnector.
"""
if not isinstance(obj, BECConnector):
return obj
# Respect RPC = False # Respect RPC = False
if hasattr(obj, "RPC") and obj.RPC is False: if getattr(obj, "RPC", True) is False:
return None return None
return self._serialize_bec_connector(obj) return self._serialize_bec_connector(obj)
return obj
def emit_heartbeat(self): def emit_heartbeat(self):
logger.trace(f"Emitting heartbeat for {self.gui_id}") logger.trace(f"Emitting heartbeat for {self.gui_id}")