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

fix(widgets/dock): BECDockArea close overwrites the default pyqtgraph Container close + minor improvements

Browse Source
This commit is contained in:
wyzula_j
2024-05-01 22:11:06 +02:00
parent fcd6ef0975
commit ceae979f37
24 changed files with 930 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bec_widgets/cli/__init__.py
View File

@ -1,2 +1,2 @@
from .auto_updates import AutoUpdates, ScanInfo
from .client import BECFigure
from .client import BECDockArea, BECFigure

BIN
bec_widgets/cli/bec_widgets_icon.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 5.6 KiB

After

Width:  |  Height:  |  Size: 1.7 MiB

185
bec_widgets/cli/client.py
View File

@ -2,7 +2,7 @@
from typing import Literal, Optional, overload
from bec_widgets.cli.client_utils import BECFigureClientMixin, RPCBase, rpc_call
from bec_widgets.cli.client_utils import BECGuiClientMixin, RPCBase, rpc_call
class BECPlotBase(RPCBase):
@ -393,7 +393,7 @@ class BECWaveform(RPCBase):
"""
class BECFigure(RPCBase, BECFigureClientMixin):
class BECFigure(RPCBase):
@property
@rpc_call
def rpc_id(self) -> "str":
@ -426,7 +426,9 @@ class BECFigure(RPCBase, BECFigureClientMixin):
@rpc_call
def widgets(self) -> "dict":
"""
None
All widgets within the figure with gui ids as keys.
Returns:
dict: All widgets within the figure.
"""
@rpc_call
@ -1310,31 +1312,160 @@ class BECMotorMap(RPCBase):
class BECDock(RPCBase):
@rpc_call
def add_widget(self, widget: "QWidget", row=None, col=0, rowspan=1, colspan=1):
"""
None
"""
@property
@rpc_call
def widget_list(self) -> "list":
"""
Get the widgets in the dock.
Returns:
widgets(list): The widgets in the dock.
"""
@rpc_call
def show_title_bar(self):
"""
Hide the title bar of the dock.
"""
@rpc_call
def hide_title_bar(self):
"""
Hide the title bar of the dock.
"""
@rpc_call
def get_widgets_positions(self) -> "dict":
"""
Get the positions of the widgets in the dock.
Returns:
dict: The positions of the widgets in the dock as dict -> {(row, col, rowspan, colspan):widget}
"""
@rpc_call
def set_title(self, title: "str"):
"""
Set the title of the dock.
Args:
title(str): The title of the dock.
"""
@rpc_call
def add_widget_bec(
self,
widget_type: "str",
row=None,
col=0,
rowspan=1,
colspan=1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
):
"""
Add a widget to the dock.
Args:
widget_type(str): The widget to add. Only BEC RPC widgets from RPCWidgetHandler are allowed.
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 list_eligible_widgets(self) -> "list":
"""
List all widgets that can be added to the dock.
Returns:
list: The list of eligible widgets.
"""
@rpc_call
def move_widget(self, widget: "QWidget", new_row: "int", new_col: "int"):
"""
Move a widget to a new position in the layout.
Args:
widget(QWidget): The widget to move.
new_row(int): The new row to move the widget to.
new_col(int): The new column to move the widget to.
"""
@rpc_call
def remove_widget(self, widget: "QWidget"):
"""
Remove a widget from the dock.
Args:
widget(QWidget): The widget to remove.
"""
@rpc_call
def remove(self):
"""
Remove the dock from the parent dock area.
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
None
"""
class BECDockArea(RPCBase):
class BECDockArea(RPCBase, BECGuiClientMixin):
@property
@rpc_call
def panels(self) -> "dict":
"""
Get the docks in the dock area.
Returns:
dock_dict(dict): The docks in the dock area.
"""
@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 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.
"""
@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.
"""
@rpc_call
def add_dock(
self,
name: "str" = None,
position: "Literal['bottom', 'top', 'left', 'right', 'above', 'below']" = None,
relative_to: "Optional[BECDock]" = None,
closable: "bool" = False,
prefix: "str" = "dock",
widget: "QWidget" = None,
row: "int" = None,
col: "int" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
) -> "BECDock":
@ -1345,32 +1476,42 @@ class BECDockArea(RPCBase):
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.
prefix(str): The prefix for the dock name if no name is provided.
widget(QWidget): The widget to be added to the dock.
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 remove_dock_by_id(self, dock_id: "str"):
"""
None
"""
@rpc_call
def clear_all(self):
"""
None
Close all docks and remove all temp areas.
"""
@property
@rpc_call
def dock_dict(self) -> "dict":
def detach_dock(self, dock_name: "str") -> "BECDock":
"""
None
Undock a dock from the dock area.
Args:
dock(BECDock): The dock to undock.
Returns:
BECDock: The undocked dock.
"""
@rpc_call
def attach_all(self):
"""
Return all floating docks to the dock area.
"""
@rpc_call
def get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""

19
bec_widgets/cli/client_utils.py
View File

@ -22,7 +22,7 @@ from bec_widgets.cli.auto_updates import AutoUpdates
from bec_widgets.utils.bec_dispatcher import BECDispatcher
if TYPE_CHECKING:
from bec_widgets.cli.client import BECFigure
from bec_widgets.cli.client import BECDockArea, BECFigure
def rpc_call(func):
@ -56,7 +56,7 @@ def rpc_call(func):
return wrapper
class BECFigureClientMixin:
class BECGuiClientMixin:
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._process = None
@ -94,7 +94,7 @@ class BECFigureClientMixin:
)
@staticmethod
def _handle_msg_update(msg: MessageObject, parent: BECFigureClientMixin) -> None:
def _handle_msg_update(msg: MessageObject, parent: BECGuiClientMixin) -> None:
if parent.update_script is not None:
# pylint: disable=protected-access
parent._update_script_msg_parser(msg.value)
@ -139,8 +139,19 @@ class BECFigureClientMixin:
config = self._client._service_config.redis
monitor_module = importlib.import_module("bec_widgets.cli.server")
monitor_path = monitor_module.__file__
gui_class = self.__class__.__name__
command = [sys.executable, "-u", monitor_path, "--id", self._gui_id, "--config", config]
command = [
sys.executable,
"-u",
monitor_path,
"--id",
self._gui_id,
"--config",
config,
"--gui_class",
gui_class,
]
self._process = subprocess.Popen(
command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)

