bec_widgets.utils.rpc_server
============================

.. py:module:: bec_widgets.utils.rpc_server


Attributes
----------

.. autoapisummary::

   bec_widgets.utils.rpc_server.T
   bec_widgets.utils.rpc_server.messages


Exceptions
----------

.. autoapisummary::

   bec_widgets.utils.rpc_server.RegistryNotReadyError


Classes
-------

.. autoapisummary::

   bec_widgets.utils.rpc_server.RPCServer
   bec_widgets.utils.rpc_server.SingleshotRPCRepeat


Functions
---------

.. autoapisummary::

   bec_widgets.utils.rpc_server.rpc_exception_hook


Module Contents
---------------

.. py:exception:: RegistryNotReadyError

   Bases: :py:obj:`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.


.. py: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')

   .. py:method:: _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
      :staticmethod:



   .. py:method:: _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``.



   .. py:method:: _serialize_bec_connector(connector: bec_widgets.utils.bec_connector.BECConnector, wait=False) -> dict

      Create the serialization dict for a single BECConnector.

      :param connector: The BECConnector to serialize.
      :type connector: BECConnector
      :param wait: If True, wait until the object is registered in the RPC register.
      :type wait: bool

      :returns: The serialized BECConnector object.
      :rtype: dict



   .. py:method:: 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.

      :param cb: The callback to be added. It should accept a dictionary of all the
      :type cb: Callable
      :param registered RPC objects as an argument.:



   .. py:method:: broadcast_registry_update(connections: dict) -> None

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



   .. py:method:: emit_heartbeat() -> None

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



   .. py:method:: get_object_from_config(config: dict)


   .. py:method:: on_rpc_update(msg: dict, metadata: dict)


   .. py:method:: run_rpc(obj, method, args, kwargs)


   .. py:method:: run_system_rpc(method: str, args: list, kwargs: dict)


   .. py:method:: send_response(request_id: str, accepted: bool, msg: dict)


   .. py:method:: serialize_object(obj: T) -> None | dict | T

      Serialize all BECConnector objects.

      :param obj: The object to be serialized.

      :returns: The serialized object or None if the object is not a BECConnector.
      :rtype: None | dict | T



   .. py:method:: 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.

      :param request_id: The ID of the request.
      :type request_id: str
      :param res: The result of the RPC call.
      :type res: object



   .. py:method:: shutdown()


   .. py:attribute:: client
      :type:  bec_lib.client.BECClient


   .. py:attribute:: dispatcher


   .. py:attribute:: gui_id


   .. py:attribute:: rpc_register


   .. py:attribute:: status


.. py:class:: SingleshotRPCRepeat(max_delay: int = 2000)

   .. py:method:: __iadd__(delay: int)


   .. py:attribute:: accumulated_delay
      :value: 0



   .. py:attribute:: max_delay
      :value: 2000



.. py:function:: rpc_exception_hook(err_func)

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


.. py:data:: T

.. py:data:: messages