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

feat(server,launcher)!: RPC server separated with the launcher window introduced

Browse Source
This commit is contained in:
wyzula_j
2025-03-25 20:48:19 +01:00
parent 39164feb18
commit 5f27a90989
27 changed files with 1420 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
bec_widgets/applications/bw_launch.py Normal file
View File

@ -0,0 +1,6 @@
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
def dock_area(name: str | None = None):
_dock_area = BECDockArea(name=name)
return _dock_area

35
bec_widgets/applications/launch_dialog.ui Normal file
View File

@ -0,0 +1,35 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>400</width>
<height>300</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QGridLayout" name="gridLayout">
<item row="0" column="0">
<widget class="QPushButton" name="open_dock_area">
<property name="text">
<string>PushButton</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QPushButton" name="pushButton_2">
<property name="text">
<string>PushButton</string>
</property>
</widget>
</item>
</layout>
</widget>
<resources/>
<connections/>
</ui>

171
bec_widgets/applications/launch_window.py Normal file
View File

@ -0,0 +1,171 @@
import os
from bec_lib.logger import bec_logger
from qtpy.QtCore import QSize
from qtpy.QtGui import QAction, QActionGroup
from qtpy.QtWidgets import QApplication, QMainWindow, QSizePolicy, QStyle
import bec_widgets
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
logger = bec_logger.logger
MODULE_PATH = os.path.dirname(bec_widgets.__file__)
class LaunchWindow(BECWidget, QMainWindow):
def __init__(self, gui_id: str = None, *args, **kwargs):
BECWidget.__init__(self, gui_id=gui_id, **kwargs)
QMainWindow.__init__(self, *args, **kwargs)
self.app = QApplication.instance()
self.resize(500, 300)
self.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Fixed)
self._init_ui()
def _init_ui(self):
# Set the window title
self.setWindowTitle("BEC Launcher")
# Load ui file
ui_file_path = os.path.join(MODULE_PATH, "applications/launch_dialog.ui")
self.load_ui(ui_file_path)
# Set Menu and Status bar
self._setup_menu_bar()
# BEC Specific UI
self._init_bec_specific_ui()
# TODO can be implemented for toolbar
def load_ui(self, ui_file):
loader = UILoader(self)
self.ui = loader.loader(ui_file)
self.setCentralWidget(self.ui)
self.ui.open_dock_area.setText("Open Dock Area")
self.ui.open_dock_area.clicked.connect(lambda: self.launch("dock_area"))
def _init_bec_specific_ui(self):
if getattr(self.app, "gui_id", None):
self.statusBar().showMessage(f"App ID: {self.app.gui_id}")
else:
logger.warning(
"Application is not a BECApplication instance. Status bar will not show App ID. Please initialize the application with BECApplication."
)
def list_app_hierarchy(self):
"""
List the hierarchy of the application.
"""
self.app.list_hierarchy()
def _setup_menu_bar(self):
"""
Setup the menu bar for the main window.
"""
menu_bar = self.menuBar()
########################################
# Theme menu
theme_menu = menu_bar.addMenu("Theme")
theme_group = QActionGroup(self)
light_theme_action = QAction("Light Theme", self, checkable=True)
dark_theme_action = QAction("Dark Theme", self, checkable=True)
theme_group.addAction(light_theme_action)
theme_group.addAction(dark_theme_action)
theme_group.setExclusive(True)
theme_menu.addAction(light_theme_action)
theme_menu.addAction(dark_theme_action)
# Connect theme actions
light_theme_action.triggered.connect(lambda: self.change_theme("light"))
dark_theme_action.triggered.connect(lambda: self.change_theme("dark"))
# Set the default theme
# TODO can be fetched from app
dark_theme_action.setChecked(True)
########################################
# Help menu
help_menu = menu_bar.addMenu("Help")
help_icon = QApplication.style().standardIcon(QStyle.SP_MessageBoxQuestion)
bug_icon = QApplication.style().standardIcon(QStyle.SP_MessageBoxInformation)
bec_docs = QAction("BEC Docs", self)
bec_docs.setIcon(help_icon)
widgets_docs = QAction("BEC Widgets Docs", self)
widgets_docs.setIcon(help_icon)
bug_report = QAction("Bug Report", self)
bug_report.setIcon(bug_icon)
bec_docs.triggered.connect(BECWebLinksMixin.open_bec_docs)
widgets_docs.triggered.connect(BECWebLinksMixin.open_bec_widgets_docs)
bug_report.triggered.connect(BECWebLinksMixin.open_bec_bug_report)
help_menu.addAction(bec_docs)
help_menu.addAction(widgets_docs)
help_menu.addAction(bug_report)
debug_bar = menu_bar.addMenu(f"DEBUG {self.__class__.__name__}")
list_hierarchy = QAction("List App Hierarchy", self)
list_hierarchy.triggered.connect(self.list_app_hierarchy)
debug_bar.addAction(list_hierarchy)
def change_theme(self, theme):
apply_theme(theme)
def launch(
self,
launch_script: str,
name: str | None = None,
geometry: tuple[int, int, int, int] | None = None,
) -> "BECDockArea":
"""Create a new dock area.
Args:
name(str): The name of the dock area.
geometry(tuple): The geometry parameters to be passed to the dock area.
Returns:
BECDockArea: The newly created dock area.
"""
from bec_widgets.applications.bw_launch import dock_area
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:
if name in existing_dock_areas:
raise ValueError(
f"Name {name} must be unique for dock areas, but already exists: {existing_dock_areas}."
)
else:
name = "dock_area"
name = WidgetContainerUtils.generate_unique_name(name, existing_dock_areas)
dock_area = dock_area(name) # BECDockArea(name=name)
dock_area.resize(dock_area.minimumSizeHint())
# TODO Should we simply use the specified name as title here?
dock_area.window().setWindowTitle(f"BEC - {name}")
logger.info(f"Created new dock area: {name}")
logger.info(f"Existing dock areas: {geometry}")
if geometry is not None:
dock_area.setGeometry(*geometry)
dock_area.show()
return dock_area
def show_launcher(self):
self.show()
def hide_launcher(self):
self.hide()
def cleanup(self):
super().close()

65
bec_widgets/cli/client.py
View File

