bec_widgets.utils.rpc_server#

Attributes#

T

messages

Exceptions#

RegistryNotReadyError

Raised when trying to access an object from the RPC registry that is not yet registered.

Classes#

RPCServer

SingleshotRPCRepeat

Functions#

rpc_exception_hook(err_func)

This context replaces the popup message box for error display with a specific hook

Module Contents#

exception RegistryNotReadyError#

Bases: Exception

Raised when trying to access an object from the RPC registry that is not yet registered.

Initialize self. See help(type(self)) for accurate signature.

class RPCServer(gui_id: str, dispatcher: bec_widgets.utils.BECDispatcher | None = None, client: bec_lib.client.BECClient | None = None, config=None, gui_class_id: str = 'bec')#
static _launch_dock_area(name: str | None = None, geometry: tuple[int, int, int, int] | None = None, startup_profile: str | Literal['restore', 'skip'] | None = None) → qtpy.QtWidgets.QWidget | None#
_resolve_rpc_target(obj, method: str) → tuple[object, object]#

Resolve a method/property access target for RPC execution.

Primary target is the object itself. If not found there and the class defines RPC_CONTENT_CLASS, unresolved method names can be delegated to the content widget referenced by RPC_CONTENT_ATTR (default content), but only when the method is explicitly listed in the content class USER_ACCESS.

_serialize_bec_connector(connector: bec_widgets.utils.bec_connector.BECConnector, wait=False) → dict#

Create the serialization dict for a single BECConnector.

Parameters:

  • connector (BECConnector) – The BECConnector to serialize.

  • wait (bool) – If True, wait until the object is registered in the RPC register.

Returns:

The serialized BECConnector object.

Return type:

dict

add_registry_update_callback(cb: Callable) → None#

Add a callback to be called whenever the registry is updated. The specified callback is called whenever the registry is updated.

Parameters:

  • cb (Callable) – The callback to be added. It should accept a dictionary of all the

  • argument. (registered RPC objects as an)

broadcast_registry_update(connections: dict) → None#

Broadcast the registry update to all the callbacks. This method is called whenever the registry is updated.

emit_heartbeat() → None#

Emit a heartbeat message to the GUI server. This method is called periodically to indicate that the server is still running.

get_object_from_config(config: dict)#
on_rpc_update(msg: dict, metadata: dict)#
run_rpc(obj, method, args, kwargs)#
run_system_rpc(method: str, args: list, kwargs: dict)#
send_response(request_id: str, accepted: bool, msg: dict)#
serialize_object(obj: T) → None | dict | T#

Serialize all BECConnector objects.

Parameters:

obj – The object to be serialized.

Returns:

The serialized object or None if the object is not a BECConnector.

Return type:

None | dict | T

serialize_result_and_send(request_id: str, res: object)#

Serialize the result of an RPC call and send it back to the client.

Note: If the object is not yet registered in the RPC registry, this method will retry serialization after a short delay, up to a maximum delay. In order to avoid processEvents calls in the middle of serialization, QTimer.singleShot is used. This allows the target event to ‘float’ to the next event loop iteration until the object is registered. The ‘jump’ to the next event loop is indicated by raising a RegistryNotReadyError, see _serialize_bec_connector.

Parameters:

  • request_id (str) – The ID of the request.

  • res (object) – The result of the RPC call.

shutdown()#
client: bec_lib.client.BECClient#
dispatcher#
gui_id#
rpc_register#
status#
class SingleshotRPCRepeat(max_delay: int = 2000)#
__iadd__(delay: int)#
accumulated_delay = 0#
max_delay = 2000#
rpc_exception_hook(err_func)#

This context replaces the popup message box for error display with a specific hook

T#
messages#