1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-08 09:47:52 +02:00

Compare commits

...

1 Commits

Author SHA1 Message Date
e3eb2466d7 WIP first try 2026-03-13 11:04:48 +01:00
2 changed files with 47 additions and 1 deletions

View File

@@ -106,6 +106,8 @@ class RPCServer:
self._registry_update_callbacks = []
self._broadcasted_data = {}
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
logger.success(f"Server started with gui_id: {self.gui_id}")
@@ -240,8 +242,25 @@ class RPCServer:
return {"system.launch_dock_area": True}
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(
self,
name: str | None = None,
geometry: tuple[int, int, int, int] | None = None,
startup_profile: str | Literal["restore", "skip"] | None = None,
@@ -263,12 +282,14 @@ class RPCServer:
if isinstance(result_widget, BECMainWindow):
apply_window_geometry(result_widget, geometry)
result_widget.show()
self._track_top_level_rpc_widget(result_widget, result_widget)
else:
window = BECMainWindowNoRPC()
window.setCentralWidget(result_widget)
window.setWindowTitle(f"BEC - {result_widget.objectName()}")
apply_window_geometry(window, geometry)
window.show()
self._track_top_level_rpc_widget(result_widget, window)
return result_widget
def serialize_result_and_send(self, request_id: str, res: object):

View File

@@ -7,6 +7,7 @@ from qtpy.QtWidgets import QWidget
from bec_widgets.cli.server import GUIServer
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 .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", ["creator"], {}) is None
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