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

feat(multi-waveform): new widget added

This commit is contained in:
2024-10-15 23:17:47 +02:00
committed by wyzula_j
parent ec39dae273
commit f3a39a69e2
17 changed files with 2073 additions and 10 deletions

View File

@ -22,6 +22,7 @@ class Widgets(str, enum.Enum):
BECFigure = "BECFigure" BECFigure = "BECFigure"
BECImageWidget = "BECImageWidget" BECImageWidget = "BECImageWidget"
BECMotorMapWidget = "BECMotorMapWidget" BECMotorMapWidget = "BECMotorMapWidget"
BECMultiWaveformWidget = "BECMultiWaveformWidget"
BECProgressBar = "BECProgressBar" BECProgressBar = "BECProgressBar"
BECQueue = "BECQueue" BECQueue = "BECQueue"
BECStatusBox = "BECStatusBox" BECStatusBox = "BECStatusBox"
@ -1574,6 +1575,436 @@ class BECMotorMapWidget(RPCBase):
""" """
class BECMultiWaveform(RPCBase):
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@property
@rpc_call
def curves(self) -> collections.deque:
"""
Get the curves of the plot widget as a deque.
Returns:
deque: Deque of curves.
"""
@rpc_call
def set_monitor(self, monitor: str):
"""
Set the monitor for the plot widget.
Args:
monitor (str): The monitor to set.
"""
@rpc_call
def set_opacity(self, opacity: int):
"""
Set the opacity of the curve on the plot.
Args:
opacity(int): The opacity of the curve. 0-100.
"""
@rpc_call
def set_curve_limit(self, max_trace: int, flush_buffer: bool = False):
"""
Set the maximum number of traces to display on the plot.
Args:
max_trace (int): The maximum number of traces to display.
flush_buffer (bool): Flush the buffer.
"""
@rpc_call
def set_curve_highlight(self, index: int):
"""
Set the curve highlight based on visible curves.
Args:
index (int): The index of the curve to highlight among visible curves.
"""
@rpc_call
def set_colormap(self, colormap: str):
"""
Set the colormap for the scatter plot z gradient.
Args:
colormap(str): Colormap for the scatter plot.
"""
@rpc_call
def set(self, **kwargs) -> "None":
"""
Set the properties of the plot widget.
Args:
**kwargs: Keyword arguments for the properties to be set.
Possible properties:
- title: str
- x_label: str
- y_label: str
- x_scale: Literal["linear", "log"]
- y_scale: Literal["linear", "log"]
- x_lim: tuple
- y_lim: tuple
- legend_label_size: int
"""
@rpc_call
def set_title(self, title: "str", size: "int" = None):
"""
Set the title of the plot widget.
Args:
title(str): Title of the plot widget.
size(int): Font size of the title.
"""
@rpc_call
def set_x_label(self, label: "str", size: "int" = None):
"""
Set the label of the x-axis.
Args:
label(str): Label of the x-axis.
size(int): Font size of the label.
"""
@rpc_call
def set_y_label(self, label: "str", size: "int" = None):
"""
Set the label of the y-axis.
Args:
label(str): Label of the y-axis.
size(int): Font size of the label.
"""
@rpc_call
def set_x_scale(self, scale: "Literal['linear', 'log']" = "linear"):
"""
Set the scale of the x-axis.
Args:
scale(Literal["linear", "log"]): Scale of the x-axis.
"""
@rpc_call
def set_y_scale(self, scale: "Literal['linear', 'log']" = "linear"):
"""
Set the scale of the y-axis.
Args:
scale(Literal["linear", "log"]): Scale of the y-axis.
"""
@rpc_call
def set_x_lim(self, *args) -> "None":
"""
Set the limits of the x-axis. This method can accept either two separate arguments
for the minimum and maximum x-axis values, or a single tuple containing both limits.
Usage:
set_x_lim(x_min, x_max)
set_x_lim((x_min, x_max))
Args:
*args: A variable number of arguments. Can be two integers (x_min and x_max)
or a single tuple with two integers.
"""
@rpc_call
def set_y_lim(self, *args) -> "None":
"""
Set the limits of the y-axis. This method can accept either two separate arguments
for the minimum and maximum y-axis values, or a single tuple containing both limits.
Usage:
set_y_lim(y_min, y_max)
set_y_lim((y_min, y_max))
Args:
*args: A variable number of arguments. Can be two integers (y_min and y_max)
or a single tuple with two integers.
"""
@rpc_call
def set_grid(self, x: "bool" = False, y: "bool" = False):
"""
Set the grid of the plot widget.
Args:
x(bool): Show grid on the x-axis.
y(bool): Show grid on the y-axis.
"""
@rpc_call
def set_colormap(self, colormap: str):
"""
Set the colormap for the scatter plot z gradient.
Args:
colormap(str): Colormap for the scatter plot.
"""
@rpc_call
def enable_fps_monitor(self, enable: "bool" = True):
"""
Enable the FPS monitor.
Args:
enable(bool): True to enable, False to disable.
"""
@rpc_call
def lock_aspect_ratio(self, lock):
"""
Lock aspect ratio.
Args:
lock(bool): True to lock, False to unlock.
"""
@rpc_call
def get_all_data(self, output: Literal["dict", "pandas"] = "dict") -> dict:
"""
Extract all curve data into a dictionary or a pandas DataFrame.
Args:
output (Literal["dict", "pandas"]): Format of the output data.
Returns:
dict | pd.DataFrame: Data of all curves in the specified format.
"""
@rpc_call
def export(self):
"""
Show the Export Dialog of the plot widget.
"""
@rpc_call
def remove(self):
"""
Remove the plot widget from the figure.
"""
class BECMultiWaveformWidget(RPCBase):
@property
@rpc_call
def curves(self) -> list[pyqtgraph.graphicsItems.PlotDataItem.PlotDataItem]:
"""
Get the curves of the plot widget as a list
Returns:
list: List of curves.
"""
@rpc_call
def set_monitor(self, monitor: str) -> None:
"""
Set the monitor of the plot widget.
Args:
monitor(str): The monitor to set.
"""
@rpc_call
def set_curve_highlight(self, index: int) -> None:
"""
Set the curve highlight of the plot widget by index
Args:
index(int): The index of the curve to highlight.
"""
@rpc_call
def set_opacity(self, opacity: int) -> None:
"""
Set the opacity of the plot widget.
Args:
opacity(int): The opacity to set.
"""
@rpc_call
def set_curve_limit(self, curve_limit: int) -> None:
"""
Set the maximum number of traces to display on the plot widget.
Args:
curve_limit(int): The maximum number of traces to display.
"""
@rpc_call
def set_buffer_flush(self, flush_buffer: bool) -> None:
"""
Set the buffer flush property of the plot widget.
Args:
flush_buffer(bool): True to flush the buffer, False to not flush the buffer.
"""
@rpc_call
def set_highlight_last_curve(self, enable: bool) -> None:
"""
Enable or disable highlighting of the last curve.
Args:
enable(bool): True to enable highlighting of the last curve, False to disable.
"""
@rpc_call
def set_colormap(self, colormap: str) -> None:
"""
Set the colormap of the plot widget.
Args:
colormap(str): The colormap to set.
"""
@rpc_call
def set(self, **kwargs):
"""
Set the properties of the plot widget.
Args:
**kwargs: Keyword arguments for the properties to be set.
Possible properties:
- title: str
- x_label: str
- y_label: str
- x_scale: Literal["linear", "log"]
- y_scale: Literal["linear", "log"]
- x_lim: tuple
- y_lim: tuple
- legend_label_size: int
"""
@rpc_call
def set_title(self, title: str):
"""
Set the title of the plot widget.
Args:
title(str): The title to set.
"""
@rpc_call
def set_x_label(self, x_label: str):
"""
Set the x-axis label of the plot widget.
Args:
x_label(str): The x-axis label to set.
"""
@rpc_call
def set_y_label(self, y_label: str):
"""
Set the y-axis label of the plot widget.
Args:
y_label(str): The y-axis label to set.
"""
@rpc_call
def set_x_scale(self, x_scale: Literal["linear", "log"]):
"""
Set the x-axis scale of the plot widget.
Args:
x_scale(str): The x-axis scale to set.
"""
@rpc_call
def set_y_scale(self, y_scale: Literal["linear", "log"]):
"""
Set the y-axis scale of the plot widget.
Args:
y_scale(str): The y-axis scale to set.
"""
@rpc_call
def set_x_lim(self, x_lim: tuple):
"""
Set x-axis limits of the plot widget.
Args:
x_lim(tuple): The x-axis limits to set.
"""
@rpc_call
def set_y_lim(self, y_lim: tuple):
"""
Set y-axis limits of the plot widget.
Args:
y_lim(tuple): The y-axis limits to set.
"""
@rpc_call
def set_grid(self, x_grid: bool, y_grid: bool):
"""
Set the grid of the plot widget.
Args:
x_grid(bool): True to enable the x-grid, False to disable.
y_grid(bool): True to enable the y-grid, False to disable.
"""
@rpc_call
def set_colormap(self, colormap: str) -> None:
"""
Set the colormap of the plot widget.
Args:
colormap(str): The colormap to set.
"""
@rpc_call
def enable_fps_monitor(self, enabled: bool):
"""
Enable or disable the FPS monitor
Args:
enabled(bool): True to enable the FPS monitor, False to disable.
"""
@rpc_call
def lock_aspect_ratio(self, lock: bool):
"""
Lock the aspect ratio of the plot widget.
Args:
lock(bool): True to lock the aspect ratio, False to unlock.
"""
@rpc_call
def export(self):
"""
Export the plot widget.
"""
class BECPlotBase(RPCBase): class BECPlotBase(RPCBase):
@property @property
@rpc_call @rpc_call