@ -678,6 +678,14 @@ class DarkModeButton(RPCBase):
"""
class DemoApp(RPCBase):
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
class DeviceBrowser(RPCBase):
@rpc_call
def remove(self):
@ -3557,3 +3565,60 @@ class WebsiteWidget(RPCBase):
"""
Go forward in the history
"""
class WindowWithUi(RPCBase):
"""This is just testing app wiht UI file which could be connected to RPC."""
@rpc_call
def new_dock_area(
self, name: str | None = None, geometry: tuple[int, int, int, int] | None = None
) -> "BECDockArea":
"""
Create a new dock area.
Args:
name(str): The name of the dock area.
geometry(tuple): The geometry parameters to be passed to the dock area.
Returns:
BECDockArea: The newly created dock area.
"""
@property
@rpc_call
def all_connections(self) -> list:
"""
None
"""
@rpc_call
def change_theme(self, theme):
"""
None
"""
@property
@rpc_call
def dock_area(self):
"""
None
"""
@rpc_call
def register_all_rpc(self):
"""
None
"""
@property
@rpc_call
def widget_list(self) -> list:
"""
Return a list of all widgets in the application.
"""
@rpc_call
def list_app_hierarchy(self):
"""
List the hierarchy of the application.
"""

218
bec_widgets/cli/client_utils.py
View File

@ -10,11 +10,11 @@ import threading
import time
from contextlib import contextmanager
from threading import Lock
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Literal, TypeAlias
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from bec_lib.utils.import_utils import lazy_import_from
from rich.console import Console
from rich.table import Table
@ -28,7 +28,11 @@ else:
logger = bec_logger.logger
IGNORE_WIDGETS = ["BECDockArea", "BECDock"]
IGNORE_WIDGETS = ["LaunchWindow"]
RegistryState: TypeAlias = dict[
Literal["gui_id", "name", "widget_class", "config", "__rpc__"], str | bool | dict
]
# pylint: disable=redefined-outer-scope
@ -67,7 +71,7 @@ def _get_output(process, logger) -> None:
def _start_plot_process(
gui_id: str, gui_class: type, gui_class_id: str, config: dict | str, logger=None
gui_id: str, gui_class_id: str, config: dict | str, gui_class: str = "launcher", logger=None
) -> tuple[subprocess.Popen[str], threading.Thread | None]:
"""
Start the plot in a new process.
@ -82,7 +86,7 @@ def _start_plot_process(
"--id",
gui_id,
"--gui_class",
gui_class.__name__,
gui_class,
"--gui_class_id",
gui_class_id,
"--hide",
@ -199,21 +203,25 @@ class BECGuiClient(RPCBase):
self._auto_updates_enabled = True
self._auto_updates = None
self._killed = False
self._top_level: dict[str, client.BECDockArea] = {}
self._top_level: dict[str, RPCReference] = {}
self._startup_timeout = 0
self._gui_started_timer = None
self._gui_started_event = threading.Event()
self._process = None
self._process_output_processing_thread = None
self._exposed_widgets = []
self._server_registry = {}
self._ipython_registry = {}
self._server_registry: dict[str, RegistryState] = {}
self._ipython_registry: dict[str, RPCReference] = {}
self.available_widgets = AvailableWidgetsNamespace()
####################
#### Client API ####
####################
@property
def launcher(self) -> RPCBase:
"""The launcher object."""
return RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self, name="launcher")
def connect_to_gui_server(self, gui_id: str) -> None:
"""Connect to a GUI server"""
# Unregister the old callback
@ -221,21 +229,25 @@ class BECGuiClient(RPCBase):
MessageEndpoints.gui_registry_state(self._gui_id), cb=self._handle_registry_update
)
self._gui_id = gui_id
# Get the registry state
msgs = self._client.connector.xread(
MessageEndpoints.gui_registry_state(self._gui_id), count=1
)
if msgs:
self._handle_registry_update(msgs[0])
# reset the namespace
self._update_dynamic_namespace({})
self._server_registry = {}
self._top_level = {}
self._ipython_registry = {}
# Register the new callback
self._client.connector.register(
MessageEndpoints.gui_registry_state(self._gui_id), cb=self._handle_registry_update
MessageEndpoints.gui_registry_state(self._gui_id),
cb=self._handle_registry_update,
parent=self,
from_start=True,
)
@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:
@ -275,12 +287,12 @@ class BECGuiClient(RPCBase):
self.start(wait=True)
if wait:
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client = RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self)
widget = rpc_client._run_rpc(
"new_dock_area", name, geometry
"launch", "dock_area", name, geometry
) # pylint: disable=protected-access
return widget
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client = RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self)
widget = rpc_client._run_rpc(
"new_dock_area", name, geometry
) # pylint: disable=protected-access
@ -352,11 +364,13 @@ class BECGuiClient(RPCBase):
# Wait for 'bec' gui to be registered, this may take some time
# After 60s timeout. Should this raise an exception on timeout?
while time.time() < time.time() + timeout:
if len(list(self._server_registry.keys())) == 0:
if len(list(self._server_registry.keys())) < 2 or not hasattr(
self, self._default_dock_name
):
time.sleep(0.1)
else:
break
self._do_show_all()
self._gui_started_event.set()
def _start_server(self, wait: bool = False) -> None:
@ -369,7 +383,6 @@ class BECGuiClient(RPCBase):
self._gui_started_event.clear()
self._process, self._process_output_processing_thread = _start_plot_process(
self._gui_id,
self.__class__,
gui_class_id=self._default_dock_name,
config=self._client._service_config.config, # pylint: disable=protected-access
logger=logger,
@ -380,7 +393,7 @@ class BECGuiClient(RPCBase):
if callable(callback):
callback()
finally:
threading.current_thread().cancel()
threading.current_thread().cancel() # type: ignore
self._gui_started_timer = RepeatTimer(
0.5, lambda: self._gui_is_alive() and gui_started_callback(self._gui_post_startup)
@ -390,25 +403,25 @@ class BECGuiClient(RPCBase):
if wait:
self._gui_started_event.wait()
def _dump(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
return rpc_client._run_rpc("_dump")
def _start(self, wait: bool = False) -> None:
self._killed = False
self._client.connector.register(
MessageEndpoints.gui_registry_state(self._gui_id), cb=self._handle_registry_update
MessageEndpoints.gui_registry_state(self._gui_id),
cb=self._handle_registry_update,
parent=self,
)
return self._start_server(wait=wait)
def _handle_registry_update(self, msg: StreamMessage) -> None:
@staticmethod
def _handle_registry_update(msg: StreamMessage, parent: BECGuiClient) -> None:
# This was causing a deadlock during shutdown, not sure why.
# with self._lock:
self = parent
self._server_registry = msg["data"].state
self._update_dynamic_namespace()
self._update_dynamic_namespace(self._server_registry)
def _do_show_all(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client = RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self)
rpc_client._run_rpc("show") # pylint: disable=protected-access
for window in self._top_level.values():
window.show()
@ -419,112 +432,54 @@ class BECGuiClient(RPCBase):
def _hide_all(self):
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client = RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self)
rpc_client._run_rpc("hide") # pylint: disable=protected-access
if not self._killed:
for window in self._top_level.values():
window.hide()
def _update_dynamic_namespace(self):
"""Update the dynamic name space"""
# Clear the top level
self._top_level.clear()
# First we update the name space based on the new registry state
self._add_registry_to_namespace()
# Then we clear the ipython registry from old objects
self._cleanup_ipython_registry()
def _update_dynamic_namespace(self, server_registry: dict):
"""
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.
def _cleanup_ipython_registry(self):
"""Cleanup the ipython registry"""
names_in_registry = list(self._ipython_registry.keys())
names_in_server_state = list(self._server_registry.keys())
remove_ids = list(set(names_in_registry) - set(names_in_server_state))
for widget_id in remove_ids:
self._ipython_registry.pop(widget_id)
self._cleanup_rpc_references_on_rpc_base(remove_ids)
# Clear the exposed widgets
self._exposed_widgets.clear() # No longer needed I think
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
# 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
def _cleanup_rpc_references_on_rpc_base(self, remove_ids: list[str]) -> None:
"""Cleanup the rpc references on the RPCBase object"""
if not remove_ids:
return
for widget in self._ipython_registry.values():
to_delete = []
for attr_name, gui_id in widget._rpc_references.items():
if gui_id in remove_ids:
to_delete.append(attr_name)
for attr_name in to_delete:
if hasattr(widget, attr_name):
delattr(widget, attr_name)
if attr_name.startswith("elements."):
delattr(widget.elements, attr_name.split(".")[1])
widget._rpc_references.pop(attr_name)
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)
def _set_dynamic_attributes(self, obj: object, name: str, value: Any) -> None:
"""Add an object to the namespace"""
setattr(obj, name, value)
def _update_rpc_references(self, widget: RPCBase, name: str, gui_id: str) -> None:
"""Update the RPC references"""
widget._rpc_references[name] = gui_id
def _add_registry_to_namespace(self) -> None:
"""Add registry to namespace"""
# Add dock areas
dock_area_states = [
state
for state in self._server_registry.values()
if state["widget_class"] == "BECDockArea"
removed_widgets = [
widget._name for widget in self._top_level.values() if widget._is_deleted()
]
for state in dock_area_states:
dock_area_ref = self._add_widget(state, self)
dock_area = self._ipython_registry.get(dock_area_ref._gui_id)
if not hasattr(dock_area, "elements"):
self._set_dynamic_attributes(dock_area, "elements", WidgetNameSpace())
self._set_dynamic_attributes(self, dock_area.widget_name, dock_area_ref)
# Keep track of rpc references on RPCBase object
self._update_rpc_references(self, dock_area.widget_name, dock_area_ref._gui_id)
# Add dock_area to the top level
self._top_level[dock_area_ref.widget_name] = dock_area_ref
self._exposed_widgets.append(dock_area_ref._gui_id)
# Add docks
dock_states = [
state
for state in self._server_registry.values()
if state["config"].get("parent_id", "") == dock_area_ref._gui_id
]
for state in dock_states:
dock_ref = self._add_widget(state, dock_area)
dock = self._ipython_registry.get(dock_ref._gui_id)
self._set_dynamic_attributes(dock_area, dock_ref.widget_name, dock_ref)
# Keep track of rpc references on RPCBase object
self._update_rpc_references(dock_area, dock_ref.widget_name, dock_ref._gui_id)
# Keep track of exposed docks
self._exposed_widgets.append(dock_ref._gui_id)
for widget_name in removed_widgets:
# the check is not strictly necessary, but better safe
# than sorry; who knows what the user has done
if hasattr(self, widget_name):
delattr(self, widget_name)
# Add widgets
widget_states = [
state
for state in self._server_registry.values()
if state["config"].get("parent_id", "") == dock_ref._gui_id
]
for state in widget_states:
widget_ref = self._add_widget(state, dock)
self._set_dynamic_attributes(dock, widget_ref.widget_name, widget_ref)
self._set_dynamic_attributes(
dock_area.elements, widget_ref.widget_name, widget_ref
)
# Keep track of rpc references on RPCBase object
self._update_rpc_references(
dock_area, f"elements.{widget_ref.widget_name}", widget_ref._gui_id
)
self._update_rpc_references(dock, widget_ref.widget_name, widget_ref._gui_id)
# Keep track of exposed widgets
self._exposed_widgets.append(widget_ref._gui_id)
for gui_id, widget_ref in top_level_widgets.items():
setattr(self, widget_ref._name, widget_ref)
def _add_widget(self, state: dict, parent: object) -> RPCReference:
self._top_level = top_level_widgets
def _add_widget(self, state: dict, parent: object) -> RPCReference | None:
"""Add a widget to the namespace
Args:
@ -533,12 +488,11 @@ class BECGuiClient(RPCBase):
"""
name = state["name"]
gui_id = state["gui_id"]
try:
widget_class = getattr(client, state["widget_class"])
except AttributeError as e:
raise AttributeError(
f"Failed to find user widget {state['widget_class']} in the client - did you run bw-generate-cli to generate the plugin files?"
) from e
if state["widget_class"] in IGNORE_WIDGETS:
return
widget_class = getattr(client, state["widget_class"], None)
if widget_class is None:
return
obj = self._ipython_registry.get(gui_id)
if obj is None:
widget = widget_class(gui_id=gui_id, name=name, parent=parent)

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

@ -14,6 +14,9 @@ if TYPE_CHECKING: # pragma: no cover
from bec_lib import messages
from bec_lib.connector import MessageObject
from bec_widgets.cli.client_utils import BECGuiClient
import bec_widgets.cli.client as client
else:
client = lazy_import("bec_widgets.cli.client") # avoid circular import
@ -88,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)
@ -114,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__(
@ -152,7 +159,7 @@ class RPCBase:
return self._name
@property
def _root(self):
def _root(self) -> BECGuiClient:
"""
Get the root widget. This is the BECFigure widget that holds
the anchor gui_id.
@ -163,7 +170,7 @@ class RPCBase:
parent = parent._parent
return parent
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=3, **kwargs) -> Any:
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=300, **kwargs) -> Any:
"""
Run the RPC call.
@ -236,13 +243,14 @@ class RPCBase:
cls = getattr(client, cls)
# The namespace of the object will be updated dynamically on the client side
# Therefor it is important to check if the object is already in the registry
# Therefore it is important to check if the object is already in the registry
# If yes, we return the reference to the object, otherwise we create a new object
# pylint: disable=protected-access
if msg_result["gui_id"] in self._root._ipython_registry:
return RPCReference(self._root._ipython_registry, msg_result["gui_id"])
ret = cls(parent=self, **msg_result)
self._root._ipython_registry[ret._gui_id] = ret
self._refresh_references()
obj = RPCReference(self._root._ipython_registry, ret._gui_id)
return obj
# return ret
@ -258,3 +266,22 @@ class RPCBase:
if heart.status == messages.BECStatus.RUNNING:
return True
return False
def _refresh_references(self):
"""
Refresh the references.
"""
with self._root._lock:
references = {}
for key, val in self._root._server_registry.items():
parent_id = val["config"].get("parent_id")
if parent_id == self._gui_id:
references[key] = {"gui_id": val["config"]["gui_id"], "name": val["name"]}
removed_references = set(self._rpc_references.keys()) - set(references.keys())
for key in removed_references:
delattr(self, self._rpc_references[key]["name"])
self._rpc_references = references
for key, val in references.items():
setattr(
self, val["name"], RPCReference(self._root._ipython_registry, val["gui_id"])
)

7
bec_widgets/cli/rpc/rpc_register.py
View File

@ -1,7 +1,7 @@
from __future__ import annotations
from functools import wraps
from threading import Lock, RLock
from threading import RLock
from typing import TYPE_CHECKING, Callable
from weakref import WeakValueDictionary
@ -77,7 +77,7 @@ class RPCRegister:
self._rpc_register[rpc.gui_id] = rpc
@broadcast_update
def remove_rpc(self, rpc: str):
def remove_rpc(self, rpc: BECConnector):
"""
Remove an RPC object from the register.
@ -113,7 +113,7 @@ class RPCRegister:
return connections
def get_names_of_rpc_by_class_type(
self, cls: BECWidget | BECConnector | BECDock | BECDockArea
self, cls: type[BECWidget] | type[BECConnector] | type[BECDock] | type[BECDockArea]
) -> list[str]:
"""Get all the names of the widgets.
@ -170,6 +170,7 @@ class RPCRegisterBroadcast:
def __exit__(self, *exc):
"""Exit the context manager"""
self._call_depth -= 1 # Remove nested calls
if self._call_depth == 0: # Last one to exit is repsonsible for broadcasting
self.rpc_register._skip_broadcast = False

407
bec_widgets/cli/server.py
View File

@ -1,189 +1,27 @@
from __future__ import annotations
import functools
import argparse
import json
import os
import signal
import sys
import traceback
import types
from contextlib import contextmanager, redirect_stderr, redirect_stdout
from typing import Union
from contextlib import redirect_stderr, redirect_stdout
from typing import cast
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.service_config import ServiceConfig
from bec_lib.utils.import_utils import lazy_import
from qtpy.QtCore import Qt, QTimer
from redis.exceptions import RedisError
from qtpy.QtCore import QSize, Qt
from qtpy.QtGui import QIcon
from qtpy.QtWidgets import QApplication
import bec_widgets
from bec_widgets.applications.launch_window import LaunchWindow
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.error_popups import ErrorPopupUtility
from bec_widgets.widgets.containers.dock import BECDockArea
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
from bec_widgets.utils.bec_dispatcher import BECDispatcher
messages = lazy_import("bec_lib.messages")
logger = bec_logger.logger
@contextmanager
def rpc_exception_hook(err_func):
"""This context replaces the popup message box for error display with a specific hook"""
# get error popup utility singleton
popup = ErrorPopupUtility()
# save current setting
old_exception_hook = popup.custom_exception_hook
# install err_func, if it is a callable
# IMPORTANT, Keep self here, because this method is overwriting the custom_exception_hook
# of the ErrorPopupUtility (popup instance) class.
def custom_exception_hook(self, exc_type, value, tb, **kwargs):
err_func({"error": popup.get_error_message(exc_type, value, tb)})
popup.custom_exception_hook = types.MethodType(custom_exception_hook, popup)
try:
yield popup
finally:
# restore state of error popup utility singleton
popup.custom_exception_hook = old_exception_hook
class BECWidgetsCLIServer:
def __init__(
self,
gui_id: str,
dispatcher: BECDispatcher = None,
client=None,
config=None,
gui_class: type[BECDockArea] = BECDockArea,
gui_class_id: str = "bec",
) -> None:
self.status = messages.BECStatus.BUSY
self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher
self.client = self.dispatcher.client if client is None else client
self.client.start()
self.gui_id = gui_id
# register broadcast callback
self.rpc_register = RPCRegister()
self.rpc_register.add_callback(self.broadcast_registry_update)
self.dispatcher.connect_slot(
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
)
# Setup QTimer for heartbeat
self._heartbeat_timer = QTimer()
self._heartbeat_timer.timeout.connect(self.emit_heartbeat)
self._heartbeat_timer.start(200)
self.status = messages.BECStatus.RUNNING
with RPCRegister.delayed_broadcast():
self.gui = gui_class(parent=None, name=gui_class_id, gui_id=gui_class_id)
logger.success(f"Server started with gui_id: {self.gui_id}")
# Create initial object -> BECFigure or BECDockArea
def on_rpc_update(self, msg: dict, metadata: dict):
request_id = metadata.get("request_id")
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)
except Exception as e:
logger.error(f"Error while executing RPC instruction: {traceback.format_exc()}")
self.send_response(request_id, False, {"error": str(e)})
else:
logger.debug(f"RPC instruction executed successfully: {res}")
self.send_response(request_id, True, {"result": res})
def send_response(self, request_id: str, accepted: bool, msg: dict):
self.client.connector.set_and_publish(
MessageEndpoints.gui_instruction_response(request_id),
messages.RequestResponseMessage(accepted=accepted, message=msg),
expire=60,
)
def get_object_from_config(self, config: dict):
gui_id = config.get("gui_id")
obj = self.rpc_register.get_rpc_by_id(gui_id)
if obj is None:
raise ValueError(f"Object with gui_id {gui_id} not found")
return obj
def run_rpc(self, obj, method, args, kwargs):
# Run with rpc registry broadcast, but only once
with RPCRegister.delayed_broadcast():
logger.debug(f"Running RPC instruction: {method} with args: {args}, kwargs: {kwargs}")
method_obj = getattr(obj, method)
# check if the method accepts args and kwargs
if not callable(method_obj):
if not args:
res = method_obj
else:
setattr(obj, method, args[0])
res = None
else:
res = method_obj(*args, **kwargs)
if isinstance(res, list):
res = [self.serialize_object(obj) for obj in res]
elif isinstance(res, dict):
res = {key: self.serialize_object(val) for key, val in res.items()}
else:
res = self.serialize_object(res)
return res
def serialize_object(self, obj):
if isinstance(obj, BECConnector):
config = obj.config.model_dump()
config["parent_id"] = obj.parent_id # add parent_id to config
return {
"gui_id": obj.gui_id,
"name": (
obj._name if hasattr(obj, "_name") else obj.__class__.__name__
), # pylint: disable=protected-access
"widget_class": obj.__class__.__name__,
"config": config,
"__rpc__": True,
}
return obj
def emit_heartbeat(self):
logger.trace(f"Emitting heartbeat for {self.gui_id}")
try:
self.client.connector.set(
MessageEndpoints.gui_heartbeat(self.gui_id),
messages.StatusMessage(name=self.gui_id, status=self.status, info={}),
expire=10,
)
except RedisError as exc:
logger.error(f"Error while emitting heartbeat: {exc}")
def broadcast_registry_update(self, connections: dict):
"""
Broadcast the updated registry to all clients.
"""
# We only need to broadcast the dock areas
data = {key: self.serialize_object(val) for key, val in connections.items()}
self.client.connector.xadd(
MessageEndpoints.gui_registry_state(self.gui_id),
msg_dict={"data": messages.GUIRegistryStateMessage(state=data)},
max_size=1, # only single message in stream
)
def shutdown(self): # TODO not sure if needed when cleanup is done at level of BECConnector
self.status = messages.BECStatus.IDLE
self._heartbeat_timer.stop()
self.emit_heartbeat()
logger.info("Succeded in shutting down gui")
self.client.shutdown()
MODULE_PATH = os.path.dirname(bec_widgets.__file__)
class SimpleFileLikeFromLogOutputFunc:
@ -204,40 +42,136 @@ class SimpleFileLikeFromLogOutputFunc:
return
def _start_server(
gui_id: str, gui_class: BECDockArea, gui_class_id: str = "bec", config: str | None = None
):
if config:
try:
config = json.loads(config)
service_config = ServiceConfig(config=config)
except (json.JSONDecodeError, TypeError):
service_config = ServiceConfig(config_path=config)
else:
# if no config is provided, use the default config
service_config = ServiceConfig()
class GUIServer:
"""
This class is used to start the BEC GUI and is the main entry point for launching BEC Widgets in a subprocess.
"""
# bec_logger.configure(
# service_config.redis,
# QtRedisConnector,
# service_name="BECWidgetsCLIServer",
# service_config=service_config.service_config,
# )
server = BECWidgetsCLIServer(
gui_id=gui_id, config=service_config, gui_class=gui_class, gui_class_id=gui_class_id
)
return server
def __init__(self, args):
self.config = args.config
self.gui_id = args.id
self.gui_class = args.gui_class
self.gui_class_id = args.gui_class_id
self.hide = args.hide
self.app: QApplication | None = None
self.launcher_window: LaunchWindow | None = None
self.dispatcher: BECDispatcher | None = None
def start(self):
"""
Start the GUI server.
"""
bec_logger.level = bec_logger.LOGLEVEL.INFO
if self.hide:
# pylint: disable=protected-access
bec_logger._stderr_log_level = bec_logger.LOGLEVEL.ERROR
bec_logger._update_sinks()
with redirect_stdout(SimpleFileLikeFromLogOutputFunc(logger.info)): # type: ignore
with redirect_stderr(SimpleFileLikeFromLogOutputFunc(logger.error)): # type: ignore
self._run()
def _get_service_config(self) -> ServiceConfig:
if self.config:
try:
config = json.loads(self.config)
service_config = ServiceConfig(config=config)
except (json.JSONDecodeError, TypeError):
service_config = ServiceConfig(config_path=config)
else:
# if no config is provided, use the default config
service_config = ServiceConfig()
return service_config
def _turn_off_the_lights(self, connections: dict):
"""
If there is only one connection remaining, it is the launcher, so we show it.
Once the launcher is closed as the last window, we quit the application.
"""
self.launcher_window = cast(LaunchWindow, self.launcher_window)
if len(connections) <= 1:
self.launcher_window.show()
self.launcher_window.activateWindow()
self.launcher_window.raise_()
if self.app:
self.app.setQuitOnLastWindowClosed(True)
else:
self.launcher_window.hide()
if self.app:
self.app.setQuitOnLastWindowClosed(False)
def _run(self):
"""
Run the GUI server.
"""
self.app = QApplication(sys.argv)
self.app.setApplicationName("BEC")
self.app.gui_id = self.gui_id # type: ignore
self.setup_bec_icon()
service_config = self._get_service_config()
self.dispatcher = BECDispatcher(config=service_config)
self.dispatcher.start_cli_server(gui_id=self.gui_id)
self.launcher_window = LaunchWindow(gui_id=f"{self.gui_id}:launcher")
self.launcher_window.setAttribute(Qt.WA_ShowWithoutActivating) # type: ignore
self.app.aboutToQuit.connect(self.shutdown)
self.app.setQuitOnLastWindowClosed(False)
register = RPCRegister()
register.callbacks.append(self._turn_off_the_lights)
register.broadcast()
if self.gui_class:
# If the server is started with a specific gui class, we launch it.
# This will automatically hide the launcher.
self.launcher_window.launch(self.gui_class, name=self.gui_class_id)
def sigint_handler(*args):
# display message, for people to let it terminate gracefully
print("Caught SIGINT, exiting")
# Widgets should be all closed.
with RPCRegister.delayed_broadcast():
for widget in QApplication.instance().topLevelWidgets(): # type: ignore
widget.close()
if self.app:
self.app.quit()
# gui.bec.close()
# win.shutdown()
signal.signal(signal.SIGINT, sigint_handler)
signal.signal(signal.SIGTERM, sigint_handler)
sys.exit(self.app.exec())
def setup_bec_icon(self):
"""
Set the BEC icon for the application
"""
if self.app is None:
return
icon = QIcon()
icon.addFile(
os.path.join(MODULE_PATH, "assets", "app_icons", "bec_widgets_icon.png"),
size=QSize(48, 48),
)
self.app.setWindowIcon(icon)
def shutdown(self):
"""
Shutdown the GUI server.
"""
if self.dispatcher:
self.dispatcher.stop_cli_server()
self.dispatcher.disconnect_all()
def main():
import argparse
import os
from qtpy.QtCore import QSize
from qtpy.QtGui import QIcon
from qtpy.QtWidgets import QApplication
import bec_widgets
"""
Main entry point for subprocesses that start a GUI server.
"""
parser = argparse.ArgumentParser(description="BEC Widgets CLI Server")
parser.add_argument("--id", type=str, default="test", help="The id of the server")
@ -257,69 +191,12 @@ def main():
args = parser.parse_args()
bec_logger.level = bec_logger.LOGLEVEL.INFO
if args.hide:
# pylint: disable=protected-access
bec_logger._stderr_log_level = bec_logger.LOGLEVEL.ERROR
bec_logger._update_sinks()
if args.gui_class == "BECDockArea":
gui_class = BECDockArea
else:
print(
"Please specify a valid gui_class to run. Use -h for help."
"\n Starting with default gui_class BECFigure."
)
gui_class = BECDockArea
with redirect_stdout(SimpleFileLikeFromLogOutputFunc(logger.info)):
with redirect_stderr(SimpleFileLikeFromLogOutputFunc(logger.error)):
app = QApplication(sys.argv)
# set close on last window, only if not under control of client ;
# indeed, Qt considers a hidden window a closed window, so if all windows
# are hidden by default it exits
app.setQuitOnLastWindowClosed(not args.hide)
module_path = os.path.dirname(bec_widgets.__file__)
icon = QIcon()
icon.addFile(
os.path.join(module_path, "assets", "app_icons", "bec_widgets_icon.png"),
size=QSize(48, 48),
)
app.setWindowIcon(icon)
# store gui id within QApplication object, to make it available to all widgets
app.gui_id = args.id
# args.id = "abff6"
server = _start_server(args.id, gui_class, args.gui_class_id, args.config)
win = BECMainWindow(gui_id=f"{server.gui_id}:window")
win.setAttribute(Qt.WA_ShowWithoutActivating)
win.setWindowTitle("BEC")
RPCRegister().add_rpc(win)
gui = server.gui
win.setCentralWidget(gui)
if not args.hide:
win.show()
app.aboutToQuit.connect(server.shutdown)
def sigint_handler(*args):
# display message, for people to let it terminate gracefully
print("Caught SIGINT, exiting")
# Widgets should be all closed.
with RPCRegister.delayed_broadcast():
for widget in QApplication.instance().topLevelWidgets():
widget.close()
app.quit()
# gui.bec.close()
# win.shutdown()
signal.signal(signal.SIGINT, sigint_handler)
signal.signal(signal.SIGTERM, sigint_handler)
sys.exit(app.exec())
server = GUIServer(args)
server.start()
if __name__ == "__main__":
# import sys
# sys.argv = ["bec_widgets", "--gui_class", "MainWindow"]
main()

58
bec_widgets/utils/bec_dispatcher.py
View File

@ -1,6 +1,8 @@
from __future__ import annotations
import collections
import random
import string
from collections.abc import Callable
from typing import TYPE_CHECKING, Union
@ -17,6 +19,8 @@ logger = bec_logger.logger
if TYPE_CHECKING:
from bec_lib.endpoints import EndpointInfo
from bec_widgets.utils.cli_server import CLIServer
class QtThreadSafeCallback(QObject):
cb_signal = pyqtSignal(dict, dict)
@ -73,14 +77,16 @@ class BECDispatcher:
_instance = None
_initialized = False
client: BECClient
cli_server: CLIServer | None = None
def __new__(cls, client=None, config: str = None, *args, **kwargs):
def __new__(cls, client=None, config: str | ServiceConfig | None = None, *args, **kwargs):
if cls._instance is None:
cls._instance = super(BECDispatcher, cls).__new__(cls)
cls._initialized = False
return cls._instance
def __init__(self, client=None, config: str | ServiceConfig = None):
def __init__(self, client=None, config: str | ServiceConfig | None = None):
if self._initialized:
return
@ -112,6 +118,9 @@ class BECDispatcher:
@classmethod
def reset_singleton(cls):
"""
Reset the singleton instance of the BECDispatcher.
"""
cls._instance = None
cls._initialized = False
@ -178,4 +187,49 @@ class BECDispatcher:
*args: Arbitrary positional arguments
**kwargs: Arbitrary keyword arguments
"""
# pylint: disable=protected-access
self.disconnect_topics(self.client.connector._topics_cb)
def start_cli_server(self, gui_id: str | None = None):
"""
Start the CLI server.
Args:
gui_id(str, optional): The GUI ID. Defaults to None. If None, a unique identifier will be generated.
"""
# pylint: disable=import-outside-toplevel
from bec_widgets.utils.cli_server import CLIServer
if gui_id is None:
gui_id = self.generate_unique_identifier()
if not self.client.started:
logger.error("Cannot start CLI server without a running client")
return
self.cli_server = CLIServer(gui_id, dispatcher=self, client=self.client)
logger.success("Started CLI server with gui_id: {gui_id}")
def stop_cli_server(self):
"""
Stop the CLI server.
"""
if self.cli_server is None:
logger.error("Cannot stop CLI server without starting it first")
return
self.cli_server.shutdown()
self.cli_server = None
logger.success("Stopped CLI server")
@staticmethod
def generate_unique_identifier(length: int = 4) -> str:
"""
Generate a unique identifier for the application.
Args:
length: The length of the identifier. Defaults to 4.
Returns:
str: The unique identifier.
"""
allowed_chars = string.ascii_lowercase + string.digits
return "".join(random.choices(allowed_chars, k=length))

