diff --git a/bec_widgets/cli/client_utils.py b/bec_widgets/cli/client_utils.py index 37527b70..61a59f3c 100644 --- a/bec_widgets/cli/client_utils.py +++ b/bec_widgets/cli/client_utils.py @@ -355,17 +355,53 @@ class BECGuiClient(RPCBase): self.start(wait=True) if wait: with wait_for_server(self): - widget = self.launcher._run_rpc( - "launch", + return self._new_impl( + name=name, + geometry=geometry, launch_script=launch_script, + profile=profile, + start_empty=start_empty, + **kwargs, + ) + return self._new_impl( + name=name, + geometry=geometry, + launch_script=launch_script, + profile=profile, + start_empty=start_empty, + **kwargs, + ) + + def _new_impl( + self, + *, + name: str | None, + geometry: tuple[int, int, int, int] | None, + launch_script: str, + profile: str | None, + start_empty: bool, + **kwargs, + ): + if launch_script == "dock_area": + try: + return self.launcher._run_rpc( + "system.launch_dock_area", name=name, geometry=geometry, profile=profile, start_empty=start_empty, **kwargs, - ) # pylint: disable=protected-access - return widget - widget = self.launcher._run_rpc( + ) + except ValueError as exc: + error = str(exc) + if ( + "Unknown system RPC method: system.launch_dock_area" not in error + and "has no attribute 'system.launch_dock_area'" not in error + ): + raise + logger.debug("Server does not support system.launch_dock_area; using launcher RPC") + + return self.launcher._run_rpc( "launch", launch_script=launch_script, name=name, @@ -374,7 +410,6 @@ class BECGuiClient(RPCBase): start_empty=start_empty, **kwargs, ) # pylint: disable=protected-access - return widget def delete(self, name: str) -> None: """Delete a dock area and its parent window. diff --git a/bec_widgets/utils/rpc_server.py b/bec_widgets/utils/rpc_server.py index 5bffa051..d95585ba 100644 --- a/bec_widgets/utils/rpc_server.py +++ b/bec_widgets/utils/rpc_server.py @@ -11,13 +11,17 @@ from bec_lib.endpoints import MessageEndpoints from bec_lib.logger import bec_logger from bec_lib.utils.import_utils import lazy_import from qtpy.QtCore import Qt, QTimer +from qtpy.QtWidgets import QWidget from redis.exceptions import RedisError from bec_widgets.cli.rpc.rpc_register import RPCRegister from bec_widgets.utils import BECDispatcher from bec_widgets.utils.bec_connector import BECConnector +from bec_widgets.utils.container_utils import WidgetContainerUtils from bec_widgets.utils.error_popups import ErrorPopupUtility -from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow +from bec_widgets.utils.screen_utils import apply_window_geometry +from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea +from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow, BECMainWindowNoRPC if TYPE_CHECKING: # pragma: no cover from bec_lib import messages @@ -114,11 +118,14 @@ class RPCServer: logger.debug(f"Received RPC instruction: {msg}, metadata: {metadata}") with rpc_exception_hook(functools.partial(self.send_response, request_id, False)): try: - obj = self.get_object_from_config(msg["parameter"]) method = msg["action"] args = msg["parameter"].get("args", []) kwargs = msg["parameter"].get("kwargs", {}) - res = self.run_rpc(obj, method, args, kwargs) + if method.startswith("system."): + res = self.run_system_rpc(method, args, kwargs) + else: + obj = self.get_object_from_config(msg["parameter"]) + res = self.run_rpc(obj, method, args, kwargs) except Exception: content = traceback.format_exc() logger.error(f"Error while executing RPC instruction: {content}") @@ -186,6 +193,47 @@ class RPCServer: res = method_obj(*args, **kwargs) return res + def run_system_rpc(self, method: str, args: list, kwargs: dict): + if method == "system.launch_dock_area": + return self._launch_dock_area(*args, **kwargs) + if method == "system.list_capabilities": + return {"system.launch_dock_area": True} + raise ValueError(f"Unknown system RPC method: {method}") + + @staticmethod + def _launch_dock_area( + name: str | None = None, + geometry: tuple[int, int, int, int] | None = None, + profile: str | None = None, + start_empty: bool = False, + ) -> QWidget | None: + from bec_widgets.applications import bw_launch + + with RPCRegister.delayed_broadcast() as rpc_register: + existing_dock_areas = rpc_register.get_names_of_rpc_by_class_type(BECDockArea) + if name is not None: + WidgetContainerUtils.raise_for_invalid_name(name) + if name in existing_dock_areas: + name = WidgetContainerUtils.generate_unique_name(name, existing_dock_areas) + else: + name = WidgetContainerUtils.generate_unique_name("dock_area", existing_dock_areas) + + result_widget = bw_launch.dock_area( + object_name=name, profile=profile, start_empty=start_empty + ) + result_widget.window().setWindowTitle(f"BEC - {name}") + + if isinstance(result_widget, BECMainWindow): + apply_window_geometry(result_widget, geometry) + result_widget.show() + else: + window = BECMainWindowNoRPC() + window.setCentralWidget(result_widget) + window.setWindowTitle(f"BEC - {result_widget.objectName()}") + apply_window_geometry(window, geometry) + window.show() + return result_widget + def serialize_result_and_send(self, request_id: str, res: object): """ Serialize the result of an RPC call and send it back to the client.