mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-05-09 08:12:15 +02:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e3eb2466d7 |
@@ -106,6 +106,8 @@ class RPCServer:
|
|||||||
self._registry_update_callbacks = []
|
self._registry_update_callbacks = []
|
||||||
self._broadcasted_data = {}
|
self._broadcasted_data = {}
|
||||||
self._rpc_singleshot_repeats: dict[str, SingleshotRPCRepeat] = {}
|
self._rpc_singleshot_repeats: dict[str, SingleshotRPCRepeat] = {}
|
||||||
|
self._top_level_rpc_windows: dict[str, QWidget] = {}
|
||||||
|
self._top_level_rpc_widgets: dict[str, QWidget] = {}
|
||||||
|
|
||||||
self.status = messages.BECStatus.RUNNING
|
self.status = messages.BECStatus.RUNNING
|
||||||
logger.success(f"Server started with gui_id: {self.gui_id}")
|
logger.success(f"Server started with gui_id: {self.gui_id}")
|
||||||
@@ -240,8 +242,25 @@ class RPCServer:
|
|||||||
return {"system.launch_dock_area": True}
|
return {"system.launch_dock_area": True}
|
||||||
raise ValueError(f"Unknown system RPC method: {method}")
|
raise ValueError(f"Unknown system RPC method: {method}")
|
||||||
|
|
||||||
@staticmethod
|
def _track_top_level_rpc_widget(self, widget: QWidget, window: QWidget | None = None) -> None:
|
||||||
|
gui_id = getattr(widget, "gui_id", None)
|
||||||
|
if not gui_id:
|
||||||
|
return
|
||||||
|
|
||||||
|
self._top_level_rpc_widgets[gui_id] = widget
|
||||||
|
if window is not None:
|
||||||
|
self._top_level_rpc_windows[gui_id] = window
|
||||||
|
|
||||||
|
def _cleanup_refs(*_args, tracked_gui_id=gui_id):
|
||||||
|
self._top_level_rpc_widgets.pop(tracked_gui_id, None)
|
||||||
|
self._top_level_rpc_windows.pop(tracked_gui_id, None)
|
||||||
|
|
||||||
|
widget.destroyed.connect(_cleanup_refs)
|
||||||
|
if window is not None and window is not widget:
|
||||||
|
window.destroyed.connect(_cleanup_refs)
|
||||||
|
|
||||||
def _launch_dock_area(
|
def _launch_dock_area(
|
||||||
|
self,
|
||||||
name: str | None = None,
|
name: str | None = None,
|
||||||
geometry: tuple[int, int, int, int] | None = None,
|
geometry: tuple[int, int, int, int] | None = None,
|
||||||
startup_profile: str | Literal["restore", "skip"] | None = None,
|
startup_profile: str | Literal["restore", "skip"] | None = None,
|
||||||
@@ -263,12 +282,14 @@ class RPCServer:
|
|||||||
if isinstance(result_widget, BECMainWindow):
|
if isinstance(result_widget, BECMainWindow):
|
||||||
apply_window_geometry(result_widget, geometry)
|
apply_window_geometry(result_widget, geometry)
|
||||||
result_widget.show()
|
result_widget.show()
|
||||||
|
self._track_top_level_rpc_widget(result_widget, result_widget)
|
||||||
else:
|
else:
|
||||||
window = BECMainWindowNoRPC()
|
window = BECMainWindowNoRPC()
|
||||||
window.setCentralWidget(result_widget)
|
window.setCentralWidget(result_widget)
|
||||||
window.setWindowTitle(f"BEC - {result_widget.objectName()}")
|
window.setWindowTitle(f"BEC - {result_widget.objectName()}")
|
||||||
apply_window_geometry(window, geometry)
|
apply_window_geometry(window, geometry)
|
||||||
window.show()
|
window.show()
|
||||||
|
self._track_top_level_rpc_widget(result_widget, window)
|
||||||
return result_widget
|
return result_widget
|
||||||
|
|
||||||
def serialize_result_and_send(self, request_id: str, res: object):
|
def serialize_result_and_send(self, request_id: str, res: object):
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ from qtpy.QtWidgets import QWidget
|
|||||||
|
|
||||||
from bec_widgets.cli.server import GUIServer
|
from bec_widgets.cli.server import GUIServer
|
||||||
from bec_widgets.utils.bec_connector import BECConnector
|
from bec_widgets.utils.bec_connector import BECConnector
|
||||||
|
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindowNoRPC
|
||||||
from bec_widgets.utils.rpc_server import RegistryNotReadyError, RPCServer, SingleshotRPCRepeat
|
from bec_widgets.utils.rpc_server import RegistryNotReadyError, RPCServer, SingleshotRPCRepeat
|
||||||
|
|
||||||
from .client_mocks import mocked_client
|
from .client_mocks import mocked_client
|
||||||
@@ -171,3 +172,27 @@ def test_run_rpc_delegates_to_rpc_content_class(rpc_server):
|
|||||||
assert rpc_server.run_rpc(view, "mode", [], {}) == "initial"
|
assert rpc_server.run_rpc(view, "mode", [], {}) == "initial"
|
||||||
assert rpc_server.run_rpc(view, "mode", ["creator"], {}) is None
|
assert rpc_server.run_rpc(view, "mode", ["creator"], {}) is None
|
||||||
assert view.content.mode == "creator"
|
assert view.content.mode == "creator"
|
||||||
|
|
||||||
|
|
||||||
|
def test_launch_dock_area_keeps_strong_references(rpc_server, qtbot):
|
||||||
|
class DummyDockArea(BECConnector, QWidget):
|
||||||
|
def __init__(self, parent=None, client=None, **kwargs):
|
||||||
|
super().__init__(parent=parent, client=client, root_widget=True, **kwargs)
|
||||||
|
|
||||||
|
with patch("bec_widgets.applications.bw_launch.dock_area") as mock_launch_dock_area:
|
||||||
|
mock_launch_dock_area.return_value = DummyDockArea(
|
||||||
|
client=rpc_server.client, object_name="dock_area"
|
||||||
|
)
|
||||||
|
|
||||||
|
result = rpc_server._launch_dock_area(name="dock_area")
|
||||||
|
assert result is mock_launch_dock_area.return_value
|
||||||
|
|
||||||
|
gui_id = result.gui_id
|
||||||
|
assert rpc_server._top_level_rpc_widgets[gui_id] is result
|
||||||
|
assert isinstance(rpc_server._top_level_rpc_windows[gui_id], BECMainWindowNoRPC)
|
||||||
|
|
||||||
|
window = rpc_server._top_level_rpc_windows[gui_id]
|
||||||
|
window.close()
|
||||||
|
window.deleteLater()
|
||||||
|
qtbot.waitUntil(lambda: gui_id not in rpc_server._top_level_rpc_windows, timeout=3000)
|
||||||
|
assert gui_id not in rpc_server._top_level_rpc_widgets
|
||||||
|
|||||||
Reference in New Issue
Block a user