View File

@ -56,6 +56,7 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
# "cm": self.colormap, # "cm": self.colormap,
"im": self.im, "im": self.im,
"mm": self.mm, "mm": self.mm,
"mw": self.mw,
} }
) )
@ -167,9 +168,11 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
self.im.image("waveform", "1d") self.im.image("waveform", "1d")
self.d2 = self.dock.add_dock(name="dock_2", position="bottom") self.d2 = self.dock.add_dock(name="dock_2", position="bottom")
self.wf = self.d2.add_widget("BECWaveformWidget", row=0, col=0) self.wf = self.d2.add_widget("BECFigure", row=0, col=0)
self.wf.plot(x_name="samx", y_name="bpm3a")
self.wf.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel") self.mw = self.wf.multi_waveform(monitor="waveform") # , config=config)
# self.wf.plot(x_name="samx", y_name="bpm3a")
# self.wf.plot(x_name="samx", y_name="bpm4i", dap="GaussianModel")
# self.bar = self.d2.add_widget("RingProgressBar", row=0, col=1) # self.bar = self.d2.add_widget("RingProgressBar", row=0, col=1)
# self.bar.set_diameter(200) # self.bar.set_diameter(200)
@ -210,6 +213,7 @@ if __name__ == "__main__": # pragma: no cover
win = JupyterConsoleWindow() win = JupyterConsoleWindow()
win.show() win.show()
win.resize(1200, 800)
app.aboutToQuit.connect(win.close) app.aboutToQuit.connect(win.close)
sys.exit(app.exec_()) sys.exit(app.exec_())

View File

@ -6,13 +6,16 @@ from qtpy.QtCore import QObject, Qt, Signal, Slot
from qtpy.QtWidgets import QApplication from qtpy.QtWidgets import QApplication
class NonDownsamplingScatterPlotItem(pg.ScatterPlotItem): class CrosshairScatterItem(pg.ScatterPlotItem):
def setDownsampling(self, ds=None, auto=None, method=None): def setDownsampling(self, ds=None, auto=None, method=None):
pass pass
def setClipToView(self, state): def setClipToView(self, state):
pass pass
def setAlpha(self, *args, **kwargs):
pass
class Crosshair(QObject): class Crosshair(QObject):
# QT Position of mouse cursor # QT Position of mouse cursor
@ -123,7 +126,7 @@ class Crosshair(QObject):
continue continue
pen = item.opts["pen"] pen = item.opts["pen"]
color = pen.color() if hasattr(pen, "color") else pg.mkColor(pen) color = pen.color() if hasattr(pen, "color") else pg.mkColor(pen)
marker_moved = NonDownsamplingScatterPlotItem( marker_moved = CrosshairScatterItem(
size=10, pen=pg.mkPen(color), brush=pg.mkBrush(None) size=10, pen=pg.mkPen(color), brush=pg.mkBrush(None)
) )
marker_moved.skip_auto_range = True marker_moved.skip_auto_range = True
@ -132,7 +135,7 @@ class Crosshair(QObject):
# Create glowing effect markers for clicked events # Create glowing effect markers for clicked events
for size, alpha in [(18, 64), (14, 128), (10, 255)]: for size, alpha in [(18, 64), (14, 128), (10, 255)]:
marker_clicked = NonDownsamplingScatterPlotItem( marker_clicked = CrosshairScatterItem(
size=size, size=size,
pen=pg.mkPen(None), pen=pg.mkPen(None),
brush=pg.mkBrush(color.red(), color.green(), color.blue(), alpha), brush=pg.mkBrush(color.red(), color.green(), color.blue(), alpha),

View File

@ -24,6 +24,7 @@ from bec_widgets.widgets.dark_mode_button.dark_mode_button import DarkModeButton
from bec_widgets.widgets.dock.dock import BECDock, DockConfig from bec_widgets.widgets.dock.dock import BECDock, DockConfig
from bec_widgets.widgets.image.image_widget import BECImageWidget from bec_widgets.widgets.image.image_widget import BECImageWidget
from bec_widgets.widgets.motor_map.motor_map_widget import BECMotorMapWidget from bec_widgets.widgets.motor_map.motor_map_widget import BECMotorMapWidget
from bec_widgets.widgets.multi_waveform.multi_waveform_widget import BECMultiWaveformWidget
from bec_widgets.widgets.positioner_box.positioner_box import PositionerBox from bec_widgets.widgets.positioner_box.positioner_box import PositionerBox
from bec_widgets.widgets.ring_progress_bar.ring_progress_bar import RingProgressBar from bec_widgets.widgets.ring_progress_bar.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.scan_control.scan_control import ScanControl from bec_widgets.widgets.scan_control.scan_control import ScanControl
@ -85,6 +86,11 @@ class BECDockArea(BECWidget, QWidget):
tooltip="Add Waveform", tooltip="Add Waveform",
filled=True, filled=True,
), ),
"multi_waveform": MaterialIconAction(
icon_name=BECMultiWaveformWidget.ICON_NAME,
tooltip="Add Multi Waveform",
filled=True,
),
"image": MaterialIconAction( "image": MaterialIconAction(
icon_name=BECImageWidget.ICON_NAME, tooltip="Add Image", filled=True icon_name=BECImageWidget.ICON_NAME, tooltip="Add Image", filled=True
), ),
@ -154,6 +160,9 @@ class BECDockArea(BECWidget, QWidget):
self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect( self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect(
lambda: self.add_dock(widget="BECWaveformWidget", prefix="waveform") lambda: self.add_dock(widget="BECWaveformWidget", prefix="waveform")
) )
self.toolbar.widgets["menu_plots"].widgets["multi_waveform"].triggered.connect(
lambda: self.add_dock(widget="BECMultiWaveformWidget", prefix="multi_waveform")
)
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.add_dock(widget="BECImageWidget", prefix="image")
) )

View File