6
bec_widgets/cli/generate_cli.py
View File

@ -22,7 +22,7 @@ else:
class ClientGenerator:
def __init__(self):
self.header = """# This file was automatically generated by generate_cli.py\n
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECGuiClientMixin
from typing import Literal, Optional, overload"""
self.content = ""
@ -53,9 +53,9 @@ from typing import Literal, Optional, overload"""
# from {module} import {class_name}"""
# Generate the content
if cls.__name__ == "BECFigure":
if cls.__name__ == "BECDockArea":
self.content += f"""
class {class_name}(RPCBase, BECFigureClientMixin):"""
class {class_name}(RPCBase, BECGuiClientMixin):"""
else:
self.content += f"""
class {class_name}(RPCBase):"""

26
bec_widgets/cli/rpc_wigdet_handler.py Normal file
View File

@ -0,0 +1,26 @@
from bec_widgets.utils import BECConnector
from bec_widgets.widgets.figure import BECFigure
class RPCWidgetHandler:
"""Handler class for creating widgets from RPC messages."""
widget_classes = {
"BECFigure": BECFigure,
}
@staticmethod
def create_widget(widget_type, **kwargs) -> BECConnector:
"""
Create a widget from an RPC message.
Args:
widget_type(str): The type of the widget.
**kwargs: The keyword arguments for the widget.
Returns:
widget(BECConnector): The created widget.
"""
widget_class = RPCWidgetHandler.widget_classes.get(widget_type)
if widget_class:
return widget_class(**kwargs)
raise ValueError(f"Unknown widget type: {widget_type}")

16
bec_widgets/cli/server.py
View File

@ -1,7 +1,7 @@
import inspect
import threading
import time
from typing import Literal
from typing import Literal, Union
from bec_lib import MessageEndpoints, messages
from qtpy.QtCore import QTimer
@ -23,7 +23,7 @@ class BECWidgetsCLIServer:
dispatcher: BECDispatcher = None,
client=None,
config=None,
gui_class: BECFigure | BECDockArea = BECFigure,
gui_class: Union["BECFigure", "BECDockArea"] = BECFigure,
) -> None:
self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher
self.client = self.dispatcher.client if client is None else client
@ -109,7 +109,7 @@ class BECWidgetsCLIServer:
expire=10,
)
def shutdown(self):
def shutdown(self): # TODO not sure if needed when cleanup is done at level of BECConnector
self._shutdown_event = True
self._heartbeat_timer.stop()
self.client.shutdown()
@ -117,6 +117,7 @@ class BECWidgetsCLIServer:
if __name__ == "__main__": # pragma: no cover
import argparse
import os
import sys
from qtpy.QtCore import QSize
@ -125,8 +126,9 @@ if __name__ == "__main__": # pragma: no cover
app = QApplication(sys.argv)
app.setApplicationName("BEC Figure")
current_path = os.path.dirname(__file__)
icon = QIcon()
icon.addFile("bec_widgets_icon.png", size=QSize(48, 48))
icon.addFile(os.path.join(current_path, "bec_widgets_icon.png"), size=QSize(48, 48))
app.setWindowIcon(icon)
win = QMainWindow()
@ -155,10 +157,10 @@ if __name__ == "__main__": # pragma: no cover
gui_class = BECFigure
server = BECWidgetsCLIServer(gui_id=args.id, config=args.config, gui_class=gui_class)
# server = BECWidgetsCLIServer(gui_id="test", config=args.config, gui_class=gui_class)
fig = server.gui
win.setCentralWidget(fig)
gui = server.gui
win.setCentralWidget(gui)
win.resize(800, 600)
win.show()
app.aboutToQuit.connect(server.shutdown)

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

@ -49,7 +49,7 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
self.register = RPCRegister()
self.register.add_rpc(self.figure)
print("Registered objects:", dict(self.register.list_all_connections()))
# console push
self.console.kernel_manager.kernel.shell.push(
{
@ -62,6 +62,9 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
"d1": self.d1,
"d2": self.d2,
"d3": self.d3,
"b2a": self.button_2_a,
"b2b": self.button_2_b,
"b2c": self.button_2_c,
"bec": self.figure.client,
"scans": self.figure.client.scans,
"dev": self.figure.client.device_manager.devices,
@ -107,6 +110,9 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
def _init_dock(self):
self.button_1 = QtWidgets.QPushButton("Button 1 ")
self.button_2_a = QtWidgets.QPushButton("Button to be added at place 0,0 in d3")
self.button_2_b = QtWidgets.QPushButton("button after without postions specified")
self.button_2_c = QtWidgets.QPushButton("button super late")
self.button_3 = QtWidgets.QPushButton("Button above Figure ")
self.label_1 = QtWidgets.QLabel("some scan info label with useful information")
@ -123,6 +129,15 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
self.d3.add_widget(self.button_3)
self.d3.add_widget(self.fig_dock3)
self.dock.save_state()
def closeEvent(self, event):
"""Override to handle things when main window is closed."""
self.dock.cleanup()
self.figure.clear_all()
self.figure.client.shutdown()
super().closeEvent(event)
if __name__ == "__main__": # pragma: no cover
import sys
@ -140,4 +155,5 @@ if __name__ == "__main__": # pragma: no cover
win = JupyterConsoleWindow()
win.show()
app.aboutToQuit.connect(win.close)
sys.exit(app.exec_())

12
bec_widgets/utils/bec_connector.py
View File

@ -148,6 +148,14 @@ class BECConnector:
else:
return self.config
def closeEvent(self, event):
def cleanup(self):
"""Cleanup the widget."""
self.rpc_register.remove_rpc(self)
all_connections = self.rpc_register.list_all_connections()
if len(all_connections) == 0:
print("No more connections. Shutting down GUI BEC client.")
self.client.shutdown()
super().closeEvent(event)
# def closeEvent(self, event):
# self.cleanup()
# super().closeEvent(event)

217
bec_widgets/widgets/dock/dock.py
View File

@ -1,13 +1,18 @@
from __future__ import annotations
from typing import Literal, Optional
from typing import TYPE_CHECKING, Literal, Optional
from pydantic import Field
from pyqtgraph.dockarea import Dock
from qtpy.QtWidgets import QWidget
from bec_widgets.cli.rpc_wigdet_handler import RPCWidgetHandler
from bec_widgets.utils import BECConnector, ConnectionConfig, GridLayoutManager
if TYPE_CHECKING:
from qtpy.QtWidgets import QWidget
from bec_widgets.widgets import BECDockArea
class DockConfig(ConnectionConfig):
widgets: dict[str, ConnectionConfig] = Field({}, description="The widgets in the dock.")
@ -20,18 +25,30 @@ class DockConfig(ConnectionConfig):
class BECDock(BECConnector, Dock):
USER_ACCESS = ["add_widget", "widget_list"]
USER_ACCESS = [
"rpc_id",
"widget_list",
"show_title_bar",
"hide_title_bar",
"get_widgets_positions",
"set_title",
"add_widget_bec",
"list_eligible_widgets",
"move_widget",
"remove_widget",
"remove",
"attach",
"detach",
]
def __init__(
self,
parent: Optional[QWidget] = None,
parent_dock_area: Optional["BECDockArea"] = None,
config: Optional[
DockConfig
] = None, # TODO ATM connection config -> will be changed when I will know what I want to use there
name: Optional[str] = None,
parent: QWidget | None = None,
parent_dock_area: BECDockArea | None = None,
config: DockConfig | None = None,
name: str | None = None,
client=None,
gui_id: Optional[str] = None,
gui_id: str | None = None,
**kwargs,
) -> None:
if config is None:
@ -47,23 +64,124 @@ class BECDock(BECConnector, Dock):
self.parent_dock_area = parent_dock_area
# Signals
self.sigClosed.connect(self._remove_from_dock_area)
# Layout Manager
self.layout_manager = GridLayoutManager(self.layout)
def dropEvent(self, event):
source = event.source()
old_area = source.area
self.setOrientation("horizontal", force=True)
super().dropEvent(event)
if old_area in self.parent_dock_area.tempAreas and old_area != self.parent_dock_area:
self.parent_dock_area.removeTempArea(old_area)
def float(self):
"""
Float the dock.
Overwrites the default pyqtgraph dock float.
"""
# need to check if the dock is temporary and if it is the only dock in the area
# fixes bug in pyqtgraph detaching
if self.area.temporary == True and len(self.area.docks) <= 1:
return
elif self.area.temporary == True and len(self.area.docks) > 1:
self.area.docks.pop(self.name(), None)
super().float()
else:
super().float()
@property
def widget_list(self) -> list:
"""
Get the widgets in the dock.
Returns:
widgets(list): The widgets in the dock.
"""
return self.widgets
@widget_list.setter
def widget_list(self, value: list):
self.widgets = value
def get_widgets_positions(self):
def hide_title_bar(self):
"""
Hide the title bar of the dock.
"""
# self.hideTitleBar() #TODO pyqtgraph looks bugged ATM, doing my implementation
self.label.hide()
self.labelHidden = True
def show_title_bar(self):
"""
Hide the title bar of the dock.
"""
# self.showTitleBar() #TODO pyqtgraph looks bugged ATM, doing my implementation
self.label.show()
self.labelHidden = False
def set_title(self, title: str):
"""
Set the title of the dock.
Args:
title(str): The title of the dock.
"""
self.parent_dock_area.docks[title] = self.parent_dock_area.docks.pop(self.name())
self.setTitle(title)
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}
"""
return self.layout_manager.get_widgets_positions()
def list_eligible_widgets(
self,
) -> list: # TODO can be moved to some util mixin like container class for rpc widgets
"""
List all widgets that can be added to the dock.
Returns:
list: The list of eligible widgets.
"""
return list(RPCWidgetHandler.widget_classes.keys())
def add_widget_bec(
self,
widget_type: str,
row=None,
col=0,
rowspan=1,
colspan=1,
shift: Literal["down", "up", "left", "right"] = "down",
):
"""
Add a widget to the dock.
Args:
widget_type(str): The widget to add. Only BEC RPC widgets from RPCWidgetHandler are allowed.
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.
"""
if row is None:
row = self.layout.rowCount()
if self.layout_manager.is_position_occupied(row, col):
self.layout_manager.shift_widgets(shift, start_row=row)
widget = RPCWidgetHandler.create_widget(widget_type)
self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
return widget
def add_widget(
self,
widget: QWidget,
@ -73,6 +191,17 @@ class BECDock(BECConnector, Dock):
colspan=1,
shift: Literal["down", "up", "left", "right"] = "down",
):
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
colspan(int): The number of columns the widget should span.
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
if row is None:
row = self.layout.rowCount()
@ -81,6 +210,60 @@ class BECDock(BECConnector, Dock):
self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
def _remove_from_dock_area(self):
"""Remove this dock from the DockArea it lives inside."""
self.parent_dock_area.docks.pop(self.name())
def move_widget(self, widget: QWidget, new_row: int, new_col: int):
"""
Move a widget to a new position in the layout.
Args:
widget(QWidget): The widget to move.
new_row(int): The new row to move the widget to.
new_col(int): The new column to move the widget to.
"""
self.layout_manager.move_widget(widget, new_row, new_col)
def attach(self):
"""
Attach the dock to the parent dock area.
"""
self.parent_dock_area.removeTempArea(self.area)
def detach(self):
"""
Detach the dock from the parent dock area.
"""
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)
widget.close()
def remove(self):
"""
Remove the dock from the parent dock area.
"""
# self.cleanup()
self.parent_dock_area.remove_dock(self.name())
def cleanup(self):
"""
Clean up the dock, including all its widgets.
"""
for widget in self.widgets:
if hasattr(widget, "cleanup"):
widget.cleanup()
super().cleanup()
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()

