bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

wip

Browse Source
This commit is contained in:
wakonig_k
2025-03-28 14:07:09 +01:00
committed by wyzula-jan
parent 97109f71c4
commit 1701bc3f80
6 changed files with 40 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
bec_widgets/applications/bw_launch.py
View File

@ -1,35 +1,6 @@
import os
from qtpy.QtWidgets import QApplication, QMainWindow
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
class Launcher(BECWidget, QMainWindow):
def __init__(self, gui_id: str = None, *args, **kwargs):
BECWidget.__init__(self, gui_id=gui_id, **kwargs)
QMainWindow.__init__(self, *args, **kwargs)
ui_file_path = os.path.join(os.path.dirname(__file__), "launcher.ui")
self.load_ui(ui_file_path)
def load_ui(self, ui_file):
loader = UILoader(self)
self.ui = loader.loader(ui_file)
self.setCentralWidget(self.ui)
def dock_area(name: str | None = None):
dock_area = BECDockArea(name=name)
return dock_area
def launcher():
launcher = Launcher()
return launcher
if __name__ == "__main__":
dock_area()
_dock_area = BECDockArea(name=name)
return _dock_area

24
bec_widgets/applications/launch_window.py
View File

@ -111,30 +111,6 @@ class LaunchWindow(BECWidget, QMainWindow):
def change_theme(self, theme):
apply_theme(theme)
def _dump(self):
"""Return a dictionary with informations about the application state, for use in tests"""
# TODO: ModularToolBar and something else leak top-level widgets (3 or 4 QMenu + 2 QWidget);
# so, a filtering based on title is applied here, but the solution is to not have those widgets
# as top-level (so for now, a window with no title does not appear in _dump() result)
# NOTE: the main window itself is excluded, since we want to dump dock areas
info = {
tlw.gui_id: {
"title": tlw.windowTitle(),
"visible": tlw.isVisible(),
"class": str(type(tlw)),
}
for tlw in QApplication.instance().topLevelWidgets()
if tlw is not self and tlw.windowTitle()
}
# Add the main window dock area
info[self.centralWidget().gui_id] = {
"title": self.windowTitle(),
"visible": self.isVisible(),
"class": str(type(self.centralWidget())),
}
return info
def launch(
self,
launch_script: str,

50
bec_widgets/cli/client_utils.py
View File

@ -247,7 +247,7 @@ class BECGuiClient(RPCBase):
@property
def windows(self) -> dict:
"""Dictionary with dock areas in the GUI."""
return self._top_level
return {widget._name: widget for widget in self._top_level.values()}
@property
def window_list(self) -> list:
@ -400,10 +400,6 @@ class BECGuiClient(RPCBase):
if wait:
self._gui_started_event.wait()
def _dump(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self)
return rpc_client._run_rpc("_dump")
def _start(self, wait: bool = False) -> None:
self._killed = False
self._client.connector.register(
@ -440,34 +436,42 @@ class BECGuiClient(RPCBase):
window.hide()
def _update_dynamic_namespace(self, server_registry: dict):
"""Update the dynamic name space"""
for state in server_registry.values():
if state["widget_class"] in IGNORE_WIDGETS:
"""
Update the dynamic name space with the given server registry.
Setting the server registry to an empty dictionary will remove all widgets from the namespace.
Args:
server_registry (dict): The server registry
"""
top_level_widgets: dict[str, RPCReference] = {}
for gui_id, state in server_registry.items():
widget = self._add_widget(state, self)
if widget is None:
# ignore widgets that are not supported
continue
self._add_widget(state, self)
# get all top-level widgets. These are widgets that have no parent
if not state["config"].get("parent_id"):
top_level_widgets[gui_id] = widget
for widget in self._ipython_registry.values():
remove_from_registry = []
for gui_id, widget in self._ipython_registry.items():
if gui_id not in server_registry:
remove_from_registry.append(gui_id)
widget._refresh_references()
for gui_id in remove_from_registry:
self._ipython_registry.pop(gui_id)
# get all top-level widgets. These are widgets that have no parent
top_level_widgets = [
state
for state in server_registry.values()
if not state["config"].get("parent_id") and state["widget_class"] not in IGNORE_WIDGETS
removed_widgets = [
widget._name for widget in self._top_level.values() if widget._is_deleted()
]
removed_widgets = set(self._top_level.keys()) - set(
state["name"] for state in top_level_widgets
)
for widget_name in removed_widgets:
delattr(self, widget_name)
for state in top_level_widgets:
setattr(self, state["name"], self._ipython_registry[state["gui_id"]])
for gui_id, widget_ref in top_level_widgets.items():
setattr(self, widget_ref._name, widget_ref)
self._top_level = {
state["name"]: self._ipython_registry[state["gui_id"]] for state in top_level_widgets
}
self._top_level = top_level_widgets
def _add_widget(self, state: dict, parent: object) -> RPCReference | None:
"""Add a widget to the namespace

6
bec_widgets/cli/rpc/rpc_base.py
View File

@ -91,10 +91,11 @@ class RPCReference:
def __init__(self, registry: dict, gui_id: str) -> None:
self._registry = registry
self._gui_id = gui_id
self._name = self._registry[self._gui_id]._name
@check_for_deleted_widget
def __getattr__(self, name):
if name in ["_registry", "_gui_id"]:
if name in ["_registry", "_gui_id", "_is_deleted", "_name"]:
return super().__getattribute__(name)
return self._registry[self._gui_id].__getattribute__(name)
@ -117,6 +118,9 @@ class RPCReference:
return []
return self._registry[self._gui_id].__dir__()
def _is_deleted(self) -> bool:
return self._gui_id not in self._registry
class RPCBase:
def __init__(

7
bec_widgets/utils/cli_server.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import functools
import traceback
import types
from contextlib import contextmanager
from typing import TYPE_CHECKING
@ -10,7 +11,6 @@ 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 QTimer
from qtpy.QtWidgets import QApplication
from redis.exceptions import RedisError
from bec_widgets.cli.rpc.rpc_register import RPCRegister
@ -95,8 +95,9 @@ class CLIServer:
kwargs = msg["parameter"].get("kwargs", {})
res = self.run_rpc(obj, method, args, kwargs)
except Exception as e:
logger.error(f"Error while executing RPC instruction: {e}")
self.send_response(request_id, False, {"error": str(e)})
content = traceback.format_exc()
logger.error(f"Error while executing RPC instruction: {content}")
self.send_response(request_id, False, {"error": content})
else:
logger.debug(f"RPC instruction executed successfully: {res}")
self.send_response(request_id, True, {"result": res})

22
tests/end-2-end/test_bec_dock_rpc_e2e.py
View File

@ -144,7 +144,7 @@ def test_ring_bar(qtbot, connected_client_gui_obj):
def test_rpc_gui_obj(connected_client_gui_obj, qtbot):
gui = connected_client_gui_obj
assert len(gui.windows) == 1
qtbot.waitUntil(lambda: len(gui.windows) == 1, timeout=3000)
assert gui.windows["bec"] is gui.bec
mw = gui.bec
assert mw.__class__.__name__ == "RPCReference"
@ -155,22 +155,6 @@ def test_rpc_gui_obj(connected_client_gui_obj, qtbot):
assert gui._ipython_registry[xw._gui_id].__class__.__name__ == "BECDockArea"
assert len(gui.windows) == 2
gui_info = gui._dump()
mw_info = gui_info[mw._gui_id]
assert mw_info["title"] == "BEC"
assert mw_info["visible"]
xw_info = gui_info[xw._gui_id]
assert xw_info["title"] == "BEC - X"
assert xw_info["visible"]
gui.hide()
gui_info = gui._dump() #
assert not any(windows["visible"] for windows in gui_info.values())
gui.show()
gui_info = gui._dump()
assert all(windows["visible"] for windows in gui_info.values())
assert gui._gui_is_alive()
gui.kill_server()
assert not gui._gui_is_alive()
@ -186,10 +170,8 @@ def test_rpc_gui_obj(connected_client_gui_obj, qtbot):
qtbot.waitUntil(wait_for_gui_started, timeout=3000)
# gui.windows should have bec with gui_id 'bec'
assert len(gui.windows) == 1
assert gui.windows["bec"]._gui_id == mw._gui_id
# communication should work, main dock area should have same id and be visible
gui_info = gui._dump()
assert gui_info[mw._gui_id]["visible"]
yw = gui.new("Y")
assert len(gui.windows) == 2