@ -11,6 +11,7 @@ from bec_lib.logger import bec_logger
from pydantic import Field, ValidationError, field_validator from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Signal as pyqtSignal from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtWidgets import QWidget from qtpy.QtWidgets import QWidget
from tornado.gen import multi
from typeguard import typechecked from typeguard import typechecked
from bec_widgets.utils import ConnectionConfig, WidgetContainerUtils from bec_widgets.utils import ConnectionConfig, WidgetContainerUtils
@ -18,6 +19,10 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.colors import apply_theme
from bec_widgets.widgets.figure.plots.image.image import BECImageShow, ImageConfig from bec_widgets.widgets.figure.plots.image.image import BECImageShow, ImageConfig
from bec_widgets.widgets.figure.plots.motor_map.motor_map import BECMotorMap, MotorMapConfig from bec_widgets.widgets.figure.plots.motor_map.motor_map import BECMotorMap, MotorMapConfig
from bec_widgets.widgets.figure.plots.multi_waveform.multi_waveform import (
BECMultiWaveform,
BECMultiWaveformConfig,
)
from bec_widgets.widgets.figure.plots.plot_base import BECPlotBase, SubplotConfig from bec_widgets.widgets.figure.plots.plot_base import BECPlotBase, SubplotConfig
from bec_widgets.widgets.figure.plots.waveform.waveform import BECWaveform, Waveform1DConfig from bec_widgets.widgets.figure.plots.waveform.waveform import BECWaveform, Waveform1DConfig
@ -64,6 +69,7 @@ class WidgetHandler:
"BECWaveform": (BECWaveform, Waveform1DConfig), "BECWaveform": (BECWaveform, Waveform1DConfig),
"BECImageShow": (BECImageShow, ImageConfig), "BECImageShow": (BECImageShow, ImageConfig),
"BECMotorMap": (BECMotorMap, MotorMapConfig), "BECMotorMap": (BECMotorMap, MotorMapConfig),
"BECMultiWaveform": (BECMultiWaveform, BECMultiWaveformConfig),
} }
def create_widget( def create_widget(
@ -134,8 +140,14 @@ class BECFigure(BECWidget, pg.GraphicsLayoutWidget):
"BECWaveform": BECWaveform, "BECWaveform": BECWaveform,
"BECImageShow": BECImageShow, "BECImageShow": BECImageShow,
"BECMotorMap": BECMotorMap, "BECMotorMap": BECMotorMap,
"BECMultiWaveform": BECMultiWaveform,
}
widget_method_map = {
"BECWaveform": "plot",
"BECImageShow": "image",
"BECMotorMap": "motor_map",
"BECMultiWaveform": "multi_waveform",
} }
widget_method_map = {"BECWaveform": "plot", "BECImageShow": "image", "BECMotorMap": "motor_map"}
clean_signal = pyqtSignal() clean_signal = pyqtSignal()
@ -445,10 +457,27 @@ class BECFigure(BECWidget, pg.GraphicsLayoutWidget):
return motor_map return motor_map
def multi_waveform(
self,
monitor: str = None,
new: bool = False,
row: int | None = None,
col: int | None = None,
config: dict | None = None,
**axis_kwargs,
):
multi_waveform = self.subplot_factory(
widget_type="BECMultiWaveform", config=config, row=row, col=col, new=new, **axis_kwargs
)
if config is not None:
return multi_waveform
multi_waveform.set_monitor(monitor)
return multi_waveform
def subplot_factory( def subplot_factory(
self, self,
widget_type: Literal[ widget_type: Literal[
"BECPlotBase", "BECWaveform", "BECImageShow", "BECMotorMap" "BECPlotBase", "BECWaveform", "BECImageShow", "BECMotorMap", "BECMultiWaveform"
] = "BECPlotBase", ] = "BECPlotBase",
row: int = None, row: int = None,
col: int = None, col: int = None,
@ -500,7 +529,7 @@ class BECFigure(BECWidget, pg.GraphicsLayoutWidget):
def add_widget( def add_widget(
self, self,
widget_type: Literal[ widget_type: Literal[
"BECPlotBase", "BECWaveform", "BECImageShow", "BECMotorMap" "BECPlotBase", "BECWaveform", "BECImageShow", "BECMotorMap", "BECMultiWaveform"
] = "BECPlotBase", ] = "BECPlotBase",
widget_id: str = None, widget_id: str = None,
row: int = None, row: int = None,

View File

@ -0,0 +1,330 @@
from collections import deque
from typing import Literal, Optional
import pyqtgraph as pg
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from pydantic import Field, field_validator
from pyqtgraph.exporters import MatplotlibExporter
from qtpy.QtCore import Signal, Slot
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import Colors
from bec_widgets.widgets.figure.plots.plot_base import BECPlotBase, SubplotConfig
logger = bec_logger.logger
class BECMultiWaveformConfig(SubplotConfig):
color_palette: Optional[str] = Field(
"magma", description="The color palette of the figure widget.", validate_default=True
)
curve_limit: Optional[int] = Field(
200, description="The maximum number of curves to display on the plot."
)
flush_buffer: Optional[bool] = Field(
False, description="Flush the buffer of the plot widget when the curve limit is reached."
)
monitor: Optional[str] = Field(
None, description="The monitor to set for the plot widget."
) # TODO validate monitor in bec -> maybe make it as SignalData class for validation purpose
curve_width: Optional[int] = Field(1, description="The width of the curve on the plot.")
opacity: Optional[int] = Field(50, description="The opacity of the curve on the plot.")
highlight_last_curve: Optional[bool] = Field(
True, description="Highlight the last curve on the plot."
)
model_config: dict = {"validate_assignment": True}
_validate_color_map_z = field_validator("color_palette")(Colors.validate_color_map)
class BECMultiWaveform(BECPlotBase):
monitor_signal_updated = Signal()
USER_ACCESS = [
"_rpc_id",
"_config_dict",
"curves",
"set_monitor",
"set_opacity",
"set_curve_limit",
"set_curve_highlight",
"set_colormap",
"set",
"set_title",
"set_x_label",
"set_y_label",
"set_x_scale",
"set_y_scale",
"set_x_lim",
"set_y_lim",
"set_grid",
"set_colormap",
"enable_fps_monitor",
"lock_aspect_ratio",
"export",
"get_all_data",
"remove",
]
def __init__(
self,
parent: Optional[QWidget] = None,
parent_figure=None,
config: Optional[BECMultiWaveformConfig] = None,
client=None,
gui_id: Optional[str] = None,
):
if config is None:
config = BECMultiWaveformConfig(widget_class=self.__class__.__name__)
super().__init__(
parent=parent, parent_figure=parent_figure, config=config, client=client, gui_id=gui_id
)
self.old_scan_id = None
self.scan_id = None
self.monitor = None
self.connected = False
self.current_highlight_index = 0
self._curves = deque()
self.number_of_visible_curves = 0
# Get bec shortcuts dev, scans, queue, scan_storage, dap
self.get_bec_shortcuts()
@property
def curves(self) -> deque:
"""
Get the curves of the plot widget as a deque.
Returns:
deque: Deque of curves.
"""
return self._curves
@curves.setter
def curves(self, value: deque):
self._curves = value
@property
def highlight_last_curve(self) -> bool:
"""
Get the highlight_last_curve property.
Returns:
bool: The highlight_last_curve property.
"""
return self.config.highlight_last_curve
@highlight_last_curve.setter
def highlight_last_curve(self, value: bool):
self.config.highlight_last_curve = value
def set_monitor(self, monitor: str):
"""
Set the monitor for the plot widget.
Args:
monitor (str): The monitor to set.
"""
self.config.monitor = monitor
self._connect_monitor()
def _connect_monitor(self):
"""
Connect the monitor to the plot widget.
"""
try:
previous_monitor = self.monitor
except AttributeError:
previous_monitor = None
if previous_monitor and self.connected is True:
self.bec_dispatcher.disconnect_slot(
self.on_monitor_1d_update, MessageEndpoints.device_monitor_1d(previous_monitor)
)
if self.config.monitor and self.connected is False:
self.bec_dispatcher.connect_slot(
self.on_monitor_1d_update, MessageEndpoints.device_monitor_1d(self.config.monitor)
)
self.connected = True
self.monitor = self.config.monitor
@Slot(dict, dict)
def on_monitor_1d_update(self, msg: dict, metadata: dict):
"""
Update the plot widget with the monitor data.
Args:
msg(dict): The message data.
metadata(dict): The metadata of the message.
"""
data = msg.get("data", None)
current_scan_id = metadata.get("scan_id", None)
if current_scan_id != self.scan_id:
self.scan_id = current_scan_id
self.plot_item.clear()
self.curves.clear()
# Always create a new curve and add it
curve = pg.PlotDataItem()
curve.setData(data)
self.plot_item.addItem(curve)
self.curves.append(curve)
# Max Trace and scale colors
self.set_curve_limit(self.config.curve_limit, self.config.flush_buffer)
self.monitor_signal_updated.emit()
@Slot(int)
def set_curve_highlight(self, index: int):
"""
Set the curve highlight based on visible curves.
Args:
index (int): The index of the curve to highlight among visible curves.
"""
visible_curves = [curve for curve in self.curves if curve.isVisible()]
num_visible_curves = len(visible_curves)
self.number_of_visible_curves = num_visible_curves
if num_visible_curves == 0:
return # No curves to highlight
if index >= num_visible_curves:
index = num_visible_curves - 1
elif index < 0:
index = num_visible_curves + index
self.current_highlight_index = index
num_colors = num_visible_curves
colors = Colors.evenly_spaced_colors(
colormap=self.config.color_palette, num=num_colors, format="HEX"
)
for i, curve in enumerate(visible_curves):
curve.setPen()
if i == self.current_highlight_index:
curve.setPen(pg.mkPen(color=colors[i], width=5))
curve.setAlpha(alpha=1, auto=False)
curve.setZValue(1)
else:
curve.setPen(pg.mkPen(color=colors[i], width=1))
curve.setAlpha(alpha=self.config.opacity / 100, auto=False)
curve.setZValue(0)
@Slot(int)
def set_opacity(self, opacity: int):
"""
Set the opacity of the curve on the plot.
Args:
opacity(int): The opacity of the curve. 0-100.
"""
self.config.opacity = max(0, min(100, opacity))
self.set_curve_highlight(self.current_highlight_index)
@Slot(int, bool)
def set_curve_limit(self, max_trace: int, flush_buffer: bool = False):
"""
Set the maximum number of traces to display on the plot.
Args:
max_trace (int): The maximum number of traces to display.
flush_buffer (bool): Flush the buffer.
"""
self.config.curve_limit = max_trace
self.config.flush_buffer = flush_buffer
if self.config.curve_limit is None:
self.scale_colors()
return
if self.config.flush_buffer:
# Remove excess curves from the plot and the deque
while len(self.curves) > self.config.curve_limit:
curve = self.curves.popleft()
self.plot_item.removeItem(curve)
else:
# Hide or show curves based on the new max_trace
num_curves_to_show = min(self.config.curve_limit, len(self.curves))
for i, curve in enumerate(self.curves):
if i < len(self.curves) - num_curves_to_show:
curve.hide()
else:
curve.show()
self.scale_colors()
def scale_colors(self):
"""
Scale the colors of the curves based on the current colormap.
"""
if self.config.highlight_last_curve:
self.set_curve_highlight(-1) # Use -1 to highlight the last visible curve
else:
self.set_curve_highlight(self.current_highlight_index)
def set_colormap(self, colormap: str):
"""
Set the colormap for the curves.
Args:
colormap(str): Colormap for the curves.
"""
self.config.color_palette = colormap
self.set_curve_highlight(self.current_highlight_index)
def get_all_data(self, output: Literal["dict", "pandas"] = "dict") -> dict:
"""
Extract all curve data into a dictionary or a pandas DataFrame.
Args:
output (Literal["dict", "pandas"]): Format of the output data.
Returns:
dict | pd.DataFrame: Data of all curves in the specified format.
"""
data = {}
try:
import pandas as pd
except ImportError:
pd = None
if output == "pandas":
logger.warning(
"Pandas is not installed. "
"Please install pandas using 'pip install pandas'."
"Output will be dictionary instead."
)
output = "dict"
curve_keys = []
curves_list = list(self.curves)
for i, curve in enumerate(curves_list):
x_data, y_data = curve.getData()
if x_data is not None or y_data is not None:
key = f"curve_{i}"
curve_keys.append(key)
if output == "dict":
data[key] = {"x": x_data.tolist(), "y": y_data.tolist()}
elif output == "pandas" and pd is not None:
data[key] = pd.DataFrame({"x": x_data, "y": y_data})
if output == "pandas" and pd is not None:
combined_data = pd.concat([data[key] for key in curve_keys], axis=1, keys=curve_keys)
return combined_data
return data
def export_to_matplotlib(self):
"""
Export current waveform to matplotlib GUI. Available only if matplotlib is installed in the environment.
"""
MatplotlibExporter(self.plot_item).export()
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.widgets.figure import BECFigure
app = QApplication(sys.argv)
widget = BECFigure()
widget.multi_waveform(monitor="waveform")
widget.show()
sys.exit(app.exec_())

View File

@ -0,0 +1 @@
{'files': ['multi_waveform_widget.py','multi-waveform_controls.ui']}

View File

@ -0,0 +1,54 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.multi_waveform.multi_waveform_widget import BECMultiWaveformWidget
DOM_XML = """
<ui language='c++'>
<widget class='BECMultiWaveformWidget' name='bec_multi_waveform_widget'>
</widget>
</ui>
"""
class BECMultiWaveformWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = BECMultiWaveformWidget(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Plots"
def icon(self):
return designer_material_icon(BECMultiWaveformWidget.ICON_NAME)
def includeFile(self):
return "bec_multi_waveform_widget"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "BECMultiWaveformWidget"
def toolTip(self):
return "BECMultiWaveformWidget"
def whatsThis(self):
return self.toolTip()

View File

@ -0,0 +1,99 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>561</width>
<height>86</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QGridLayout" name="gridLayout">
<item row="0" column="0">
<widget class="QLabel" name="label_curve_index">
<property name="text">
<string>Curve Index</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QSlider" name="slider_index">
<property name="orientation">
<enum>Qt::Orientation::Horizontal</enum>
</property>
</widget>
</item>
<item row="0" column="2">
<widget class="QSpinBox" name="spinbox_index"/>
</item>
<item row="0" column="3" colspan="3">
<widget class="QCheckBox" name="checkbox_highlight">
<property name="text">
<string>Highlight always last curve</string>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_opacity">
<property name="text">
<string>Opacity</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QSlider" name="slider_opacity">
<property name="maximum">
<number>100</number>
</property>
<property name="orientation">
<enum>Qt::Orientation::Horizontal</enum>
</property>
</widget>
</item>
<item row="1" column="3">
<widget class="QLabel" name="label_max_trace">
<property name="text">
<string>Max Trace</string>
</property>
</widget>
</item>
<item row="1" column="4">
<widget class="QSpinBox" name="spinbox_max_trace">
<property name="toolTip">
<string>How many curves should be displayed</string>
</property>
<property name="maximum">
<number>500</number>
</property>
<property name="value">
<number>200</number>
</property>
</widget>
</item>
<item row="1" column="5">
<widget class="QCheckBox" name="checkbox_flush_buffer">
<property name="toolTip">
<string>If hiddne curves should be deleted.</string>
</property>
<property name="text">
<string>Flush Buffer</string>
</property>
</widget>
</item>
<item row="1" column="2">
<widget class="QSpinBox" name="spinbox_opacity">
<property name="maximum">
<number>100</number>
</property>
</widget>
</item>
</layout>
</widget>
<resources/>
<connections/>
</ui>

View File

@ -0,0 +1,533 @@
import os
from typing import Literal
import pyqtgraph as pg
from bec_lib.device import ReadoutPriority
from bec_lib.logger import bec_logger
from qtpy.QtCore import Slot
from qtpy.QtWidgets import QVBoxLayout, QWidget
from bec_widgets.qt_utils.error_popups import SafeSlot
from bec_widgets.qt_utils.settings_dialog import SettingsDialog
from bec_widgets.qt_utils.toolbar import (
DeviceSelectionAction,
MaterialIconAction,
ModularToolBar,
SeparatorAction,
WidgetAction,
)
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.base_classes.device_input_base import BECDeviceFilter
from bec_widgets.widgets.colormap_widget.colormap_widget import BECColorMapWidget
from bec_widgets.widgets.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.figure.plots.axis_settings import AxisSettings
from bec_widgets.widgets.figure.plots.multi_waveform.multi_waveform import BECMultiWaveformConfig
logger = bec_logger.logger
class BECMultiWaveformWidget(BECWidget, QWidget):
ICON_NAME = "ssid_chart"
USER_ACCESS = [
"curves",
"set_monitor",
"set_curve_highlight",
"set_opacity",
"set_curve_limit",
"set_buffer_flush",
"set_highlight_last_curve",
"set_colormap",
"set",
"set_title",
"set_x_label",
"set_y_label",
"set_x_scale",
"set_y_scale",
"set_x_lim",
"set_y_lim",
"set_grid",
"set_colormap",
"enable_fps_monitor",
"lock_aspect_ratio",
"get_all_data",
"export",
]
def __init__(
self,
parent: QWidget | None = None,
config: BECMultiWaveformConfig | dict = None,
client=None,
gui_id: str | None = None,
) -> None:
if config is None:
config = BECMultiWaveformConfig(widget_class=self.__class__.__name__)
else:
if isinstance(config, dict):
config = BECMultiWaveformConfig(**config)
super().__init__(client=client, gui_id=gui_id)
QWidget.__init__(self, parent)
self.layout = QVBoxLayout(self)
self.layout.setSpacing(0)
self.layout.setContentsMargins(0, 0, 0, 0)
self.fig = BECFigure()
self.colormap_button = BECColorMapWidget(cmap="magma")
self.toolbar = ModularToolBar(
actions={
"monitor": DeviceSelectionAction(
"",
DeviceComboBox(
device_filter=BECDeviceFilter.DEVICE,
readout_priority_filter=ReadoutPriority.ASYNC,
),
),
"connect": MaterialIconAction(icon_name="link", tooltip="Connect Device"),
"separator_0": SeparatorAction(),
"colormap": WidgetAction(widget=self.colormap_button),
"separator_1": SeparatorAction(),
"save": MaterialIconAction(icon_name="save", tooltip="Open Export Dialog"),
"matplotlib": MaterialIconAction(
icon_name="photo_library", tooltip="Open Matplotlib Plot"
),
"separator_2": SeparatorAction(),
"drag_mode": MaterialIconAction(
icon_name="drag_pan", tooltip="Drag Mouse Mode", checkable=True
),
"rectangle_mode": MaterialIconAction(
icon_name="frame_inspect", tooltip="Rectangle Zoom Mode", checkable=True
),
"auto_range": MaterialIconAction(
icon_name="open_in_full", tooltip="Autorange Plot"
),
"crosshair": MaterialIconAction(
icon_name="point_scan", tooltip="Show Crosshair", checkable=True
),
"separator_3": SeparatorAction(),
"fps_monitor": MaterialIconAction(
icon_name="speed", tooltip="Show FPS Monitor", checkable=True
),
"axis_settings": MaterialIconAction(
icon_name="settings", tooltip="Open Configuration Dialog"
),
},
target_widget=self,
)
self.layout.addWidget(self.toolbar)
self.layout.addWidget(self.fig)
self.waveform = self.fig.multi_waveform() # FIXME config should be injected here
self.config = config
self.create_multi_waveform_controls()
self._hook_actions()
self.waveform.monitor_signal_updated.connect(self.update_controls_limits)
def create_multi_waveform_controls(self):
"""
Create the controls for the multi waveform widget.
"""
current_path = os.path.dirname(__file__)
self.controls = UILoader(self).loader(
os.path.join(current_path, "multi_waveform_controls.ui")
)
self.layout.addWidget(self.controls)
# Hook default controls properties
self.controls.checkbox_highlight.setChecked(self.config.highlight_last_curve)
self.controls.spinbox_opacity.setValue(self.config.opacity)
self.controls.slider_opacity.setValue(self.config.opacity)
self.controls.spinbox_max_trace.setValue(self.config.curve_limit)
self.controls.checkbox_flush_buffer.setChecked(self.config.flush_buffer)
# Connect signals
self.controls.spinbox_max_trace.valueChanged.connect(self.set_curve_limit)
self.controls.checkbox_flush_buffer.toggled.connect(self.set_buffer_flush)
self.controls.slider_opacity.valueChanged.connect(self.controls.spinbox_opacity.setValue)
self.controls.spinbox_opacity.valueChanged.connect(self.controls.slider_opacity.setValue)
self.controls.slider_opacity.valueChanged.connect(self.set_opacity)
self.controls.spinbox_opacity.valueChanged.connect(self.set_opacity)
self.controls.slider_index.valueChanged.connect(self.controls.spinbox_index.setValue)
self.controls.spinbox_index.valueChanged.connect(self.controls.slider_index.setValue)
self.controls.slider_index.valueChanged.connect(self.set_curve_highlight)
self.controls.spinbox_index.valueChanged.connect(self.set_curve_highlight)
self.controls.checkbox_highlight.toggled.connect(self.set_highlight_last_curve)
# Trigger first round of settings
self.set_curve_limit(self.config.curve_limit)
self.set_opacity(self.config.opacity)
self.set_highlight_last_curve(self.config.highlight_last_curve)
@Slot()
def update_controls_limits(self):
"""
Update the limits of the controls.
"""
num_curves = len(self.waveform.curves)
if num_curves == 0:
num_curves = 1 # Avoid setting max to 0
current_index = num_curves - 1
self.controls.slider_index.setMinimum(0)
self.controls.slider_index.setMaximum(self.waveform.number_of_visible_curves - 1)
self.controls.spinbox_index.setMaximum(self.waveform.number_of_visible_curves - 1)
if self.controls.checkbox_highlight.isChecked():
self.controls.slider_index.setValue(current_index)
self.controls.spinbox_index.setValue(current_index)
def _hook_actions(self):
self.toolbar.widgets["connect"].action.triggered.connect(self._connect_action)
# Separator 0
self.toolbar.widgets["save"].action.triggered.connect(self.export)
self.toolbar.widgets["matplotlib"].action.triggered.connect(self.export_to_matplotlib)
self.toolbar.widgets["colormap"].widget.colormap_changed_signal.connect(self.set_colormap)
# Separator 1
self.toolbar.widgets["drag_mode"].action.triggered.connect(self.enable_mouse_pan_mode)
self.toolbar.widgets["rectangle_mode"].action.triggered.connect(
self.enable_mouse_rectangle_mode
)
self.toolbar.widgets["auto_range"].action.triggered.connect(self._auto_range_from_toolbar)
self.toolbar.widgets["crosshair"].action.triggered.connect(self.waveform.toggle_crosshair)
# Separator 2
self.toolbar.widgets["fps_monitor"].action.triggered.connect(self.enable_fps_monitor)
self.toolbar.widgets["axis_settings"].action.triggered.connect(self.show_axis_settings)
###################################
# Dialog Windows
###################################
@SafeSlot(popup_error=True)
def _connect_action(self):
monitor_combo = self.toolbar.widgets["monitor"].device_combobox
monitor_name = monitor_combo.currentText()
self.set_monitor(monitor=monitor_name)
monitor_combo.setStyleSheet("QComboBox { background-color: " "; }")
def show_axis_settings(self):
dialog = SettingsDialog(
self,
settings_widget=AxisSettings(),
window_title="Axis Settings",
config=self.waveform._config_dict["axis"],
)
dialog.exec()
########################################
# User Access Methods from MultiWaveform
########################################
@property
def curves(self) -> list[pg.PlotDataItem]:
"""
Get the curves of the plot widget as a list
Returns:
list: List of curves.
"""
return list(self.waveform.curves)
@curves.setter
def curves(self, value: list[pg.PlotDataItem]):
self.waveform.curves = value
@SafeSlot(popup_error=True)
def set_monitor(self, monitor: str) -> None:
"""
Set the monitor of the plot widget.
Args:
monitor(str): The monitor to set.
"""
self.waveform.set_monitor(monitor)
if self.toolbar.widgets["monitor"].device_combobox.currentText() != monitor:
self.toolbar.widgets["monitor"].device_combobox.setCurrentText(monitor)
self.toolbar.widgets["monitor"].device_combobox.setStyleSheet(
"QComboBox { background-color: " "; }"
)
@SafeSlot(int)
def set_curve_highlight(self, index: int) -> None:
"""
Set the curve highlight of the plot widget by index
Args:
index(int): The index of the curve to highlight.
"""
if self.controls.checkbox_highlight.isChecked():
# If always highlighting the last curve, set index to -1
self.waveform.set_curve_highlight(-1)
else:
self.waveform.set_curve_highlight(index)
@SafeSlot(int)
def set_opacity(self, opacity: int) -> None:
"""
Set the opacity of the plot widget.
Args:
opacity(int): The opacity to set.
"""
self.waveform.set_opacity(opacity)
@SafeSlot(int)
def set_curve_limit(self, curve_limit: int) -> None:
"""
Set the maximum number of traces to display on the plot widget.
Args:
curve_limit(int): The maximum number of traces to display.
"""
flush_buffer = self.controls.checkbox_flush_buffer.isChecked()
self.waveform.set_curve_limit(curve_limit, flush_buffer)
self.update_controls_limits()
@SafeSlot(bool)
def set_buffer_flush(self, flush_buffer: bool) -> None:
"""
Set the buffer flush property of the plot widget.
Args:
flush_buffer(bool): True to flush the buffer, False to not flush the buffer.
"""
curve_limit = self.controls.spinbox_max_trace.value()
self.waveform.set_curve_limit(curve_limit, flush_buffer)
self.update_controls_limits()
@SafeSlot(bool)
def set_highlight_last_curve(self, enable: bool) -> None:
"""
Enable or disable highlighting of the last curve.
Args:
enable(bool): True to enable highlighting of the last curve, False to disable.
"""
self.waveform.config.highlight_last_curve = enable
if enable:
self.controls.slider_index.setEnabled(False)
self.controls.spinbox_index.setEnabled(False)
self.controls.checkbox_highlight.setChecked(True)
self.waveform.set_curve_highlight(-1)
else:
self.controls.slider_index.setEnabled(True)
self.controls.spinbox_index.setEnabled(True)
self.controls.checkbox_highlight.setChecked(False)
index = self.controls.spinbox_index.value()
self.waveform.set_curve_highlight(index)
@SafeSlot()
def set_colormap(self, colormap: str) -> None:
"""
Set the colormap of the plot widget.
Args:
colormap(str): The colormap to set.
"""
self.waveform.set_colormap(colormap)
###################################
# User Access Methods from PlotBase
###################################
def set(self, **kwargs):
"""
Set the properties of the plot widget.
Args:
**kwargs: Keyword arguments for the properties to be set.
Possible properties:
- title: str
- x_label: str
- y_label: str
- x_scale: Literal["linear", "log"]
- y_scale: Literal["linear", "log"]
- x_lim: tuple
- y_lim: tuple
- legend_label_size: int
"""
self.waveform.set(**kwargs)
def set_title(self, title: str):
"""
Set the title of the plot widget.
Args:
title(str): The title to set.
"""
self.waveform.set_title(title)
def set_x_label(self, x_label: str):
"""
Set the x-axis label of the plot widget.
Args:
x_label(str): The x-axis label to set.
"""
self.waveform.set_x_label(x_label)
def set_y_label(self, y_label: str):
"""
Set the y-axis label of the plot widget.
Args:
y_label(str): The y-axis label to set.
"""
self.waveform.set_y_label(y_label)
def set_x_scale(self, x_scale: Literal["linear", "log"]):
"""
Set the x-axis scale of the plot widget.
Args:
x_scale(str): The x-axis scale to set.
"""
self.waveform.set_x_scale(x_scale)
def set_y_scale(self, y_scale: Literal["linear", "log"]):
"""
Set the y-axis scale of the plot widget.
Args:
y_scale(str): The y-axis scale to set.
"""
self.waveform.set_y_scale(y_scale)
def set_x_lim(self, x_lim: tuple):
"""
Set x-axis limits of the plot widget.
Args:
x_lim(tuple): The x-axis limits to set.
"""
self.waveform.set_x_lim(x_lim)
def set_y_lim(self, y_lim: tuple):
"""
Set y-axis limits of the plot widget.
Args:
y_lim(tuple): The y-axis limits to set.
"""
self.waveform.set_y_lim(y_lim)
def set_legend_label_size(self, legend_label_size: int):
"""
Set the legend label size of the plot widget.
Args:
legend_label_size(int): The legend label size to set.
"""
self.waveform.set_legend_label_size(legend_label_size)
def set_auto_range(self, enabled: bool, axis: str = "xy"):
"""
Set the auto range of the plot widget.
Args:
enabled(bool): True to enable auto range, False to disable.
axis(str): The axis to set the auto range for. Default is "xy".
"""
self.waveform.set_auto_range(enabled, axis)
def enable_fps_monitor(self, enabled: bool):
"""
Enable or disable the FPS monitor
Args:
enabled(bool): True to enable the FPS monitor, False to disable.
"""
self.waveform.enable_fps_monitor(enabled)
if self.toolbar.widgets["fps_monitor"].action.isChecked() != enabled:
self.toolbar.widgets["fps_monitor"].action.setChecked(enabled)
@SafeSlot()
def _auto_range_from_toolbar(self):
"""
Set the auto range of the plot widget from the toolbar.
"""
self.waveform.set_auto_range(True, "xy")
def set_grid(self, x_grid: bool, y_grid: bool):
"""
Set the grid of the plot widget.
Args:
x_grid(bool): True to enable the x-grid, False to disable.
y_grid(bool): True to enable the y-grid, False to disable.
"""
self.waveform.set_grid(x_grid, y_grid)
def set_outer_axes(self, show: bool):
"""
Set the outer axes of the plot widget.
Args:
show(bool): True to show the outer axes, False to hide.
"""
self.waveform.set_outer_axes(show)
def lock_aspect_ratio(self, lock: bool):
"""
Lock the aspect ratio of the plot widget.
Args:
lock(bool): True to lock the aspect ratio, False to unlock.
"""
self.waveform.lock_aspect_ratio(lock)
@SafeSlot()
def enable_mouse_rectangle_mode(self):
"""
Enable the mouse rectangle mode of the plot widget.
"""
self.toolbar.widgets["rectangle_mode"].action.setChecked(True)
self.toolbar.widgets["drag_mode"].action.setChecked(False)
self.waveform.plot_item.getViewBox().setMouseMode(pg.ViewBox.RectMode)
@SafeSlot()
def enable_mouse_pan_mode(self):
"""
Enable the mouse pan mode of the plot widget.
"""
self.toolbar.widgets["drag_mode"].action.setChecked(True)
self.toolbar.widgets["rectangle_mode"].action.setChecked(False)
self.waveform.plot_item.getViewBox().setMouseMode(pg.ViewBox.PanMode)
def export(self):
"""
Export the plot widget.
"""
self.waveform.export()
def export_to_matplotlib(self):
"""
Export the plot widget to matplotlib.
"""
try:
import matplotlib as mpl
except ImportError:
self.warning_util.show_warning(
title="Matplotlib not installed",
message="Matplotlib is required for this feature.",
detailed_text="Please install matplotlib in your Python environment by using 'pip install matplotlib'.",
)
return
self.waveform.export_to_matplotlib()
#######################################
# User Access Methods from BECConnector
######################################
def cleanup(self):
self.fig.cleanup()
return super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = BECMultiWaveformWidget()
widget.show()
sys.exit(app.exec())

View File

@ -0,0 +1,17 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.multi_waveform.bec_multi_waveform_widget_plugin import (
BECMultiWaveformWidgetPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(BECMultiWaveformWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
main()

View File

@ -65,16 +65,18 @@ def test_add_remove_bec_figure_to_dock(bec_dock_area):
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")
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.widgets) == 1
assert len(d0.widget_list) == 1 assert len(d0.widget_list) == 1
assert len(fig.widgets) == 3 assert len(fig.widgets) == 4
assert fig.config.widget_class == "BECFigure" assert fig.config.widget_class == "BECFigure"
assert plt.config.widget_class == "BECWaveform" assert plt.config.widget_class == "BECWaveform"
assert im.config.widget_class == "BECImageShow" assert im.config.widget_class == "BECImageShow"
assert mm.config.widget_class == "BECMotorMap" assert mm.config.widget_class == "BECMotorMap"
assert mw.config.widget_class == "BECMultiWaveform"
def test_close_docks(bec_dock_area, qtbot): def test_close_docks(bec_dock_area, qtbot):

View File

@ -6,6 +6,7 @@ import pytest
from bec_widgets.widgets.figure import BECFigure from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.figure.plots.image.image import BECImageShow from bec_widgets.widgets.figure.plots.image.image import BECImageShow
from bec_widgets.widgets.figure.plots.motor_map.motor_map import BECMotorMap from bec_widgets.widgets.figure.plots.motor_map.motor_map import BECMotorMap
from bec_widgets.widgets.figure.plots.multi_waveform.multi_waveform import BECMultiWaveform
from bec_widgets.widgets.figure.plots.waveform.waveform import BECWaveform from bec_widgets.widgets.figure.plots.waveform.waveform import BECWaveform
from .client_mocks import mocked_client from .client_mocks import mocked_client
@ -63,10 +64,12 @@ def test_add_different_types_of_widgets(qtbot, mocked_client):
plt = bec_figure.plot(x_name="samx", y_name="bpm4i") plt = bec_figure.plot(x_name="samx", y_name="bpm4i")
im = bec_figure.image("eiger") im = bec_figure.image("eiger")
motor_map = bec_figure.motor_map("samx", "samy") motor_map = bec_figure.motor_map("samx", "samy")
multi_waveform = bec_figure.multi_waveform("waveform")
assert plt.__class__ == BECWaveform assert plt.__class__ == BECWaveform
assert im.__class__ == BECImageShow assert im.__class__ == BECImageShow
assert motor_map.__class__ == BECMotorMap assert motor_map.__class__ == BECMotorMap
assert multi_waveform.__class__ == BECMultiWaveform
def test_access_widgets_access_errors(qtbot, mocked_client): def test_access_widgets_access_errors(qtbot, mocked_client):

View File

@ -0,0 +1,253 @@
from unittest import mock
import numpy as np
import pytest
from bec_lib.endpoints import messages
from bec_widgets.utils import Colors
from bec_widgets.widgets.figure import BECFigure
from .client_mocks import mocked_client
from .conftest import create_widget
def test_set_monitor(qtbot, mocked_client):
"""Test that setting the monitor connects the appropriate slot."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("waveform1d")
assert multi_waveform.config.monitor == "waveform1d"
assert multi_waveform.connected is True
data_0 = np.random.rand(100)
msg = messages.DeviceMonitor1DMessage(
device="waveform1d", data=data_0, metadata={"scan_id": "12345"}
)
multi_waveform.on_monitor_1d_update(msg.content, msg.metadata)
data_waveform = multi_waveform.get_all_data()
print(data_waveform)
assert len(data_waveform) == 1
assert np.array_equal(data_waveform["curve_0"]["y"], data_0)
data_1 = np.random.rand(100)
msg = messages.DeviceMonitor1DMessage(
device="waveform1d", data=data_1, metadata={"scan_id": "12345"}
)
multi_waveform.on_monitor_1d_update(msg.content, msg.metadata)
data_waveform = multi_waveform.get_all_data()
assert len(data_waveform) == 2
assert np.array_equal(data_waveform["curve_0"]["y"], data_0)
assert np.array_equal(data_waveform["curve_1"]["y"], data_1)
def test_on_monitor_1d_update(qtbot, mocked_client):
"""Test that data updates add curves to the plot."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("test_monitor")
# Simulate receiving data updates
test_data = np.array([1, 2, 3, 4, 5])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
# Call the on_monitor_1d_update method
multi_waveform.on_monitor_1d_update(msg, metadata)
# Check that a curve has been added
assert len(multi_waveform.curves) == 1
# Check that the data in the curve is correct
curve = multi_waveform.curves[-1]
x_data, y_data = curve.getData()
assert np.array_equal(y_data, test_data)
# Simulate another data update
test_data_2 = np.array([6, 7, 8, 9, 10])
msg2 = {"data": test_data_2}
metadata2 = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg2, metadata2)
# Check that another curve has been added
assert len(multi_waveform.curves) == 2
# Check that the data in the curve is correct
curve2 = multi_waveform.curves[-1]
x_data2, y_data2 = curve2.getData()
assert np.array_equal(y_data2, test_data_2)
def test_set_curve_limit_no_flush(qtbot, mocked_client):
"""Test set_curve_limit with flush_buffer=False."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("test_monitor")
# Simulate adding multiple curves
for i in range(5):
test_data = np.array([i, i + 1, i + 2])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Check that there are 5 curves
assert len(multi_waveform.curves) == 5
# Set curve limit to 3 with flush_buffer=False
multi_waveform.set_curve_limit(3, flush_buffer=False)
# Check that curves are hidden, but not removed
assert len(multi_waveform.curves) == 5
visible_curves = [curve for curve in multi_waveform.curves if curve.isVisible()]
assert len(visible_curves) == 3
# The first two curves should be hidden
assert not multi_waveform.curves[0].isVisible()
assert not multi_waveform.curves[1].isVisible()
assert multi_waveform.curves[2].isVisible()
assert multi_waveform.curves[3].isVisible()
assert multi_waveform.curves[4].isVisible()
def test_set_curve_limit_flush(qtbot, mocked_client):
"""Test set_curve_limit with flush_buffer=True."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("test_monitor")
# Simulate adding multiple curves
for i in range(5):
test_data = np.array([i, i + 1, i + 2])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Check that there are 5 curves
assert len(multi_waveform.curves) == 5
# Set curve limit to 3 with flush_buffer=True
multi_waveform.set_curve_limit(3, flush_buffer=True)
# Check that only 3 curves remain
assert len(multi_waveform.curves) == 3
# The curves should be the last 3 added
x_data, y_data = multi_waveform.curves[0].getData()
assert np.array_equal(y_data, [2, 3, 4])
x_data, y_data = multi_waveform.curves[1].getData()
assert np.array_equal(y_data, [3, 4, 5])
x_data, y_data = multi_waveform.curves[2].getData()
assert np.array_equal(y_data, [4, 5, 6])
def test_set_curve_highlight(qtbot, mocked_client):
"""Test that the correct curve is highlighted."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("test_monitor")
# Simulate adding multiple curves
for i in range(3):
test_data = np.array([i, i + 1, i + 2])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Set highlight_last_curve to False
multi_waveform.highlight_last_curve = False
multi_waveform.set_curve_highlight(1) # Highlight the second curve (index 1)
# Check that the second curve is highlighted
visible_curves = [curve for curve in multi_waveform.curves if curve.isVisible()]
# Reverse the list to match indexing in set_curve_highlight
visible_curves = list(reversed(visible_curves))
for i, curve in enumerate(visible_curves):
pen = curve.opts["pen"]
width = pen.width()
if i == 1:
# Highlighted curve should have width 5
assert width == 5
else:
assert width == 1
def test_set_opacity(qtbot, mocked_client):
"""Test that setting opacity updates the curves."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("waveform1d")
# Simulate adding a curve
test_data = np.array([1, 2, 3])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Set opacity to 30
multi_waveform.set_opacity(30)
assert multi_waveform.config.opacity == 30
def test_set_colormap(qtbot, mocked_client):
"""Test that setting the colormap updates the curve colors."""
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("waveform1d")
# Simulate adding multiple curves
for i in range(3):
test_data = np.array([i, i + 1, i + 2])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Set a new colormap
multi_waveform.set_opacity(100)
multi_waveform.set_colormap("viridis")
# Check that the colors of the curves have changed accordingly
visible_curves = [curve for curve in multi_waveform.curves if curve.isVisible()]
# Get the colors applied
colors = Colors.evenly_spaced_colors(colormap="viridis", num=len(visible_curves), format="HEX")
for i, curve in enumerate(visible_curves):
pen = curve.opts["pen"]
pen_color = pen.color().name()
expected_color = colors[i]
# Compare pen color to expected color
assert pen_color.lower() == expected_color.lower()
def test_export_to_matplotlib(qtbot, mocked_client):
"""Test that export_to_matplotlib can be called without errors."""
try:
import matplotlib
except ImportError:
pytest.skip("Matplotlib not installed")
# Create a BECFigure
bec_figure = create_widget(qtbot, BECFigure, client=mocked_client)
# Add a multi_waveform plot
multi_waveform = bec_figure.multi_waveform()
multi_waveform.set_monitor("test_monitor")
# Simulate adding a curve
test_data = np.array([1, 2, 3])
msg = {"data": test_data}
metadata = {"scan_id": "scan_1"}
multi_waveform.on_monitor_1d_update(msg, metadata)
# Call export_to_matplotlib
with mock.patch("pyqtgraph.exporters.MatplotlibExporter.export") as mock_export:
multi_waveform.export_to_matplotlib()
mock_export.assert_called_once()

View File

@ -0,0 +1,295 @@
from unittest.mock import MagicMock, patch
import pytest
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QApplication
from bec_widgets.qt_utils.settings_dialog import SettingsDialog
from bec_widgets.utils.colors import apply_theme, get_theme_palette, set_theme
from bec_widgets.widgets.figure.plots.axis_settings import AxisSettings
from bec_widgets.widgets.multi_waveform.multi_waveform_widget import BECMultiWaveformWidget
from .client_mocks import mocked_client
@pytest.fixture
def multi_waveform_widget(qtbot, mocked_client):
widget = BECMultiWaveformWidget(client=mocked_client())
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
return widget
@pytest.fixture
def mock_waveform(multi_waveform_widget):
waveform_mock = MagicMock()
multi_waveform_widget.waveform = waveform_mock
return waveform_mock
def test_multi_waveform_widget_init(multi_waveform_widget):
assert multi_waveform_widget is not None
assert multi_waveform_widget.client is not None
assert isinstance(multi_waveform_widget, BECMultiWaveformWidget)
assert multi_waveform_widget.config.widget_class == "BECMultiWaveformWidget"
###################################
# Wrapper methods for Waveform
###################################
def test_multi_waveform_widget_set_monitor(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_monitor("waveform1d")
mock_waveform.set_monitor.assert_called_once_with("waveform1d")
def test_multi_waveform_widget_set_curve_highlight_last_active(
multi_waveform_widget, mock_waveform
):
multi_waveform_widget.set_curve_highlight(1)
mock_waveform.set_curve_highlight.assert_called_once_with(-1)
def test_multi_waveform_widget_set_curve_highlight_last_not_active(
multi_waveform_widget, mock_waveform
):
multi_waveform_widget.set_highlight_last_curve(False)
multi_waveform_widget.set_curve_highlight(1)
mock_waveform.set_curve_highlight.assert_called_with(1)
def test_multi_waveform_widget_set_opacity(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_opacity(50)
mock_waveform.set_opacity.assert_called_once_with(50)
def test_multi_waveform_widget_set_curve_limit(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_curve_limit(10)
mock_waveform.set_curve_limit.assert_called_once_with(
10, multi_waveform_widget.controls.checkbox_flush_buffer.isChecked()
)
def test_multi_waveform_widget_set_buffer_flush(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_buffer_flush(True)
mock_waveform.set_curve_limit.assert_called_once_with(
multi_waveform_widget.controls.spinbox_max_trace.value(), True
)
def test_multi_waveform_widget_set_highlight_last_curve(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_highlight_last_curve(True)
assert multi_waveform_widget.waveform.config.highlight_last_curve is True
assert not multi_waveform_widget.controls.slider_index.isEnabled()
assert not multi_waveform_widget.controls.spinbox_index.isEnabled()
mock_waveform.set_curve_highlight.assert_called_once_with(-1)
def test_multi_waveform_widget_set_colormap(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set_colormap("viridis")
mock_waveform.set_colormap.assert_called_once_with("viridis")
def test_multi_waveform_widget_set_base(multi_waveform_widget, mock_waveform):
multi_waveform_widget.set(
title="Test Title",
x_label="X Label",
y_label="Y Label",
x_scale="linear",
y_scale="log",
x_lim=(0, 10),
y_lim=(0, 10),
)
mock_waveform.set.assert_called_once_with(
title="Test Title",
x_label="X Label",
y_label="Y Label",
x_scale="linear",
y_scale="log",
x_lim=(0, 10),
y_lim=(0, 10),
)
###################################
# Toolbar interactions
###################################
def test_toolbar_connect_action_triggered(multi_waveform_widget, qtbot):
action_connect = multi_waveform_widget.toolbar.widgets["connect"].action
device_combobox = multi_waveform_widget.toolbar.widgets["monitor"].device_combobox
device_combobox.addItem("test_monitor")
device_combobox.setCurrentText("test_monitor")
with patch.object(multi_waveform_widget, "set_monitor") as mock_set_monitor:
action_connect.trigger()
mock_set_monitor.assert_called_once_with(monitor="test_monitor")
def test_toolbar_drag_mode_action_triggered(multi_waveform_widget, qtbot):
action_drag = multi_waveform_widget.toolbar.widgets["drag_mode"].action
action_rectangle = multi_waveform_widget.toolbar.widgets["rectangle_mode"].action
action_drag.trigger()
assert action_drag.isChecked() == True
assert action_rectangle.isChecked() == False
def test_toolbar_rectangle_mode_action_triggered(multi_waveform_widget, qtbot):
action_drag = multi_waveform_widget.toolbar.widgets["drag_mode"].action
action_rectangle = multi_waveform_widget.toolbar.widgets["rectangle_mode"].action
action_rectangle.trigger()
assert action_drag.isChecked() == False
assert action_rectangle.isChecked() == True
def test_toolbar_auto_range_action_triggered(multi_waveform_widget, mock_waveform, qtbot):
action = multi_waveform_widget.toolbar.widgets["auto_range"].action
action.trigger()
qtbot.wait(200)
mock_waveform.set_auto_range.assert_called_once_with(True, "xy")
###################################
# Control Panel interactions
###################################
def test_controls_opacity_slider(multi_waveform_widget, mock_waveform):
multi_waveform_widget.controls.slider_opacity.setValue(75)
mock_waveform.set_opacity.assert_called_with(75)
assert multi_waveform_widget.controls.spinbox_opacity.value() == 75
def test_controls_opacity_spinbox(multi_waveform_widget, mock_waveform):
multi_waveform_widget.controls.spinbox_opacity.setValue(25)
mock_waveform.set_opacity.assert_called_with(25)
assert multi_waveform_widget.controls.slider_opacity.value() == 25
def test_controls_max_trace_spinbox(multi_waveform_widget, mock_waveform):
multi_waveform_widget.controls.spinbox_max_trace.setValue(15)
mock_waveform.set_curve_limit.assert_called_with(
15, multi_waveform_widget.controls.checkbox_flush_buffer.isChecked()
)
def test_controls_flush_buffer_checkbox(multi_waveform_widget, mock_waveform):
multi_waveform_widget.controls.checkbox_flush_buffer.setChecked(True)
mock_waveform.set_curve_limit.assert_called_with(
multi_waveform_widget.controls.spinbox_max_trace.value(), True
)
def test_controls_highlight_checkbox(multi_waveform_widget, mock_waveform):
multi_waveform_widget.controls.checkbox_highlight.setChecked(False)
assert multi_waveform_widget.waveform.config.highlight_last_curve is False
assert multi_waveform_widget.controls.slider_index.isEnabled()
assert multi_waveform_widget.controls.spinbox_index.isEnabled()
index = multi_waveform_widget.controls.spinbox_index.value()
mock_waveform.set_curve_highlight.assert_called_with(index)
###################################
# Axis Settings Dialog Tests
###################################
def show_axis_dialog(qtbot, multi_waveform_widget):
axis_dialog = SettingsDialog(
multi_waveform_widget,
settings_widget=AxisSettings(),
window_title="Axis Settings",
config=multi_waveform_widget.waveform._config_dict["axis"],
)
qtbot.addWidget(axis_dialog)
qtbot.waitExposed(axis_dialog)
return axis_dialog
def test_axis_dialog_with_axis_limits(qtbot, multi_waveform_widget):
multi_waveform_widget.set(
title="Test Title",
x_label="X Label",
y_label="Y Label",
x_scale="linear",
y_scale="log",
x_lim=(0, 10),
y_lim=(0, 10),
)
axis_dialog = show_axis_dialog(qtbot, multi_waveform_widget)
assert axis_dialog is not None
assert axis_dialog.widget.ui.plot_title.text() == "Test Title"
assert axis_dialog.widget.ui.x_label.text() == "X Label"
assert axis_dialog.widget.ui.y_label.text() == "Y Label"
assert axis_dialog.widget.ui.x_scale.currentText() == "linear"
assert axis_dialog.widget.ui.y_scale.currentText() == "log"
assert axis_dialog.widget.ui.x_min.value() == 0
assert axis_dialog.widget.ui.x_max.value() == 10
assert axis_dialog.widget.ui.y_min.value() == 0
assert axis_dialog.widget.ui.y_max.value() == 10
def test_axis_dialog_set_properties(qtbot, multi_waveform_widget):
axis_dialog = show_axis_dialog(qtbot, multi_waveform_widget)
axis_dialog.widget.ui.plot_title.setText("New Title")
axis_dialog.widget.ui.x_label.setText("New X Label")
axis_dialog.widget.ui.y_label.setText("New Y Label")
axis_dialog.widget.ui.x_scale.setCurrentText("log")
axis_dialog.widget.ui.y_scale.setCurrentText("linear")
axis_dialog.widget.ui.x_min.setValue(5)
axis_dialog.widget.ui.x_max.setValue(15)
axis_dialog.widget.ui.y_min.setValue(5)
axis_dialog.widget.ui.y_max.setValue(15)
axis_dialog.accept()
assert multi_waveform_widget.waveform.config.axis.title == "New Title"
assert multi_waveform_widget.waveform.config.axis.x_label == "New X Label"
assert multi_waveform_widget.waveform.config.axis.y_label == "New Y Label"
assert multi_waveform_widget.waveform.config.axis.x_scale == "log"
assert multi_waveform_widget.waveform.config.axis.y_scale == "linear"
assert multi_waveform_widget.waveform.config.axis.x_lim == (5, 15)
assert multi_waveform_widget.waveform.config.axis.y_lim == (5, 15)
###################################
# Theme Update Test
###################################
def test_multi_waveform_widget_theme_update(qtbot, multi_waveform_widget):
"""Test theme update for multi waveform widget."""
qapp = QApplication.instance()
# Set the theme to dark
set_theme("dark")
palette = get_theme_palette()
waveform_color_dark = multi_waveform_widget.waveform.plot_item.getAxis("left").pen().color()
bg_color = multi_waveform_widget.fig.backgroundBrush().color()
assert bg_color == QColor("black")
assert waveform_color_dark == palette.text().color()
# Set the theme to light
set_theme("light")
palette = get_theme_palette()
waveform_color_light = multi_waveform_widget.waveform.plot_item.getAxis("left").pen().color()
bg_color = multi_waveform_widget.fig.backgroundBrush().color()
assert bg_color == QColor("white")
assert waveform_color_light == palette.text().color()
assert waveform_color_dark != waveform_color_light
# Set the theme to auto and simulate OS theme change
set_theme("auto")
qapp.theme_signal.theme_updated.emit("dark")
apply_theme("dark")
waveform_color = multi_waveform_widget.waveform.plot_item.getAxis("left").pen().color()
bg_color = multi_waveform_widget.fig.backgroundBrush().color()
assert bg_color == QColor("black")
assert waveform_color == waveform_color_dark