161
bec_widgets/widgets/dock/dock_area.py
View File

@ -1,17 +1,18 @@
from __future__ import annotations
from typing import Literal, Optional
from weakref import WeakValueDictionary
from pydantic import Field
from pyqtgraph.dockarea.DockArea import DockArea
from qtpy.QtCore import Qt
from qtpy.QtGui import QPainter, QPaintEvent
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector, ConnectionConfig, WidgetContainerUtils
from .dock import BECDock, DockConfig
# from bec_widgets.widgets import BECDock
class DockAreaConfig(ConnectionConfig):
docks: dict[str, DockConfig] = Field({}, description="The docks in the dock area.")
@ -19,18 +20,23 @@ class DockAreaConfig(ConnectionConfig):
class BECDockArea(BECConnector, DockArea):
USER_ACCESS = [
"panels",
"save_state",
"remove_dock",
"restore_state",
"add_dock",
"remove_dock_by_id",
"clear_all",
"dock_dict",
"detach_dock",
"attach_all",
"get_all_rpc",
]
def __init__(
self,
parent: Optional[QWidget] = None,
config: Optional[DockAreaConfig] = None,
parent: QWidget | None = None,
config: DockAreaConfig | None = None,
client=None,
gui_id: Optional[str] = None,
gui_id: str = None,
) -> None:
if config is None:
config = DockAreaConfig(widget_class=self.__class__.__name__)
@ -41,41 +47,80 @@ class BECDockArea(BECConnector, DockArea):
super().__init__(client=client, config=config, gui_id=gui_id)
DockArea.__init__(self, parent=parent)
self._last_state = None # TODO not sure if this will ever work
self._instructions_visible = True
def paintEvent(self, event: QPaintEvent):
super().paintEvent(event)
if self._instructions_visible:
painter = QPainter(self)
painter.drawText(self.rect(), Qt.AlignCenter, "Add docks using 'add_dock' method")
@property
def dock_dict(self) -> dict:
def panels(self) -> dict:
"""
Get the docks in the dock area.
Returns:
dock_dict(dict): The docks in the dock area.
"""
return dict(self.docks)
@dock_dict.setter
def dock_dict(self, value: dict):
from weakref import WeakValueDictionary
@panels.setter
def panels(self, value: dict):
self.docks = WeakValueDictionary(value)
def remove_dock_by_id(self, dock_id: str):
if dock_id in self.docks:
dock_to_remove = self.docks[dock_id]
dock_to_remove.close()
else:
raise ValueError(f"Dock with id {dock_id} does not exist.")
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.
"""
if state is None:
state = self._last_state
self.restoreState(state, missing=missing, extra=extra)
def save_state(self) -> dict:
"""
Save the state of the dock area.
Returns:
dict: The state of the dock area.
"""
self._last_state = self.saveState()
return self._last_state
def remove_dock(self, name: str):
for id, dock in self.docks.items():
dock_name = dock.name()
if dock_name == name:
"""
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.docks.pop(name, None)
if dock:
dock.close()
break
if len(self.docks) <= 1:
for dock in self.docks.values():
dock.hide_title_bar()
else:
raise ValueError(f"Dock with name {name} does not exist.")
def add_dock(
self,
name: str = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = None,
relative_to: Optional[BECDock] = None,
relative_to: BECDock | None = None,
closable: bool = False,
prefix: str = "dock",
widget: QWidget = None,
widget: str | QWidget | None = None,
row: int = None,
col: int = None,
col: int = 0,
rowspan: int = 1,
colspan: int = 1,
) -> BECDock:
@ -86,13 +131,13 @@ class BECDockArea(BECConnector, DockArea):
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.
prefix(str): The prefix for the dock name if no name is provided.
widget(QWidget): The widget to be added to the dock.
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.
"""
@ -107,17 +152,69 @@ class BECDockArea(BECConnector, DockArea):
if position is None:
position = "bottom"
dock = BECDock(name=name, parent_dock_area=self, closable=True)
dock = BECDock(name=name, parent_dock_area=self, closable=closable)
dock.config.position = position
self.config.docks[name] = dock.config
self.addDock(dock=dock, position=position, relativeTo=relative_to)
if widget is not None:
dock.addWidget(widget) # , row, col, rowspan, colspan)
if len(self.docks) <= 1:
dock.hide_title_bar()
elif len(self.docks) > 1:
for dock in self.docks.values():
dock.show_title_bar()
if widget is not None and isinstance(widget, str):
dock.add_widget_bec(
widget_type=widget, row=row, col=col, rowspan=rowspan, colspan=colspan
)
elif widget is not None and isinstance(widget, QWidget):
dock.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
if self._instructions_visible:
self._instructions_visible = False
self.update()
return dock
def detach_dock(self, dock_name: str) -> BECDock:
"""
Undock a dock from the dock area.
Args:
dock_name(str): The dock to undock.
Returns:
BECDock: The undocked dock.
"""
dock = self.docks[dock_name]
self.floatDock(dock)
return dock
def attach_all(self):
"""
Return all floating docks to the dock area.
"""
while self.tempAreas:
for temp_area in self.tempAreas:
self.removeTempArea(temp_area)
def clear_all(self):
for dock in self.docks.values():
dock.close()
"""
Close all docks and remove all temp areas.
"""
self.attach_all()
for dock in dict(self.docks).values():
dock.remove()
def cleanup(self):
"""
Cleanup the dock area.
"""
self.clear_all()
super().cleanup()
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()

43
bec_widgets/widgets/figure/figure.py
View File

@ -1,20 +1,18 @@
# pylint: disable = no-name-in-module,missing-module-docstring
from __future__ import annotations
import itertools
import os
import uuid
from collections import defaultdict
from typing import Literal, Optional, Type
from typing import Literal, Optional
import numpy as np
import pyqtgraph as pg
import qdarktheme
from pydantic import Field
from pyqtgraph.Qt import uic
from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector, BECDispatcher, ConnectionConfig, WidgetContainerUtils
from bec_widgets.utils import BECConnector, ConnectionConfig, WidgetContainerUtils
from bec_widgets.widgets.plots import (
BECImageShow,
BECMotorMap,
@ -166,14 +164,29 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
@widget_list.setter
def widget_list(self, value: list[BECPlotBase]):
"""
Access all widget in BECFigure as a list
Returns:
list[BECPlotBase]: List of all widgets in the figure.
"""
self._axes = value
@property
def widgets(self) -> dict:
"""
All widgets within the figure with gui ids as keys.
Returns:
dict: All widgets within the figure.
"""
return self._widgets
@widgets.setter
def widgets(self, value: dict):
"""
All widgets within the figure with gui ids as keys.
Returns:
dict: All widgets within the figure.
"""
self._widgets = value
def add_plot(
@ -204,7 +217,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
config(dict): Additional configuration for the widget.
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = str(uuid.uuid4())
waveform = self.add_widget(
widget_type="Waveform1D",
widget_id=widget_id,
@ -430,7 +443,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
BECImageShow: The image widget.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = str(uuid.uuid4())
if config is None:
config = ImageConfig(
widget_class="BECImageShow",
@ -513,7 +526,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Returns:
BECMotorMap: The motor map widget.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = str(uuid.uuid4())
if config is None:
config = MotorMapConfig(
widget_class="BECMotorMap",
@ -554,7 +567,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
if not widget_id:
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = str(uuid.uuid4())
if widget_id in self._widgets:
raise ValueError(f"Widget with ID '{widget_id}' already exists.")
@ -767,12 +780,16 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
def clear_all(self):
"""Clear all widgets from the figure and reset to default state"""
for widget in self._widgets.values():
widget.cleanup()
self.clear()
for widget in list(self._widgets.values()):
widget.remove()
# self.clear()
self._widgets = defaultdict(dict)
self.grid = []
theme = self.config.theme
self.config = FigureConfig(
widget_class=self.__class__.__name__, gui_id=self.gui_id, theme=theme
)
def cleanup(self):
self.clear_all()
super().cleanup()

6
bec_widgets/widgets/plots/image.py
View File

@ -288,10 +288,6 @@ class BECImageItem(BECConnector, pg.ImageItem):
else:
raise ValueError("style should be 'simple' or 'full'")
def cleanup(self):
"""Clean up widget."""
self.rpc_register.remove_rpc(self)
class BECImageShow(BECPlotBase):
USER_ACCESS = [
@ -806,7 +802,7 @@ class BECImageShow(BECPlotBase):
for image in self.images:
image.cleanup()
self.rpc_register.remove_rpc(self)
super().cleanup()
class ImageProcessor:

2
bec_widgets/widgets/plots/motor_map.py
View File

@ -425,4 +425,4 @@ class BECMotorMap(BECPlotBase):
def cleanup(self):
"""Cleanup the widget."""
self._disconnect_current_motors()
self.rpc_register.remove_rpc(self)
super().cleanup()

1
bec_widgets/widgets/plots/plot_base.py
View File

@ -248,3 +248,4 @@ class BECPlotBase(BECConnector, pg.GraphicsLayout):
def cleanup(self):
"""Cleanup the plot widget."""
super().cleanup()

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

@ -229,14 +229,10 @@ class BECCurve(BECConnector, pg.PlotDataItem):
x_data, y_data = self.getData()
return x_data, y_data
def cleanup(self):
"""Cleanup the curve."""
self.rpc_register.remove_rpc(self)
def remove(self):
"""Remove the curve from the plot."""
self.cleanup()
self.parent_item.removeItem(self)
self.cleanup()
class BECWaveform(BECPlotBase):
@ -799,4 +795,4 @@ class BECWaveform(BECPlotBase):
self.bec_dispatcher.disconnect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())
for curve in self.curves:
curve.cleanup()
self.rpc_register.remove_rpc(self)
super().cleanup()

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

@ -3,6 +3,7 @@ import pytest
from bec_widgets.cli.rpc_register import RPCRegister
from bec_widgets.cli.server import BECWidgetsCLIServer
from bec_widgets.utils import BECDispatcher
from bec_widgets.widgets import BECDockArea, BECFigure
@pytest.fixture(autouse=True)
@ -12,11 +13,25 @@ def rpc_register():
@pytest.fixture
def rpc_server(qtbot, bec_client_lib, threads_check):
def rpc_server_figure(qtbot, bec_client_lib, threads_check):
dispatcher = BECDispatcher(client=bec_client_lib) # Has to init singleton with fixture client
server = BECWidgetsCLIServer(gui_id="figure")
qtbot.addWidget(server.fig)
qtbot.waitExposed(server.fig)
server = BECWidgetsCLIServer(gui_id="figure", gui_class=BECFigure)
qtbot.addWidget(server.gui)
qtbot.waitExposed(server.gui)
qtbot.wait(1000) # 1s long to wait until gui is ready
yield server
dispatcher.disconnect_all()
server.client.shutdown()
server.shutdown()
dispatcher.reset_singleton()
@pytest.fixture
def rpc_server_dock(qtbot, bec_client_lib, threads_check):
dispatcher = BECDispatcher(client=bec_client_lib) # Has to init singleton with fixture client
server = BECWidgetsCLIServer(gui_id="figure", gui_class=BECDockArea)
qtbot.addWidget(server.gui)
qtbot.waitExposed(server.gui)
qtbot.wait(1000) # 1s long to wait until gui is ready
yield server
dispatcher.disconnect_all()

145
tests/end-2-end/test_bec_dock_rpc_e2e.py Normal file
View File

@ -0,0 +1,145 @@
import numpy as np
import pytest
from bec_lib import MessageEndpoints
from bec_widgets.cli.client import BECDockArea, BECFigure, BECImageShow, BECMotorMap, BECWaveform
def test_rpc_add_dock_with_figure_e2e(rpc_server_dock, qtbot):
dock = BECDockArea(rpc_server_dock.gui_id)
dock_server = rpc_server_dock.gui
# BEC client shortcuts
client = rpc_server_dock.client
dev = client.device_manager.devices
scans = client.scans
queue = client.queue
# Create 3 docks
d0 = dock.add_dock("dock_0")
d1 = dock.add_dock("dock_1")
d2 = dock.add_dock("dock_2")
assert len(dock_server.docks) == 3
# Add 3 figures with some widgets
fig0 = d0.add_widget_bec("BECFigure")
fig1 = d1.add_widget_bec("BECFigure")
fig2 = d2.add_widget_bec("BECFigure")
assert len(dock_server.docks) == 3
assert len(dock_server.docks["dock_0"].widgets) == 1
assert len(dock_server.docks["dock_1"].widgets) == 1
assert len(dock_server.docks["dock_2"].widgets) == 1
assert fig1.__class__.__name__ == "BECFigure"
assert fig1.__class__ == BECFigure
assert fig2.__class__.__name__ == "BECFigure"
assert fig2.__class__ == BECFigure
mm = fig0.motor_map("samx", "samy")
plt = fig1.plot("samx", "bpm4i")
im = fig2.image("eiger")
assert mm.__class__.__name__ == "BECMotorMap"
assert mm.__class__ == BECMotorMap
assert plt.__class__.__name__ == "BECWaveform"
assert plt.__class__ == BECWaveform
assert im.__class__.__name__ == "BECImageShow"
assert im.__class__ == BECImageShow
assert mm.config_dict["signals"] == {
"source": "device_readback",
"x": {
"name": "samx",
"entry": "samx",
"unit": None,
"modifier": None,
"limits": [-50.0, 50.0],
},
"y": {
"name": "samy",
"entry": "samy",
"unit": None,
"modifier": None,
"limits": [-50.0, 50.0],
},
"z": None,
}
assert plt.config_dict["curves"]["bpm4i-bpm4i"]["signals"] == {
"source": "scan_segment",
"x": {"name": "samx", "entry": "samx", "unit": None, "modifier": None, "limits": None},
"y": {"name": "bpm4i", "entry": "bpm4i", "unit": None, "modifier": None, "limits": None},
"z": None,
}
assert im.config_dict["images"]["eiger"]["monitor"] == "eiger"
# check initial position of motor map
initial_pos_x = dev.samx.read()["samx"]["value"]
initial_pos_y = dev.samy.read()["samy"]["value"]
# Try to make a scan
status = scans.line_scan(dev.samx, -5, 5, steps=10, exp_time=0.05, relative=False)
# wait for scan to finish
while not status.status == "COMPLETED":
qtbot.wait(200)
# plot
plt_last_scan_data = queue.scan_storage.storage[-1].data
plt_data = plt.get_all_data()
assert plt_data["bpm4i-bpm4i"]["x"] == plt_last_scan_data["samx"]["samx"].val
assert plt_data["bpm4i-bpm4i"]["y"] == plt_last_scan_data["bpm4i"]["bpm4i"].val
# image
last_image_device = client.connector.get_last(MessageEndpoints.device_monitor("eiger"))[
"data"
].data
qtbot.wait(500)
last_image_plot = im.images[0].get_data()
np.testing.assert_equal(last_image_device, last_image_plot)
# motor map
final_pos_x = dev.samx.read()["samx"]["value"]
final_pos_y = dev.samy.read()["samy"]["value"]
# check final coordinates of motor map
motor_map_data = mm.get_data()
np.testing.assert_equal(
[motor_map_data["x"][0], motor_map_data["y"][0]], [initial_pos_x, initial_pos_y]
)
np.testing.assert_equal(
[motor_map_data["x"][-1], motor_map_data["y"][-1]], [final_pos_x, final_pos_y]
)
def test_dock_manipulations_e2e(rpc_server_dock, qtbot):
dock = BECDockArea(rpc_server_dock.gui_id)
dock_server = rpc_server_dock.gui
d0 = dock.add_dock("dock_0")
d1 = dock.add_dock("dock_1")
d2 = dock.add_dock("dock_2")
assert len(dock_server.docks) == 3
d0.detach()
dock.detach_dock("dock_2")
assert len(dock_server.docks) == 3
assert len(dock_server.tempAreas) == 2
d0.attach()
assert len(dock_server.docks) == 3
assert len(dock_server.tempAreas) == 1
d2.remove()
qtbot.wait(200)
assert len(dock_server.docks) == 2
docks_list = list(dict(dock_server.docks).keys())
assert ["dock_0", "dock_1"] == docks_list
dock.clear_all()
assert len(dock_server.docks) == 0
assert len(dock_server.tempAreas) == 0

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

@ -3,27 +3,11 @@ import pytest
from bec_lib import MessageEndpoints
from bec_widgets.cli.client import BECFigure, BECImageShow, BECMotorMap, BECWaveform
from bec_widgets.cli.server import BECWidgetsCLIServer
from bec_widgets.utils import BECDispatcher
@pytest.fixture
def rpc_server(qtbot, bec_client_lib, threads_check):
dispatcher = BECDispatcher(client=bec_client_lib) # Has to init singleton with fixture client
server = BECWidgetsCLIServer(gui_id="id_test")
qtbot.addWidget(server.gui)
qtbot.waitExposed(server.gui)
qtbot.wait(1000) # 1s long to wait until gui is ready
yield server
dispatcher.disconnect_all()
server.client.shutdown()
server.shutdown()
dispatcher.reset_singleton()
def test_rpc_waveform1d_custom_curve(rpc_server, qtbot):
fig = BECFigure(rpc_server.gui_id)
fig_server = rpc_server.gui
def test_rpc_waveform1d_custom_curve(rpc_server_figure, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
fig_server = rpc_server_figure.gui
ax = fig.add_plot()
curve = ax.add_curve_custom([1, 2, 3], [1, 2, 3])
@ -32,12 +16,12 @@ def test_rpc_waveform1d_custom_curve(rpc_server, qtbot):
curve.set_color("blue")
assert len(fig_server.widgets) == 1
assert len(fig_server.widgets["widget_1"].curves) == 1
assert len(fig_server.widgets[ax.rpc_id].curves) == 1
def test_rpc_plotting_shortcuts_init_configs(rpc_server, qtbot):
fig = BECFigure(rpc_server.gui_id)
fig_server = rpc_server.gui
def test_rpc_plotting_shortcuts_init_configs(rpc_server_figure, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
fig_server = rpc_server_figure.gui
plt = fig.plot("samx", "bpm4i")
im = fig.image("eiger")
@ -91,15 +75,15 @@ def test_rpc_plotting_shortcuts_init_configs(rpc_server, qtbot):
}
def test_rpc_waveform_scan(rpc_server, qtbot):
fig = BECFigure(rpc_server.gui_id)
def test_rpc_waveform_scan(rpc_server_figure, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
# add 3 different curves to track
plt = fig.plot("samx", "bpm4i")
fig.plot("samx", "bpm3a")
fig.plot("samx", "bpm4d")
client = rpc_server.client
client = rpc_server_figure.client
dev = client.device_manager.devices
scans = client.scans
queue = client.queue
@ -124,12 +108,12 @@ def test_rpc_waveform_scan(rpc_server, qtbot):
assert plt_data["bpm4d-bpm4d"]["y"] == last_scan_data["bpm4d"]["bpm4d"].val
def test_rpc_image(rpc_server, qtbot):
fig = BECFigure(rpc_server.gui_id)
def test_rpc_image(rpc_server_figure, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
im = fig.image("eiger")
client = rpc_server.client
client = rpc_server_figure.client
dev = client.device_manager.devices
scans = client.scans
@ -149,13 +133,13 @@ def test_rpc_image(rpc_server, qtbot):
np.testing.assert_equal(last_image_device, last_image_plot)
def test_rpc_motor_map(rpc_server, qtbot):
fig = BECFigure(rpc_server.gui_id)
fig_server = rpc_server.gui
def test_rpc_motor_map(rpc_server_figure, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
fig_server = rpc_server_figure.gui
motor_map = fig.motor_map("samx", "samy")
client = rpc_server.client
client = rpc_server_figure.client
dev = client.device_manager.devices
scans = client.scans

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

@ -18,9 +18,9 @@ def find_deepest_value(d: dict):
return d
def test_rpc_register_list_connections(rpc_server, rpc_register, qtbot):
fig = BECFigure(rpc_server.gui_id)
fig_server = rpc_server.fig
def test_rpc_register_list_connections(rpc_server_figure, rpc_register, qtbot):
fig = BECFigure(rpc_server_figure.gui_id)
fig_server = rpc_server_figure.gui
plt = fig.plot("samx", "bpm4i")
im = fig.image("eiger")

114
tests/unit_tests/test_bec_dock.py Normal file
View File

@ -0,0 +1,114 @@
# pylint: disable=missing-function-docstring, missing-module-docstring, unused-import
import pytest
from bec_widgets.widgets import BECDock, BECDockArea
from .client_mocks import mocked_client
@pytest.fixture
def bec_dock_area(qtbot, mocked_client):
widget = BECDockArea(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
widget.close()
def test_bec_dock_area_init(bec_dock_area):
assert bec_dock_area is not None
assert bec_dock_area.client is not None
assert isinstance(bec_dock_area, BECDockArea)
assert bec_dock_area.config.widget_class == "BECDockArea"
def test_bec_dock_area_add_remove_dock(bec_dock_area, qtbot):
initial_count = len(bec_dock_area.docks)
# Adding 3 docks
d0 = bec_dock_area.add_dock()
d1 = bec_dock_area.add_dock()
d2 = bec_dock_area.add_dock()
# Check if the docks were added
assert len(bec_dock_area.docks) == initial_count + 3
assert d0.name() in dict(bec_dock_area.docks)
assert d1.name() in dict(bec_dock_area.docks)
assert d2.name() in dict(bec_dock_area.docks)
assert bec_dock_area.docks[d0.name()].config.widget_class == "BECDock"
assert bec_dock_area.docks[d1.name()].config.widget_class == "BECDock"
assert bec_dock_area.docks[d2.name()].config.widget_class == "BECDock"
# Check panels API for getting docks to CLI
assert bec_dock_area.panels == dict(bec_dock_area.docks)
# Remove docks
d0_name = d0.name()
bec_dock_area.remove_dock(d0_name) # TODO fix this, works in jupyter console
qtbot.wait(200)
d1.remove()
qtbot.wait(200)
assert len(bec_dock_area.docks) == initial_count + 1
assert d0.name() not in dict(bec_dock_area.docks)
assert d1.name() not in dict(bec_dock_area.docks)
assert d2.name() in dict(bec_dock_area.docks)
def test_add_remove_bec_figure_to_dock(bec_dock_area):
d0 = bec_dock_area.add_dock()
fig = d0.add_widget_bec("BECFigure")
plt = fig.plot("samx", "bpm4i")
im = fig.image("eiger")
mm = fig.motor_map("samx", "samy")
assert len(bec_dock_area.docks) == 1
assert len(d0.widgets) == 1
assert len(d0.widget_list) == 1
assert len(fig.widgets) == 3
assert fig.config.widget_class == "BECFigure"
assert plt.config.widget_class == "BECWaveform"
assert im.config.widget_class == "BECImageShow"
assert mm.config.widget_class == "BECMotorMap"
def test_dock_area_errors(bec_dock_area):
d0 = bec_dock_area.add_dock(name="dock_0")
with pytest.raises(ValueError) as excinfo:
bec_dock_area.add_dock(name="dock_0")
assert "Dock with name dock_0 already exists." in str(excinfo.value)
def test_close_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_2")
bec_dock_area.clear_all()
qtbot.wait(200)
assert len(bec_dock_area.docks) == 0
def test_undock_and_dock_docks(bec_dock_area, qtbot):
d0 = bec_dock_area.add_dock(name="dock_0")
d1 = bec_dock_area.add_dock(name="dock_1")
d2 = bec_dock_area.add_dock(name="dock_4")
d3 = bec_dock_area.add_dock(name="dock_3")
d0.detach()
bec_dock_area.detach_dock("dock_1")
d2.detach()
assert len(bec_dock_area.docks) == 4
assert len(bec_dock_area.tempAreas) == 3
d0.attach()
assert len(bec_dock_area.docks) == 4
assert len(bec_dock_area.tempAreas) == 2
bec_dock_area.attach_all()
assert len(bec_dock_area.docks) == 4
assert len(bec_dock_area.tempAreas) == 0

42
tests/unit_tests/test_bec_figure.py
View File

@ -1,6 +1,4 @@
# pylint: disable=missing-function-docstring, missing-module-docstring, unused-import
import os
from unittest.mock import MagicMock
import numpy as np
import pytest
@ -48,12 +46,12 @@ def test_bec_figure_add_remove_plot(bec_figure):
# Check if the widgets were added
assert len(bec_figure._widgets) == initial_count + 3
assert "widget_1" in bec_figure._widgets
assert "widget_2" in bec_figure._widgets
assert "widget_3" in bec_figure._widgets
assert bec_figure._widgets["widget_1"].config.widget_class == "BECWaveform"
assert bec_figure._widgets["widget_2"].config.widget_class == "BECWaveform"
assert bec_figure._widgets["widget_3"].config.widget_class == "BECPlotBase"
assert w0.gui_id in bec_figure._widgets
assert w1.gui_id in bec_figure._widgets
assert w2.gui_id in bec_figure._widgets
assert bec_figure._widgets[w0.gui_id].config.widget_class == "BECWaveform"
assert bec_figure._widgets[w1.gui_id].config.widget_class == "BECWaveform"
assert bec_figure._widgets[w2.gui_id].config.widget_class == "BECPlotBase"
# Check accessing positions by the grid in figure
assert bec_figure[0, 0] == w0
@ -61,11 +59,11 @@ def test_bec_figure_add_remove_plot(bec_figure):
assert bec_figure[2, 0] == w2
# Removing 1 widget
bec_figure.remove(widget_id="widget_1")
bec_figure.remove(widget_id=w0.gui_id)
assert len(bec_figure._widgets) == initial_count + 2
assert "widget_1" not in bec_figure._widgets
assert "widget_3" in bec_figure._widgets
assert bec_figure._widgets["widget_2"].config.widget_class == "BECWaveform"
assert w0.gui_id not in bec_figure._widgets
assert w2.gui_id in bec_figure._widgets
assert bec_figure._widgets[w1.gui_id].config.widget_class == "BECWaveform"
def test_add_different_types_of_widgets(bec_figure):
@ -121,20 +119,20 @@ def test_remove_plots(bec_figure):
# remove by coordinates
bec_figure[0, 0].remove()
assert "widget_1" not in bec_figure._widgets
assert w1.gui_id not in bec_figure._widgets
# remove by widget_id
bec_figure.remove(widget_id="widget_2")
assert "widget_2" not in bec_figure._widgets
bec_figure.remove(widget_id=w2.gui_id)
assert w2.gui_id not in bec_figure._widgets
# remove by widget object
w3.remove()
assert "widget_3" not in bec_figure._widgets
assert w3.gui_id not in bec_figure._widgets
# check the remaining widget 4
assert bec_figure[0, 0] == w4
assert bec_figure["widget_4"] == w4
assert "widget_4" in bec_figure._widgets
assert bec_figure[w4.gui_id] == w4
assert w4.gui_id in bec_figure._widgets
assert len(bec_figure._widgets) == 1
@ -143,8 +141,8 @@ def test_remove_plots_by_coordinates_ints(bec_figure):
w2 = bec_figure.add_plot(row=0, col=1)
bec_figure.remove(0, 0)
assert "widget_1" not in bec_figure._widgets
assert "widget_2" in bec_figure._widgets
assert w1.gui_id not in bec_figure._widgets
assert w2.gui_id in bec_figure._widgets
assert bec_figure[0, 0] == w2
assert len(bec_figure._widgets) == 1
@ -154,8 +152,8 @@ def test_remove_plots_by_coordinates_tuple(bec_figure):
w2 = bec_figure.add_plot(row=0, col=1)
bec_figure.remove(coordinates=(0, 0))
assert "widget_1" not in bec_figure._widgets
assert "widget_2" in bec_figure._widgets
assert w1.gui_id not in bec_figure._widgets
assert w2.gui_id in bec_figure._widgets
assert bec_figure[0, 0] == w2
assert len(bec_figure._widgets) == 1

2
tests/unit_tests/test_generate_cli_client.py
View File

@ -40,7 +40,7 @@ def test_client_generator_with_black_formatting():
'''\
# This file was automatically generated by generate_cli.py
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECGuiClientMixin
from typing import Literal, Optional, overload
class MockBECWaveform1D(RPCBase):

2
tests/unit_tests/test_waveform1d.py
View File

@ -141,7 +141,7 @@ def test_getting_curve(bec_figure):
c1_expected_config = CurveConfig(
widget_class="BECCurve",
gui_id="test_curve",
parent_id="widget_1",
parent_id=w1.gui_id,
label="bpm4i-bpm4i",
color="#cc4778",
symbol="o",