10
bec_widgets/utils/bec_widget.py
View File

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import darkdetect
from bec_lib.logger import bec_logger
@ -56,7 +56,6 @@ class BECWidget(BECConnector):
"""
if not isinstance(self, QWidget):
raise RuntimeError(f"{repr(self)} is not a subclass of QWidget")
super().__init__(
client=client,
config=config,
@ -78,6 +77,13 @@ class BECWidget(BECConnector):
logger.debug(f"Subscribing to theme updates for {self.__class__.__name__}")
self._connect_to_theme_change()
def _ensure_bec_app(self):
# pylint: disable=import-outside-toplevel
from bec_widgets.utils.bec_qapp import BECApplication
app = BECApplication.from_qapplication()
return app
def _connect_to_theme_change(self):
"""Connect to the theme change signal."""
qapp = QApplication.instance()

187
bec_widgets/utils/cli_server.py Normal file
View File

@ -0,0 +1,187 @@
from __future__ import annotations
import functools
import traceback
import types
from contextlib import contextmanager
from typing import TYPE_CHECKING
from bec_lib.client import BECClient
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 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.error_popups import ErrorPopupUtility
if TYPE_CHECKING:
from bec_lib import messages
else:
messages = lazy_import("bec_lib.messages")
logger = bec_logger.logger
@contextmanager
def rpc_exception_hook(err_func):
"""This context replaces the popup message box for error display with a specific hook"""
# get error popup utility singleton
popup = ErrorPopupUtility()
# save current setting
old_exception_hook = popup.custom_exception_hook
# install err_func, if it is a callable
# IMPORTANT, Keep self here, because this method is overwriting the custom_exception_hook
# of the ErrorPopupUtility (popup instance) class.
def custom_exception_hook(self, exc_type, value, tb, **kwargs):
err_func({"error": popup.get_error_message(exc_type, value, tb)})
popup.custom_exception_hook = types.MethodType(custom_exception_hook, popup)
try:
yield popup
finally:
# restore state of error popup utility singleton
popup.custom_exception_hook = old_exception_hook
class CLIServer:
client: BECClient
def __init__(
self,
gui_id: str,
dispatcher: BECDispatcher | None = None,
client: BECClient | None = None,
config=None,
gui_class_id: str = "bec",
) -> None:
self.status = messages.BECStatus.BUSY
self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher
self.client = self.dispatcher.client if client is None else client
self.client.start()
self.gui_id = gui_id
# register broadcast callback
self.rpc_register = RPCRegister()
self.rpc_register.add_callback(self.broadcast_registry_update)
self.dispatcher.connect_slot(
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
)
# Setup QTimer for heartbeat
self._heartbeat_timer = QTimer()
self._heartbeat_timer.timeout.connect(self.emit_heartbeat)
self._heartbeat_timer.start(200)
self.status = messages.BECStatus.RUNNING
logger.success(f"Server started with gui_id: {self.gui_id}")
def on_rpc_update(self, msg: dict, metadata: dict):
request_id = metadata.get("request_id")
if request_id is None:
logger.error("Received RPC instruction without request_id")
return
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)
except Exception:
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})
def send_response(self, request_id: str, accepted: bool, msg: dict):
self.client.connector.set_and_publish(
MessageEndpoints.gui_instruction_response(request_id),
messages.RequestResponseMessage(accepted=accepted, message=msg),
expire=60,
)
def get_object_from_config(self, config: dict):
gui_id = config.get("gui_id")
obj = self.rpc_register.get_rpc_by_id(gui_id)
if obj is None:
raise ValueError(f"Object with gui_id {gui_id} not found")
return obj
def run_rpc(self, obj, method, args, kwargs):
# Run with rpc registry broadcast, but only once
with RPCRegister.delayed_broadcast():
logger.debug(f"Running RPC instruction: {method} with args: {args}, kwargs: {kwargs}")
method_obj = getattr(obj, method)
# check if the method accepts args and kwargs
if not callable(method_obj):
if not args:
res = method_obj
else:
setattr(obj, method, args[0])
res = None
else:
res = method_obj(*args, **kwargs)
if isinstance(res, list):
res = [self.serialize_object(obj) for obj in res]
elif isinstance(res, dict):
res = {key: self.serialize_object(val) for key, val in res.items()}
else:
res = self.serialize_object(res)
return res
def serialize_object(self, obj):
if isinstance(obj, BECConnector):
config = obj.config.model_dump()
config["parent_id"] = obj.parent_id # add parent_id to config
return {
"gui_id": obj.gui_id,
"name": (
obj._name if hasattr(obj, "_name") else obj.__class__.__name__
), # pylint: disable=protected-access
"widget_class": obj.__class__.__name__,
"config": config,
"__rpc__": True,
}
return obj
def emit_heartbeat(self):
logger.trace(f"Emitting heartbeat for {self.gui_id}")
try:
self.client.connector.set(
MessageEndpoints.gui_heartbeat(self.gui_id),
messages.StatusMessage(name=self.gui_id, status=self.status, info={}),
expire=10,
)
except RedisError as exc:
logger.error(f"Error while emitting heartbeat: {exc}")
def broadcast_registry_update(self, connections: dict):
"""
Broadcast the updated registry to all clients.
"""
# We only need to broadcast the dock areas
data = {key: self.serialize_object(val) for key, val in connections.items()}
logger.info(f"Broadcasting registry update: {data} for {self.gui_id}")
self.client.connector.xadd(
MessageEndpoints.gui_registry_state(self.gui_id),
msg_dict={"data": messages.GUIRegistryStateMessage(state=data)},
max_size=1, # only single message in stream
)
def shutdown(self): # TODO not sure if needed when cleanup is done at level of BECConnector
self.status = messages.BECStatus.IDLE
self._heartbeat_timer.stop()
self.emit_heartbeat()
logger.info("Succeded in shutting down CLI server")
self.client.shutdown()

2
bec_widgets/utils/toolbar.py
View File

@ -858,7 +858,7 @@ class MainWindow(QMainWindow): # pragma: no cover
# For theme testing
self.dark_button = DarkModeButton(toolbar=True)
self.dark_button = DarkModeButton(parent=self, toolbar=True)
dark_mode_action = WidgetAction(label=None, widget=self.dark_button)
self.toolbar.add_action("dark_mode", dark_mode_action, self)

20
bec_widgets/utils/ui_loader.py
View File

@ -1,11 +1,14 @@
import os
import inspect
from bec_lib.logger import bec_logger
from qtpy import PYQT6, PYSIDE6, QT_VERSION
from qtpy.QtCore import QFile, QIODevice
from bec_widgets.utils.generate_designer_plugin import DesignerPluginInfo
from bec_widgets.utils.plugin_utils import get_custom_classes
logger = bec_logger.logger
if PYSIDE6:
from PySide6.QtUiTools import QUiLoader
@ -18,10 +21,19 @@ if PYSIDE6:
def createWidget(self, class_name, parent=None, name=""):
if class_name in self.custom_widgets:
widget = self.custom_widgets[class_name](parent)
# check if the custom widget has a parent_id argument
if "parent_id" in inspect.signature(self.custom_widgets[class_name]).parameters:
gui_id = getattr(self.baseinstance, "gui_id", None)
widget = self.custom_widgets[class_name](self.baseinstance, parent_id=gui_id)
else:
logger.warning(
f"Custom widget {class_name} does not have a parent_id argument. "
)
widget = self.custom_widgets[class_name](self.baseinstance)
widget.setObjectName(name)
return widget
return super().createWidget(class_name, parent, name)
return super().createWidget(class_name, self.baseinstance, name)
class UILoader:
@ -51,7 +63,7 @@ class UILoader:
Returns:
QWidget: The loaded widget.
"""
parent = parent or self.parent
loader = CustomUiLoader(parent, self.custom_widgets)
file = QFile(ui_file)
if not file.open(QIODevice.ReadOnly):

