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

feat!: namespace update for gui, dock_area and docks.

Browse Source
This commit is contained in:
wakonig_k
2025-03-13 10:06:11 +01:00
committed by wyzula-jan
parent f895dd4863
commit 08f5a1cceb
26 changed files with 1105 additions and 892 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
bec_widgets/cli/auto_updates.py
View File

@ -35,9 +35,9 @@ class AutoUpdates:
Create a default dock for the auto updates.
"""
self.dock_name = "default_figure"
self._default_dock = self.gui.add_dock(self.dock_name)
self._default_dock.add_widget("BECFigure")
self._default_fig = self._default_dock.widget_list[0]
self._default_dock = self.gui.new(self.dock_name)
self._default_dock.new("BECFigure")
self._default_fig = self._default_dock.elements_list[0]
@staticmethod
def get_scan_info(msg) -> ScanInfo:

671
bec_widgets/cli/client.py
View File

@ -50,27 +50,12 @@ class Widgets(str, enum.Enum):
class AbortButton(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""A button that abort the scan."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
@ -230,14 +215,7 @@ class BECDock(RPCBase):
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def widget_list(self) -> "list[BECWidget]":
def element_list(self) -> "list[BECWidget]":
"""
Get the widgets in the dock.
@ -245,27 +223,58 @@ class BECDock(RPCBase):
widgets(list): The widgets in the dock.
"""
@property
@rpc_call
def elements(self) -> "dict[str, BECWidget]":
"""
Get the widgets in the dock.
Returns:
widgets(dict): The widgets in the dock.
"""
@rpc_call
def new(
self,
widget: "BECWidget | str",
name: "str | None" = None,
row: "int | None" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
) -> "BECWidget":
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add. It can not be BECDock or BECDockArea.
name(str): The name of the widget.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
colspan(int): The number of columns the widget should span.
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
@rpc_call
def show(self):
"""
Show the dock.
"""
@rpc_call
def hide(self):
"""
Hide the dock.
"""
@rpc_call
def show_title_bar(self):
"""
Hide the title bar of the dock.
"""
@rpc_call
def hide_title_bar(self):
"""
Hide the title bar of the dock.
"""
@rpc_call
def get_widgets_positions(self) -> "dict":
"""
Get the positions of the widgets in the dock.
Returns:
dict: The positions of the widgets in the dock as dict -> {(row, col, rowspan, colspan):widget}
"""
@rpc_call
def set_title(self, title: "str"):
"""
@ -276,29 +285,13 @@ class BECDock(RPCBase):
"""
@rpc_call
def add_widget(
self,
widget: "BECWidget | str",
row=None,
col=0,
rowspan=1,
colspan=1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
) -> "BECWidget":
def hide_title_bar(self):
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
colspan(int): The number of columns the widget should span.
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
Hide the title bar of the dock.
"""
@rpc_call
def list_eligible_widgets(self) -> "list":
def available_widgets(self) -> "list":
"""
List all widgets that can be added to the dock.
@ -307,23 +300,18 @@ class BECDock(RPCBase):
"""
@rpc_call
def move_widget(self, widget: "QWidget", new_row: "int", new_col: "int"):
"""
Move a widget to a new position in the layout.
Args:
widget(QWidget): The widget to move.
new_row(int): The new row to move the widget to.
new_col(int): The new column to move the widget to.
"""
@rpc_call
def remove_widget(self, widget_rpc_id: "str"):
def delete(self, widget_name: "str") -> "None":
"""
Remove a widget from the dock.
Args:
widget_rpc_id(str): The ID of the widget to remove.
widget_name(str): Delete the widget with the given name.
"""
@rpc_call
def delete_all(self):
"""
Remove all widgets from the dock.
"""
@rpc_call
@ -346,21 +334,50 @@ class BECDock(RPCBase):
class BECDockArea(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
def new(
self,
name: "str | None" = None,
widget: "str | QWidget | None" = None,
widget_name: "str | None" = None,
position: "Literal['bottom', 'top', 'left', 'right', 'above', 'below']" = "bottom",
relative_to: "BECDock | None" = None,
closable: "bool" = True,
floating: "bool" = False,
row: "int | None" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
) -> "BECDock":
"""
Get the configuration of the widget.
Add a dock to the dock area. Dock has QGridLayout as layout manager by default.
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
colspan(int): The colspan of the added widget.
Returns:
dict: The configuration of the widget.
BECDock: The created dock.
"""
@property
@rpc_call
def selected_device(self) -> "str":
def show(self):
"""
None
Show all windows including floating docks.
"""
@rpc_call
def hide(self):
"""
Hide all windows including floating docks.
"""
@property
@ -372,76 +389,35 @@ class BECDockArea(RPCBase):
dock_dict(dict): The docks in the dock area.
"""
@property
@rpc_call
def save_state(self) -> "dict":
def panel_list(self) -> "list[BECDock]":
"""
Save the state of the dock area.
Get the docks in the dock area.
Returns:
dict: The state of the dock area.
list: The docks in the dock area.
"""
@rpc_call
def remove_dock(self, name: "str"):
def delete(self, dock_name: "str"):
"""
Remove a dock by name and ensure it is properly closed and cleaned up.
Delete a dock by name.
Args:
name(str): The name of the dock to remove.
dock_name(str): The name of the dock to delete.
"""
@rpc_call
def restore_state(
self, state: "dict" = None, missing: "Literal['ignore', 'error']" = "ignore", extra="bottom"
):
def delete_all(self) -> "None":
"""
Restore the state of the dock area. If no state is provided, the last state is restored.
Args:
state(dict): The state to restore.
missing(Literal['ignore','error']): What to do if a dock is missing.
extra(str): Extra docks that are in the dockarea but that are not mentioned in state will be added to the bottom of the dockarea, unless otherwise specified by the extra argument.
Delete all docks.
"""
@rpc_call
def add_dock(
self,
name: "str" = None,
position: "Literal['bottom', 'top', 'left', 'right', 'above', 'below']" = None,
relative_to: "BECDock | None" = None,
closable: "bool" = True,
floating: "bool" = False,
prefix: "str" = "dock",
widget: "str | QWidget | None" = None,
row: "int" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
) -> "BECDock":
def remove(self) -> "None":
"""
Add a dock to the dock area. Dock has QGridLayout as layout manager by default.
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
prefix(str): The prefix for the dock name if no name is provided.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
colspan(int): The colspan of the added widget.
Returns:
BECDock: The created dock.
"""
@rpc_call
def clear_all(self):
"""
Close all docks and remove all temp areas.
Remove the dock area.
"""
@rpc_call
@ -462,40 +438,35 @@ class BECDockArea(RPCBase):
Return all floating docks to the dock area.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def temp_areas(self) -> "list":
"""
Get the temporary areas in the dock area.
Returns:
list: The temporary areas in the dock area.
"""
@rpc_call
def show(self):
"""
Show all windows including floating docks.
"""
@rpc_call
def hide(self):
"""
Hide all windows including floating docks.
"""
@rpc_call
def delete(self):
def selected_device(self) -> "str":
"""
None
"""
@rpc_call
def save_state(self) -> "dict":
"""
Save the state of the dock area.
Returns:
dict: The state of the dock area.
"""
@rpc_call
def restore_state(
self, state: "dict" = None, missing: "Literal['ignore', 'error']" = "ignore", extra="bottom"
):
"""
Restore the state of the dock area. If no state is provided, the last state is restored.
Args:
state(dict): The state to restore.
missing(Literal['ignore','error']): What to do if a dock is missing.
extra(str): Extra docks that are in the dockarea but that are not mentioned in state will be added to the bottom of the dockarea, unless otherwise specified by the extra argument.
"""
class BECFigure(RPCBase):
@property
@ -1409,27 +1380,10 @@ class BECImageWidget(RPCBase):
class BECMainWindow(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
def remove(self):
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
@ -2230,6 +2184,8 @@ class BECPlotBase(RPCBase):
class BECProgressBar(RPCBase):
"""A custom progress bar with smooth transitions. The displayed text can be customized using a template."""
@rpc_call
def set_value(self, value):
"""
@ -2281,52 +2237,22 @@ class BECProgressBar(RPCBase):
class BECQueue(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Widget to display the BEC queue."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class BECStatusBox(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""An autonomous widget to display the status of BEC services."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
@ -2845,6 +2771,8 @@ class Curve(RPCBase):
class DapComboBox(RPCBase):
"""The DAPComboBox widget is an extension to the QComboBox with all avaialble DAP model from BEC."""
@rpc_call
def select_y_axis(self, y_axis: str):
"""
@ -2883,156 +2811,66 @@ class DarkModeButton(RPCBase):
class DeviceBrowser(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
def remove(self):
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class DeviceComboBox(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Combobox widget for device input with autocomplete for device names."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class DeviceInputBase(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Mixin base class for device input widgets."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class DeviceLineEdit(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class DeviceSignalInputBase(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Mixin base class for device signal input widgets."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class LMFitDialog(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Dialog for displaying the fit summary and params for LMFit DAP processes"""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class LogPanel(RPCBase):
"""Displays a log panel"""
@rpc_call
def set_plain_text(self, text: str) -> None:
"""
@ -3095,6 +2933,8 @@ class PositionIndicator(RPCBase):
class PositionerBox(RPCBase):
"""Simple Widget to control a positioner in box form"""
@rpc_call
def set_positioner(self, positioner: "str | Positioner"):
"""
@ -3106,6 +2946,8 @@ class PositionerBox(RPCBase):
class PositionerBox2D(RPCBase):
"""Simple Widget to control two positioners in box form"""
@rpc_call
def set_positioner_hor(self, positioner: "str | Positioner"):
"""
@ -3126,31 +2968,18 @@ class PositionerBox2D(RPCBase):
class PositionerBoxBase(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Contains some core logic for positioner box widgets"""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class PositionerControlLine(RPCBase):
"""A widget that controls a single device."""
@rpc_call
def set_positioner(self, positioner: "str | Positioner"):
"""
@ -3162,6 +2991,8 @@ class PositionerControlLine(RPCBase):
class PositionerGroup(RPCBase):
"""Simple Widget to control a positioner in box form"""
@rpc_call
def set_positioners(self, device_names: "str"):
"""
@ -3172,52 +3003,22 @@ class PositionerGroup(RPCBase):
class ResetButton(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""A button that resets the scan queue."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class ResumeButton(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""A button that continue scan queue."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
@ -3501,131 +3302,56 @@ class RingProgressBar(RPCBase):
class ScanControl(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
def remove(self):
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class ScanMetadata(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Dynamically generates a form for inclusion of metadata for a scan. Uses the"""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class SignalComboBox(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class SignalLineEdit(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class StopButton(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
"""A button that stops the current scan."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
Cleanup the BECConnector
"""
class TextBox(RPCBase):
"""A widget that displays text in plain and HTML format"""
@rpc_call
def set_plain_text(self, text: str) -> None:
"""
@ -3645,7 +3371,10 @@ class TextBox(RPCBase):
"""
class VSCodeEditor(RPCBase): ...
class VSCodeEditor(RPCBase):
"""A widget to display the VSCode editor."""
...
class Waveform(RPCBase):
@ -4060,6 +3789,8 @@ class Waveform(RPCBase):
class WebsiteWidget(RPCBase):
"""A simple widget to display a website"""
@rpc_call
def set_url(self, url: str) -> None:
"""

348
bec_widgets/cli/client_utils.py
View File

@ -1,3 +1,5 @@
"""Client utilities for the BEC GUI."""
from __future__ import annotations
import importlib
@ -7,13 +9,15 @@ import os
import select
import subprocess
import threading
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import TYPE_CHECKING
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.utils.import_utils import isinstance_based_on_class_name, lazy_import, lazy_import_from
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from rich.console import Console
from rich.table import Table
import bec_widgets.cli.client as client
from bec_widgets.cli.auto_updates import AutoUpdates
@ -23,16 +27,17 @@ if TYPE_CHECKING:
from bec_lib import messages
from bec_lib.connector import MessageObject
from bec_lib.device import DeviceBase
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_lib.redis_connector import StreamMessage
else:
messages = lazy_import("bec_lib.messages")
# from bec_lib.connector import MessageObject
MessageObject = lazy_import_from("bec_lib.connector", ("MessageObject",))
BECDispatcher = lazy_import_from("bec_widgets.utils.bec_dispatcher", ("BECDispatcher",))
StreamMessage = lazy_import_from("bec_lib.redis_connector", ("StreamMessage",))
logger = bec_logger.logger
IGNORE_WIDGETS = ["BECDockArea", "BECDock"]
def _filter_output(output: str) -> str:
"""
@ -67,7 +72,9 @@ def _get_output(process, logger) -> None:
logger.error(f"Error reading process output: {str(e)}")
def _start_plot_process(gui_id: str, gui_class: type, config: dict | str, logger=None) -> None:
def _start_plot_process(
gui_id: str, gui_class: type, gui_class_id: str, config: dict | str, logger=None
) -> None:
"""
Start the plot in a new process.
@ -76,7 +83,16 @@ def _start_plot_process(gui_id: str, gui_class: type, config: dict | str, logger
process will not be captured.
"""
# pylint: disable=subprocess-run-check
command = ["bec-gui-server", "--id", gui_id, "--gui_class", gui_class.__name__, "--hide"]
command = [
"bec-gui-server",
"--id",
gui_id,
"--gui_class",
gui_class.__name__,
"--gui_class_id",
gui_class_id,
"--hide",
]
if config:
if isinstance(config, dict):
config = json.dumps(config)
@ -111,16 +127,20 @@ def _start_plot_process(gui_id: str, gui_class: type, config: dict | str, logger
class RepeatTimer(threading.Timer):
"""RepeatTimer class."""
def run(self):
while not self.finished.wait(self.interval):
self.function(*self.args, **self.kwargs)
# pylint: disable=protected-access
@contextmanager
def wait_for_server(client):
def wait_for_server(client: BECGuiClient):
"""Context manager to wait for the server to start."""
timeout = client._startup_timeout
if not timeout:
if client.gui_is_alive():
if client._gui_is_alive():
# there is hope, let's wait a bit
timeout = 1
else:
@ -138,42 +158,63 @@ def wait_for_server(client):
yield
### ----------------------------
### NOTE
### it is far easier to extend the 'delete' method on the client side,
### to know when the client is deleted, rather than listening to server
### to get notified. However, 'generate_cli.py' cannot add extra stuff
### in the generated client module. So, here a class with the same name
### is created, and client module is patched.
class WidgetNameSpace:
def __repr__(self):
console = Console()
table = Table(title="Available widgets for BEC CLI usage")
table.add_column("Widget Name", justify="left", style="magenta")
table.add_column("Description", justify="left")
for attr, value in self.__dict__.items():
docs = value.__doc__
docs = docs if docs else "No description available"
table.add_row(attr, docs)
console.print(table)
return f""
class AvailableWidgetsNamespace:
"""Namespace for available widgets in the BEC GUI."""
def __init__(self):
for widget in client.Widgets:
name = widget.value
if name in IGNORE_WIDGETS:
continue
setattr(self, name, name)
def __repr__(self):
console = Console()
table = Table(title="Available widgets for BEC CLI usage")
table.add_column("Widget Name", justify="left", style="magenta")
table.add_column("Description", justify="left")
for attr_name, _ in self.__dict__.items():
docs = getattr(client, attr_name).__doc__
docs = docs if docs else "No description available"
table.add_row(attr_name, docs if len(docs.strip()) > 0 else "No description available")
console.print(table)
return "" # f"<{self.__class__.__name__}>"
class BECDockArea(client.BECDockArea):
def delete(self):
if self is BECGuiClient._top_level["main"].widget:
raise RuntimeError("Cannot delete main window")
super().delete()
try:
del BECGuiClient._top_level[self._gui_id]
except KeyError:
# if a dock area is not at top level
pass
"""Extend the BECDockArea class and add namespaces to access widgets of docks."""
client.BECDockArea = BECDockArea
### ----------------------------
@dataclass
class WidgetDesc:
title: str
widget: BECDockArea
def __init__(self, gui_id=None, config=None, name=None, parent=None):
super().__init__(gui_id, config, name, parent)
# Add namespaces for DockArea
self.elements = WidgetNameSpace()
class BECGuiClient(RPCBase):
"""BEC GUI client class. Container for GUI applications within Python."""
_top_level = {}
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._default_dock_name = "bec"
self._auto_updates_enabled = True
self._auto_updates = None
self._killed = False
self._startup_timeout = 0
self._gui_started_timer = None
self._gui_started_event = threading.Event()
@ -181,14 +222,21 @@ class BECGuiClient(RPCBase):
self._process_output_processing_thread = None
@property
def windows(self):
def windows(self) -> dict:
"""Dictionary with dock ares in the GUI."""
return self._top_level
@property
def auto_updates(self):
if self._auto_updates_enabled:
with wait_for_server(self):
return self._auto_updates
def window_list(self) -> list:
"""List with dock areas in the GUI."""
return list(self._top_level.values())
# FIXME AUTO UPDATES
# @property
# def auto_updates(self):
# if self._auto_updates_enabled:
# with wait_for_server(self):
# return self._auto_updates
def _get_update_script(self) -> AutoUpdates | None:
eps = imd.entry_points(group="bec.widgets.auto_updates")
@ -199,71 +247,73 @@ class BECGuiClient(RPCBase):
# if the module is not found, we skip it
if spec is None:
continue
return ep.load()(gui=self._top_level["main"].widget)
return ep.load()(gui=self._top_level["main"])
except Exception as e:
logger.error(f"Error loading auto update script from plugin: {str(e)}")
return None
@property
def selected_device(self):
"""
Selected device for the plot.
"""
auto_update_config_ep = MessageEndpoints.gui_auto_update_config(self._gui_id)
auto_update_config = self._client.connector.get(auto_update_config_ep)
if auto_update_config:
return auto_update_config.selected_device
return None
# FIXME AUTO UPDATES
# @property
# def selected_device(self) -> str | None:
# """
# Selected device for the plot.
# """
# auto_update_config_ep = MessageEndpoints.gui_auto_update_config(self._gui_id)
# auto_update_config = self._client.connector.get(auto_update_config_ep)
# if auto_update_config:
# return auto_update_config.selected_device
# return None
@selected_device.setter
def selected_device(self, device: str | DeviceBase):
if isinstance_based_on_class_name(device, "bec_lib.device.DeviceBase"):
self._client.connector.set_and_publish(
MessageEndpoints.gui_auto_update_config(self._gui_id),
messages.GUIAutoUpdateConfigMessage(selected_device=device.name),
)
elif isinstance(device, str):
self._client.connector.set_and_publish(
MessageEndpoints.gui_auto_update_config(self._gui_id),
messages.GUIAutoUpdateConfigMessage(selected_device=device),
)
else:
raise ValueError("Device must be a string or a device object")
# @selected_device.setter
# def selected_device(self, device: str | DeviceBase):
# if isinstance_based_on_class_name(device, "bec_lib.device.DeviceBase"):
# self._client.connector.set_and_publish(
# MessageEndpoints.gui_auto_update_config(self._gui_id),
# messages.GUIAutoUpdateConfigMessage(selected_device=device.name),
# )
# elif isinstance(device, str):
# self._client.connector.set_and_publish(
# MessageEndpoints.gui_auto_update_config(self._gui_id),
# messages.GUIAutoUpdateConfigMessage(selected_device=device),
# )
# else:
# raise ValueError("Device must be a string or a device object")
def _start_update_script(self) -> None:
self._client.connector.register(MessageEndpoints.scan_status(), cb=self._handle_msg_update)
# FIXME AUTO UPDATES
# def _start_update_script(self) -> None:
# self._client.connector.register(MessageEndpoints.scan_status(), cb=self._handle_msg_update)
def _handle_msg_update(self, msg: MessageObject) -> None:
if self.auto_updates is not None:
# pylint: disable=protected-access
return self._update_script_msg_parser(msg.value)
# def _handle_msg_update(self, msg: StreamMessage) -> None:
# if self.auto_updates is not None:
# # pylint: disable=protected-access
# return self._update_script_msg_parser(msg.value)
def _update_script_msg_parser(self, msg: messages.BECMessage) -> None:
if isinstance(msg, messages.ScanStatusMessage):
if not self.gui_is_alive():
return
if self._auto_updates_enabled:
return self.auto_updates.do_update(msg)
# def _update_script_msg_parser(self, msg: messages.BECMessage) -> None:
# if isinstance(msg, messages.ScanStatusMessage):
# if not self._gui_is_alive():
# return
# if self._auto_updates_enabled:
# return self.auto_updates.do_update(msg)
def _gui_post_startup(self):
self._top_level["main"] = WidgetDesc(
title="BEC Widgets", widget=BECDockArea(gui_id=self._gui_id)
# if self._auto_updates_enabled:
# if self._auto_updates is None:
# auto_updates = self._get_update_script()
# if auto_updates is None:
# AutoUpdates.create_default_dock = True
# AutoUpdates.enabled = True
# auto_updates = AutoUpdates(self._top_level["main"].widget)
# if auto_updates.create_default_dock:
# auto_updates.start_default_dock()
# self._start_update_script()
# self._auto_updates = auto_updates
self._top_level[self._default_dock_name] = BECDockArea(
gui_id=f"{self._default_dock_name}", name=self._default_dock_name, parent=self
)
if self._auto_updates_enabled:
if self._auto_updates is None:
auto_updates = self._get_update_script()
if auto_updates is None:
AutoUpdates.create_default_dock = True
AutoUpdates.enabled = True
auto_updates = AutoUpdates(self._top_level["main"].widget)
if auto_updates.create_default_dock:
auto_updates.start_default_dock()
self._start_update_script()
self._auto_updates = auto_updates
self._do_show_all()
self._gui_started_event.set()
def start_server(self, wait=False) -> None:
def _start_server(self, wait: bool = False) -> None:
"""
Start the GUI server, and execute callback when it is launched
"""
@ -272,7 +322,11 @@ class BECGuiClient(RPCBase):
self._startup_timeout = 5
self._gui_started_event.clear()
self._process, self._process_output_processing_thread = _start_plot_process(
self._gui_id, self.__class__, self._client._service_config.config, logger=logger
self._gui_id,
self.__class__,
gui_class_id=self._default_dock_name,
config=self._client._service_config.config, # pylint: disable=protected-access
logger=logger,
)
def gui_started_callback(callback):
@ -283,7 +337,7 @@ class BECGuiClient(RPCBase):
threading.current_thread().cancel()
self._gui_started_timer = RepeatTimer(
0.5, lambda: self.gui_is_alive() and gui_started_callback(self._gui_post_startup)
0.5, lambda: self._gui_is_alive() and gui_started_callback(self._gui_post_startup)
)
self._gui_started_timer.start()
@ -295,53 +349,97 @@ class BECGuiClient(RPCBase):
return rpc_client._run_rpc("_dump")
def start(self):
return self.start_server()
return self._start_server()
def _do_show_all(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client._run_rpc("show")
rpc_client._run_rpc("show") # pylint: disable=protected-access
for window in self._top_level.values():
window.widget.show()
window.show()
def show_all(self):
def _show_all(self):
with wait_for_server(self):
return self._do_show_all()
def hide_all(self):
def _hide_all(self):
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client._run_rpc("hide")
for window in self._top_level.values():
window.widget.hide()
rpc_client._run_rpc("hide") # pylint: disable=protected-access
# because of the registry callbacks, we may have
# dock areas that are already killed, but not yet
# removed from the registry state
if not self._killed:
for window in self._top_level.values():
window.hide()
def show(self):
"""Show the GUI window."""
if self._process is not None:
return self.show_all()
return self._show_all()
# backward compatibility: show() was also starting server
return self.start_server(wait=True)
return self._start_server(wait=True)
def hide(self):
return self.hide_all()
"""Hide the GUI window."""
return self._hide_all()
@property
def main(self):
"""Return client to main dock area (in main window)"""
with wait_for_server(self):
return self._top_level["main"].widget
def new(
self,
name: str | None = None,
wait: bool = True,
geometry: tuple[int, int, int, int] | None = None,
) -> BECDockArea:
"""Create a new top-level dock area.
def new(self, title):
"""Ask main window to create a new top-level dock area"""
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
widget = rpc_client._run_rpc("new_dock_area", title)
self._top_level[widget._gui_id] = WidgetDesc(title=title, widget=widget)
return widget
def close(self) -> None:
Args:
name(str, optional): The name of the dock area. Defaults to None.
wait(bool, optional): Whether to wait for the server to start. Defaults to True.
geometry(tuple[int, int, int, int] | None): The geometry of the dock area (pos_x, pos_y, w, h)
Returns:
BECDockArea: The new dock area.
"""
Close the gui window.
if len(self.window_list) == 0:
self.show()
if wait:
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
widget = rpc_client._run_rpc(
"new_dock_area", name, geometry
) # pylint: disable=protected-access
self._top_level[widget.widget_name] = widget
return widget
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
widget = rpc_client._run_rpc(
"new_dock_area", name, geometry
) # pylint: disable=protected-access
self._top_level[widget.widget_name] = widget
return widget
def delete(self, name: str) -> None:
"""Delete a dock area.
Args:
name(str): The name of the dock area.
"""
widget = self.windows.get(name)
if widget is None:
raise ValueError(f"Dock area {name} not found.")
widget._run_rpc("close") # pylint: disable=protected-access
def delete_all(self) -> None:
"""Delete all dock areas."""
for widget_name in self.windows.keys():
self.delete(widget_name)
def close(self):
"""Deprecated. Use kill_server() instead."""
# FIXME, deprecated in favor of kill, will be removed in the future
self.kill_server()
def kill_server(self) -> None:
"""Kill the GUI server."""
self._top_level.clear()
self._killed = True
if self._gui_started_timer is not None:
self._gui_started_timer.cancel()
@ -357,3 +455,17 @@ class BECGuiClient(RPCBase):
self._process_output_processing_thread.join()
self._process.wait()
self._process = None
if __name__ == "__main__":
from bec_lib.client import BECClient
from bec_lib.service_config import ServiceConfig
config = ServiceConfig()
client = BECClient(config)
client.start()
# Test the client_utils.py module
gui = BECGuiClient()
gui.start()
print(gui.window_list)

12
bec_widgets/cli/generate_cli.py
View File

@ -95,9 +95,21 @@ class {class_name}(RPCBase):"""
self.content += f"""
class {class_name}(RPCBase):"""
if cls.__doc__:
# We only want the first line of the docstring
# But skip the first line if it's a blank line
first_line = cls.__doc__.split("\n")[0]
if first_line:
class_docs = first_line
else:
class_docs = cls.__doc__.split("\n")[1]
self.content += f"""
\"\"\"{class_docs}\"\"\"
"""
if not cls.USER_ACCESS:
self.content += """...
"""
for method in cls.USER_ACCESS:
is_property_setter = False
obj = getattr(cls, method, None)

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

@ -3,7 +3,7 @@ from __future__ import annotations
import threading
import uuid
from functools import wraps
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, cast
from bec_lib.client import BECClient
from bec_lib.endpoints import MessageEndpoints
@ -44,7 +44,7 @@ def rpc_call(func):
for key, val in kwargs.items():
if hasattr(val, "name"):
kwargs[key] = val.name
if not self.gui_is_alive():
if not self._root._gui_is_alive():
raise RuntimeError("GUI is not alive")
return self._run_rpc(func.__name__, *args, **kwargs)
@ -61,10 +61,17 @@ class RPCResponseTimeoutError(Exception):
class RPCBase:
def __init__(self, gui_id: str = None, config: dict = None, parent=None) -> None:
def __init__(
self,
gui_id: str | None = None,
config: dict | None = None,
name: str | None = None,
parent=None,
) -> None:
self._client = BECClient() # BECClient is a singleton; here, we simply get the instance
self._config = config if config is not None else {}
self._gui_id = gui_id if gui_id is not None else str(uuid.uuid4())[:5]
self._name = name if name is not None else str(uuid.uuid4())[:5]
self._parent = parent
self._msg_wait_event = threading.Event()
self._rpc_response = None
@ -74,7 +81,20 @@ class RPCBase:
def __repr__(self):
type_ = type(self)
qualname = type_.__qualname__
return f"<{qualname} object at {hex(id(self))}>"
return f"<{qualname} with name: {self.widget_name}>"
def remove(self):
"""
Remove the widget.
"""
self._run_rpc("remove")
@property
def widget_name(self):
"""
Get the widget name.
"""
return self._name
@property
def _root(self):
@ -88,7 +108,7 @@ class RPCBase:
parent = parent._parent
return parent
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=3, **kwargs):
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=3, **kwargs) -> Any:
"""
Run the RPC call.
@ -165,7 +185,7 @@ class RPCBase:
return cls(parent=self, **msg_result)
return msg_result
def gui_is_alive(self):
def _gui_is_alive(self):
"""
Check if the GUI is alive.
"""

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

@ -1,10 +1,21 @@
from __future__ import annotations
from functools import wraps
from threading import Lock
from typing import TYPE_CHECKING, Callable
from weakref import WeakValueDictionary
from bec_lib.logger import bec_logger
from qtpy.QtCore import QObject
if TYPE_CHECKING:
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.containers.dock.dock import BECDock
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
logger = bec_logger.logger
class RPCRegister:
"""
@ -49,7 +60,7 @@ class RPCRegister:
raise ValueError(f"RPC object {rpc} must have a 'gui_id' attribute.")
self._rpc_register.pop(rpc.gui_id, None)
def get_rpc_by_id(self, gui_id: str) -> QObject:
def get_rpc_by_id(self, gui_id: str) -> QObject | None:
"""
Get an RPC object by its ID.
@ -57,11 +68,25 @@ class RPCRegister:
gui_id(str): The ID of the RPC object to be retrieved.
Returns:
QObject: The RPC object with the given ID.
QObject | None: The RPC object with the given ID or None
"""
rpc_object = self._rpc_register.get(gui_id, None)
return rpc_object
def get_rpc_by_name(self, name: str) -> QObject | None:
"""
Get an RPC object by its name.
Args:
name(str): The name of the RPC object to be retrieved.
Returns:
QObject | None: The RPC object with the given name.
"""
rpc_object = [rpc for rpc in self._rpc_register if rpc._name == name]
rpc_object = rpc_object[0] if len(rpc_object) > 0 else None
return rpc_object
def list_all_connections(self) -> dict:
"""
List all the registered RPC objects.
@ -73,6 +98,19 @@ class RPCRegister:
connections = dict(self._rpc_register)
return connections
def get_names_of_rpc_by_class_type(
self, cls: BECWidget | BECConnector | BECDock | BECDockArea
) -> list[str]:
"""Get all the names of the widgets.
Args:
cls(BECWidget | BECConnector): The class of the RPC object to be retrieved.
"""
# This retrieves any rpc objects that are subclass of BECWidget,
# i.e. curve and image items are excluded
widgets = [rpc for rpc in self._rpc_register.values() if isinstance(rpc, cls)]
return [widget._name for widget in widgets]
@classmethod
def reset_singleton(cls):
"""

22
bec_widgets/cli/rpc/rpc_widget_handler.py
View File

@ -1,6 +1,9 @@
from __future__ import annotations
from bec_widgets.utils import BECConnector
from typing import Any
from bec_widgets.cli.client_utils import IGNORE_WIDGETS
from bec_widgets.utils.bec_widget import BECWidget
class RPCWidgetHandler:
@ -10,7 +13,7 @@ class RPCWidgetHandler:
self._widget_classes = None
@property
def widget_classes(self):
def widget_classes(self) -> dict[str, Any]:
"""
Get the available widget classes.
@ -19,7 +22,7 @@ class RPCWidgetHandler:
"""
if self._widget_classes is None:
self.update_available_widgets()
return self._widget_classes
return self._widget_classes # type: ignore
def update_available_widgets(self):
"""
@ -31,24 +34,27 @@ class RPCWidgetHandler:
from bec_widgets.utils.plugin_utils import get_custom_classes
clss = get_custom_classes("bec_widgets")
self._widget_classes = {cls.__name__: cls for cls in clss.widgets}
self._widget_classes = {
cls.__name__: cls for cls in clss.widgets if cls.__name__ not in IGNORE_WIDGETS
}
def create_widget(self, widget_type, **kwargs) -> BECConnector:
def create_widget(self, widget_type, name: str | None = None, **kwargs) -> BECWidget:
"""
Create a widget from an RPC message.
Args:
widget_type(str): The type of the widget.
name (str): The name of the widget.
**kwargs: The keyword arguments for the widget.
Returns:
widget(BECConnector): The created widget.
widget(BECWidget): The created widget.
"""
if self._widget_classes is None:
self.update_available_widgets()
widget_class = self._widget_classes.get(widget_type)
widget_class = self._widget_classes.get(widget_type) # type: ignore
if widget_class:
return widget_class(**kwargs)
return widget_class(name=name, **kwargs)
raise ValueError(f"Unknown widget type: {widget_type}")

38
bec_widgets/cli/server.py
View File

@ -15,6 +15,7 @@ from bec_lib.utils.import_utils import lazy_import
from qtpy.QtCore import Qt, QTimer
from redis.exceptions import RedisError
from bec_widgets.cli.rpc import rpc_register
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.utils import BECDispatcher
@ -36,6 +37,8 @@ def rpc_exception_hook(err_func):
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)})
@ -56,16 +59,18 @@ class BECWidgetsCLIServer:
dispatcher: BECDispatcher = None,
client=None,
config=None,
gui_class: Union[BECFigure, BECDockArea] = BECFigure,
gui_class: Union[BECFigure, 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
self.gui = gui_class(gui_id=self.gui_id)
# register broadcast callback
self.rpc_register = RPCRegister()
self.rpc_register.add_rpc(self.gui)
self.gui = gui_class(parent=None, name=gui_class_id, gui_id=gui_class_id)
# self.rpc_register.add_rpc(self.gui)
self.dispatcher.connect_slot(
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
@ -78,6 +83,7 @@ class BECWidgetsCLIServer:
self.status = messages.BECStatus.RUNNING
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")
@ -135,6 +141,9 @@ class BECWidgetsCLIServer:
if isinstance(obj, BECConnector):
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": obj.config.model_dump(),
"__rpc__": True,
@ -179,7 +188,12 @@ class SimpleFileLikeFromLogOutputFunc:
return
def _start_server(gui_id: str, gui_class: Union[BECFigure, BECDockArea], config: str | None = None):
def _start_server(
gui_id: str,
gui_class: Union[BECFigure, BECDockArea],
gui_class_id: str = "bec",
config: str | None = None,
):
if config:
try:
config = json.loads(config)
@ -196,7 +210,9 @@ def _start_server(gui_id: str, gui_class: Union[BECFigure, BECDockArea], config:
# service_name="BECWidgetsCLIServer",
# service_config=service_config.service_config,
# )
server = BECWidgetsCLIServer(gui_id=gui_id, config=service_config, gui_class=gui_class)
server = BECWidgetsCLIServer(
gui_id=gui_id, config=service_config, gui_class=gui_class, gui_class_id=gui_class_id
)
return server
@ -217,6 +233,12 @@ def main():
type=str,
help="Name of the gui class to be rendered. Possible values: \n- BECFigure\n- BECDockArea",
)
parser.add_argument(
"--gui_class_id",
type=str,
default="bec",
help="The id of the gui class that is added to the QApplication",
)
parser.add_argument("--config", type=str, help="Config file or config string.")
parser.add_argument("--hide", action="store_true", help="Hide on startup")
@ -256,14 +278,14 @@ def main():
# store gui id within QApplication object, to make it available to all widgets
app.gui_id = args.id
server = _start_server(args.id, gui_class, args.config)
# 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 Widgets")
win.setWindowTitle("BEC")
RPCRegister().add_rpc(win)
gui = server.gui
win.setCentralWidget(gui)
if not args.hide:

12
bec_widgets/examples/jupyter_console/jupyter_console_window.py
View File

@ -198,14 +198,18 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
def _init_dock(self):
self.d0 = self.dock.add_dock(name="dock_0")
self.mm = self.d0.add_widget("BECMotorMapWidget")
self.d0 = self.dock.new(name="dock_0")
self.mm = self.d0.new("BECMotorMapWidget")
self.mm.change_motors("samx", "samy")
self.d1 = self.dock.add_dock(name="dock_1", position="right")
self.im = self.d1.add_widget("BECImageWidget")
self.d1 = self.dock.new(name="dock_1", position="right")
self.im = self.d1.new("BECImageWidget")
self.im.image("waveform", "1d")
self.d2 = self.dock.new(name="dock_2", position="bottom")
self.wf = self.d2.new("BECFigure", row=0, col=0)
self.mw = self.wf.multi_waveform(monitor="waveform") # , config=config)
self.mw = None # self.wf.multi_waveform(monitor="waveform") # , config=config)
self.dock.save_state()

52
bec_widgets/utils/bec_connector.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
import os
import time
import uuid
from datetime import datetime
from typing import TYPE_CHECKING, Optional
from bec_lib.logger import bec_logger
@ -15,6 +16,7 @@ from qtpy.QtWidgets import QApplication
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.qt_utils.error_popups import SafeSlot as pyqtSlot
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.yaml_dialog import load_yaml, load_yaml_gui, save_yaml, save_yaml_gui
if TYPE_CHECKING:
@ -39,8 +41,7 @@ class ConnectionConfig(BaseModel):
"""Generate a GUI ID if none is provided."""
if v is None:
widget_class = values.data["widget_class"]
v = f"{widget_class}_{str(time.time())}"
return v
v = f"{widget_class}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S_%f')}"
return v
@ -75,7 +76,13 @@ class BECConnector:
USER_ACCESS = ["_config_dict", "_get_all_rpc", "_rpc_id"]
EXIT_HANDLERS = {}
def __init__(self, client=None, config: ConnectionConfig = None, gui_id: str = None):
def __init__(
self,
client=None,
config: ConnectionConfig | None = None,
gui_id: str | None = None,
name: str | None = None,
):
# BEC related connections
self.bec_dispatcher = BECDispatcher(client=client)
self.client = self.bec_dispatcher.client if client is None else client
@ -103,15 +110,22 @@ class BECConnector:
)
self.config = ConnectionConfig(widget_class=self.__class__.__name__)
# I feel that we should not allow BECConnector to be created with a custom gui_id
# because this would break with the logic in the RPCRegister of retrieving widgets by type
# iterating over all widgets and checkinf if the register widget starts with the string that is passsed.
# If the gui_id is randomly generated, this would break since that widget would have a
# gui_id that is generated in a different way.
if gui_id:
self.config.gui_id = gui_id
self.gui_id = gui_id
self.gui_id: str = gui_id
else:
self.gui_id = self.config.gui_id
# register widget to rpc register
# be careful: when registering, and the object is not a BECWidget,
# cleanup has to be called manually since there is no 'closeEvent'
self.gui_id: str = self.config.gui_id # type: ignore
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__
self.rpc_register = RPCRegister()
self.rpc_register.add_rpc(self)
@ -195,6 +209,7 @@ class BECConnector:
"""
self.config = config
# FIXME some thoughts are required to decide how thhis should work with rpc registry
def apply_config(self, config: dict, generate_new_id: bool = True) -> None:
"""
Apply the configuration to the widget.
@ -207,11 +222,12 @@ class BECConnector:
if generate_new_id is True:
gui_id = str(uuid.uuid4())
self.rpc_register.remove_rpc(self)
self.set_gui_id(gui_id)
self._set_gui_id(gui_id)
self.rpc_register.add_rpc(self)
else:
self.gui_id = self.config.gui_id
# FIXME some thoughts are required to decide how thhis should work with rpc registry
def load_config(self, path: str | None = None, gui: bool = False):
"""
Load the configuration of the widget from YAML.
@ -248,8 +264,8 @@ class BECConnector:
file_path = os.path.join(path, f"{self.__class__.__name__}_config.yaml")
save_yaml(file_path, self._config_dict)
@pyqtSlot(str)
def set_gui_id(self, gui_id: str) -> None:
# @pyqtSlot(str)
def _set_gui_id(self, gui_id: str) -> None:
"""
Set the GUI ID for the widget.
@ -288,9 +304,21 @@ class BECConnector:
Args:
config (ConnectionConfig | dict): Configuration settings.
"""
gui_id = getattr(config, "gui_id", None)
if isinstance(config, dict):
config = ConnectionConfig(**config)
self.config = config
if gui_id and config.gui_id != gui_id: # Recreating config should not overwrite the gui_id
self.config.gui_id = gui_id
def remove(self):
"""Cleanup the BECConnector"""
if hasattr(self, "close"):
self.close()
if hasattr(self, "deleteLater"):
self.deleteLater()
else:
self.rpc_register.remove_rpc(self)
def get_config(self, dict_output: bool = True) -> dict | BaseModel:
"""

31
bec_widgets/utils/bec_widget.py
View File

@ -1,5 +1,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
import darkdetect
from bec_lib.logger import bec_logger
from qtpy.QtCore import Slot
@ -7,6 +9,10 @@ from qtpy.QtWidgets import QApplication, QWidget
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import set_theme
from bec_widgets.utils.container_utils import WidgetContainerUtils
if TYPE_CHECKING:
from bec_widgets.widgets.containers.dock import BECDock
logger = bec_logger.logger
@ -17,13 +23,17 @@ class BECWidget(BECConnector):
# The icon name is the name of the icon in the icon theme, typically a name taken
# from fonts.google.com/icons. Override this in subclasses to set the icon name.
ICON_NAME = "widgets"
USER_ACCESS = ["remove"]
# pylint: disable=too-many-arguments
def __init__(
self,
client=None,
config: ConnectionConfig = None,
gui_id: str = None,
gui_id: str | None = None,
theme_update: bool = False,
name: str | None = None,
parent_dock: BECDock | None = None,
**kwargs,
):
"""
@ -45,9 +55,15 @@ 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, gui_id=gui_id, **kwargs)
# Set the theme to auto if it is not set yet
# Create a default name if None is provided
if name is None:
name = "bec_widget_init_without_name"
# name = self.__class__.__name__
# Check for invalid chars in the name
if not WidgetContainerUtils.has_name_valid_chars(name):
raise ValueError(f"Name {name} contains invalid characters.")
super().__init__(client=client, config=config, gui_id=gui_id, name=name)
self._parent_dock = parent_dock
app = QApplication.instance()
if not hasattr(app, "theme"):
# DO NOT SET THE THEME TO AUTO! Otherwise, the qwebengineview will segfault
@ -88,10 +104,13 @@ class BECWidget(BECConnector):
def cleanup(self):
"""Cleanup the widget."""
# needed here instead of closeEvent, to be checked why
# However, all widgets need to call super().cleanup() in their cleanup method
self.rpc_register.remove_rpc(self)
def closeEvent(self, event):
self.rpc_register.remove_rpc(self)
"""Wrap the close even to ensure the rpc_register is cleaned up."""
try:
self.cleanup()
finally:
super().closeEvent(event)
super().closeEvent(event) # pylint: disable=no-member

49
bec_widgets/utils/container_utils.py
View File

@ -1,30 +1,55 @@
from __future__ import annotations
import itertools
from typing import Type
from typing import Literal, Type
from qtpy.QtWidgets import QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
class WidgetContainerUtils:
# We need one handler that checks if a WIDGET of a given name is already created for that DOCKAREA
# 1. If the name exists, then it depends whether the name was auto-generated -> add _1 to the name
# or alternatively raise an error that it can't be added again ( just raise an error)
# 2. Dock names in between docks should also be unique
@staticmethod
def generate_unique_widget_id(container: dict, prefix: str = "widget") -> str:
"""
Generate a unique widget ID.
def has_name_valid_chars(name: str) -> bool:
"""Check if the name is valid.
Args:
container(dict): The container of widgets.
prefix(str): The prefix of the widget ID.
name(str): The name to be checked.
Returns:
widget_id(str): The unique widget ID.
bool: True if the name is valid, False otherwise.
"""
existing_ids = set(container.keys())
for i in itertools.count(1):
widget_id = f"{prefix}_{i}"
if widget_id not in existing_ids:
return widget_id
if not name or len(name) > 256:
return False # Don't accept empty names or names longer than 256 characters
check_value = name.replace("_", "").replace("-", "")
if not check_value.isalnum() or not check_value.isascii():
return False
return True
@staticmethod
def generate_unique_name(name: str, list_of_names: list[str] | None = None) -> str:
"""Generate a unique ID.
Args:
name(str): The name of the widget.
Returns:
tuple (str): The unique name
"""
if list_of_names is None:
list_of_names = []
ii = 0
while ii < 1000: # 1000 is arbritrary!
name_candidate = f"{name}_{ii}"
if name_candidate not in list_of_names:
return name_candidate
ii += 1
raise ValueError("Could not generate a unique name after within 1000 attempts.")
@staticmethod
def find_first_widget_by_class(

207
bec_widgets/widgets/containers/dock/dock.py
View File

@ -1,25 +1,33 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Literal, Optional
from typing import TYPE_CHECKING, Any, Literal, Optional, cast
from bec_lib.logger import bec_logger
from pydantic import Field
from pyqtgraph.dockarea import Dock, DockLabel
from qtpy import QtCore, QtGui
from bec_widgets.cli.client_utils import IGNORE_WIDGETS
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils import ConnectionConfig, GridLayoutManager
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.container_utils import WidgetContainerUtils
logger = bec_logger.logger
if TYPE_CHECKING:
from qtpy.QtWidgets import QWidget
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
class DockConfig(ConnectionConfig):
widgets: dict[str, Any] = Field({}, description="The widgets in the dock.")
position: Literal["bottom", "top", "left", "right", "above", "below"] = Field(
"bottom", description="The position of the dock."
)
parent_dock_area: Optional[str] = Field(
parent_dock_area: Optional[str] | None = Field(
None, description="The GUI ID of parent dock area of the dock."
)
@ -103,16 +111,17 @@ class BECDock(BECWidget, Dock):
ICON_NAME = "widgets"
USER_ACCESS = [
"_config_dict",
"_rpc_id",
"widget_list",
"element_list",
"elements",
"new",
"show",
"hide",
"show_title_bar",
"hide_title_bar",
"get_widgets_positions",
"set_title",
"add_widget",
"list_eligible_widgets",
"move_widget",
"remove_widget",
"hide_title_bar",
"available_widgets",
"delete",
"delete_all",
"remove",
"attach",
"detach",
@ -121,7 +130,7 @@ class BECDock(BECWidget, Dock):
def __init__(
self,
parent: QWidget | None = None,
parent_dock_area: QWidget | None = None,
parent_dock_area: BECDockArea | None = None,
config: DockConfig | None = None,
name: str | None = None,
client=None,
@ -129,21 +138,24 @@ class BECDock(BECWidget, Dock):
closable: bool = True,
**kwargs,
) -> None:
if config is None:
config = DockConfig(
widget_class=self.__class__.__name__, parent_dock_area=parent_dock_area.gui_id
widget_class=self.__class__.__name__,
parent_dock_area=parent_dock_area.gui_id if parent_dock_area else None,
)
else:
if isinstance(config, dict):
config = DockConfig(**config)
self.config = config
super().__init__(client=client, config=config, gui_id=gui_id)
super().__init__(
client=client, config=config, gui_id=gui_id, name=name
) # Name was checked and created in BEC Widget
label = CustomDockLabel(text=name, closable=closable)
Dock.__init__(self, name=name, label=label, **kwargs)
Dock.__init__(self, name=name, label=label, parent=self, **kwargs)
# Dock.__init__(self, name=name, **kwargs)
self.parent_dock_area = parent_dock_area
# Layout Manager
self.layout_manager = GridLayoutManager(self.layout)
@ -173,7 +185,18 @@ class BECDock(BECWidget, Dock):
super().float()
@property
def widget_list(self) -> list[BECWidget]:
def elements(self) -> dict[str, BECWidget]:
"""
Get the widgets in the dock.
Returns:
widgets(dict): The widgets in the dock.
"""
# pylint: disable=protected-access
return dict((widget._name, widget) for widget in self.element_list)
@property
def element_list(self) -> list[BECWidget]:
"""
Get the widgets in the dock.
@ -182,10 +205,6 @@ class BECDock(BECWidget, Dock):
"""
return self.widgets
@widget_list.setter
def widget_list(self, value: list[BECWidget]):
self.widgets = value
def hide_title_bar(self):
"""
Hide the title bar of the dock.
@ -194,6 +213,20 @@ class BECDock(BECWidget, Dock):
self.label.hide()
self.labelHidden = True
def show(self):
"""
Show the dock.
"""
super().show()
self.show_title_bar()
def hide(self):
"""
Hide the dock.
"""
self.hide_title_bar()
super().hide()
def show_title_bar(self):
"""
Hide the title bar of the dock.
@ -211,7 +244,6 @@ class BECDock(BECWidget, Dock):
"""
self.orig_area.docks[title] = self.orig_area.docks.pop(self.name())
self.setTitle(title)
self._name = title
def get_widgets_positions(self) -> dict:
"""
@ -222,7 +254,7 @@ class BECDock(BECWidget, Dock):
"""
return self.layout_manager.get_widgets_positions()
def list_eligible_widgets(
def available_widgets(
self,
) -> list: # TODO can be moved to some util mixin like container class for rpc widgets
"""
@ -233,20 +265,29 @@ class BECDock(BECWidget, Dock):
"""
return list(widget_handler.widget_classes.keys())
def add_widget(
def _get_list_of_widget_name_of_parent_dock_area(self):
docks = self.parent_dock_area.panel_list
widgets = []
for dock in docks:
widgets.extend(dock.elements.keys())
return widgets
def new(
self,
widget: BECWidget | str,
row=None,
col=0,
rowspan=1,
colspan=1,
name: str | None = None,
row: int | None = None,
col: int = 0,
rowspan: int = 1,
colspan: int = 1,
shift: Literal["down", "up", "left", "right"] = "down",
) -> BECWidget:
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add.
widget(QWidget): The widget to add. It can not be BECDock or BECDockArea.
name(str): The name of the widget.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
@ -254,15 +295,39 @@ class BECDock(BECWidget, Dock):
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
if row is None:
# row = cast(int, self.layout.rowCount()) # type:ignore
row = self.layout.rowCount()
# row = cast(int, row)
if self.layout_manager.is_position_occupied(row, col):
self.layout_manager.shift_widgets(shift, start_row=row)
existing_widgets_parent_dock = self._get_list_of_widget_name_of_parent_dock_area()
if name is not None: # Name is provided
if name in existing_widgets_parent_dock:
# pylint: disable=protected-access
raise ValueError(
f"Name {name} must be unique for widgets, but already exists in DockArea "
f"with name: {self.parent_dock_area._name} and id {self.parent_dock_area.gui_id}."
)
else: # Name is not provided
widget_class_name = widget if isinstance(widget, str) else widget.__class__.__name__
name = WidgetContainerUtils.generate_unique_name(
name=widget_class_name, list_of_names=existing_widgets_parent_dock
)
# Check that Widget is not BECDock or BECDockArea
widget_class_name = widget if isinstance(widget, str) else widget.__class__.__name__
if widget_class_name in IGNORE_WIDGETS:
raise ValueError(f"Widget {widget} can not be added to dock.")
if isinstance(widget, str):
widget = widget_handler.create_widget(widget)
widget = cast(
BECWidget,
widget_handler.create_widget(widget_type=widget, name=name, parent_dock=self),
)
else:
widget = widget
widget._name = name # pylint: disable=protected-access
self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
@ -294,37 +359,72 @@ class BECDock(BECWidget, Dock):
"""
self.float()
def remove_widget(self, widget_rpc_id: str):
"""
Remove a widget from the dock.
Args:
widget_rpc_id(str): The ID of the widget to remove.
"""
widget = self.rpc_register.get_rpc_by_id(widget_rpc_id)
self.layout.removeWidget(widget)
self.config.widgets.pop(widget_rpc_id, None)
widget.close()
def remove(self):
"""
Remove the dock from the parent dock area.
"""
# self.cleanup()
self.parent_dock_area.remove_dock(self.name())
self.parent_dock_area.delete(self._name)
def delete(self, widget_name: str) -> None:
"""
Remove a widget from the dock.
Args:
widget_name(str): Delete the widget with the given name.
"""
# pylint: disable=protected-access
widgets = [widget for widget in self.widgets if widget._name == widget_name]
if len(widgets) == 0:
logger.warning(
f"Widget with name {widget_name} not found in dock {self.name()}. "
f"Checking if gui_id was passed as widget_name."
)
# Try to find the widget in the RPC register, maybe the gui_id was passed as widget_name
widget = self.rpc_register.get_rpc_by_id(widget_name)
if widget is None:
logger.warning(
f"Widget not found for name or gui_id: {widget_name} in dock {self.name()}"
)
return
else:
widget = widgets[0]
self.layout.removeWidget(widget)
self.config.widgets.pop(widget._name, None)
if widget in self.widgets:
self.widgets.remove(widget)
widget.close()
# self._broadcast_update()
def delete_all(self):
"""
Remove all widgets from the dock.
"""
for widget in self.widgets:
self.delete(widget._name) # pylint: disable=protected-access
def cleanup(self):
"""
Clean up the dock, including all its widgets.
"""
for widget in self.widgets:
if hasattr(widget, "cleanup"):
widget.cleanup()
# Remove the dock from the parent dock area
if self.parent_dock_area:
self.parent_dock_area.dock_area.docks.pop(self.name(), None)
self.parent_dock_area.config.docks.pop(self.name(), None)
self.delete_all()
self.widgets.clear()
self.label.close()
self.label.deleteLater()
super().cleanup()
# def closeEvent(self, event): # pylint: disable=uselsess-parent-delegation
# """Close Event for dock and cleanup.
# This wrapper ensures that the BECWidget close event is triggered.
# If removed, the closeEvent from pyqtgraph will be triggered, which
# is not calling super().closeEvent(event) and will not trigger the BECWidget close event.
# """
# return super().closeEvent(event)
def close(self):
"""
Close the dock area and cleanup.
@ -332,4 +432,15 @@ class BECDock(BECWidget, Dock):
"""
self.cleanup()
super().close()
self.parent_dock_area.dock_area.docks.pop(self.name(), None)
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication([])
dock = BECDock(name="dock")
dock.show()
app.exec_()
sys.exit(app.exec_())

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

@ -4,12 +4,14 @@ from typing import Literal, Optional
from weakref import WeakValueDictionary
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from pydantic import Field
from pyqtgraph.dockarea.DockArea import DockArea
from qtpy.QtCore import QSize, Qt
from qtpy.QtGui import QPainter, QPaintEvent
from qtpy.QtWidgets import QApplication, QSizePolicy, QVBoxLayout, QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import SafeSlot
from bec_widgets.qt_utils.toolbar import (
ExpandableMenuAction,
@ -33,6 +35,8 @@ from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatus
from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
logger = bec_logger.logger
class DockAreaConfig(ConnectionConfig):
docks: dict[str, DockConfig] = Field({}, description="The docks in the dock area.")
@ -44,21 +48,19 @@ class DockAreaConfig(ConnectionConfig):
class BECDockArea(BECWidget, QWidget):
PLUGIN = True
USER_ACCESS = [
"_config_dict",
"selected_device",
"panels",
"save_state",
"remove_dock",
"restore_state",
"add_dock",
"clear_all",
"detach_dock",
"attach_all",
"_get_all_rpc",
"temp_areas",
"new",
"show",
"hide",
"panels",
"panel_list",
"delete",
"delete_all",
"remove",
"detach_dock",
"attach_all",
"selected_device",
"save_state",
"restore_state",
]
def __init__(
@ -67,6 +69,8 @@ class BECDockArea(BECWidget, QWidget):
config: DockAreaConfig | None = None,
client=None,
gui_id: str = None,
name: str | None = None,
**kwargs,
) -> None:
if config is None:
config = DockAreaConfig(widget_class=self.__class__.__name__)
@ -74,8 +78,9 @@ class BECDockArea(BECWidget, QWidget):
if isinstance(config, dict):
config = DockAreaConfig(**config)
self.config = config
super().__init__(client=client, config=config, gui_id=gui_id)
super().__init__(client=client, config=config, gui_id=gui_id, name=name, **kwargs)
QWidget.__init__(self, parent=parent)
self._parent = parent
self.layout = QVBoxLayout(self)
self.layout.setSpacing(5)
self.layout.setContentsMargins(0, 0, 0, 0)
@ -169,41 +174,41 @@ class BECDockArea(BECWidget, QWidget):
def _hook_toolbar(self):
# Menu Plot
self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect(
lambda: self.add_dock(widget="Waveform", prefix="waveform")
lambda: self._create_widget_from_toolbar(widget_name="Waveform")
)
self.toolbar.widgets["menu_plots"].widgets["multi_waveform"].triggered.connect(
lambda: self.add_dock(widget="BECMultiWaveformWidget", prefix="multi_waveform")
lambda: self._create_widget_from_toolbar(widget_name="BECMultiWaveformWidget")
)
self.toolbar.widgets["menu_plots"].widgets["image"].triggered.connect(
lambda: self.add_dock(widget="BECImageWidget", prefix="image")
lambda: self._create_widget_from_toolbar(widget_name="BECImageWidget")
)
self.toolbar.widgets["menu_plots"].widgets["motor_map"].triggered.connect(
lambda: self.add_dock(widget="BECMotorMapWidget", prefix="motor_map")
lambda: self._create_widget_from_toolbar(widget_name="BECMotorMapWidget")
)
# Menu Devices
self.toolbar.widgets["menu_devices"].widgets["scan_control"].triggered.connect(
lambda: self.add_dock(widget="ScanControl", prefix="scan_control")
lambda: self._create_widget_from_toolbar(widget_name="ScanControl")
)
self.toolbar.widgets["menu_devices"].widgets["positioner_box"].triggered.connect(
lambda: self.add_dock(widget="PositionerBox", prefix="positioner_box")
lambda: self._create_widget_from_toolbar(widget_name="PositionerBox")
)
# Menu Utils
self.toolbar.widgets["menu_utils"].widgets["queue"].triggered.connect(
lambda: self.add_dock(widget="BECQueue", prefix="queue")
lambda: self._create_widget_from_toolbar(widget_name="BECQueue")
)
self.toolbar.widgets["menu_utils"].widgets["status"].triggered.connect(
lambda: self.add_dock(widget="BECStatusBox", prefix="status")
lambda: self._create_widget_from_toolbar(widget_name="BECStatusBox")
)
self.toolbar.widgets["menu_utils"].widgets["vs_code"].triggered.connect(
lambda: self.add_dock(widget="VSCodeEditor", prefix="vs_code")
lambda: self._create_widget_from_toolbar(widget_name="VSCodeEditor")
)
self.toolbar.widgets["menu_utils"].widgets["progress_bar"].triggered.connect(
lambda: self.add_dock(widget="RingProgressBar", prefix="progress_bar")
lambda: self._create_widget_from_toolbar(widget_name="RingProgressBar")
)
self.toolbar.widgets["menu_utils"].widgets["log_panel"].triggered.connect(
lambda: self.add_dock(widget="LogPanel", prefix="log_panel")
lambda: self._create_widget_from_toolbar(widget_name="LogPanel")
)
# Icons
@ -211,6 +216,11 @@ class BECDockArea(BECWidget, QWidget):
self.toolbar.widgets["save_state"].action.triggered.connect(self.save_state)
self.toolbar.widgets["restore_state"].action.triggered.connect(self.restore_state)
@SafeSlot()
def _create_widget_from_toolbar(self, widget_name: str) -> None:
dock_name = WidgetContainerUtils.generate_unique_name(widget_name, self.panels.keys())
self.new(name=dock_name, widget=widget_name)
def paintEvent(self, event: QPaintEvent): # TODO decide if we want any default instructions
super().paintEvent(event)
if self._instructions_visible:
@ -218,7 +228,7 @@ class BECDockArea(BECWidget, QWidget):
painter.drawText(
self.rect(),
Qt.AlignCenter,
"Add docks using 'add_dock' method from CLI\n or \n Add widget docks using the toolbar",
"Add docks using 'new' method from CLI\n or \n Add widget docks using the toolbar",
)
@property
@ -243,7 +253,17 @@ class BECDockArea(BECWidget, QWidget):
@panels.setter
def panels(self, value: dict[str, BECDock]):
self.dock_area.docks = WeakValueDictionary(value)
self.dock_area.docks = WeakValueDictionary(value) # This can not work can it?
@property
def panel_list(self) -> list[BECDock]:
"""
Get the docks in the dock area.
Returns:
list: The docks in the dock area.
"""
return list(self.dock_area.docks.values())
@property
def temp_areas(self) -> list:
@ -287,36 +307,17 @@ class BECDockArea(BECWidget, QWidget):
self.config.docks_state = last_state
return last_state
def remove_dock(self, name: str):
"""
Remove a dock by name and ensure it is properly closed and cleaned up.
Args:
name(str): The name of the dock to remove.
"""
dock = self.dock_area.docks.pop(name, None)
self.config.docks.pop(name, None)
if dock:
dock.close()
dock.deleteLater()
if len(self.dock_area.docks) <= 1:
for dock in self.dock_area.docks.values():
dock.hide_title_bar()
else:
raise ValueError(f"Dock with name {name} does not exist.")
@SafeSlot(popup_error=True)
def add_dock(
def new(
self,
name: str = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = None,
name: str | None = None,
widget: str | QWidget | None = None,
widget_name: str | None = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = "bottom",
relative_to: BECDock | None = None,
closable: bool = True,
floating: bool = False,
prefix: str = "dock",
widget: str | QWidget | None = None,
row: int = None,
row: int | None = None,
col: int = 0,
rowspan: int = 1,
colspan: int = 1,
@ -326,12 +327,11 @@ class BECDockArea(BECWidget, QWidget):
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
prefix(str): The prefix for the dock name if no name is provided.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
@ -340,21 +340,20 @@ class BECDockArea(BECWidget, QWidget):
Returns:
BECDock: The created dock.
"""
if name is None:
name = WidgetContainerUtils.generate_unique_widget_id(
container=self.dock_area.docks, prefix=prefix
)
if name in set(self.dock_area.docks.keys()):
raise ValueError(f"Dock with name {name} already exists.")
if position is None:
position = "bottom"
dock_names = [dock._name for dock in self.panel_list] # pylint: disable=protected-access
if name is not None: # Name is provided
if name in dock_names:
raise ValueError(
f"Name {name} must be unique for docks, but already exists in DockArea "
f"with name: {self._name} and id {self.gui_id}."
)
else: # Name is not provided
name = WidgetContainerUtils.generate_unique_name(name="dock", list_of_names=dock_names)
dock = BECDock(name=name, parent_dock_area=self, closable=closable)
dock.config.position = position
self.config.docks[name] = dock.config
self.config.docks[dock.name()] = dock.config
# The dock.name is equal to the name passed to BECDock
self.dock_area.addDock(dock=dock, position=position, relativeTo=relative_to)
if len(self.dock_area.docks) <= 1:
@ -363,10 +362,11 @@ class BECDockArea(BECWidget, QWidget):
for dock in self.dock_area.docks.values():
dock.show_title_bar()
if widget is not None and isinstance(widget, str):
dock.add_widget(widget=widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
elif widget is not None and isinstance(widget, QWidget):
dock.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
if widget is not None:
# Check if widget name exists.
dock.new(
widget=widget, name=widget_name, row=row, col=col, rowspan=rowspan, colspan=colspan
)
if (
self._instructions_visible
): # TODO still decide how initial instructions should be handled
@ -404,49 +404,26 @@ class BECDockArea(BECWidget, QWidget):
Remove a temporary area from the dock area.
This is a patched method of pyqtgraph's removeTempArea
"""
if area not in self.dock_area.tempAreas:
# FIXME add some context for the logging, I am not sure which object is passed.
# It looks like a pyqtgraph.DockArea
logger.info(f"Attempted to remove dock_area, but was not floating.")
return
self.dock_area.tempAreas.remove(area)
area.window().close()
area.window().deleteLater()
def clear_all(self):
"""
Close all docks and remove all temp areas.
"""
self.attach_all()
for dock in dict(self.dock_area.docks).values():
dock.remove()
self.dock_area.docks.clear()
def cleanup(self):
"""
Cleanup the dock area.
"""
self.clear_all()
self.delete_all()
self.toolbar.close()
self.toolbar.deleteLater()
self.dock_area.close()
self.dock_area.deleteLater()
super().cleanup()
def closeEvent(self, event):
if self.parent() is None:
# we are at top-level (independent window)
if self.isVisible():
# we are visible => user clicked on [X]
# (when closeEvent is called from shutdown procedure,
# everything is hidden first)
# so, let's ignore "close", and do hide instead
event.ignore()
self.setVisible(False)
def close(self):
"""
Close the dock area and cleanup.
Has to be implemented to overwrite pyqtgraph event accept in Container close.
"""
self.cleanup()
super().close()
def show(self):
"""Show all windows including floating docks."""
super().show()
@ -465,17 +442,52 @@ class BECDockArea(BECWidget, QWidget):
continue
docks.window().hide()
def delete(self):
self.hide()
self.deleteLater()
def delete_all(self) -> None:
"""
Delete all docks.
"""
self.attach_all()
for dock_name in self.panels.keys():
self.delete(dock_name)
def delete(self, dock_name: str):
"""
Delete a dock by name.
Args:
dock_name(str): The name of the dock to delete.
"""
dock = self.dock_area.docks.pop(dock_name, None)
self.config.docks.pop(dock_name, None)
if dock:
dock.close()
dock.deleteLater()
if len(self.dock_area.docks) <= 1:
for dock in self.dock_area.docks.values():
dock.hide_title_bar()
else:
raise ValueError(f"Dock with name {dock_name} does not exist.")
# self._broadcast_update()
def remove(self) -> None:
"""Remove the dock area."""
self.close()
if __name__ == "__main__": # pragma: no cover
import sys
from bec_widgets.utils.colors import set_theme
app = QApplication([])
set_theme("auto")
dock_area = BECDockArea()
dock_1 = dock_area.new(name="dock_0", widget="Waveform")
# dock_1 = dock_area.new(name="dock_0", widget="Waveform")
dock_area.new(widget="Waveform")
dock_area.show()
dock_area.setGeometry(100, 100, 800, 600)
app.topLevelWidgets()
app.exec_()
sys.exit(app.exec_())

13
bec_widgets/widgets/containers/figure/figure.py
View File

@ -78,13 +78,7 @@ class WidgetHandler:
}
def create_widget(
self,
widget_type: str,
widget_id: str,
parent_figure,
parent_id: str,
config: dict = None,
**axis_kwargs,
self, widget_type: str, parent_figure, parent_id: str, config: dict = None, **axis_kwargs
) -> BECPlotBase:
"""
Create and configure a widget based on its type.
@ -109,7 +103,6 @@ class WidgetHandler:
widget_config_dict = {
"widget_class": widget_class.__name__,
"parent_id": parent_id,
"gui_id": widget_id,
**(config if config is not None else {}),
}
widget_config = config_class(**widget_config_dict)
@ -568,13 +561,13 @@ class BECFigure(BECWidget, pg.GraphicsLayoutWidget):
widget = self.widget_handler.create_widget(
widget_type=widget_type,
widget_id=widget_id,
parent_figure=self,
parent_id=self.gui_id,
config=config,
**axis_kwargs,
)
widget.set_gui_id(widget_id)
widget_id = widget.gui_id
widget.config.row = row
widget.config.col = col

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

@ -1,12 +1,17 @@
from bec_lib.logger import bec_logger
from qtpy.QtWidgets import QApplication, QMainWindow
from bec_widgets.utils import BECConnector
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
logger = bec_logger.logger
class BECMainWindow(QMainWindow, BECConnector):
def __init__(self, *args, **kwargs):
BECConnector.__init__(self, **kwargs)
class BECMainWindow(BECWidget, QMainWindow):
def __init__(self, gui_id: str = None, *args, **kwargs):
BECWidget.__init__(self, gui_id=gui_id, **kwargs)
QMainWindow.__init__(self, *args, **kwargs)
def _dump(self):
@ -33,9 +38,38 @@ class BECMainWindow(QMainWindow, BECConnector):
}
return info
def new_dock_area(self, name):
dock_area = BECDockArea()
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.
"""
rpc_register = RPCRegister()
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 = BECDockArea(name=name)
dock_area.resize(dock_area.minimumSizeHint())
dock_area.window().setWindowTitle(name)
# 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 cleanup(self):
# TODO
super().close()

2
bec_widgets/widgets/plots_next_gen/plot_base.py
View File

@ -942,7 +942,7 @@ class PlotBase(BECWidget, QWidget):
self.axis_settings_dialog.close()
self.axis_settings_dialog = None
self.cleanup_pyqtgraph()
self.rpc_register.remove_rpc(self)
super().cleanup()
def cleanup_pyqtgraph(self):
"""Cleanup pyqtgraph items."""

5
bec_widgets/widgets/plots_next_gen/waveform/waveform.py
View File

@ -117,10 +117,13 @@ class Waveform(PlotBase):
client=None,
gui_id: str | None = None,
popups: bool = True,
**kwargs,
):
if config is None:
config = WaveformConfig(widget_class=self.__class__.__name__)
super().__init__(parent=parent, config=config, client=client, gui_id=gui_id, popups=popups)
super().__init__(
parent=parent, config=config, client=client, gui_id=gui_id, popups=popups, **kwargs
)
# For PropertyManager identification
self.setObjectName("Waveform")

23
tests/end-2-end/conftest.py
View File

@ -27,7 +27,9 @@ def gui_id():
@contextmanager
def plot_server(gui_id, klass, client_lib):
dispatcher = BECDispatcher(client=client_lib) # Has to init singleton with fixture client
process, _ = _start_plot_process(gui_id, klass, client_lib._client._service_config.config_path)
process, _ = _start_plot_process(
gui_id, klass, gui_class_id="bec", config=client_lib._client._service_config.config_path
)
try:
while client_lib._client.connector.get(MessageEndpoints.gui_heartbeat(gui_id)) is None:
time.sleep(0.3)
@ -42,6 +44,7 @@ def plot_server(gui_id, klass, client_lib):
@pytest.fixture
def connected_client_figure(gui_id, bec_client_lib):
with plot_server(gui_id, BECFigure, bec_client_lib) as server:
yield server
@ -49,10 +52,11 @@ def connected_client_figure(gui_id, bec_client_lib):
def connected_client_gui_obj(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id)
try:
gui.start_server(wait=True)
gui.start(wait=True)
# gui._start_server(wait=True)
yield gui
finally:
gui.close()
gui.kill_server()
@pytest.fixture
@ -60,17 +64,18 @@ def connected_client_dock(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id)
gui._auto_updates_enabled = False
try:
gui.start_server(wait=True)
yield gui.main
gui.start(wait=True)
gui.window_list[0]
yield gui.window_list[0]
finally:
gui.close()
gui.kill_server()
@pytest.fixture
def connected_client_dock_w_auto_updates(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id)
try:
gui.start_server(wait=True)
yield gui, gui.main
gui._start_server(wait=True)
yield gui, gui.window_list[0]
finally:
gui.close()
gui.kill_server()

45
tests/end-2-end/test_bec_figure_rpc_e2e.py
View File

@ -1,14 +1,24 @@
import time
import numpy as np
import pytest
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform
from bec_widgets.tests.utils import check_remote_data_size
def test_rpc_waveform1d_custom_curve(connected_client_figure):
fig = BECFigure(connected_client_figure)
@pytest.fixture
def connected_figure(connected_client_gui_obj):
gui = connected_client_gui_obj
dock = gui.bec.new("dock")
fig = dock.new(name="fig", widget="BECFigure")
return fig
def test_rpc_waveform1d_custom_curve(connected_figure):
fig = connected_figure
# fig = BECFigure(connected_client_figure)
ax = fig.plot()
curve = ax.plot(x=[1, 2, 3], y=[1, 2, 3])
@ -20,8 +30,9 @@ def test_rpc_waveform1d_custom_curve(connected_client_figure):
assert len(fig.widgets[ax._rpc_id].curves) == 1
def test_rpc_plotting_shortcuts_init_configs(connected_client_figure, qtbot):
fig = BECFigure(connected_client_figure)
def test_rpc_plotting_shortcuts_init_configs(connected_figure, qtbot):
fig = connected_figure
# fig = BECFigure(connected_client_figure)
plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger")
@ -78,9 +89,9 @@ def test_rpc_plotting_shortcuts_init_configs(connected_client_figure, qtbot):
}
def test_rpc_waveform_scan(qtbot, connected_client_figure, bec_client_lib):
fig = BECFigure(connected_client_figure)
def test_rpc_waveform_scan(qtbot, connected_figure, bec_client_lib):
# fig = BECFigure(connected_client_figure)
fig = connected_figure
# add 3 different curves to track
plt = fig.plot(x_name="samx", y_name="bpm4i")
fig.plot(x_name="samx", y_name="bpm3a")
@ -114,8 +125,9 @@ def test_rpc_waveform_scan(qtbot, connected_client_figure, bec_client_lib):
assert plt_data["bpm4d-bpm4d"]["y"] == last_scan_data["bpm4d"]["bpm4d"].val
def test_rpc_image(connected_client_figure, bec_client_lib):
fig = BECFigure(connected_client_figure)
def test_rpc_image(connected_figure, bec_client_lib):
# fig = BECFigure(connected_client_figure)
fig = connected_figure
im = fig.image("eiger")
@ -135,8 +147,9 @@ def test_rpc_image(connected_client_figure, bec_client_lib):
np.testing.assert_equal(last_image_device, last_image_plot)
def test_rpc_motor_map(connected_client_figure, bec_client_lib):
fig = BECFigure(connected_client_figure)
def test_rpc_motor_map(connected_figure, bec_client_lib):
# fig = BECFigure(connected_client_figure)
fig = connected_figure
motor_map = fig.motor_map("samx", "samy")
@ -164,9 +177,10 @@ def test_rpc_motor_map(connected_client_figure, bec_client_lib):
)
def test_dap_rpc(connected_client_figure, bec_client_lib, qtbot):
def test_dap_rpc(connected_figure, bec_client_lib, qtbot):
fig = BECFigure(connected_client_figure)
fig = connected_figure
# fig = BECFigure(connected_client_figure)
plt = fig.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel")
client = bec_client_lib
@ -204,8 +218,9 @@ def test_dap_rpc(connected_client_figure, bec_client_lib, qtbot):
qtbot.waitUntil(wait_for_fit, timeout=10000)
def test_removing_subplots(connected_client_figure, bec_client_lib):
fig = BECFigure(connected_client_figure)
def test_removing_subplots(connected_figure, bec_client_lib):
# fig = BECFigure(connected_client_figure)
fig = connected_figure
plt = fig.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel")
im = fig.image(monitor="eiger")
mm = fig.motor_map(motor_x="samx", motor_y="samy")

10
tests/end-2-end/test_rpc_register_e2e.py
View File

@ -3,8 +3,9 @@ import pytest
from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform
def test_rpc_register_list_connections(connected_client_figure):
fig = BECFigure(connected_client_figure)
def test_rpc_register_list_connections(connected_client_gui_obj):
gui = connected_client_gui_obj
fig = gui.bec.new("fig").new(name="fig", widget="BECFigure")
plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger")
@ -36,5 +37,6 @@ def test_rpc_register_list_connections(connected_client_figure):
**image_item_expected,
}
assert len(all_connections) == 9
assert all_connections == all_connections_expected
assert len(all_connections) == 9 + 3 # gui, dock_area, dock
# In the old implementation, gui , dock_area and dock were not included in the _get_all_rpc() method
# assert all_connections == all_connections_expected

4
tests/unit_tests/test_bec_connector.py
View File

@ -30,7 +30,7 @@ def test_bec_connector_init_with_gui_id(mocked_client):
def test_bec_connector_set_gui_id(bec_connector):
bec_connector.set_gui_id("test_gui_id")
bec_connector._set_gui_id("test_gui_id")
assert bec_connector.config.gui_id == "test_gui_id"
@ -40,7 +40,7 @@ def test_bec_connector_change_config(bec_connector):
def test_bec_connector_get_obj_by_id(bec_connector):
bec_connector.set_gui_id("test_gui_id")
bec_connector._set_gui_id("test_gui_id")
assert bec_connector.get_obj_by_id("test_gui_id") == bec_connector
assert bec_connector.get_obj_by_id("test_gui_id_2") is None

68
tests/unit_tests/test_bec_dock.py
View File

@ -28,9 +28,9 @@ def test_bec_dock_area_add_remove_dock(bec_dock_area, qtbot):
initial_count = len(bec_dock_area.dock_area.docks)
# Adding 3 docks
d0 = bec_dock_area.add_dock()
d1 = bec_dock_area.add_dock()
d2 = bec_dock_area.add_dock()
d0 = bec_dock_area.new()
d1 = bec_dock_area.new()
d2 = bec_dock_area.new()
# Check if the docks were added
assert len(bec_dock_area.dock_area.docks) == initial_count + 3
@ -46,7 +46,7 @@ def test_bec_dock_area_add_remove_dock(bec_dock_area, qtbot):
# Remove docks
d0_name = d0.name()
bec_dock_area.remove_dock(d0_name)
bec_dock_area.delete(d0_name)
qtbot.wait(200)
d1.remove()
qtbot.wait(200)
@ -58,16 +58,16 @@ def test_bec_dock_area_add_remove_dock(bec_dock_area, qtbot):
def test_add_remove_bec_figure_to_dock(bec_dock_area):
d0 = bec_dock_area.add_dock()
fig = d0.add_widget("BECFigure")
d0 = bec_dock_area.new()
fig = d0.new("BECFigure")
plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger")
mm = fig.motor_map("samx", "samy")
mw = fig.multi_waveform("waveform1d")
assert len(bec_dock_area.dock_area.docks) == 1
assert len(d0.widgets) == 1
assert len(d0.widget_list) == 1
assert len(d0.elements) == 1
assert len(d0.element_list) == 1
assert len(fig.widgets) == 4
assert fig.config.widget_class == "BECFigure"
@ -78,20 +78,20 @@ def test_add_remove_bec_figure_to_dock(bec_dock_area):
def test_close_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_2")
d0 = bec_dock_area.new(name="dock_0")
d1 = bec_dock_area.new(name="dock_1")
d2 = bec_dock_area.new(name="dock_2")
bec_dock_area.clear_all()
bec_dock_area.delete_all()
qtbot.wait(200)
assert len(bec_dock_area.dock_area.docks) == 0
def test_undock_and_dock_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_4")
d3 = bec_dock_area.add_dock(name="dock_3")
d0 = bec_dock_area.new(name="dock_0")
d1 = bec_dock_area.new(name="dock_1")
d2 = bec_dock_area.new(name="dock_4")
d3 = bec_dock_area.new(name="dock_3")
d0.detach()
bec_dock_area.detach_dock("dock_1")
@ -114,28 +114,31 @@ def test_undock_and_dock_docks(bec_dock_area, qtbot):
###################################
def test_toolbar_add_plot_waveform(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["waveform"].trigger()
assert "waveform_1" in bec_dock_area.panels
assert bec_dock_area.panels["waveform_1"].widgets[0].config.widget_class == "Waveform"
assert "Waveform_0" in bec_dock_area.panels
assert bec_dock_area.panels["Waveform_0"].widgets[0].config.widget_class == "Waveform"
def test_toolbar_add_plot_image(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["image"].trigger()
assert "image_1" in bec_dock_area.panels
assert bec_dock_area.panels["image_1"].widgets[0].config.widget_class == "BECImageWidget"
assert "BECImageWidget_0" in bec_dock_area.panels
assert (
bec_dock_area.panels["BECImageWidget_0"].widgets[0].config.widget_class == "BECImageWidget"
)
def test_toolbar_add_plot_motor_map(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["motor_map"].trigger()
assert "motor_map_1" in bec_dock_area.panels
assert bec_dock_area.panels["motor_map_1"].widgets[0].config.widget_class == "BECMotorMapWidget"
assert "BECMotorMapWidget_0" in bec_dock_area.panels
assert (
bec_dock_area.panels["BECMotorMapWidget_0"].widgets[0].config.widget_class
== "BECMotorMapWidget"
)
def test_toolbar_add_device_positioner_box(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_devices"].widgets["positioner_box"].trigger()
assert "positioner_box_1" in bec_dock_area.panels
assert (
bec_dock_area.panels["positioner_box_1"].widgets[0].config.widget_class == "PositionerBox"
)
assert "PositionerBox_0" in bec_dock_area.panels
assert bec_dock_area.panels["PositionerBox_0"].widgets[0].config.widget_class == "PositionerBox"
def test_toolbar_add_utils_queue(bec_dock_area, bec_queue_msg_full):
@ -143,19 +146,20 @@ def test_toolbar_add_utils_queue(bec_dock_area, bec_queue_msg_full):
MessageEndpoints.scan_queue_status(), bec_queue_msg_full
)
bec_dock_area.toolbar.widgets["menu_utils"].widgets["queue"].trigger()
assert "queue_1" in bec_dock_area.panels
assert bec_dock_area.panels["queue_1"].widgets[0].config.widget_class == "BECQueue"
assert "BECQueue_0" in bec_dock_area.panels
assert bec_dock_area.panels["BECQueue_0"].widgets[0].config.widget_class == "BECQueue"
def test_toolbar_add_utils_status(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_utils"].widgets["status"].trigger()
assert "status_1" in bec_dock_area.panels
assert bec_dock_area.panels["status_1"].widgets[0].config.widget_class == "BECStatusBox"
assert "BECStatusBox_0" in bec_dock_area.panels
assert bec_dock_area.panels["BECStatusBox_0"].widgets[0].config.widget_class == "BECStatusBox"
def test_toolbar_add_utils_progress_bar(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_utils"].widgets["progress_bar"].trigger()
assert "progress_bar_1" in bec_dock_area.panels
assert "RingProgressBar_0" in bec_dock_area.panels
assert (
bec_dock_area.panels["progress_bar_1"].widgets[0].config.widget_class == "RingProgressBar"
bec_dock_area.panels["RingProgressBar_0"].widgets[0].config.widget_class
== "RingProgressBar"
)

29
tests/unit_tests/test_client_utils.py
View File

@ -12,7 +12,7 @@ from bec_widgets.tests.utils import FakeDevice
def cli_figure():
fig = BECFigure(gui_id="test")
with mock.patch.object(fig, "_run_rpc") as mock_rpc_call:
with mock.patch.object(fig, "gui_is_alive", return_value=True):
with mock.patch.object(fig, "_gui_is_alive", return_value=True):
yield fig, mock_rpc_call
@ -40,8 +40,17 @@ def test_rpc_call_accepts_device_as_input(cli_figure):
)
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", BECFigure, config)
command = ["bec-gui-server", "--id", "gui_id", "--gui_class", "BECFigure", "--hide"]
_start_plot_process("gui_id", BECFigure, "bec", config)
command = [
"bec-gui-server",
"--id",
"gui_id",
"--gui_class",
"BECFigure",
"--gui_class_id",
"bec",
"--hide",
]
if call_config:
command.extend(["--config", call_config])
mock_popen.assert_called_once_with(
@ -66,20 +75,24 @@ def test_client_utils_passes_client_config_to_server(bec_dispatcher):
mixin = BECGuiClient()
mixin._client = bec_dispatcher.client
mixin._gui_id = "gui_id"
mixin.gui_is_alive = mock.MagicMock()
mixin.gui_is_alive.side_effect = [True]
mixin._gui_is_alive = mock.MagicMock()
mixin._gui_is_alive.side_effect = [True]
try:
yield mixin
finally:
mixin.close()
mixin.kill_server()
with bec_client_mixin() as mixin:
with mock.patch("bec_widgets.cli.client_utils._start_plot_process") as mock_start_plot:
mock_start_plot.return_value = [mock.MagicMock(), mock.MagicMock()]
mixin.start_server(
mixin._start_server(
wait=False
) # the started event will not be set, wait=True would block forever
mock_start_plot.assert_called_once_with(
"gui_id", BECGuiClient, mixin._client._service_config.config, logger=mock.ANY
"gui_id",
BECGuiClient,
gui_class_id="bec",
config=mixin._client._service_config.config,
logger=mock.ANY,
)

2
tests/unit_tests/test_plot_base.py
View File

@ -14,7 +14,7 @@ def test_init_plot_base(qtbot, mocked_client):
plot_base = bec_figure.add_widget(widget_type="BECPlotBase", widget_id="test_plot")
assert plot_base is not None
assert plot_base.config.widget_class == "BECPlotBase"
assert plot_base.config.gui_id == "test_plot"
assert plot_base.config.gui_id == plot_base.gui_id
def test_plot_base_axes_by_separate_methods(qtbot, mocked_client):

12
tests/unit_tests/test_rpc_server.py
View File

@ -20,8 +20,10 @@ def test_rpc_server_start_server_without_service_config(mocked_cli_server):
"""
mock_server, mock_config, _ = mocked_cli_server
_start_server("gui_id", BECFigure, None)
mock_server.assert_called_once_with(gui_id="gui_id", config=mock_config(), gui_class=BECFigure)
_start_server("gui_id", BECFigure, config=None)
mock_server.assert_called_once_with(
gui_id="gui_id", config=mock_config(), gui_class=BECFigure, gui_class_id="bec"
)
@pytest.mark.parametrize(
@ -37,5 +39,7 @@ def test_rpc_server_start_server_with_service_config(mocked_cli_server, config,
"""
mock_server, mock_config, _ = mocked_cli_server
config = mock_config(**call_config)
_start_server("gui_id", BECFigure, config)
mock_server.assert_called_once_with(gui_id="gui_id", config=config, gui_class=BECFigure)
_start_server("gui_id", BECFigure, config=config)
mock_server.assert_called_once_with(
gui_id="gui_id", config=config, gui_class=BECFigure, gui_class_id="bec"
)