bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 11:41:49 +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 b085ef6e73
commit ac3c5a38e4
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. Create a default dock for the auto updates.
""" """
self.dock_name = "default_figure" self.dock_name = "default_figure"
self._default_dock = self.gui.add_dock(self.dock_name) self._default_dock = self.gui.new(self.dock_name)
self._default_dock.add_widget("BECFigure") self._default_dock.new("BECFigure")
self._default_fig = self._default_dock.widget_list[0] self._default_fig = self._default_dock.elements_list[0]
@staticmethod @staticmethod
def get_scan_info(msg) -> ScanInfo: 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): class AbortButton(RPCBase):
@property """A button that abort the scan."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
@ -230,14 +215,7 @@ class BECDock(RPCBase):
@property @property
@rpc_call @rpc_call
def _rpc_id(self) -> "str": def element_list(self) -> "list[BECWidget]":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def widget_list(self) -> "list[BECWidget]":
""" """
Get the widgets in the dock. Get the widgets in the dock.
@ -245,27 +223,58 @@ class BECDock(RPCBase):
widgets(list): The widgets in the dock. 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 @rpc_call
def show_title_bar(self): def show_title_bar(self):
""" """
Hide the title bar of the dock. 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 @rpc_call
def set_title(self, title: "str"): def set_title(self, title: "str"):
""" """
@ -276,29 +285,13 @@ class BECDock(RPCBase):
""" """
@rpc_call @rpc_call
def add_widget( def hide_title_bar(self):
self,
widget: "BECWidget | str",
row=None,
col=0,
rowspan=1,
colspan=1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
) -> "BECWidget":
""" """
Add a widget to the dock. Hide the title bar of 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.
""" """
@rpc_call @rpc_call
def list_eligible_widgets(self) -> "list": def available_widgets(self) -> "list":
""" """
List all widgets that can be added to the dock. List all widgets that can be added to the dock.
@ -307,23 +300,18 @@ class BECDock(RPCBase):
""" """
@rpc_call @rpc_call
def move_widget(self, widget: "QWidget", new_row: "int", new_col: "int"): def delete(self, widget_name: "str") -> "None":
"""
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"):
""" """
Remove a widget from the dock. Remove a widget from the dock.
Args: 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 @rpc_call
@ -346,21 +334,50 @@ class BECDock(RPCBase):
class BECDockArea(RPCBase): class BECDockArea(RPCBase):
@property
@rpc_call @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: Returns:
dict: The configuration of the widget. BECDock: The created dock.
""" """
@property
@rpc_call @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 @property
@ -372,76 +389,35 @@ class BECDockArea(RPCBase):
dock_dict(dict): The docks in the dock area. dock_dict(dict): The docks in the dock area.
""" """
@property
@rpc_call @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: Returns:
dict: The state of the dock area. list: The docks in the dock area.
""" """
@rpc_call @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: Args:
name(str): The name of the dock to remove. dock_name(str): The name of the dock to delete.
""" """
@rpc_call @rpc_call
def restore_state( def delete_all(self) -> "None":
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. Delete all docks.
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.
""" """
@rpc_call @rpc_call
def add_dock( def remove(self) -> "None":
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":
""" """
Add a dock to the dock area. Dock has QGridLayout as layout manager by default. Remove the dock area.
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.
""" """
@rpc_call @rpc_call
@ -462,40 +438,35 @@ class BECDockArea(RPCBase):
Return all floating docks to the dock area. Return all floating docks to the dock area.
""" """
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property @property
@rpc_call @rpc_call
def temp_areas(self) -> "list": def selected_device(self) -> "str":
"""
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):
""" """
None 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): class BECFigure(RPCBase):
@property @property
@ -1409,27 +1380,10 @@ class BECImageWidget(RPCBase):
class BECMainWindow(RPCBase): class BECMainWindow(RPCBase):
@property
@rpc_call @rpc_call
def _config_dict(self) -> "dict": def remove(self):
""" """
Get the configuration of the widget. Cleanup the BECConnector
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.
""" """
@ -2230,6 +2184,8 @@ class BECPlotBase(RPCBase):
class BECProgressBar(RPCBase): class BECProgressBar(RPCBase):
"""A custom progress bar with smooth transitions. The displayed text can be customized using a template."""
@rpc_call @rpc_call
def set_value(self, value): def set_value(self, value):
""" """
@ -2281,52 +2237,22 @@ class BECProgressBar(RPCBase):
class BECQueue(RPCBase): class BECQueue(RPCBase):
@property """Widget to display the BEC queue."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class BECStatusBox(RPCBase): class BECStatusBox(RPCBase):
@property """An autonomous widget to display the status of BEC services."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
@ -2845,6 +2771,8 @@ class Curve(RPCBase):
class DapComboBox(RPCBase): class DapComboBox(RPCBase):
"""The DAPComboBox widget is an extension to the QComboBox with all avaialble DAP model from BEC."""
@rpc_call @rpc_call
def select_y_axis(self, y_axis: str): def select_y_axis(self, y_axis: str):
""" """
@ -2883,156 +2811,66 @@ class DarkModeButton(RPCBase):
class DeviceBrowser(RPCBase): class DeviceBrowser(RPCBase):
@property
@rpc_call @rpc_call
def _config_dict(self) -> "dict": def remove(self):
""" """
Get the configuration of the widget. Cleanup the BECConnector
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.
""" """
class DeviceComboBox(RPCBase): class DeviceComboBox(RPCBase):
@property """Combobox widget for device input with autocomplete for device names."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class DeviceInputBase(RPCBase): class DeviceInputBase(RPCBase):
@property """Mixin base class for device input widgets."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class DeviceLineEdit(RPCBase): class DeviceLineEdit(RPCBase):
@property """Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class DeviceSignalInputBase(RPCBase): class DeviceSignalInputBase(RPCBase):
@property """Mixin base class for device signal input widgets."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class LMFitDialog(RPCBase): class LMFitDialog(RPCBase):
@property """Dialog for displaying the fit summary and params for LMFit DAP processes"""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class LogPanel(RPCBase): class LogPanel(RPCBase):
"""Displays a log panel"""
@rpc_call @rpc_call
def set_plain_text(self, text: str) -> None: def set_plain_text(self, text: str) -> None:
""" """
@ -3095,6 +2933,8 @@ class PositionIndicator(RPCBase):
class PositionerBox(RPCBase): class PositionerBox(RPCBase):
"""Simple Widget to control a positioner in box form"""
@rpc_call @rpc_call
def set_positioner(self, positioner: "str | Positioner"): def set_positioner(self, positioner: "str | Positioner"):
""" """
@ -3106,6 +2946,8 @@ class PositionerBox(RPCBase):
class PositionerBox2D(RPCBase): class PositionerBox2D(RPCBase):
"""Simple Widget to control two positioners in box form"""
@rpc_call @rpc_call
def set_positioner_hor(self, positioner: "str | Positioner"): def set_positioner_hor(self, positioner: "str | Positioner"):
""" """
@ -3126,31 +2968,18 @@ class PositionerBox2D(RPCBase):
class PositionerBoxBase(RPCBase): class PositionerBoxBase(RPCBase):
@property """Contains some core logic for positioner box widgets"""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class PositionerControlLine(RPCBase): class PositionerControlLine(RPCBase):
"""A widget that controls a single device."""
@rpc_call @rpc_call
def set_positioner(self, positioner: "str | Positioner"): def set_positioner(self, positioner: "str | Positioner"):
""" """
@ -3162,6 +2991,8 @@ class PositionerControlLine(RPCBase):
class PositionerGroup(RPCBase): class PositionerGroup(RPCBase):
"""Simple Widget to control a positioner in box form"""
@rpc_call @rpc_call
def set_positioners(self, device_names: "str"): def set_positioners(self, device_names: "str"):
""" """
@ -3172,52 +3003,22 @@ class PositionerGroup(RPCBase):
class ResetButton(RPCBase): class ResetButton(RPCBase):
@property """A button that resets the scan queue."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class ResumeButton(RPCBase): class ResumeButton(RPCBase):
@property """A button that continue scan queue."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
@ -3501,131 +3302,56 @@ class RingProgressBar(RPCBase):
class ScanControl(RPCBase): class ScanControl(RPCBase):
@property
@rpc_call @rpc_call
def _config_dict(self) -> "dict": def remove(self):
""" """
Get the configuration of the widget. Cleanup the BECConnector
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.
""" """
class ScanMetadata(RPCBase): class ScanMetadata(RPCBase):
@property """Dynamically generates a form for inclusion of metadata for a scan. Uses the"""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class SignalComboBox(RPCBase): class SignalComboBox(RPCBase):
@property """Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class SignalLineEdit(RPCBase): class SignalLineEdit(RPCBase):
@property """Line edit widget for device input with autocomplete for device names."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class StopButton(RPCBase): class StopButton(RPCBase):
@property """A button that stops the current scan."""
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call @rpc_call
def _get_all_rpc(self) -> "dict": def remove(self):
""" """
Get all registered RPC objects. Cleanup the BECConnector
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
""" """
class TextBox(RPCBase): class TextBox(RPCBase):
"""A widget that displays text in plain and HTML format"""
@rpc_call @rpc_call
def set_plain_text(self, text: str) -> None: 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): class Waveform(RPCBase):
@ -4060,6 +3789,8 @@ class Waveform(RPCBase):
class WebsiteWidget(RPCBase): class WebsiteWidget(RPCBase):
"""A simple widget to display a website"""
@rpc_call @rpc_call
def set_url(self, url: str) -> None: 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 from __future__ import annotations
import importlib import importlib
@ -7,13 +9,15 @@ import os
import select import select
import subprocess import subprocess
import threading import threading
import time
from contextlib import contextmanager from contextlib import contextmanager
from dataclasses import dataclass
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger 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 import bec_widgets.cli.client as client
from bec_widgets.cli.auto_updates import AutoUpdates from bec_widgets.cli.auto_updates import AutoUpdates
@ -23,16 +27,17 @@ if TYPE_CHECKING:
from bec_lib import messages from bec_lib import messages
from bec_lib.connector import MessageObject from bec_lib.connector import MessageObject
from bec_lib.device import DeviceBase from bec_lib.device import DeviceBase
from bec_lib.redis_connector import StreamMessage
from bec_widgets.utils.bec_dispatcher import BECDispatcher
else: else:
messages = lazy_import("bec_lib.messages") messages = lazy_import("bec_lib.messages")
# from bec_lib.connector import MessageObject # from bec_lib.connector import MessageObject
MessageObject = lazy_import_from("bec_lib.connector", ("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 logger = bec_logger.logger
IGNORE_WIDGETS = ["BECDockArea", "BECDock"]
def _filter_output(output: str) -> str: 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)}") 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. 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. process will not be captured.
""" """
# pylint: disable=subprocess-run-check # 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 config:
if isinstance(config, dict): if isinstance(config, dict):
config = json.dumps(config) 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): class RepeatTimer(threading.Timer):
"""RepeatTimer class."""
def run(self): def run(self):
while not self.finished.wait(self.interval): while not self.finished.wait(self.interval):
self.function(*self.args, **self.kwargs) self.function(*self.args, **self.kwargs)
# pylint: disable=protected-access
@contextmanager @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 timeout = client._startup_timeout
if not timeout: if not timeout:
if client.gui_is_alive(): if client._gui_is_alive():
# there is hope, let's wait a bit # there is hope, let's wait a bit
timeout = 1 timeout = 1
else: else:
@ -138,42 +158,63 @@ def wait_for_server(client):
yield yield
### ---------------------------- class WidgetNameSpace:
### NOTE def __repr__(self):
### it is far easier to extend the 'delete' method on the client side, console = Console()
### to know when the client is deleted, rather than listening to server table = Table(title="Available widgets for BEC CLI usage")
### to get notified. However, 'generate_cli.py' cannot add extra stuff table.add_column("Widget Name", justify="left", style="magenta")
### in the generated client module. So, here a class with the same name table.add_column("Description", justify="left")
### is created, and client module is patched. 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): class BECDockArea(client.BECDockArea):
def delete(self): """Extend the BECDockArea class and add namespaces to access widgets of docks."""
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
def __init__(self, gui_id=None, config=None, name=None, parent=None):
client.BECDockArea = BECDockArea super().__init__(gui_id, config, name, parent)
### ---------------------------- # Add namespaces for DockArea
self.elements = WidgetNameSpace()
@dataclass
class WidgetDesc:
title: str
widget: BECDockArea
class BECGuiClient(RPCBase): class BECGuiClient(RPCBase):
"""BEC GUI client class. Container for GUI applications within Python."""
_top_level = {} _top_level = {}
def __init__(self, **kwargs) -> None: def __init__(self, **kwargs) -> None:
super().__init__(**kwargs) super().__init__(**kwargs)
self._default_dock_name = "bec"
self._auto_updates_enabled = True self._auto_updates_enabled = True
self._auto_updates = None self._auto_updates = None
self._killed = False
self._startup_timeout = 0 self._startup_timeout = 0
self._gui_started_timer = None self._gui_started_timer = None
self._gui_started_event = threading.Event() self._gui_started_event = threading.Event()
@ -181,14 +222,21 @@ class BECGuiClient(RPCBase):
self._process_output_processing_thread = None self._process_output_processing_thread = None
@property @property
def windows(self): def windows(self) -> dict:
"""Dictionary with dock ares in the GUI."""
return self._top_level return self._top_level
@property @property
def auto_updates(self): def window_list(self) -> list:
if self._auto_updates_enabled: """List with dock areas in the GUI."""
with wait_for_server(self): return list(self._top_level.values())
return self._auto_updates
# 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: def _get_update_script(self) -> AutoUpdates | None:
eps = imd.entry_points(group="bec.widgets.auto_updates") 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 the module is not found, we skip it
if spec is None: if spec is None:
continue continue
return ep.load()(gui=self._top_level["main"].widget) return ep.load()(gui=self._top_level["main"])
except Exception as e: except Exception as e:
logger.error(f"Error loading auto update script from plugin: {str(e)}") logger.error(f"Error loading auto update script from plugin: {str(e)}")
return None return None
@property # FIXME AUTO UPDATES
def selected_device(self): # @property
""" # def selected_device(self) -> str | None:
Selected device for the plot. # """
""" # 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) # auto_update_config_ep = MessageEndpoints.gui_auto_update_config(self._gui_id)
if auto_update_config: # auto_update_config = self._client.connector.get(auto_update_config_ep)
return auto_update_config.selected_device # if auto_update_config:
return None # return auto_update_config.selected_device
# return None
@selected_device.setter # @selected_device.setter
def selected_device(self, device: str | DeviceBase): # def selected_device(self, device: str | DeviceBase):
if isinstance_based_on_class_name(device, "bec_lib.device.DeviceBase"): # if isinstance_based_on_class_name(device, "bec_lib.device.DeviceBase"):
self._client.connector.set_and_publish( # self._client.connector.set_and_publish(
MessageEndpoints.gui_auto_update_config(self._gui_id), # MessageEndpoints.gui_auto_update_config(self._gui_id),
messages.GUIAutoUpdateConfigMessage(selected_device=device.name), # messages.GUIAutoUpdateConfigMessage(selected_device=device.name),
) # )
elif isinstance(device, str): # elif isinstance(device, str):
self._client.connector.set_and_publish( # self._client.connector.set_and_publish(
MessageEndpoints.gui_auto_update_config(self._gui_id), # MessageEndpoints.gui_auto_update_config(self._gui_id),
messages.GUIAutoUpdateConfigMessage(selected_device=device), # messages.GUIAutoUpdateConfigMessage(selected_device=device),
) # )
else: # else:
raise ValueError("Device must be a string or a device object") # raise ValueError("Device must be a string or a device object")
def _start_update_script(self) -> None: # FIXME AUTO UPDATES
self._client.connector.register(MessageEndpoints.scan_status(), cb=self._handle_msg_update) # 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: # def _handle_msg_update(self, msg: StreamMessage) -> None:
if self.auto_updates is not None: # if self.auto_updates is not None:
# pylint: disable=protected-access # # pylint: disable=protected-access
return self._update_script_msg_parser(msg.value) # return self._update_script_msg_parser(msg.value)
def _update_script_msg_parser(self, msg: messages.BECMessage) -> None: # def _update_script_msg_parser(self, msg: messages.BECMessage) -> None:
if isinstance(msg, messages.ScanStatusMessage): # if isinstance(msg, messages.ScanStatusMessage):
if not self.gui_is_alive(): # if not self._gui_is_alive():
return # return
if self._auto_updates_enabled: # if self._auto_updates_enabled:
return self.auto_updates.do_update(msg) # return self.auto_updates.do_update(msg)
def _gui_post_startup(self): def _gui_post_startup(self):
self._top_level["main"] = WidgetDesc( # if self._auto_updates_enabled:
title="BEC Widgets", widget=BECDockArea(gui_id=self._gui_id) # 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._do_show_all()
self._gui_started_event.set() 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 Start the GUI server, and execute callback when it is launched
""" """
@ -272,7 +322,11 @@ class BECGuiClient(RPCBase):
self._startup_timeout = 5 self._startup_timeout = 5
self._gui_started_event.clear() self._gui_started_event.clear()
self._process, self._process_output_processing_thread = _start_plot_process( 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): def gui_started_callback(callback):
@ -283,7 +337,7 @@ class BECGuiClient(RPCBase):
threading.current_thread().cancel() threading.current_thread().cancel()
self._gui_started_timer = RepeatTimer( 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() self._gui_started_timer.start()
@ -295,53 +349,97 @@ class BECGuiClient(RPCBase):
return rpc_client._run_rpc("_dump") return rpc_client._run_rpc("_dump")
def start(self): def start(self):
return self.start_server() return self._start_server()
def _do_show_all(self): def _do_show_all(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self) rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client._run_rpc("show") rpc_client._run_rpc("show") # pylint: disable=protected-access
for window in self._top_level.values(): 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): with wait_for_server(self):
return self._do_show_all() return self._do_show_all()
def hide_all(self): def _hide_all(self):
with wait_for_server(self): with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self) rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client._run_rpc("hide") rpc_client._run_rpc("hide") # pylint: disable=protected-access
for window in self._top_level.values(): # because of the registry callbacks, we may have
window.widget.hide() # 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): def show(self):
"""Show the GUI window."""
if self._process is not None: if self._process is not None:
return self.show_all() return self._show_all()
# backward compatibility: show() was also starting server # backward compatibility: show() was also starting server
return self.start_server(wait=True) return self._start_server(wait=True)
def hide(self): def hide(self):
return self.hide_all() """Hide the GUI window."""
return self._hide_all()
@property def new(
def main(self): self,
"""Return client to main dock area (in main window)""" name: str | None = None,
with wait_for_server(self): wait: bool = True,
return self._top_level["main"].widget geometry: tuple[int, int, int, int] | None = None,
) -> BECDockArea:
"""Create a new top-level dock area.
def new(self, title): Args:
"""Ask main window to create a new top-level dock area""" name(str, optional): The name of the dock area. Defaults to None.
with wait_for_server(self): wait(bool, optional): Whether to wait for the server to start. Defaults to True.
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self) geometry(tuple[int, int, int, int] | None): The geometry of the dock area (pos_x, pos_y, w, h)
widget = rpc_client._run_rpc("new_dock_area", title) Returns:
self._top_level[widget._gui_id] = WidgetDesc(title=title, widget=widget) BECDockArea: The new dock area.
return widget
def close(self) -> None:
""" """
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._top_level.clear()
self._killed = True
if self._gui_started_timer is not None: if self._gui_started_timer is not None:
self._gui_started_timer.cancel() self._gui_started_timer.cancel()
@ -357,3 +455,17 @@ class BECGuiClient(RPCBase):
self._process_output_processing_thread.join() self._process_output_processing_thread.join()
self._process.wait() self._process.wait()
self._process = None 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""" self.content += f"""
class {class_name}(RPCBase):""" 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: if not cls.USER_ACCESS:
self.content += """... self.content += """...
""" """
for method in cls.USER_ACCESS: for method in cls.USER_ACCESS:
is_property_setter = False is_property_setter = False
obj = getattr(cls, method, None) 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 threading
import uuid import uuid
from functools import wraps 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.client import BECClient
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
@ -44,7 +44,7 @@ def rpc_call(func):
for key, val in kwargs.items(): for key, val in kwargs.items():
if hasattr(val, "name"): if hasattr(val, "name"):
kwargs[key] = 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") raise RuntimeError("GUI is not alive")
return self._run_rpc(func.__name__, *args, **kwargs) return self._run_rpc(func.__name__, *args, **kwargs)
@ -61,10 +61,17 @@ class RPCResponseTimeoutError(Exception):
class RPCBase: 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._client = BECClient() # BECClient is a singleton; here, we simply get the instance
self._config = config if config is not None else {} 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._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._parent = parent
self._msg_wait_event = threading.Event() self._msg_wait_event = threading.Event()
self._rpc_response = None self._rpc_response = None
@ -74,7 +81,20 @@ class RPCBase:
def __repr__(self): def __repr__(self):
type_ = type(self) type_ = type(self)
qualname = type_.__qualname__ 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 @property
def _root(self): def _root(self):
@ -88,7 +108,7 @@ class RPCBase:
parent = parent._parent parent = parent._parent
return 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. Run the RPC call.
@ -165,7 +185,7 @@ class RPCBase:
return cls(parent=self, **msg_result) return cls(parent=self, **msg_result)
return msg_result return msg_result
def gui_is_alive(self): def _gui_is_alive(self):
""" """
Check if the GUI is alive. 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 __future__ import annotations
from functools import wraps
from threading import Lock from threading import Lock
from typing import TYPE_CHECKING, Callable
from weakref import WeakValueDictionary from weakref import WeakValueDictionary
from bec_lib.logger import bec_logger
from qtpy.QtCore import QObject 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: class RPCRegister:
""" """
@ -49,7 +60,7 @@ class RPCRegister:
raise ValueError(f"RPC object {rpc} must have a 'gui_id' attribute.") raise ValueError(f"RPC object {rpc} must have a 'gui_id' attribute.")
self._rpc_register.pop(rpc.gui_id, None) 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. 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. gui_id(str): The ID of the RPC object to be retrieved.
Returns: 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) rpc_object = self._rpc_register.get(gui_id, None)
return rpc_object 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: def list_all_connections(self) -> dict:
""" """
List all the registered RPC objects. List all the registered RPC objects.
@ -73,6 +98,19 @@ class RPCRegister:
connections = dict(self._rpc_register) connections = dict(self._rpc_register)
return connections 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 @classmethod
def reset_singleton(cls): def reset_singleton(cls):
""" """

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

@ -1,6 +1,9 @@
from __future__ import annotations 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: class RPCWidgetHandler:
@ -10,7 +13,7 @@ class RPCWidgetHandler:
self._widget_classes = None self._widget_classes = None
@property @property
def widget_classes(self): def widget_classes(self) -> dict[str, Any]:
""" """
Get the available widget classes. Get the available widget classes.
@ -19,7 +22,7 @@ class RPCWidgetHandler:
""" """
if self._widget_classes is None: if self._widget_classes is None:
self.update_available_widgets() self.update_available_widgets()
return self._widget_classes return self._widget_classes # type: ignore
def update_available_widgets(self): def update_available_widgets(self):
""" """
@ -31,24 +34,27 @@ class RPCWidgetHandler:
from bec_widgets.utils.plugin_utils import get_custom_classes from bec_widgets.utils.plugin_utils import get_custom_classes
clss = get_custom_classes("bec_widgets") 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. Create a widget from an RPC message.
Args: Args:
widget_type(str): The type of the widget. widget_type(str): The type of the widget.
name (str): The name of the widget.
**kwargs: The keyword arguments for the widget. **kwargs: The keyword arguments for the widget.
Returns: Returns:
widget(BECConnector): The created widget. widget(BECWidget): The created widget.
""" """
if self._widget_classes is None: if self._widget_classes is None:
self.update_available_widgets() 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: if widget_class:
return widget_class(**kwargs) return widget_class(name=name, **kwargs)
raise ValueError(f"Unknown widget type: {widget_type}") 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 qtpy.QtCore import Qt, QTimer
from redis.exceptions import RedisError 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.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.utils import BECDispatcher from bec_widgets.utils import BECDispatcher
@ -36,6 +37,8 @@ def rpc_exception_hook(err_func):
old_exception_hook = popup.custom_exception_hook old_exception_hook = popup.custom_exception_hook
# install err_func, if it is a callable # 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): def custom_exception_hook(self, exc_type, value, tb, **kwargs):
err_func({"error": popup.get_error_message(exc_type, value, tb)}) err_func({"error": popup.get_error_message(exc_type, value, tb)})
@ -56,16 +59,18 @@ class BECWidgetsCLIServer:
dispatcher: BECDispatcher = None, dispatcher: BECDispatcher = None,
client=None, client=None,
config=None, config=None,
gui_class: Union[BECFigure, BECDockArea] = BECFigure, gui_class: Union[BECFigure, BECDockArea] = BECDockArea,
gui_class_id: str = "bec",
) -> None: ) -> None:
self.status = messages.BECStatus.BUSY self.status = messages.BECStatus.BUSY
self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher
self.client = self.dispatcher.client if client is None else client self.client = self.dispatcher.client if client is None else client
self.client.start() self.client.start()
self.gui_id = gui_id self.gui_id = gui_id
self.gui = gui_class(gui_id=self.gui_id) # register broadcast callback
self.rpc_register = RPCRegister() 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.dispatcher.connect_slot(
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id) self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
@ -78,6 +83,7 @@ class BECWidgetsCLIServer:
self.status = messages.BECStatus.RUNNING self.status = messages.BECStatus.RUNNING
logger.success(f"Server started with gui_id: {self.gui_id}") logger.success(f"Server started with gui_id: {self.gui_id}")
# Create initial object -> BECFigure or BECDockArea
def on_rpc_update(self, msg: dict, metadata: dict): def on_rpc_update(self, msg: dict, metadata: dict):
request_id = metadata.get("request_id") request_id = metadata.get("request_id")
@ -135,6 +141,9 @@ class BECWidgetsCLIServer:
if isinstance(obj, BECConnector): if isinstance(obj, BECConnector):
return { return {
"gui_id": obj.gui_id, "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__, "widget_class": obj.__class__.__name__,
"config": obj.config.model_dump(), "config": obj.config.model_dump(),
"__rpc__": True, "__rpc__": True,
@ -179,7 +188,12 @@ class SimpleFileLikeFromLogOutputFunc:
return 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: if config:
try: try:
config = json.loads(config) config = json.loads(config)
@ -196,7 +210,9 @@ def _start_server(gui_id: str, gui_class: Union[BECFigure, BECDockArea], config:
# service_name="BECWidgetsCLIServer", # service_name="BECWidgetsCLIServer",
# service_config=service_config.service_config, # 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 return server
@ -217,6 +233,12 @@ def main():
type=str, type=str,
help="Name of the gui class to be rendered. Possible values: \n- BECFigure\n- BECDockArea", 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("--config", type=str, help="Config file or config string.")
parser.add_argument("--hide", action="store_true", help="Hide on startup") 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 # store gui id within QApplication object, to make it available to all widgets
app.gui_id = args.id 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 = BECMainWindow(gui_id=f"{server.gui_id}:window")
win.setAttribute(Qt.WA_ShowWithoutActivating) win.setAttribute(Qt.WA_ShowWithoutActivating)
win.setWindowTitle("BEC Widgets") win.setWindowTitle("BEC")
RPCRegister().add_rpc(win) RPCRegister().add_rpc(win)
gui = server.gui gui = server.gui
win.setCentralWidget(gui) win.setCentralWidget(gui)
if not args.hide: 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): def _init_dock(self):
self.d0 = self.dock.add_dock(name="dock_0") self.d0 = self.dock.new(name="dock_0")
self.mm = self.d0.add_widget("BECMotorMapWidget") self.mm = self.d0.new("BECMotorMapWidget")
self.mm.change_motors("samx", "samy") self.mm.change_motors("samx", "samy")
self.d1 = self.dock.add_dock(name="dock_1", position="right") self.d1 = self.dock.new(name="dock_1", position="right")
self.im = self.d1.add_widget("BECImageWidget") self.im = self.d1.new("BECImageWidget")
self.im.image("waveform", "1d") 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.mw = None # self.wf.multi_waveform(monitor="waveform") # , config=config)
self.dock.save_state() self.dock.save_state()

52
bec_widgets/utils/bec_connector.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
import os import os
import time import time
import uuid import uuid
from datetime import datetime
from typing import TYPE_CHECKING, Optional from typing import TYPE_CHECKING, Optional
from bec_lib.logger import bec_logger 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.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.qt_utils.error_popups import SafeSlot as pyqtSlot 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 from bec_widgets.utils.yaml_dialog import load_yaml, load_yaml_gui, save_yaml, save_yaml_gui
if TYPE_CHECKING: if TYPE_CHECKING:
@ -39,8 +41,7 @@ class ConnectionConfig(BaseModel):
"""Generate a GUI ID if none is provided.""" """Generate a GUI ID if none is provided."""
if v is None: if v is None:
widget_class = values.data["widget_class"] widget_class = values.data["widget_class"]
v = f"{widget_class}_{str(time.time())}" v = f"{widget_class}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S_%f')}"
return v
return v return v
@ -75,7 +76,13 @@ class BECConnector:
USER_ACCESS = ["_config_dict", "_get_all_rpc", "_rpc_id"] USER_ACCESS = ["_config_dict", "_get_all_rpc", "_rpc_id"]
EXIT_HANDLERS = {} 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 # BEC related connections
self.bec_dispatcher = BECDispatcher(client=client) self.bec_dispatcher = BECDispatcher(client=client)
self.client = self.bec_dispatcher.client if client is None else 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__) 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: if gui_id:
self.config.gui_id = gui_id self.config.gui_id = gui_id
self.gui_id = gui_id self.gui_id: str = gui_id
else: else:
self.gui_id = self.config.gui_id self.gui_id: str = self.config.gui_id # type: ignore
if name is None:
# register widget to rpc register name = self.__class__.__name__
# be careful: when registering, and the object is not a BECWidget, else:
# cleanup has to be called manually since there is no 'closeEvent' 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 = RPCRegister()
self.rpc_register.add_rpc(self) self.rpc_register.add_rpc(self)
@ -195,6 +209,7 @@ class BECConnector:
""" """
self.config = config 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: def apply_config(self, config: dict, generate_new_id: bool = True) -> None:
""" """
Apply the configuration to the widget. Apply the configuration to the widget.
@ -207,11 +222,12 @@ class BECConnector:
if generate_new_id is True: if generate_new_id is True:
gui_id = str(uuid.uuid4()) gui_id = str(uuid.uuid4())
self.rpc_register.remove_rpc(self) self.rpc_register.remove_rpc(self)
self.set_gui_id(gui_id) self._set_gui_id(gui_id)
self.rpc_register.add_rpc(self) self.rpc_register.add_rpc(self)
else: else:
self.gui_id = self.config.gui_id 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): def load_config(self, path: str | None = None, gui: bool = False):
""" """
Load the configuration of the widget from YAML. 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") file_path = os.path.join(path, f"{self.__class__.__name__}_config.yaml")
save_yaml(file_path, self._config_dict) save_yaml(file_path, self._config_dict)
@pyqtSlot(str) # @pyqtSlot(str)
def set_gui_id(self, gui_id: str) -> None: def _set_gui_id(self, gui_id: str) -> None:
""" """
Set the GUI ID for the widget. Set the GUI ID for the widget.
@ -288,9 +304,21 @@ class BECConnector:
Args: Args:
config (ConnectionConfig | dict): Configuration settings. config (ConnectionConfig | dict): Configuration settings.
""" """
gui_id = getattr(config, "gui_id", None)
if isinstance(config, dict): if isinstance(config, dict):
config = ConnectionConfig(**config) config = ConnectionConfig(**config)
self.config = 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: 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 __future__ import annotations
from typing import TYPE_CHECKING
import darkdetect import darkdetect
from bec_lib.logger import bec_logger from bec_lib.logger import bec_logger
from qtpy.QtCore import Slot 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.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import set_theme 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 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 # 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. # from fonts.google.com/icons. Override this in subclasses to set the icon name.
ICON_NAME = "widgets" ICON_NAME = "widgets"
USER_ACCESS = ["remove"]
# pylint: disable=too-many-arguments
def __init__( def __init__(
self, self,
client=None, client=None,
config: ConnectionConfig = None, config: ConnectionConfig = None,
gui_id: str = None, gui_id: str | None = None,
theme_update: bool = False, theme_update: bool = False,
name: str | None = None,
parent_dock: BECDock | None = None,
**kwargs, **kwargs,
): ):
""" """
@ -45,9 +55,15 @@ class BECWidget(BECConnector):
""" """
if not isinstance(self, QWidget): if not isinstance(self, QWidget):
raise RuntimeError(f"{repr(self)} is not a subclass of QWidget") raise RuntimeError(f"{repr(self)} is not a subclass of QWidget")
super().__init__(client=client, config=config, gui_id=gui_id, **kwargs) # Create a default name if None is provided
if name is None:
# Set the theme to auto if it is not set yet 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() app = QApplication.instance()
if not hasattr(app, "theme"): if not hasattr(app, "theme"):
# DO NOT SET THE THEME TO AUTO! Otherwise, the qwebengineview will segfault # DO NOT SET THE THEME TO AUTO! Otherwise, the qwebengineview will segfault
@ -88,10 +104,13 @@ class BECWidget(BECConnector):
def cleanup(self): def cleanup(self):
"""Cleanup the widget.""" """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): def closeEvent(self, event):
self.rpc_register.remove_rpc(self) """Wrap the close even to ensure the rpc_register is cleaned up."""
try: try:
self.cleanup() self.cleanup()
finally: 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 from __future__ import annotations
import itertools import itertools
from typing import Type from typing import Literal, Type
from qtpy.QtWidgets import QWidget from qtpy.QtWidgets import QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
class WidgetContainerUtils: 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 @staticmethod
def generate_unique_widget_id(container: dict, prefix: str = "widget") -> str: def has_name_valid_chars(name: str) -> bool:
""" """Check if the name is valid.
Generate a unique widget ID.
Args: Args:
container(dict): The container of widgets. name(str): The name to be checked.
prefix(str): The prefix of the widget ID.
Returns: Returns:
widget_id(str): The unique widget ID. bool: True if the name is valid, False otherwise.
""" """
existing_ids = set(container.keys()) if not name or len(name) > 256:
for i in itertools.count(1): return False # Don't accept empty names or names longer than 256 characters
widget_id = f"{prefix}_{i}" check_value = name.replace("_", "").replace("-", "")
if widget_id not in existing_ids: if not check_value.isalnum() or not check_value.isascii():
return widget_id 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 @staticmethod
def find_first_widget_by_class( 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 __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 pydantic import Field
from pyqtgraph.dockarea import Dock, DockLabel from pyqtgraph.dockarea import Dock, DockLabel
from qtpy import QtCore, QtGui 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.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils import ConnectionConfig, GridLayoutManager from bec_widgets.utils import ConnectionConfig, GridLayoutManager
from bec_widgets.utils.bec_widget import BECWidget from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.container_utils import WidgetContainerUtils
logger = bec_logger.logger
if TYPE_CHECKING: if TYPE_CHECKING:
from qtpy.QtWidgets import QWidget from qtpy.QtWidgets import QWidget
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
class DockConfig(ConnectionConfig): class DockConfig(ConnectionConfig):
widgets: dict[str, Any] = Field({}, description="The widgets in the dock.") widgets: dict[str, Any] = Field({}, description="The widgets in the dock.")
position: Literal["bottom", "top", "left", "right", "above", "below"] = Field( position: Literal["bottom", "top", "left", "right", "above", "below"] = Field(
"bottom", description="The position of the dock." "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." None, description="The GUI ID of parent dock area of the dock."
) )
@ -103,16 +111,17 @@ class BECDock(BECWidget, Dock):
ICON_NAME = "widgets" ICON_NAME = "widgets"
USER_ACCESS = [ USER_ACCESS = [
"_config_dict", "_config_dict",
"_rpc_id", "element_list",
"widget_list", "elements",
"new",
"show",
"hide",
"show_title_bar", "show_title_bar",
"hide_title_bar",
"get_widgets_positions",
"set_title", "set_title",
"add_widget", "hide_title_bar",
"list_eligible_widgets", "available_widgets",
"move_widget", "delete",
"remove_widget", "delete_all",
"remove", "remove",
"attach", "attach",
"detach", "detach",
@ -121,7 +130,7 @@ class BECDock(BECWidget, Dock):
def __init__( def __init__(
self, self,
parent: QWidget | None = None, parent: QWidget | None = None,
parent_dock_area: QWidget | None = None, parent_dock_area: BECDockArea | None = None,
config: DockConfig | None = None, config: DockConfig | None = None,
name: str | None = None, name: str | None = None,
client=None, client=None,
@ -129,21 +138,24 @@ class BECDock(BECWidget, Dock):
closable: bool = True, closable: bool = True,
**kwargs, **kwargs,
) -> None: ) -> None:
if config is None: if config is None:
config = DockConfig( 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: else:
if isinstance(config, dict): if isinstance(config, dict):
config = DockConfig(**config) config = DockConfig(**config)
self.config = 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) 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) # Dock.__init__(self, name=name, **kwargs)
self.parent_dock_area = parent_dock_area self.parent_dock_area = parent_dock_area
# Layout Manager # Layout Manager
self.layout_manager = GridLayoutManager(self.layout) self.layout_manager = GridLayoutManager(self.layout)
@ -173,7 +185,18 @@ class BECDock(BECWidget, Dock):
super().float() super().float()
@property @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. Get the widgets in the dock.
@ -182,10 +205,6 @@ class BECDock(BECWidget, Dock):
""" """
return self.widgets return self.widgets
@widget_list.setter
def widget_list(self, value: list[BECWidget]):
self.widgets = value
def hide_title_bar(self): def hide_title_bar(self):
""" """
Hide the title bar of the dock. Hide the title bar of the dock.
@ -194,6 +213,20 @@ class BECDock(BECWidget, Dock):
self.label.hide() self.label.hide()
self.labelHidden = True 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): def show_title_bar(self):
""" """
Hide the title bar of the dock. 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.orig_area.docks[title] = self.orig_area.docks.pop(self.name())
self.setTitle(title) self.setTitle(title)
self._name = title
def get_widgets_positions(self) -> dict: def get_widgets_positions(self) -> dict:
""" """
@ -222,7 +254,7 @@ class BECDock(BECWidget, Dock):
""" """
return self.layout_manager.get_widgets_positions() return self.layout_manager.get_widgets_positions()
def list_eligible_widgets( def available_widgets(
self, self,
) -> list: # TODO can be moved to some util mixin like container class for rpc widgets ) -> 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()) 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, self,
widget: BECWidget | str, widget: BECWidget | str,
row=None, name: str | None = None,
col=0, row: int | None = None,
rowspan=1, col: int = 0,
colspan=1, rowspan: int = 1,
colspan: int = 1,
shift: Literal["down", "up", "left", "right"] = "down", shift: Literal["down", "up", "left", "right"] = "down",
) -> BECWidget: ) -> BECWidget:
""" """
Add a widget to the dock. Add a widget to the dock.
Args: 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. 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. col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span. 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. shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
""" """
if row is None: if row is None:
# row = cast(int, self.layout.rowCount()) # type:ignore
row = self.layout.rowCount() row = self.layout.rowCount()
# row = cast(int, row)
if self.layout_manager.is_position_occupied(row, col): if self.layout_manager.is_position_occupied(row, col):
self.layout_manager.shift_widgets(shift, start_row=row) 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): 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: else:
widget = widget widget._name = name # pylint: disable=protected-access
self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan) self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
@ -294,37 +359,72 @@ class BECDock(BECWidget, Dock):
""" """
self.float() 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): def remove(self):
""" """
Remove the dock from the parent dock area. Remove the dock from the parent dock area.
""" """
# self.cleanup() self.parent_dock_area.delete(self._name)
self.parent_dock_area.remove_dock(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): def cleanup(self):
""" """
Clean up the dock, including all its widgets. Clean up the dock, including all its widgets.
""" """
for widget in self.widgets: # Remove the dock from the parent dock area
if hasattr(widget, "cleanup"): if self.parent_dock_area:
widget.cleanup() 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.widgets.clear()
self.label.close() self.label.close()
self.label.deleteLater() self.label.deleteLater()
super().cleanup() 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): def close(self):
""" """
Close the dock area and cleanup. Close the dock area and cleanup.
@ -332,4 +432,15 @@ class BECDock(BECWidget, Dock):
""" """
self.cleanup() self.cleanup()
super().close() 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 weakref import WeakValueDictionary
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from pydantic import Field from pydantic import Field
from pyqtgraph.dockarea.DockArea import DockArea from pyqtgraph.dockarea.DockArea import DockArea
from qtpy.QtCore import QSize, Qt from qtpy.QtCore import QSize, Qt
from qtpy.QtGui import QPainter, QPaintEvent from qtpy.QtGui import QPainter, QPaintEvent
from qtpy.QtWidgets import QApplication, QSizePolicy, QVBoxLayout, QWidget 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.error_popups import SafeSlot
from bec_widgets.qt_utils.toolbar import ( from bec_widgets.qt_utils.toolbar import (
ExpandableMenuAction, 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.logpanel.logpanel import LogPanel
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
logger = bec_logger.logger
class DockAreaConfig(ConnectionConfig): class DockAreaConfig(ConnectionConfig):
docks: dict[str, DockConfig] = Field({}, description="The docks in the dock area.") docks: dict[str, DockConfig] = Field({}, description="The docks in the dock area.")
@ -44,21 +48,19 @@ class DockAreaConfig(ConnectionConfig):
class BECDockArea(BECWidget, QWidget): class BECDockArea(BECWidget, QWidget):
PLUGIN = True PLUGIN = True
USER_ACCESS = [ USER_ACCESS = [
"_config_dict", "new",
"selected_device",
"panels",
"save_state",
"remove_dock",
"restore_state",
"add_dock",
"clear_all",
"detach_dock",
"attach_all",
"_get_all_rpc",
"temp_areas",
"show", "show",
"hide", "hide",
"panels",
"panel_list",
"delete", "delete",
"delete_all",
"remove",
"detach_dock",
"attach_all",
"selected_device",
"save_state",
"restore_state",
] ]
def __init__( def __init__(
@ -67,6 +69,8 @@ class BECDockArea(BECWidget, QWidget):
config: DockAreaConfig | None = None, config: DockAreaConfig | None = None,
client=None, client=None,
gui_id: str = None, gui_id: str = None,
name: str | None = None,
**kwargs,
) -> None: ) -> None:
if config is None: if config is None:
config = DockAreaConfig(widget_class=self.__class__.__name__) config = DockAreaConfig(widget_class=self.__class__.__name__)
@ -74,8 +78,9 @@ class BECDockArea(BECWidget, QWidget):
if isinstance(config, dict): if isinstance(config, dict):
config = DockAreaConfig(**config) config = DockAreaConfig(**config)
self.config = 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) QWidget.__init__(self, parent=parent)
self._parent = parent
self.layout = QVBoxLayout(self) self.layout = QVBoxLayout(self)
self.layout.setSpacing(5) self.layout.setSpacing(5)
self.layout.setContentsMargins(0, 0, 0, 0) self.layout.setContentsMargins(0, 0, 0, 0)
@ -169,41 +174,41 @@ class BECDockArea(BECWidget, QWidget):
def _hook_toolbar(self): def _hook_toolbar(self):
# Menu Plot # Menu Plot
self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect( 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( 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( 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( 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 # Menu Devices
self.toolbar.widgets["menu_devices"].widgets["scan_control"].triggered.connect( 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( 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 # Menu Utils
self.toolbar.widgets["menu_utils"].widgets["queue"].triggered.connect( 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( 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( 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( 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( 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 # Icons
@ -211,6 +216,11 @@ class BECDockArea(BECWidget, QWidget):
self.toolbar.widgets["save_state"].action.triggered.connect(self.save_state) self.toolbar.widgets["save_state"].action.triggered.connect(self.save_state)
self.toolbar.widgets["restore_state"].action.triggered.connect(self.restore_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 def paintEvent(self, event: QPaintEvent): # TODO decide if we want any default instructions
super().paintEvent(event) super().paintEvent(event)
if self._instructions_visible: if self._instructions_visible:
@ -218,7 +228,7 @@ class BECDockArea(BECWidget, QWidget):
painter.drawText( painter.drawText(
self.rect(), self.rect(),
Qt.AlignCenter, 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 @property
@ -243,7 +253,17 @@ class BECDockArea(BECWidget, QWidget):
@panels.setter @panels.setter
def panels(self, value: dict[str, BECDock]): 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 @property
def temp_areas(self) -> list: def temp_areas(self) -> list:
@ -287,36 +307,17 @@ class BECDockArea(BECWidget, QWidget):
self.config.docks_state = last_state self.config.docks_state = last_state
return 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) @SafeSlot(popup_error=True)
def add_dock( def new(
self, self,
name: str = None, name: str | None = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = 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, relative_to: BECDock | None = None,
closable: bool = True, closable: bool = True,
floating: bool = False, floating: bool = False,
prefix: str = "dock", row: int | None = None,
widget: str | QWidget | None = None,
row: int = None,
col: int = 0, col: int = 0,
rowspan: int = 1, rowspan: int = 1,
colspan: int = 1, colspan: int = 1,
@ -326,12 +327,11 @@ class BECDockArea(BECWidget, QWidget):
Args: Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique. 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. 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. relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable. closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating. 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. row(int): The row of the added widget.
col(int): The column of the added widget. col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget. rowspan(int): The rowspan of the added widget.
@ -340,21 +340,20 @@ class BECDockArea(BECWidget, QWidget):
Returns: Returns:
BECDock: The created dock. BECDock: The created dock.
""" """
if name is None: dock_names = [dock._name for dock in self.panel_list] # pylint: disable=protected-access
name = WidgetContainerUtils.generate_unique_widget_id( if name is not None: # Name is provided
container=self.dock_area.docks, prefix=prefix if name in dock_names:
) raise ValueError(
f"Name {name} must be unique for docks, but already exists in DockArea "
if name in set(self.dock_area.docks.keys()): f"with name: {self._name} and id {self.gui_id}."
raise ValueError(f"Dock with name {name} already exists.") )
else: # Name is not provided
if position is None: name = WidgetContainerUtils.generate_unique_name(name="dock", list_of_names=dock_names)
position = "bottom"
dock = BECDock(name=name, parent_dock_area=self, closable=closable) dock = BECDock(name=name, parent_dock_area=self, closable=closable)
dock.config.position = position 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) self.dock_area.addDock(dock=dock, position=position, relativeTo=relative_to)
if len(self.dock_area.docks) <= 1: if len(self.dock_area.docks) <= 1:
@ -363,10 +362,11 @@ class BECDockArea(BECWidget, QWidget):
for dock in self.dock_area.docks.values(): for dock in self.dock_area.docks.values():
dock.show_title_bar() dock.show_title_bar()
if widget is not None and isinstance(widget, str): if widget is not None:
dock.add_widget(widget=widget, row=row, col=col, rowspan=rowspan, colspan=colspan) # Check if widget name exists.
elif widget is not None and isinstance(widget, QWidget): dock.new(
dock.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan) widget=widget, name=widget_name, row=row, col=col, rowspan=rowspan, colspan=colspan
)
if ( if (
self._instructions_visible self._instructions_visible
): # TODO still decide how initial instructions should be handled ): # 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. Remove a temporary area from the dock area.
This is a patched method of pyqtgraph's removeTempArea 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) self.dock_area.tempAreas.remove(area)
area.window().close() area.window().close()
area.window().deleteLater() 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): def cleanup(self):
""" """
Cleanup the dock area. Cleanup the dock area.
""" """
self.clear_all() self.delete_all()
self.toolbar.close() self.toolbar.close()
self.toolbar.deleteLater() self.toolbar.deleteLater()
self.dock_area.close() self.dock_area.close()
self.dock_area.deleteLater() self.dock_area.deleteLater()
super().cleanup() 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): def show(self):
"""Show all windows including floating docks.""" """Show all windows including floating docks."""
super().show() super().show()
@ -465,17 +442,52 @@ class BECDockArea(BECWidget, QWidget):
continue continue
docks.window().hide() docks.window().hide()
def delete(self): def delete_all(self) -> None:
self.hide() """
self.deleteLater() 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 if __name__ == "__main__": # pragma: no cover
import sys
from bec_widgets.utils.colors import set_theme from bec_widgets.utils.colors import set_theme
app = QApplication([]) app = QApplication([])
set_theme("auto") set_theme("auto")
dock_area = BECDockArea() 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.show()
dock_area.setGeometry(100, 100, 800, 600)
app.topLevelWidgets()
app.exec_() 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( def create_widget(
self, self, widget_type: str, parent_figure, parent_id: str, config: dict = None, **axis_kwargs
widget_type: str,
widget_id: str,
parent_figure,
parent_id: str,
config: dict = None,
**axis_kwargs,
) -> BECPlotBase: ) -> BECPlotBase:
""" """
Create and configure a widget based on its type. Create and configure a widget based on its type.
@ -109,7 +103,6 @@ class WidgetHandler:
widget_config_dict = { widget_config_dict = {
"widget_class": widget_class.__name__, "widget_class": widget_class.__name__,
"parent_id": parent_id, "parent_id": parent_id,
"gui_id": widget_id,
**(config if config is not None else {}), **(config if config is not None else {}),
} }
widget_config = config_class(**widget_config_dict) widget_config = config_class(**widget_config_dict)
@ -568,13 +561,13 @@ class BECFigure(BECWidget, pg.GraphicsLayoutWidget):
widget = self.widget_handler.create_widget( widget = self.widget_handler.create_widget(
widget_type=widget_type, widget_type=widget_type,
widget_id=widget_id,
parent_figure=self, parent_figure=self,
parent_id=self.gui_id, parent_id=self.gui_id,
config=config, config=config,
**axis_kwargs, **axis_kwargs,
) )
widget.set_gui_id(widget_id) widget_id = widget.gui_id
widget.config.row = row widget.config.row = row
widget.config.col = col 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 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 from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
logger = bec_logger.logger
class BECMainWindow(QMainWindow, BECConnector):
def __init__(self, *args, **kwargs): class BECMainWindow(BECWidget, QMainWindow):
BECConnector.__init__(self, **kwargs) def __init__(self, gui_id: str = None, *args, **kwargs):
BECWidget.__init__(self, gui_id=gui_id, **kwargs)
QMainWindow.__init__(self, *args, **kwargs) QMainWindow.__init__(self, *args, **kwargs)
def _dump(self): def _dump(self):
@ -33,9 +38,38 @@ class BECMainWindow(QMainWindow, BECConnector):
} }
return info return info
def new_dock_area(self, name): def new_dock_area(
dock_area = BECDockArea() 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.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() dock_area.show()
return dock_area 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.close()
self.axis_settings_dialog = None self.axis_settings_dialog = None
self.cleanup_pyqtgraph() self.cleanup_pyqtgraph()
self.rpc_register.remove_rpc(self) super().cleanup()
def cleanup_pyqtgraph(self): def cleanup_pyqtgraph(self):
"""Cleanup pyqtgraph items.""" """Cleanup pyqtgraph items."""

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

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

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

@ -27,7 +27,9 @@ def gui_id():
@contextmanager @contextmanager
def plot_server(gui_id, klass, client_lib): def plot_server(gui_id, klass, client_lib):
dispatcher = BECDispatcher(client=client_lib) # Has to init singleton with fixture client 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: try:
while client_lib._client.connector.get(MessageEndpoints.gui_heartbeat(gui_id)) is None: while client_lib._client.connector.get(MessageEndpoints.gui_heartbeat(gui_id)) is None:
time.sleep(0.3) time.sleep(0.3)
@ -42,6 +44,7 @@ def plot_server(gui_id, klass, client_lib):
@pytest.fixture @pytest.fixture
def connected_client_figure(gui_id, bec_client_lib): def connected_client_figure(gui_id, bec_client_lib):
with plot_server(gui_id, BECFigure, bec_client_lib) as server: with plot_server(gui_id, BECFigure, bec_client_lib) as server:
yield 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): def connected_client_gui_obj(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id) gui = BECGuiClient(gui_id=gui_id)
try: try:
gui.start_server(wait=True) gui.start(wait=True)
# gui._start_server(wait=True)
yield gui yield gui
finally: finally:
gui.close() gui.kill_server()
@pytest.fixture @pytest.fixture
@ -60,17 +64,18 @@ def connected_client_dock(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id) gui = BECGuiClient(gui_id=gui_id)
gui._auto_updates_enabled = False gui._auto_updates_enabled = False
try: try:
gui.start_server(wait=True) gui.start(wait=True)
yield gui.main gui.window_list[0]
yield gui.window_list[0]
finally: finally:
gui.close() gui.kill_server()
@pytest.fixture @pytest.fixture
def connected_client_dock_w_auto_updates(gui_id, bec_client_lib): def connected_client_dock_w_auto_updates(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id) gui = BECGuiClient(gui_id=gui_id)
try: try:
gui.start_server(wait=True) gui._start_server(wait=True)
yield gui, gui.main yield gui, gui.window_list[0]
finally: 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 time
import numpy as np import numpy as np
import pytest
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform
from bec_widgets.tests.utils import check_remote_data_size from bec_widgets.tests.utils import check_remote_data_size
def test_rpc_waveform1d_custom_curve(connected_client_figure): @pytest.fixture
fig = BECFigure(connected_client_figure) 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() ax = fig.plot()
curve = ax.plot(x=[1, 2, 3], y=[1, 2, 3]) 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 assert len(fig.widgets[ax._rpc_id].curves) == 1
def test_rpc_plotting_shortcuts_init_configs(connected_client_figure, qtbot): def test_rpc_plotting_shortcuts_init_configs(connected_figure, qtbot):
fig = BECFigure(connected_client_figure) fig = connected_figure
# fig = BECFigure(connected_client_figure)
plt = fig.plot(x_name="samx", y_name="bpm4i") plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger") 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): def test_rpc_waveform_scan(qtbot, connected_figure, bec_client_lib):
fig = BECFigure(connected_client_figure) # fig = BECFigure(connected_client_figure)
fig = connected_figure
# add 3 different curves to track # add 3 different curves to track
plt = fig.plot(x_name="samx", y_name="bpm4i") plt = fig.plot(x_name="samx", y_name="bpm4i")
fig.plot(x_name="samx", y_name="bpm3a") 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 assert plt_data["bpm4d-bpm4d"]["y"] == last_scan_data["bpm4d"]["bpm4d"].val
def test_rpc_image(connected_client_figure, bec_client_lib): def test_rpc_image(connected_figure, bec_client_lib):
fig = BECFigure(connected_client_figure) # fig = BECFigure(connected_client_figure)
fig = connected_figure
im = fig.image("eiger") 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) np.testing.assert_equal(last_image_device, last_image_plot)
def test_rpc_motor_map(connected_client_figure, bec_client_lib): def test_rpc_motor_map(connected_figure, bec_client_lib):
fig = BECFigure(connected_client_figure) # fig = BECFigure(connected_client_figure)
fig = connected_figure
motor_map = fig.motor_map("samx", "samy") 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") plt = fig.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel")
client = bec_client_lib 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) qtbot.waitUntil(wait_for_fit, timeout=10000)
def test_removing_subplots(connected_client_figure, bec_client_lib): def test_removing_subplots(connected_figure, bec_client_lib):
fig = BECFigure(connected_client_figure) # fig = BECFigure(connected_client_figure)
fig = connected_figure
plt = fig.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel") plt = fig.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel")
im = fig.image(monitor="eiger") im = fig.image(monitor="eiger")
mm = fig.motor_map(motor_x="samx", motor_y="samy") 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 from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform
def test_rpc_register_list_connections(connected_client_figure): def test_rpc_register_list_connections(connected_client_gui_obj):
fig = BECFigure(connected_client_figure) gui = connected_client_gui_obj
fig = gui.bec.new("fig").new(name="fig", widget="BECFigure")
plt = fig.plot(x_name="samx", y_name="bpm4i") plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger") im = fig.image("eiger")
@ -36,5 +37,6 @@ def test_rpc_register_list_connections(connected_client_figure):
**image_item_expected, **image_item_expected,
} }
assert len(all_connections) == 9 assert len(all_connections) == 9 + 3 # gui, dock_area, dock
assert all_connections == all_connections_expected # 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): 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" 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): 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") == bec_connector
assert bec_connector.get_obj_by_id("test_gui_id_2") is None 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) initial_count = len(bec_dock_area.dock_area.docks)
# Adding 3 docks # Adding 3 docks
d0 = bec_dock_area.add_dock() d0 = bec_dock_area.new()
d1 = bec_dock_area.add_dock() d1 = bec_dock_area.new()
d2 = bec_dock_area.add_dock() d2 = bec_dock_area.new()
# Check if the docks were added # Check if the docks were added
assert len(bec_dock_area.dock_area.docks) == initial_count + 3 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 # Remove docks
d0_name = d0.name() d0_name = d0.name()
bec_dock_area.remove_dock(d0_name) bec_dock_area.delete(d0_name)
qtbot.wait(200) qtbot.wait(200)
d1.remove() d1.remove()
qtbot.wait(200) 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): def test_add_remove_bec_figure_to_dock(bec_dock_area):
d0 = bec_dock_area.add_dock() d0 = bec_dock_area.new()
fig = d0.add_widget("BECFigure") fig = d0.new("BECFigure")
plt = fig.plot(x_name="samx", y_name="bpm4i") plt = fig.plot(x_name="samx", y_name="bpm4i")
im = fig.image("eiger") im = fig.image("eiger")
mm = fig.motor_map("samx", "samy") mm = fig.motor_map("samx", "samy")
mw = fig.multi_waveform("waveform1d") mw = fig.multi_waveform("waveform1d")
assert len(bec_dock_area.dock_area.docks) == 1 assert len(bec_dock_area.dock_area.docks) == 1
assert len(d0.widgets) == 1 assert len(d0.elements) == 1
assert len(d0.widget_list) == 1 assert len(d0.element_list) == 1
assert len(fig.widgets) == 4 assert len(fig.widgets) == 4
assert fig.config.widget_class == "BECFigure" 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): def test_close_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0") d0 = bec_dock_area.new(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1") d1 = bec_dock_area.new(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_2") d2 = bec_dock_area.new(name="dock_2")
bec_dock_area.clear_all() bec_dock_area.delete_all()
qtbot.wait(200) qtbot.wait(200)
assert len(bec_dock_area.dock_area.docks) == 0 assert len(bec_dock_area.dock_area.docks) == 0
def test_undock_and_dock_docks(bec_dock_area, qtbot): def test_undock_and_dock_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0") d0 = bec_dock_area.new(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1") d1 = bec_dock_area.new(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_4") d2 = bec_dock_area.new(name="dock_4")
d3 = bec_dock_area.add_dock(name="dock_3") d3 = bec_dock_area.new(name="dock_3")
d0.detach() d0.detach()
bec_dock_area.detach_dock("dock_1") 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): def test_toolbar_add_plot_waveform(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["waveform"].trigger() bec_dock_area.toolbar.widgets["menu_plots"].widgets["waveform"].trigger()
assert "waveform_1" in bec_dock_area.panels assert "Waveform_0" in bec_dock_area.panels
assert bec_dock_area.panels["waveform_1"].widgets[0].config.widget_class == "Waveform" assert bec_dock_area.panels["Waveform_0"].widgets[0].config.widget_class == "Waveform"
def test_toolbar_add_plot_image(bec_dock_area): def test_toolbar_add_plot_image(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["image"].trigger() bec_dock_area.toolbar.widgets["menu_plots"].widgets["image"].trigger()
assert "image_1" in bec_dock_area.panels assert "BECImageWidget_0" in bec_dock_area.panels
assert bec_dock_area.panels["image_1"].widgets[0].config.widget_class == "BECImageWidget" assert (
bec_dock_area.panels["BECImageWidget_0"].widgets[0].config.widget_class == "BECImageWidget"
)
def test_toolbar_add_plot_motor_map(bec_dock_area): def test_toolbar_add_plot_motor_map(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["motor_map"].trigger() bec_dock_area.toolbar.widgets["menu_plots"].widgets["motor_map"].trigger()
assert "motor_map_1" in bec_dock_area.panels assert "BECMotorMapWidget_0" in bec_dock_area.panels
assert bec_dock_area.panels["motor_map_1"].widgets[0].config.widget_class == "BECMotorMapWidget" assert (
bec_dock_area.panels["BECMotorMapWidget_0"].widgets[0].config.widget_class
== "BECMotorMapWidget"
)
def test_toolbar_add_device_positioner_box(bec_dock_area): def test_toolbar_add_device_positioner_box(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_devices"].widgets["positioner_box"].trigger() bec_dock_area.toolbar.widgets["menu_devices"].widgets["positioner_box"].trigger()
assert "positioner_box_1" in bec_dock_area.panels assert "PositionerBox_0" in bec_dock_area.panels
assert ( assert bec_dock_area.panels["PositionerBox_0"].widgets[0].config.widget_class == "PositionerBox"
bec_dock_area.panels["positioner_box_1"].widgets[0].config.widget_class == "PositionerBox"
)
def test_toolbar_add_utils_queue(bec_dock_area, bec_queue_msg_full): 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 MessageEndpoints.scan_queue_status(), bec_queue_msg_full
) )
bec_dock_area.toolbar.widgets["menu_utils"].widgets["queue"].trigger() bec_dock_area.toolbar.widgets["menu_utils"].widgets["queue"].trigger()
assert "queue_1" in bec_dock_area.panels assert "BECQueue_0" in bec_dock_area.panels
assert bec_dock_area.panels["queue_1"].widgets[0].config.widget_class == "BECQueue" assert bec_dock_area.panels["BECQueue_0"].widgets[0].config.widget_class == "BECQueue"
def test_toolbar_add_utils_status(bec_dock_area): def test_toolbar_add_utils_status(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_utils"].widgets["status"].trigger() bec_dock_area.toolbar.widgets["menu_utils"].widgets["status"].trigger()
assert "status_1" in bec_dock_area.panels assert "BECStatusBox_0" in bec_dock_area.panels
assert bec_dock_area.panels["status_1"].widgets[0].config.widget_class == "BECStatusBox" assert bec_dock_area.panels["BECStatusBox_0"].widgets[0].config.widget_class == "BECStatusBox"
def test_toolbar_add_utils_progress_bar(bec_dock_area): def test_toolbar_add_utils_progress_bar(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_utils"].widgets["progress_bar"].trigger() 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 ( 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(): def cli_figure():
fig = BECFigure(gui_id="test") fig = BECFigure(gui_id="test")
with mock.patch.object(fig, "_run_rpc") as mock_rpc_call: 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 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): def test_client_utils_start_plot_process(config, call_config):
with mock.patch("bec_widgets.cli.client_utils.subprocess.Popen") as mock_popen: with mock.patch("bec_widgets.cli.client_utils.subprocess.Popen") as mock_popen:
_start_plot_process("gui_id", BECFigure, config) _start_plot_process("gui_id", BECFigure, "bec", config)
command = ["bec-gui-server", "--id", "gui_id", "--gui_class", "BECFigure", "--hide"] command = [
"bec-gui-server",
"--id",
"gui_id",
"--gui_class",
"BECFigure",
"--gui_class_id",
"bec",
"--hide",
]
if call_config: if call_config:
command.extend(["--config", call_config]) command.extend(["--config", call_config])
mock_popen.assert_called_once_with( mock_popen.assert_called_once_with(
@ -66,20 +75,24 @@ def test_client_utils_passes_client_config_to_server(bec_dispatcher):
mixin = BECGuiClient() mixin = BECGuiClient()
mixin._client = bec_dispatcher.client mixin._client = bec_dispatcher.client
mixin._gui_id = "gui_id" mixin._gui_id = "gui_id"
mixin.gui_is_alive = mock.MagicMock() mixin._gui_is_alive = mock.MagicMock()
mixin.gui_is_alive.side_effect = [True] mixin._gui_is_alive.side_effect = [True]
try: try:
yield mixin yield mixin
finally: finally:
mixin.close() mixin.kill_server()
with bec_client_mixin() as mixin: with bec_client_mixin() as mixin:
with mock.patch("bec_widgets.cli.client_utils._start_plot_process") as mock_start_plot: with mock.patch("bec_widgets.cli.client_utils._start_plot_process") as mock_start_plot:
mock_start_plot.return_value = [mock.MagicMock(), mock.MagicMock()] mock_start_plot.return_value = [mock.MagicMock(), mock.MagicMock()]
mixin.start_server( mixin._start_server(
wait=False wait=False
) # the started event will not be set, wait=True would block forever ) # the started event will not be set, wait=True would block forever
mock_start_plot.assert_called_once_with( 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") plot_base = bec_figure.add_widget(widget_type="BECPlotBase", widget_id="test_plot")
assert plot_base is not None assert plot_base is not None
assert plot_base.config.widget_class == "BECPlotBase" 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): 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 mock_server, mock_config, _ = mocked_cli_server
_start_server("gui_id", BECFigure, None) _start_server("gui_id", BECFigure, config=None)
mock_server.assert_called_once_with(gui_id="gui_id", config=mock_config(), gui_class=BECFigure) mock_server.assert_called_once_with(
gui_id="gui_id", config=mock_config(), gui_class=BECFigure, gui_class_id="bec"
)
@pytest.mark.parametrize( @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 mock_server, mock_config, _ = mocked_cli_server
config = mock_config(**call_config) config = mock_config(**call_config)
_start_server("gui_id", BECFigure, config) _start_server("gui_id", BECFigure, config=config)
mock_server.assert_called_once_with(gui_id="gui_id", config=config, gui_class=BECFigure) mock_server.assert_called_once_with(
gui_id="gui_id", config=config, gui_class=BECFigure, gui_class_id="bec"
)