63
bec_widgets/utils/widget_io.py
View File

@ -18,6 +18,7 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
@ -275,39 +276,81 @@ class WidgetHierarchy:
grab_values: bool = False,
prefix: str = "",
exclude_internal_widgets: bool = True,
only_bec_widgets: bool = False,
show_parent: bool = True,
) -> None:
"""
Print the widget hierarchy to the console.
Args:
widget: Widget to print the hierarchy of
widget: Widget to print the hierarchy of.
indent(int, optional): Level of indentation.
grab_values(bool,optional): Whether to grab the values of the widgets.
prefix(stc,optional): Custom string prefix for indentation.
prefix(str,optional): Custom string prefix for indentation.
exclude_internal_widgets(bool,optional): Whether to exclude internal widgets (e.g. QComboBox in PyQt6).
only_bec_widgets(bool, optional): Whether to print only widgets that are instances of BECWidget.
show_parent(bool, optional): Whether to display which BECWidget is the parent of each discovered BECWidget.
"""
widget_info = f"{widget.__class__.__name__} ({widget.objectName()})"
if grab_values:
value = WidgetIO.get_value(widget, ignore_errors=True)
value_str = f" [value: {value}]" if value is not None else ""
widget_info += value_str
# Decide if this particular widget is to be printed
is_bec = isinstance(widget, BECWidget)
print_this = (not only_bec_widgets) or is_bec
print(prefix + widget_info)
# If it is a BECWidget and we're showing the parent, climb the chain to find the nearest BECWidget ancestor
if show_parent and is_bec:
ancestor = WidgetHierarchy._get_becwidget_ancestor(widget)
if ancestor is not None:
parent_info = f" parent={ancestor.__class__.__name__}"
else:
parent_info = " parent=None"
else:
parent_info = ""
if print_this:
widget_info = f"{widget.__class__.__name__} ({widget.objectName()}){parent_info}"
if grab_values:
value = WidgetIO.get_value(widget, ignore_errors=True)
value_str = f" [value: {value}]" if value is not None else ""
widget_info += value_str
print(prefix + widget_info)
# Always recurse so we can discover deeper BECWidgets even if the current widget is not a BECWidget
children = widget.children()
for child in children:
for i, child in enumerate(children):
# Possibly skip known internal child widgets of a QComboBox
if (
exclude_internal_widgets
and isinstance(widget, QComboBox)
and child.__class__.__name__ in ["QFrame", "QBoxLayout", "QListView"]
):
continue
child_prefix = prefix + " "
arrow = "├─ " if child != children[-1] else "└─ "
# Regardless of whether child is BECWidget or not, keep recursing, or we might miss deeper BECWidgets
WidgetHierarchy.print_widget_hierarchy(
child, indent + 1, grab_values, prefix=child_prefix + arrow
child,
indent + 1,
grab_values=grab_values,
prefix=child_prefix + arrow,
exclude_internal_widgets=exclude_internal_widgets,
only_bec_widgets=only_bec_widgets,
show_parent=show_parent,
)
@staticmethod
def _get_becwidget_ancestor(widget):
"""
Climb the parent chain to find the nearest BECWidget above this widget.
Returns None if none is found.
"""
parent = widget.parent()
while parent is not None:
if isinstance(parent, BECWidget):
return parent
parent = parent.parent()
return None
@staticmethod
def export_config_to_dict(
widget: QWidget,

6
bec_widgets/widgets/containers/dock/dock_area.py
View File

@ -91,8 +91,10 @@ class BECDockArea(BECWidget, QWidget):
self._instructions_visible = True
self.dark_mode_button = DarkModeButton(parent=self, parent_id=self.gui_id, toolbar=True)
self.dock_area = DockArea()
self.toolbar = ModularToolBar(
parent=self,
actions={
"menu_plots": ExpandableMenuAction(
label="Add Plot ",
@ -172,7 +174,7 @@ class BECDockArea(BECWidget, QWidget):
self.spacer = QWidget()
self.spacer.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
self.toolbar.addWidget(self.spacer)
self.toolbar.addWidget(DarkModeButton(toolbar=True))
self.toolbar.addWidget(self.dark_mode_button)
self._hook_toolbar()
def minimumSizeHint(self):
@ -432,6 +434,8 @@ class BECDockArea(BECWidget, QWidget):
self.delete_all()
self.toolbar.close()
self.toolbar.deleteLater()
self.dark_mode_button.close()
self.dark_mode_button.deleteLater()
self.dock_area.close()
self.dock_area.deleteLater()
super().cleanup()

0
bec_widgets/widgets/containers/main_window/addons/__init__.py Normal file
View File

15
bec_widgets/widgets/containers/main_window/addons/web_links.py Normal file
View File

@ -0,0 +1,15 @@
import webbrowser
class BECWebLinksMixin:
@staticmethod
def open_bec_docs():
webbrowser.open("https://beamline-experiment-control.readthedocs.io/en/latest/")
@staticmethod
def open_bec_widgets_docs():
webbrowser.open("https://bec.readthedocs.io/projects/bec-widgets/en/latest/")
@staticmethod
def open_bec_bug_report():
webbrowser.open("https://gitlab.psi.ch/groups/bec/-/issues/")

42
bec_widgets/widgets/containers/main_window/example_app.ui Normal file
View File

@ -0,0 +1,42 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>824</width>
<height>1234</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="Waveform" name="waveform"/>
</item>
<item>
<widget class="BECDockArea" name="dock_area_2"/>
</item>
<item>
<widget class="BECDockArea" name="dock_area"/>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>BECDockArea</class>
<extends>QWidget</extends>
<header>dock_area</header>
</customwidget>
<customwidget>
<class>Waveform</class>
<extends>QWidget</extends>
<header>waveform</header>
</customwidget>
</customwidgets>
<resources/>
<connections/>
</ui>

262
bec_widgets/widgets/containers/main_window/general_app.ui Normal file
View File

@ -0,0 +1,262 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>MainWindow</class>
<widget class="QMainWindow" name="MainWindow">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>1718</width>
<height>1139</height>
</rect>
</property>
<property name="windowTitle">
<string>MainWindow</string>
</property>
<property name="tabShape">
<enum>QTabWidget::TabShape::Rounded</enum>
</property>
<widget class="QWidget" name="centralwidget">
<layout class="QVBoxLayout" name="verticalLayout_3">
<item>
<widget class="QTabWidget" name="central_tab">
<property name="currentIndex">
<number>0</number>
</property>
<widget class="QWidget" name="dock_area_tab">
<attribute name="title">
<string>Dock Area</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout">
<property name="leftMargin">
<number>2</number>
</property>
<property name="topMargin">
<number>1</number>
</property>
<property name="rightMargin">
<number>2</number>
</property>
<property name="bottomMargin">
<number>2</number>
</property>
<item>
<widget class="BECDockArea" name="dock_area"/>
</item>
</layout>
</widget>
<widget class="QWidget" name="vscode_tab">
<attribute name="icon">
<iconset theme="QIcon::ThemeIcon::Computer"/>
</attribute>
<attribute name="title">
<string>Visual Studio Code</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout_2">
<property name="leftMargin">
<number>2</number>
</property>
<property name="topMargin">
<number>1</number>
</property>
<property name="rightMargin">
<number>2</number>
</property>
<property name="bottomMargin">
<number>2</number>
</property>
<item>
<widget class="VSCodeEditor" name="vscode"/>
</item>
</layout>
</widget>
</widget>
</item>
</layout>
</widget>
<widget class="QMenuBar" name="menubar">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>1718</width>
<height>31</height>
</rect>
</property>
<widget class="QMenu" name="menuHelp">
<property name="title">
<string>Help</string>
</property>
<addaction name="action_BEC_docs"/>
<addaction name="action_BEC_widgets_docs"/>
<addaction name="action_bug_report"/>
</widget>
<widget class="QMenu" name="menuTheme">
<property name="title">
<string>Theme</string>
</property>
<addaction name="action_light"/>
<addaction name="action_dark"/>
</widget>
<addaction name="menuTheme"/>
<addaction name="menuHelp"/>
</widget>
<widget class="QStatusBar" name="statusbar"/>
<widget class="QDockWidget" name="dock_scan_control">
<property name="windowTitle">
<string>Scan Control</string>
</property>
<attribute name="dockWidgetArea">
<number>2</number>
</attribute>
<widget class="QWidget" name="dockWidgetContents_2">
<layout class="QVBoxLayout" name="verticalLayout_4">
<item>
<widget class="ScanControl" name="scan_control"/>
</item>
</layout>
</widget>
</widget>
<widget class="QDockWidget" name="dock_status_2">
<property name="windowTitle">
<string>BEC Service Status</string>
</property>
<attribute name="dockWidgetArea">
<number>2</number>
</attribute>
<widget class="QWidget" name="dockWidgetContents_3">
<layout class="QVBoxLayout" name="verticalLayout_5">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
<widget class="BECStatusBox" name="bec_status_box_2"/>
</item>
</layout>
</widget>
</widget>
<widget class="QDockWidget" name="dock_queue">
<property name="windowTitle">
<string>Scan Queue</string>
</property>
<attribute name="dockWidgetArea">
<number>2</number>
</attribute>
<widget class="QWidget" name="dockWidgetContents_4">
<layout class="QVBoxLayout" name="verticalLayout_6">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
<widget class="BECQueue" name="bec_queue">
<row/>
<column/>
<column/>
<column/>
<item row="0" column="0"/>
<item row="0" column="1"/>
<item row="0" column="2"/>
</widget>
</item>
</layout>
</widget>
</widget>
<action name="action_BEC_docs">
<property name="icon">
<iconset theme="QIcon::ThemeIcon::DialogQuestion"/>
</property>
<property name="text">
<string>BEC Docs</string>
</property>
</action>
<action name="action_BEC_widgets_docs">
<property name="icon">
<iconset theme="QIcon::ThemeIcon::DialogQuestion"/>
</property>
<property name="text">
<string>BEC Widgets Docs</string>
</property>
</action>
<action name="action_bug_report">
<property name="icon">
<iconset theme="QIcon::ThemeIcon::DialogError"/>
</property>
<property name="text">
<string>Bug Report</string>
</property>
</action>
<action name="action_light">
<property name="checkable">
<bool>true</bool>
</property>
<property name="text">
<string>Light</string>
</property>
</action>
<action name="action_dark">
<property name="checkable">
<bool>true</bool>
</property>
<property name="text">
<string>Dark</string>
</property>
</action>
</widget>
<customwidgets>
<customwidget>
<class>WebsiteWidget</class>
<extends>QWebEngineView</extends>
<header>website_widget</header>
</customwidget>
<customwidget>
<class>BECQueue</class>
<extends>QTableWidget</extends>
<header>bec_queue</header>
</customwidget>
<customwidget>
<class>ScanControl</class>
<extends>QWidget</extends>
<header>scan_control</header>
</customwidget>
<customwidget>
<class>VSCodeEditor</class>
<extends>WebsiteWidget</extends>
<header>vs_code_editor</header>
</customwidget>
<customwidget>
<class>BECStatusBox</class>
<extends>QWidget</extends>
<header>bec_status_box</header>
</customwidget>
<customwidget>
<class>BECDockArea</class>
<extends>QWidget</extends>
<header>dock_area</header>
</customwidget>
<customwidget>
<class>QWebEngineView</class>
<extends></extends>
<header location="global">QtWebEngineWidgets/QWebEngineView</header>
</customwidget>
</customwidgets>
<resources/>
<connections/>
</ui>

184
bec_widgets/widgets/containers/main_window/main_window.py
View File

@ -1,10 +1,20 @@
import os
from typing import TYPE_CHECKING
from bec_lib.logger import bec_logger
from qtpy.QtWidgets import QApplication, QMainWindow
from qtpy.QtGui import QAction, QActionGroup
from qtpy.QtWidgets import QApplication, QMainWindow, QStyle
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_qapp import BECApplication
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
if TYPE_CHECKING:
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
logger = bec_logger.logger
@ -14,6 +24,102 @@ class BECMainWindow(BECWidget, QMainWindow):
BECWidget.__init__(self, gui_id=gui_id, **kwargs)
QMainWindow.__init__(self, *args, **kwargs)
self.app = QApplication.instance()
# self._upgrade_qapp() #TODO consider to make upgrade function to any QApplication to BECQApplication
self._init_ui()
def _init_ui(self):
# Set the window title
self.setWindowTitle("BEC")
# Set Menu and Status bar
self._setup_menu_bar()
# BEC Specific UI
self._init_bec_specific_ui()
# self.ui = UILoader
# ui_file_path = os.path.join(os.path.dirname(__file__), "general_app.ui")
# self.load_ui(ui_file_path)
# TODO can be implemented for toolbar
def load_ui(self, ui_file):
loader = UILoader(self)
self.ui = loader.loader(ui_file)
self.setCentralWidget(self.ui)
def _init_bec_specific_ui(self):
if getattr(self.app, "is_bec_app", False):
self.statusBar().showMessage(f"App ID: {self.app.gui_id}")
else:
logger.warning(
"Application is not a BECApplication instance. Status bar will not show App ID. Please initialize the application with BECApplication."
)
def list_app_hierarchy(self):
"""
List the hierarchy of the application.
"""
self.app.list_hierarchy()
def _setup_menu_bar(self):
"""
Setup the menu bar for the main window.
"""
menu_bar = self.menuBar()
########################################
# Theme menu
theme_menu = menu_bar.addMenu("Theme")
theme_group = QActionGroup(self)
light_theme_action = QAction("Light Theme", self, checkable=True)
dark_theme_action = QAction("Dark Theme", self, checkable=True)
theme_group.addAction(light_theme_action)
theme_group.addAction(dark_theme_action)
theme_group.setExclusive(True)
theme_menu.addAction(light_theme_action)
theme_menu.addAction(dark_theme_action)
# Connect theme actions
light_theme_action.triggered.connect(lambda: self.change_theme("light"))
dark_theme_action.triggered.connect(lambda: self.change_theme("dark"))
# Set the default theme
# TODO can be fetched from app
dark_theme_action.setChecked(True)
########################################
# Help menu
help_menu = menu_bar.addMenu("Help")
help_icon = QApplication.style().standardIcon(QStyle.SP_MessageBoxQuestion)
bug_icon = QApplication.style().standardIcon(QStyle.SP_MessageBoxInformation)
bec_docs = QAction("BEC Docs", self)
bec_docs.setIcon(help_icon)
widgets_docs = QAction("BEC Widgets Docs", self)
widgets_docs.setIcon(help_icon)
bug_report = QAction("Bug Report", self)
bug_report.setIcon(bug_icon)
bec_docs.triggered.connect(BECWebLinksMixin.open_bec_docs)
widgets_docs.triggered.connect(BECWebLinksMixin.open_bec_widgets_docs)
bug_report.triggered.connect(BECWebLinksMixin.open_bec_bug_report)
help_menu.addAction(bec_docs)
help_menu.addAction(widgets_docs)
help_menu.addAction(bug_report)
debug_bar = menu_bar.addMenu(f"DEBUG {self.__class__.__name__}")
list_hierarchy = QAction("List App Hierarchy", self)
list_hierarchy.triggered.connect(self.list_app_hierarchy)
debug_bar.addAction(list_hierarchy)
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);
@ -40,7 +146,7 @@ class BECMainWindow(BECWidget, QMainWindow):
def new_dock_area(
self, name: str | None = None, geometry: tuple[int, int, int, int] | None = None
) -> BECDockArea:
) -> "BECDockArea":
"""Create a new dock area.
Args:
@ -49,6 +155,8 @@ class BECMainWindow(BECWidget, QMainWindow):
Returns:
BECDockArea: The newly created dock area.
"""
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
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:
@ -59,7 +167,7 @@ class BECMainWindow(BECWidget, QMainWindow):
else:
name = "dock_area"
name = WidgetContainerUtils.generate_unique_name(name, existing_dock_areas)
dock_area = BECDockArea(name=name)
dock_area = WindowWithUi() # BECDockArea(name=name)
dock_area.resize(dock_area.minimumSizeHint())
# TODO Should we simply use the specified name as title here?
dock_area.window().setWindowTitle(f"BEC - {name}")
@ -72,3 +180,71 @@ class BECMainWindow(BECWidget, QMainWindow):
def cleanup(self):
super().close()
class WindowWithUi(BECMainWindow):
"""
This is just testing app wiht UI file which could be connected to RPC.
"""
USER_ACCESS = [
"new_dock_area",
"all_connections",
"change_theme",
"dock_area",
"register_all_rpc",
"widget_list",
"list_app_hierarchy",
]
def __init__(self, *args, name: str = None, **kwargs):
super().__init__(*args, **kwargs)
if name is None:
name = self.__class__.__name__
else:
if not WidgetContainerUtils.has_name_valid_chars(name):
raise ValueError(f"Name {name} contains invalid characters.")
self._name = name if name else self.__class__.__name__
ui_file_path = os.path.join(os.path.dirname(__file__), "example_app.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)
# TODO actually these propertiers are not much exposed now in the real CLI
@property
def dock_area(self):
dock_area = self.ui.dock_area
return dock_area
@property
def all_connections(self) -> list:
all_connections = self.rpc_register.list_all_connections()
all_connections_keys = list(all_connections.keys())
return all_connections_keys
def register_all_rpc(self):
app = QApplication.instance()
app.register_all()
@property
def widget_list(self) -> list:
"""Return a list of all widgets in the application."""
app = QApplication.instance()
all_widgets = app.list_all_bec_widgets()
return all_widgets
if __name__ == "__main__":
import sys
app = QApplication(sys.argv)
print(id(app))
# app = BECApplication(sys.argv)
# print(id(app))
main_window = WindowWithUi()
main_window.show()
sys.exit(app.exec())

6
bec_widgets/widgets/plots/multi_waveform/toolbar_bundles/monitor_selection.py
View File

@ -29,7 +29,9 @@ class MultiWaveformSelectionToolbarBundle(ToolbarBundle):
# Monitor Selection
self.monitor = DeviceComboBox(
device_filter=BECDeviceFilter.DEVICE, readout_priority_filter=ReadoutPriority.ASYNC
device_filter=BECDeviceFilter.DEVICE,
readout_priority_filter=ReadoutPriority.ASYNC,
parent_id=self.target_widget.gui_id,
)
self.monitor.addItem("", None)
self.monitor.setCurrentText("")
@ -38,7 +40,7 @@ class MultiWaveformSelectionToolbarBundle(ToolbarBundle):
self.add_action("monitor", WidgetAction(widget=self.monitor, adjust_size=False))
# Colormap Selection
self.colormap_widget = BECColorMapWidget(cmap="magma")
self.colormap_widget = BECColorMapWidget(cmap="magma", parent_id=self.target_widget.gui_id)
self.add_action("color_map", WidgetAction(widget=self.colormap_widget, adjust_size=False))
# Connect slots, a device will be connected upon change of any combobox

7
bec_widgets/widgets/plots/waveform/waveform.py
View File

@ -18,6 +18,7 @@ from bec_widgets.utils.colors import Colors, set_theme
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbar import MaterialIconAction
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
from bec_widgets.widgets.dap.lmfit_dialog.lmfit_dialog import LMFitDialog
from bec_widgets.widgets.plots.plot_base import PlotBase
from bec_widgets.widgets.plots.waveform.curve import Curve, CurveConfig, DeviceSignal
@ -1736,7 +1737,7 @@ class Waveform(PlotBase):
super().cleanup()
class DemoApp(QMainWindow): # pragma: no cover
class DemoApp(BECMainWindow): # pragma: no cover
def __init__(self):
super().__init__()
self.setWindowTitle("Waveform Demo")
@ -1759,9 +1760,9 @@ class DemoApp(QMainWindow): # pragma: no cover
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.bec_qapp import BECApplication
app = QApplication(sys.argv)
app = BECApplication(sys.argv)
set_theme("dark")
widget = DemoApp()
widget.show()

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

11
tests/end-2-end/test_plotting_framework_e2e.py
View File

@ -60,18 +60,14 @@ def test_rpc_plotting_shortcuts_init_configs(qtbot, connected_client_gui_obj):
# check if the correct devices are set
# Curve
assert c1._config["signal"] == {
assert c1._config_dict["signal"] == {
"dap": None,
"name": "bpm4i",
"entry": "bpm4i",
"dap_oversample": 1,
}
assert c1._config["source"] == "device"
assert c1._config["label"] == "bpm4i-bpm4i"
# Image Item
assert im_item._config["monitor"] == "eiger"
assert im_item._config["source"] == "auto"
assert c1._config_dict["source"] == "device"
assert c1._config_dict["label"] == "bpm4i-bpm4i"
def test_rpc_waveform_scan(qtbot, bec_client_lib, connected_client_gui_obj):
@ -93,6 +89,7 @@ def test_rpc_waveform_scan(qtbot, bec_client_lib, connected_client_gui_obj):
status = scans.line_scan(dev.samx, -5, 5, steps=10, exp_time=0.05, relative=False)
status.wait()
# FIXME if this gets flaky, we wait for status.scan.scan_id to be in client.history[-1] and then fetch data from history
item = queue.scan_storage.storage[-1]
last_scan_data = item.live_data if hasattr(item, "live_data") else item.data

4
tests/unit_tests/conftest.py
View File

@ -28,8 +28,10 @@ def qapplication(qtbot, request, testable_qtimer_class): # pylint: disable=unus
print("Test failed, skipping cleanup checks")
return
testable_qtimer_class.check_all_stopped(qtbot)
# qapp = BECApplication()
# qapp.shutdown()
testable_qtimer_class.check_all_stopped(qtbot)
qapp = QApplication.instance()
qapp.processEvents()
if hasattr(qapp, "os_listener") and qapp.os_listener:

3
tests/unit_tests/test_client_utils.py
View File

@ -31,7 +31,7 @@ def test_rpc_call_new_dock(cli_dock_area):
)
def test_client_utils_start_plot_process(config, call_config):
with mock.patch("bec_widgets.cli.client_utils.subprocess.Popen") as mock_popen:
_start_plot_process("gui_id", BECDockArea, "bec", config)
_start_plot_process("gui_id", "bec", config, gui_class="BECDockArea")
command = [
"bec-gui-server",
"--id",
@ -82,7 +82,6 @@ def test_client_utils_passes_client_config_to_server(bec_dispatcher):
) # the started event will not be set, wait=True would block forever
mock_start_plot.assert_called_once_with(
"gui_id",
BECGuiClient,
gui_class_id="bec",
config=mixin._client._service_config.config,
logger=mock.ANY,

63
tests/unit_tests/test_rpc_server.py
View File

@ -1,45 +1,56 @@
import argparse
from unittest import mock
import pytest
from bec_lib.service_config import ServiceConfig
from bec_widgets.cli.server import _start_server
from bec_widgets.widgets.containers.dock import BECDockArea
from bec_widgets.cli.server import GUIServer
@pytest.fixture
def mocked_cli_server():
with mock.patch("bec_widgets.cli.server.BECWidgetsCLIServer") as mock_server:
with mock.patch("bec_widgets.cli.server.ServiceConfig") as mock_config:
with mock.patch("bec_lib.logger.bec_logger.configure") as mock_logger:
yield mock_server, mock_config, mock_logger
def gui_server():
args = argparse.Namespace(
config=None, id="gui_id", gui_class="LaunchWindow", gui_class_id="bec", hide=False
)
return GUIServer(args=args)
def test_rpc_server_start_server_without_service_config(mocked_cli_server):
def test_gui_server_start_server_without_service_config(gui_server):
"""
Test that the server is started with the correct arguments.
"""
mock_server, mock_config, _ = mocked_cli_server
assert gui_server.config is None
assert gui_server.gui_id == "gui_id"
assert gui_server.gui_class == "LaunchWindow"
assert gui_server.gui_class_id == "bec"
assert gui_server.hide is False
_start_server("gui_id", BECDockArea, config=None)
mock_server.assert_called_once_with(
gui_id="gui_id", config=mock_config(), gui_class=BECDockArea, gui_class_id="bec"
)
def test_gui_server_get_service_config(gui_server):
"""
Test that the server is started with the correct arguments.
"""
assert gui_server._get_service_config().config is ServiceConfig().config
@pytest.mark.parametrize(
"config, call_config",
"connections, hide",
[
("/path/to/config.yml", {"config_path": "/path/to/config.yml"}),
({"key": "value"}, {"config": {"key": "value"}}),
({}, False),
({"launcher": mock.MagicMock()}, False),
({"launcher": mock.MagicMock(), "dock_area": mock.MagicMock()}, True),
],
)
def test_rpc_server_start_server_with_service_config(mocked_cli_server, config, call_config):
"""
Test that the server is started with the correct arguments.
"""
mock_server, mock_config, _ = mocked_cli_server
config = mock_config(**call_config)
_start_server("gui_id", BECDockArea, config=config)
mock_server.assert_called_once_with(
gui_id="gui_id", config=config, gui_class=BECDockArea, gui_class_id="bec"
)
def test_gui_server_turns_off_the_lights(gui_server, connections, hide):
with mock.patch.object(gui_server, "launcher_window") as mock_launcher_window:
with mock.patch.object(gui_server, "app") as mock_app:
gui_server._turn_off_the_lights(connections)
if not hide:
mock_launcher_window.show.assert_called_once()
mock_launcher_window.activateWindow.assert_called_once()
mock_launcher_window.raise_.assert_called_once()
mock_app.setQuitOnLastWindowClosed.assert_called_once_with(True)
else:
mock_launcher_window.hide.assert_called_once()
mock_app.setQuitOnLastWindowClosed.assert_called_once_with(False)