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

feat(scatter_waveform): scatter waveform widget based on new Plotbase

Browse Source
This commit is contained in:
wyzula_j
2025-03-12 15:38:56 +01:00
parent 0dd9617e6e
commit 95fcf016c3
17 changed files with 1847 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
bec_widgets/cli/client.py
View File

@ -40,6 +40,7 @@ class Widgets(str, enum.Enum):
ResumeButton = "ResumeButton"
RingProgressBar = "RingProgressBar"
ScanControl = "ScanControl"
ScatterWaveform = "ScatterWaveform"
SignalComboBox = "SignalComboBox"
SignalLineEdit = "SignalLineEdit"
StopButton = "StopButton"
@ -3778,6 +3779,349 @@ class ScanMetadata(RPCBase):
"""
class ScatterCurve(RPCBase):
@property
@rpc_call
def color_map(self) -> "str":
"""
The color map of the scatter curve.
"""
class ScatterWaveform(RPCBase):
@property
@rpc_call
def enable_toolbar(self) -> "bool":
"""
Show Toolbar.
"""
@enable_toolbar.setter
@rpc_call
def enable_toolbar(self) -> "bool":
"""
Show Toolbar.
"""
@property
@rpc_call
def enable_side_panel(self) -> "bool":
"""
Show Side Panel
"""
@enable_side_panel.setter
@rpc_call
def enable_side_panel(self) -> "bool":
"""
Show Side Panel
"""
@property
@rpc_call
def enable_fps_monitor(self) -> "bool":
"""
Enable the FPS monitor.
"""
@enable_fps_monitor.setter
@rpc_call
def enable_fps_monitor(self) -> "bool":
"""
Enable the FPS monitor.
"""
@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:
"""
@property
@rpc_call
def title(self) -> "str":
"""
Set title of the plot.
"""
@title.setter
@rpc_call
def title(self) -> "str":
"""
Set title of the plot.
"""
@property
@rpc_call
def x_label(self) -> "str":
"""
The set label for the x-axis.
"""
@x_label.setter
@rpc_call
def x_label(self) -> "str":
"""
The set label for the x-axis.
"""
@property
@rpc_call
def y_label(self) -> "str":
"""
The set label for the y-axis.
"""
@y_label.setter
@rpc_call
def y_label(self) -> "str":
"""
The set label for the y-axis.
"""
@property
@rpc_call
def x_limits(self) -> "QPointF":
"""
Get the x limits of the plot.
"""
@x_limits.setter
@rpc_call
def x_limits(self) -> "QPointF":
"""
Get the x limits of the plot.
"""
@property
@rpc_call
def y_limits(self) -> "QPointF":
"""
Get the y limits of the plot.
"""
@y_limits.setter
@rpc_call
def y_limits(self) -> "QPointF":
"""
Get the y limits of the plot.
"""
@property
@rpc_call
def x_grid(self) -> "bool":
"""
Show grid on the x-axis.
"""
@x_grid.setter
@rpc_call
def x_grid(self) -> "bool":
"""
Show grid on the x-axis.
"""
@property
@rpc_call
def y_grid(self) -> "bool":
"""
Show grid on the y-axis.
"""
@y_grid.setter
@rpc_call
def y_grid(self) -> "bool":
"""
Show grid on the y-axis.
"""
@property
@rpc_call
def inner_axes(self) -> "bool":
"""
Show inner axes of the plot widget.
"""
@inner_axes.setter
@rpc_call
def inner_axes(self) -> "bool":
"""
Show inner axes of the plot widget.
"""
@property
@rpc_call
def outer_axes(self) -> "bool":
"""
Show the outer axes of the plot widget.
"""
@outer_axes.setter
@rpc_call
def outer_axes(self) -> "bool":
"""
Show the outer axes of the plot widget.
"""
@property
@rpc_call
def lock_aspect_ratio(self) -> "bool":
"""
Lock aspect ratio of the plot widget.
"""
@lock_aspect_ratio.setter
@rpc_call
def lock_aspect_ratio(self) -> "bool":
"""
Lock aspect ratio of the plot widget.
"""
@property
@rpc_call
def auto_range_x(self) -> "bool":
"""
Set auto range for the x-axis.
"""
@auto_range_x.setter
@rpc_call
def auto_range_x(self) -> "bool":
"""
Set auto range for the x-axis.
"""
@property
@rpc_call
def auto_range_y(self) -> "bool":
"""
Set auto range for the y-axis.
"""
@auto_range_y.setter
@rpc_call
def auto_range_y(self) -> "bool":
"""
Set auto range for the y-axis.
"""
@property
@rpc_call
def x_log(self) -> "bool":
"""
Set X-axis to log scale if True, linear if False.
"""
@x_log.setter
@rpc_call
def x_log(self) -> "bool":
"""
Set X-axis to log scale if True, linear if False.
"""
@property
@rpc_call
def y_log(self) -> "bool":
"""
Set Y-axis to log scale if True, linear if False.
"""
@y_log.setter
@rpc_call
def y_log(self) -> "bool":
"""
Set Y-axis to log scale if True, linear if False.
"""
@property
@rpc_call
def legend_label_size(self) -> "int":
"""
The font size of the legend font.
"""
@legend_label_size.setter
@rpc_call
def legend_label_size(self) -> "int":
"""
The font size of the legend font.
"""
@property
@rpc_call
def main_curve(self) -> "ScatterCurve":
"""
The main scatter curve item.
"""
@property
@rpc_call
def color_map(self) -> "str":
"""
The color map of the scatter waveform.
"""
@color_map.setter
@rpc_call
def color_map(self) -> "str":
"""
The color map of the scatter waveform.
"""
@rpc_call
def plot(
self,
x_name: "str",
y_name: "str",
z_name: "str",
x_entry: "None | str" = None,
y_entry: "None | str" = None,
z_entry: "None | str" = None,
color_map: "str | None" = "magma",
label: "str | None" = None,
validate_bec: "bool" = True,
) -> "ScatterCurve":
"""
Plot the data from the device signals.
Args:
x_name (str): The name of the x device signal.
y_name (str): The name of the y device signal.
z_name (str): The name of the z device signal.
x_entry (None | str): The x entry of the device signal.
y_entry (None | str): The y entry of the device signal.
z_entry (None | str): The z entry of the device signal.
color_map (str | None): The color map of the scatter waveform.
label (str | None): The label of the curve.
validate_bec (bool): Whether to validate the device signals with current BEC instance.
Returns:
ScatterCurve: The scatter curve object.
"""
@rpc_call
def update_with_scan_history(self, scan_index: "int" = None, scan_id: "str" = None):
"""
Update the scan curves with the data from the scan storage.
Provide only one of scan_id or scan_index.
Args:
scan_id(str, optional): ScanID of the scan to be updated. Defaults to None.
scan_index(int, optional): Index of the scan to be updated. Defaults to None.
"""
@rpc_call
def clear_all(self):
"""
Clear all the curves from the plot.
"""
class SignalComboBox(RPCBase):
"""Line edit widget for device input with autocomplete for device names."""

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

@ -22,6 +22,7 @@ from bec_widgets.widgets.containers.layout_manager.layout_manager import LayoutM
from bec_widgets.widgets.editors.jupyter_console.jupyter_console import BECJupyterConsole
from bec_widgets.widgets.plots_next_gen.image.image import Image
from bec_widgets.widgets.plots_next_gen.plot_base import PlotBase
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_waveform import ScatterWaveform
from bec_widgets.widgets.plots_next_gen.waveform.waveform import Waveform
@ -67,6 +68,8 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
"pb": self.pb,
"pi": self.pi,
"wf": self.wf,
"scatter": self.scatter,
"scatter_mi": self.scatter,
}
)
@ -134,6 +137,15 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
tab_widget.addTab(sixth_tab, "Image Next Gen")
tab_widget.setCurrentIndex(5)
seventh_tab = QWidget()
seventh_tab_layout = QVBoxLayout(seventh_tab)
self.scatter = ScatterWaveform()
self.scatter_mi = self.scatter.main_curve
self.scatter.plot("samx", "samy", "bpm4i")
seventh_tab_layout.addWidget(self.scatter)
tab_widget.addTab(seventh_tab, "Scatter Waveform")
tab_widget.setCurrentIndex(6)
# add stuff to the new Waveform widget
self._init_waveform()

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

@ -28,6 +28,7 @@ from bec_widgets.widgets.editors.vscode.vscode import VSCodeEditor
from bec_widgets.widgets.plots.motor_map.motor_map_widget import BECMotorMapWidget
from bec_widgets.widgets.plots.multi_waveform.multi_waveform_widget import BECMultiWaveformWidget
from bec_widgets.widgets.plots_next_gen.image.image import Image
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_waveform import ScatterWaveform
from bec_widgets.widgets.plots_next_gen.waveform.waveform import Waveform
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.services.bec_queue.bec_queue import BECQueue
@ -96,6 +97,11 @@ class BECDockArea(BECWidget, QWidget):
"waveform": MaterialIconAction(
icon_name=Waveform.ICON_NAME, tooltip="Add Waveform", filled=True
),
"scatter_waveform": MaterialIconAction(
icon_name=ScatterWaveform.ICON_NAME,
tooltip="Add Scatter Waveform",
filled=True,
),
"multi_waveform": MaterialIconAction(
icon_name=BECMultiWaveformWidget.ICON_NAME,
tooltip="Add Multi Waveform",
@ -176,6 +182,9 @@ class BECDockArea(BECWidget, QWidget):
self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect(
lambda: self._create_widget_from_toolbar(widget_name="Waveform")
)
self.toolbar.widgets["menu_plots"].widgets["scatter_waveform"].triggered.connect(
lambda: self._create_widget_from_toolbar(widget_name="ScatterWaveform")
)
self.toolbar.widgets["menu_plots"].widgets["multi_waveform"].triggered.connect(
lambda: self._create_widget_from_toolbar(widget_name="BECMultiWaveformWidget")
)

0
bec_widgets/widgets/plots_next_gen/scatter_waveform/__init__.py Normal file
View File

17
bec_widgets/widgets/plots_next_gen/scatter_waveform/register_scatter_waveform.py Normal file
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.plots_next_gen.scatter_waveform.scatter_waveform_plugin import (
ScatterWaveformPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(ScatterWaveformPlugin())
if __name__ == "__main__": # pragma: no cover
main()

194
bec_widgets/widgets/plots_next_gen/scatter_waveform/scatter_curve.py Normal file
View File

@ -0,0 +1,194 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Literal
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger
from pydantic import BaseModel, Field, ValidationError, field_validator
from qtpy import QtCore
from bec_widgets.utils import BECConnector, Colors, ConnectionConfig
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_waveform import ScatterWaveform
logger = bec_logger.logger
# noinspection PyDataclass
class ScatterDeviceSignal(BaseModel):
"""The configuration of a signal in the scatter waveform widget."""
name: str
entry: str
model_config: dict = {"validate_assignment": True}
# noinspection PyDataclass
class ScatterCurveConfig(ConnectionConfig):
parent_id: str | None = Field(None, description="The parent plot of the curve.")
label: str | None = Field(None, description="The label of the curve.")
color: str | tuple = Field("#808080", description="The color of the curve.")
symbol: str | None = Field("o", description="The symbol of the curve.")
symbol_size: int | None = Field(7, description="The size of the symbol of the curve.")
pen_width: int | None = Field(4, description="The width of the pen of the curve.")
pen_style: Literal["solid", "dash", "dot", "dashdot"] = Field(
"solid", description="The style of the pen of the curve."
)
color_map: str | None = Field(
"magma", description="The color palette of the figure widget.", validate_default=True
)
x_device: ScatterDeviceSignal | None = Field(
None, description="The x device signal of the scatter waveform."
)
y_device: ScatterDeviceSignal | None = Field(
None, description="The y device signal of the scatter waveform."
)
z_device: ScatterDeviceSignal | None = Field(
None, description="The z device signal of the scatter waveform."
)
model_config: dict = {"validate_assignment": True}
_validate_color_palette = field_validator("color_map")(Colors.validate_color_map)
class ScatterCurve(BECConnector, pg.PlotDataItem):
"""Scatter curve item for the scatter waveform widget."""
USER_ACCESS = ["color_map"]
def __init__(
self,
parent_item: ScatterWaveform,
name: str | None = None,
config: ScatterCurveConfig | None = None,
gui_id: str | None = None,
**kwargs,
):
if config is None:
config = ScatterCurveConfig(
label=name,
widget_class=self.__class__.__name__,
parent_id=parent_item.config.gui_id,
)
self.config = config
else:
self.config = config
name = config.label
super().__init__(config=config, gui_id=gui_id)
pg.PlotDataItem.__init__(self, name=name)
self.parent_item = parent_item
self.data_z = None # color scaling needs to be cashed for changing colormap
self.apply_config()
def apply_config(self, config: dict | ScatterCurveConfig | None = None, **kwargs) -> None:
"""
Apply the configuration to the curve.
Args:
config(dict|ScatterCurveConfig, optional): The configuration to apply.
"""
if config is not None:
if isinstance(config, dict):
config = ScatterCurveConfig(**config)
self.config = config
pen_style_map = {
"solid": QtCore.Qt.SolidLine,
"dash": QtCore.Qt.DashLine,
"dot": QtCore.Qt.DotLine,
"dashdot": QtCore.Qt.DashDotLine,
}
pen_style = pen_style_map.get(self.config.pen_style, QtCore.Qt.SolidLine)
pen = pg.mkPen(color=self.config.color, width=self.config.pen_width, style=pen_style)
self.setPen(pen)
if self.config.symbol:
self.setSymbolSize(self.config.symbol_size)
self.setSymbol(self.config.symbol)
@property
def color_map(self) -> str:
"""The color map of the scatter curve."""
return self.config.color_map
@color_map.setter
def color_map(self, value: str):
"""
Set the color map of the scatter curve.
Args:
value(str): The color map to set.
"""
try:
if value != self.config.color_map:
self.config.color_map = value
self.refresh_color_map(value)
except ValidationError:
return
def set_data(
self,
x: list[float] | np.ndarray,
y: list[float] | np.ndarray,
z: list[float] | np.ndarray,
color_map: str | None = None,
):
"""
Set the data of the scatter curve.
Args:
x (list[float] | np.ndarray): The x data of the scatter curve.
y (list[float] | np.ndarray): The y data of the scatter curve.
z (list[float] | np.ndarray): The z data of the scatter curve.
color_map (str | None): The color map of the scatter curve.
"""
if color_map is None:
color_map = self.config.color_map
self.data_z = z
color_z = self._make_z_gradient(z, color_map)
try:
self.setData(x=x, y=y, symbolBrush=color_z)
except TypeError:
logger.error("Error in setData, one of the data arrays is None")
def _make_z_gradient(self, data_z: list | np.ndarray, colormap: str) -> list | None:
"""
Make a gradient color for the z values.
Args:
data_z(list|np.ndarray): Z values.
colormap(str): Colormap for the gradient color.
Returns:
list: List of colors for the z values.
"""
# Normalize z_values for color mapping
z_min, z_max = np.min(data_z), np.max(data_z)
if z_max != z_min: # Ensure that there is a range in the z values
z_values_norm = (data_z - z_min) / (z_max - z_min)
colormap = pg.colormap.get(colormap) # using colormap from global settings
colors = [colormap.map(z, mode="qcolor") for z in z_values_norm]
return colors
else:
return None
def refresh_color_map(self, color_map: str):
"""
Refresh the color map of the scatter curve.
Args:
color_map(str): The color map to use.
"""
x_data, y_data = self.getData()
if x_data is None or y_data is None:
return
if self.data_z is not None:
self.set_data(x_data, y_data, self.data_z, color_map)

518
bec_widgets/widgets/plots_next_gen/scatter_waveform/scatter_waveform.py Normal file
View File

@ -0,0 +1,518 @@
from __future__ import annotations
import json
import pyqtgraph as pg
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QHBoxLayout, QMainWindow, QWidget
from bec_widgets.qt_utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.qt_utils.settings_dialog import SettingsDialog
from bec_widgets.qt_utils.toolbar import MaterialIconAction
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.colors import set_theme
from bec_widgets.widgets.plots_next_gen.plot_base import PlotBase
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_curve import (
ScatterCurve,
ScatterCurveConfig,
ScatterDeviceSignal,
)
from bec_widgets.widgets.plots_next_gen.scatter_waveform.settings.scatter_curve_setting import (
ScatterCurveSettings,
)
logger = bec_logger.logger
# noinspection PyDataclass
class ScatterWaveformConfig(ConnectionConfig):
color_map: str | None = Field(
"magma",
description="The color map of the z scaling of scatter waveform.",
validate_default=True,
)
model_config: dict = {"validate_assignment": True}
_validate_color_palette = field_validator("color_map")(Colors.validate_color_map)
class ScatterWaveform(PlotBase):
PLUGIN = True
RPC = True
ICON_NAME = "scatter_plot"
USER_ACCESS = [
# General PlotBase Settings
"enable_toolbar",
"enable_toolbar.setter",
"enable_side_panel",
"enable_side_panel.setter",
"enable_fps_monitor",
"enable_fps_monitor.setter",
"set",
"title",
"title.setter",
"x_label",
"x_label.setter",
"y_label",
"y_label.setter",
"x_limits",
"x_limits.setter",
"y_limits",
"y_limits.setter",
"x_grid",
"x_grid.setter",
"y_grid",
"y_grid.setter",
"inner_axes",
"inner_axes.setter",
"outer_axes",
"outer_axes.setter",
"lock_aspect_ratio",
"lock_aspect_ratio.setter",
"auto_range_x",
"auto_range_x.setter",
"auto_range_y",
"auto_range_y.setter",
"x_log",
"x_log.setter",
"y_log",
"y_log.setter",
"legend_label_size",
"legend_label_size.setter",
# Scatter Waveform Specific RPC Access
"main_curve",
"color_map",
"color_map.setter",
"plot",
"update_with_scan_history",
"clear_all",
]
sync_signal_update = Signal()
new_scan = Signal()
new_scan_id = Signal(str)
scatter_waveform_property_changed = Signal()
def __init__(
self,
parent: QWidget | None = None,
config: ScatterWaveformConfig | None = None,
client=None,
gui_id: str | None = None,
popups: bool = True,
**kwargs,
):
if config is None:
config = ScatterWaveformConfig(widget_class=self.__class__.__name__)
super().__init__(
parent=parent, config=config, client=client, gui_id=gui_id, popups=popups, **kwargs
)
self._main_curve = ScatterCurve(parent_item=self)
# For PropertyManager identification
self.setObjectName("ScatterWaveform")
# Specific GUI elements
self.scatter_dialog = None
# Scan Data
self.old_scan_id = None
self.scan_id = None
self.scan_item = None
# Scan status update loop
self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
self.bec_dispatcher.connect_slot(self.on_scan_progress, MessageEndpoints.scan_progress())
# Curve update loop
self.proxy_update_sync = pg.SignalProxy(
self.sync_signal_update, rateLimit=25, slot=self.update_sync_curves
)
self._init_scatter_curve_settings()
self.update_with_scan_history(-1)
################################################################################
# Widget Specific GUI interactions
################################################################################
def _init_scatter_curve_settings(self):
"""
Initialize the scatter curve settings menu.
"""
scatter_curve_settings = ScatterCurveSettings(target_widget=self, popup=False)
self.side_panel.add_menu(
action_id="scatter_curve",
icon_name="scatter_plot",
tooltip="Show Scatter Curve Settings",
widget=scatter_curve_settings,
title="Scatter Curve Settings",
)
def add_popups(self):
"""
Add popups to the ScatterWaveform widget.
"""
super().add_popups()
scatter_curve_setting_action = MaterialIconAction(
icon_name="scatter_plot",
tooltip="Show Scatter Curve Settings",
checkable=True,
parent=self,
)
self.toolbar.add_action_to_bundle(
bundle_id="popup_bundle",
action_id="scatter_waveform_settings",
action=scatter_curve_setting_action,
target_widget=self,
)
self.toolbar.widgets["scatter_waveform_settings"].action.triggered.connect(
self.show_scatter_curve_settings
)
def show_scatter_curve_settings(self):
"""
Show the scatter curve settings dialog.
"""
scatter_settings_action = self.toolbar.widgets["scatter_waveform_settings"].action
if self.scatter_dialog is None or not self.scatter_dialog.isVisible():
scatter_settings = ScatterCurveSettings(target_widget=self, popup=True)
self.scatter_dialog = SettingsDialog(
self,
settings_widget=scatter_settings,
window_title="Scatter Curve Settings",
modal=False,
)
self.scatter_dialog.resize(620, 200)
# When the dialog is closed, update the toolbar icon and clear the reference
self.scatter_dialog.finished.connect(self._scatter_dialog_closed)
self.scatter_dialog.show()
scatter_settings_action.setChecked(True)
else:
# If already open, bring it to the front
self.scatter_dialog.raise_()
self.scatter_dialog.activateWindow()
scatter_settings_action.setChecked(True) # keep it toggled
def _scatter_dialog_closed(self):
"""
Slot for when the scatter curve settings dialog is closed.
"""
self.scatter_dialog = None
self.toolbar.widgets["scatter_waveform_settings"].action.setChecked(False)
################################################################################
# Widget Specific Properties
################################################################################
@property
def main_curve(self) -> ScatterCurve:
"""The main scatter curve item."""
return self._main_curve
@SafeProperty(str)
def color_map(self) -> str:
"""The color map of the scatter waveform."""
return self.config.color_map
@color_map.setter
def color_map(self, value: str):
"""
Set the color map of the scatter waveform.
Args:
value(str): The color map to set.
"""
try:
self.config.color_map = value
self.main_curve.color_map = value
self.scatter_waveform_property_changed.emit()
except ValidationError:
return
@SafeProperty(str, designable=False, popup_error=True)
def curve_json(self) -> str:
"""
Get the curve configuration as a JSON string.
"""
return json.dumps(self.main_curve.config.model_dump(), indent=2)
@curve_json.setter
def curve_json(self, value: str):
"""
Set the curve configuration from a JSON string.
Args:
value(str): The JSON string to set the curve configuration from.
"""
try:
config = ScatterCurveConfig(**json.loads(value))
self._add_main_scatter_curve(config)
except json.JSONDecodeError as e:
logger.error(f"Failed to decode JSON: {e}")
################################################################################
# High Level methods for API
################################################################################
@SafeSlot(popup_error=True)
def plot(
self,
x_name: str,
y_name: str,
z_name: str,
x_entry: None | str = None,
y_entry: None | str = None,
z_entry: None | str = None,
color_map: str | None = "magma",
label: str | None = None,
validate_bec: bool = True,
) -> ScatterCurve:
"""
Plot the data from the device signals.
Args:
x_name (str): The name of the x device signal.
y_name (str): The name of the y device signal.
z_name (str): The name of the z device signal.
x_entry (None | str): The x entry of the device signal.
y_entry (None | str): The y entry of the device signal.
z_entry (None | str): The z entry of the device signal.
color_map (str | None): The color map of the scatter waveform.
label (str | None): The label of the curve.
validate_bec (bool): Whether to validate the device signals with current BEC instance.
Returns:
ScatterCurve: The scatter curve object.
"""
if validate_bec:
x_entry = self.entry_validator.validate_signal(x_name, x_entry)
y_entry = self.entry_validator.validate_signal(y_name, y_entry)
z_entry = self.entry_validator.validate_signal(z_name, z_entry)
if color_map is not None:
try:
self.config.color_map = color_map
except ValidationError:
raise ValueError(
f"Invalid color map '{color_map}'. Using previously defined color map '{self.config.color_map}'."
)
if label is None:
label = f"{z_name}-{z_entry}"
config = ScatterCurveConfig(
parent_id=self.gui_id,
label=label,
color_map=color_map,
x_device=ScatterDeviceSignal(name=x_name, entry=x_entry),
y_device=ScatterDeviceSignal(name=y_name, entry=y_entry),
z_device=ScatterDeviceSignal(name=z_name, entry=z_entry),
)
# Add Curve
self._add_main_scatter_curve(config)
self.scatter_waveform_property_changed.emit()
return self._main_curve
def _add_main_scatter_curve(self, config: ScatterCurveConfig):
"""
Add the main scatter curve to the plot.
Args:
config(ScatterCurveConfig): The configuration of the scatter curve.
"""
# Apply suffix for axes
self.set_x_label_suffix(f"[{config.x_device.name}-{config.x_device.name}]")
self.set_y_label_suffix(f"[{config.y_device.name}-{config.y_device.name}]")
# To have only one main curve
if self._main_curve is not None:
self.plot_item.removeItem(self._main_curve)
self._main_curve = None
self._main_curve = ScatterCurve(
parent_item=self, config=config, gui_id=self.gui_id, name=config.label
)
self.plot_item.addItem(self._main_curve)
self.sync_signal_update.emit()
################################################################################
# BEC Update Methods
################################################################################
@SafeSlot(dict, dict)
def on_scan_status(self, msg: dict, meta: dict):
"""
Initial scan status message handler, which is triggered at the begging and end of scan.
Used for triggering the update of the sync and async curves.
Args:
msg(dict): The message content.
meta(dict): The message metadata.
"""
current_scan_id = msg.get("scan_id", None)
if current_scan_id is None:
return
if current_scan_id != self.scan_id:
self.reset()
self.new_scan.emit()
self.new_scan_id.emit(current_scan_id)
self.auto_range_x = True
self.auto_range_y = True
self.old_scan_id = self.scan_id
self.scan_id = current_scan_id
self.scan_item = self.queue.scan_storage.find_scan_by_ID(self.scan_id)
# First trigger to update the scan curves
self.sync_signal_update.emit()
@SafeSlot(dict, dict)
def on_scan_progress(self, msg: dict, meta: dict):
"""
Slot for handling scan progress messages. Used for triggering the update of the sync curves.
Args:
msg(dict): The message content.
meta(dict): The message metadata.
"""
self.sync_signal_update.emit()
@SafeSlot()
def update_sync_curves(self, _=None):
"""
Update the scan curves with the data from the scan segment.
"""
if self.scan_item is None:
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none"
data, access_key = self._fetch_scan_data_and_access()
if data == "none":
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none"
try:
x_name = self._main_curve.config.x_device.name
x_entry = self._main_curve.config.x_device.entry
y_name = self._main_curve.config.y_device.name
y_entry = self._main_curve.config.y_device.entry
z_name = self._main_curve.config.z_device.name
z_entry = self._main_curve.config.z_device.entry
except AttributeError:
return
if access_key == "val":
x_data = data.get(x_name, {}).get(x_entry, {}).get(access_key, None)
y_data = data.get(y_name, {}).get(y_entry, {}).get(access_key, None)
z_data = data.get(z_name, {}).get(z_entry, {}).get(access_key, None)
else:
x_data = data.get(x_name, {}).get(x_entry, {}).read().get("value", None)
y_data = data.get(y_name, {}).get(y_entry, {}).read().get("value", None)
z_data = data.get(z_name, {}).get(z_entry, {}).read().get("value", None)
self._main_curve.set_data(x=x_data, y=y_data, z=z_data)
def _fetch_scan_data_and_access(self):
"""
Decide whether the widget is in live or historical mode
and return the appropriate data dict and access key.
Returns:
data_dict (dict): The data structure for the current scan.
access_key (str): Either 'val' (live) or 'value' (history).
"""
if self.scan_item is None:
# Optionally fetch the latest from history if nothing is set
self.update_with_scan_history(-1)
if self.scan_item is None:
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none", "none"
if hasattr(self.scan_item, "live_data"):
# Live scan
return self.scan_item.live_data, "val"
else:
# Historical
scan_devices = self.scan_item.devices
return scan_devices, "value"
@SafeSlot(int)
@SafeSlot(str)
@SafeSlot()
def update_with_scan_history(self, scan_index: int = None, scan_id: str = None):
"""
Update the scan curves with the data from the scan storage.
Provide only one of scan_id or scan_index.
Args:
scan_id(str, optional): ScanID of the scan to be updated. Defaults to None.
scan_index(int, optional): Index of the scan to be updated. Defaults to None.
"""
if scan_index is not None and scan_id is not None:
raise ValueError("Only one of scan_id or scan_index can be provided.")
if scan_index is None and scan_id is None:
logger.warning(f"Neither scan_id or scan_number was provided, fetching the latest scan")
scan_index = -1
if scan_index is not None:
if len(self.client.history) == 0:
logger.info("No scans executed so far. Skipping scan history update.")
return
self.scan_item = self.client.history[scan_index]
metadata = self.scan_item.metadata
self.scan_id = metadata["bec"]["scan_id"]
else:
self.scan_id = scan_id
self.scan_item = self.client.history.get_by_scan_id(scan_id)
self.sync_signal_update.emit()
################################################################################
# Cleanup
################################################################################
@SafeSlot()
def clear_all(self):
"""
Clear all the curves from the plot.
"""
if self.crosshair is not None:
self.crosshair.clear_markers()
self._main_curve.clear()
class DemoApp(QMainWindow): # pragma: no cover
def __init__(self):
super().__init__()
self.setWindowTitle("Waveform Demo")
self.resize(800, 600)
self.main_widget = QWidget()
self.layout = QHBoxLayout(self.main_widget)
self.setCentralWidget(self.main_widget)
self.waveform_popup = ScatterWaveform(popups=True)
self.waveform_popup.plot("samx", "samy", "bpm4i")
self.waveform_side = ScatterWaveform(popups=False)
self.waveform_popup.plot("samx", "samy", "bpm3a")
self.layout.addWidget(self.waveform_side)
self.layout.addWidget(self.waveform_popup)
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
set_theme("dark")
widget = DemoApp()
widget.show()
widget.resize(1400, 600)
sys.exit(app.exec_())

1
bec_widgets/widgets/plots_next_gen/scatter_waveform/scatter_waveform.pyproject Normal file
View File

@ -0,0 +1 @@
{'files': ['scatter_waveform.py']}

54
bec_widgets/widgets/plots_next_gen/scatter_waveform/scatter_waveform_plugin.py Normal file
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.plots_next_gen.scatter_waveform.scatter_waveform import ScatterWaveform
DOM_XML = """
<ui language='c++'>
<widget class='ScatterWaveform' name='scatter_waveform'>
</widget>
</ui>
"""
class ScatterWaveformPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = ScatterWaveform(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "Plot Widgets Next Gen"
def icon(self):
return designer_material_icon(ScatterWaveform.ICON_NAME)
def includeFile(self):
return "scatter_waveform"
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 "ScatterWaveform"
def toolTip(self):
return "ScatterWaveform"
def whatsThis(self):
return self.toolTip()

0
bec_widgets/widgets/plots_next_gen/scatter_waveform/settings/__init__.py Normal file
View File

125
bec_widgets/widgets/plots_next_gen/scatter_waveform/settings/scatter_curve_setting.py Normal file
View File

@ -0,0 +1,125 @@
import os
from qtpy.QtWidgets import QFrame, QScrollArea, QVBoxLayout
from bec_widgets.qt_utils.error_popups import SafeSlot
from bec_widgets.qt_utils.settings_dialog import SettingWidget
from bec_widgets.utils import UILoader
class ScatterCurveSettings(SettingWidget):
def __init__(self, parent=None, target_widget=None, popup=False, *args, **kwargs):
super().__init__(parent=parent, *args, **kwargs)
# This is a settings widget that depends on the target widget
# and should mirror what is in the target widget.
# Saving settings for this widget could result in recursively setting the target widget.
self.setProperty("skip_settings", True)
self.setObjectName("ScatterCurveSettings")
current_path = os.path.dirname(__file__)
if popup:
form = UILoader().load_ui(
os.path.join(current_path, "scatter_curve_settings_horizontal.ui"), self
)
else:
form = UILoader().load_ui(
os.path.join(current_path, "scatter_curve_settings_vertical.ui"), self
)
self.target_widget = target_widget
self.popup = popup
# # Scroll area
self.scroll_area = QScrollArea(self)
self.scroll_area.setWidgetResizable(True)
self.scroll_area.setFrameShape(QFrame.NoFrame)
self.scroll_area.setWidget(form)
self.layout = QVBoxLayout(self)
self.layout.setContentsMargins(0, 0, 0, 0)
self.layout.addWidget(self.scroll_area)
self.ui = form
self.fetch_all_properties()
self.target_widget.scatter_waveform_property_changed.connect(self.fetch_all_properties)
if popup is False:
self.ui.button_apply.clicked.connect(self.accept_changes)
@SafeSlot()
def fetch_all_properties(self):
"""
Fetch all properties from the target widget and update the settings widget.
"""
if not self.target_widget:
return
# Get properties from the target widget
color_map = getattr(self.target_widget, "color_map", None)
# Default values for device properties
x_name, x_entry = None, None
y_name, y_entry = None, None
z_name, z_entry = None, None
# Safely access device properties
if hasattr(self.target_widget, "main_curve") and self.target_widget.main_curve:
if hasattr(self.target_widget.main_curve, "config"):
config = self.target_widget.main_curve.config
if hasattr(config, "x_device") and config.x_device:
x_name = getattr(config.x_device, "name", None)
x_entry = getattr(config.x_device, "entry", None)
if hasattr(config, "y_device") and config.y_device:
y_name = getattr(config.y_device, "name", None)
y_entry = getattr(config.y_device, "entry", None)
if hasattr(config, "z_device") and config.z_device:
z_name = getattr(config.z_device, "name", None)
z_entry = getattr(config.z_device, "entry", None)
# Apply the properties to the settings widget
if hasattr(self.ui, "color_map"):
self.ui.color_map.colormap = color_map
if hasattr(self.ui, "x_name"):
self.ui.x_name.set_device(x_name)
if hasattr(self.ui, "x_entry") and x_entry is not None:
self.ui.x_entry.setText(x_entry)
if hasattr(self.ui, "y_name"):
self.ui.y_name.set_device(y_name)
if hasattr(self.ui, "y_entry") and y_entry is not None:
self.ui.y_entry.setText(y_entry)
if hasattr(self.ui, "z_name"):
self.ui.z_name.set_device(z_name)
if hasattr(self.ui, "z_entry") and z_entry is not None:
self.ui.z_entry.setText(z_entry)
@SafeSlot()
def accept_changes(self):
"""
Apply all properties from the settings widget to the target widget.
"""
x_name = self.ui.x_name.text()
x_entry = self.ui.x_entry.text()
y_name = self.ui.y_name.text()
y_entry = self.ui.y_entry.text()
z_name = self.ui.z_name.text()
z_entry = self.ui.z_entry.text()
validate_bec = self.ui.validate_bec.checked
color_map = self.ui.color_map.colormap
self.target_widget.plot(
x_name=x_name,
y_name=y_name,
z_name=z_name,
x_entry=x_entry,
y_entry=y_entry,
z_entry=z_entry,
color_map=color_map,
validate_bec=validate_bec,
)

195
bec_widgets/widgets/plots_next_gen/scatter_waveform/settings/scatter_curve_settings_horizontal.ui Normal file
View File

@ -0,0 +1,195 @@
<?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>604</width>
<height>166</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<layout class="QHBoxLayout" name="horizontalLayout_2">
<item>
<widget class="QLabel" name="label_7">
<property name="text">
<string>Validate BEC</string>
</property>
</widget>
</item>
<item>
<widget class="ToggleSwitch" name="validate_bec"/>
</item>
<item>
<widget class="BECColorMapWidget" name="color_map"/>
</item>
</layout>
</item>
<item>
<layout class="QHBoxLayout" name="horizontalLayout">
<item>
<widget class="QGroupBox" name="groupBox">
<property name="title">
<string>X Device</string>
</property>
<layout class="QGridLayout" name="gridLayout">
<item row="0" column="0">
<widget class="QLabel" name="label">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="x_name"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_2">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="x_entry"/>
</item>
</layout>
</widget>
</item>
<item>
<widget class="QGroupBox" name="groupBox_2">
<property name="title">
<string>Y Device</string>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label_3">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="y_name"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_4">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="y_entry"/>
</item>
</layout>
</widget>
</item>
<item>
<widget class="QGroupBox" name="groupBox_3">
<property name="title">
<string>Z Device</string>
</property>
<layout class="QGridLayout" name="gridLayout_3">
<item row="0" column="0">
<widget class="QLabel" name="label_5">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_6">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="z_entry"/>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="z_name"/>
</item>
</layout>
</widget>
</item>
</layout>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>DeviceLineEdit</class>
<extends>QLineEdit</extends>
<header>device_line_edit</header>
</customwidget>
<customwidget>
<class>ToggleSwitch</class>
<extends>QWidget</extends>
<header>toggle_switch</header>
</customwidget>
<customwidget>
<class>BECColorMapWidget</class>
<extends>QWidget</extends>
<header>bec_color_map_widget</header>
</customwidget>
</customwidgets>
<resources/>
<connections>
<connection>
<sender>x_name</sender>
<signal>textChanged(QString)</signal>
<receiver>x_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>134</x>
<y>95</y>
</hint>
<hint type="destinationlabel">
<x>138</x>
<y>128</y>
</hint>
</hints>
</connection>
<connection>
<sender>y_name</sender>
<signal>textChanged(QString)</signal>
<receiver>y_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>351</x>
<y>91</y>
</hint>
<hint type="destinationlabel">
<x>349</x>
<y>121</y>
</hint>
</hints>
</connection>
<connection>
<sender>z_name</sender>
<signal>textChanged(QString)</signal>
<receiver>z_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>520</x>
<y>98</y>
</hint>
<hint type="destinationlabel">
<x>522</x>
<y>127</y>
</hint>
</hints>
</connection>
</connections>
</ui>

204
bec_widgets/widgets/plots_next_gen/scatter_waveform/settings/scatter_curve_settings_vertical.ui Normal file
View File

@ -0,0 +1,204 @@
<?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>233</width>
<height>427</height>
</rect>
</property>
<property name="maximumSize">
<size>
<width>16777215</width>
<height>427</height>
</size>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QPushButton" name="button_apply">
<property name="text">
<string>Apply</string>
</property>
</widget>
</item>
<item>
<widget class="BECColorMapWidget" name="color_map"/>
</item>
<item>
<layout class="QHBoxLayout" name="horizontalLayout">
<item>
<widget class="QLabel" name="label_7">
<property name="text">
<string>Validate BEC</string>
</property>
</widget>
</item>
<item>
<widget class="ToggleSwitch" name="validate_bec"/>
</item>
</layout>
</item>
<item>
<widget class="QGroupBox" name="groupBox">
<property name="title">
<string>X Device</string>
</property>
<layout class="QGridLayout" name="gridLayout">
<item row="0" column="0">
<widget class="QLabel" name="label">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="x_name"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_2">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="x_entry"/>
</item>
</layout>
</widget>
</item>
<item>
<widget class="QGroupBox" name="groupBox_2">
<property name="title">
<string>Y Device</string>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label_3">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="y_name"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_4">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="y_entry"/>
</item>
</layout>
</widget>
</item>
<item>
<widget class="QGroupBox" name="groupBox_3">
<property name="title">
<string>Z Device</string>
</property>
<layout class="QGridLayout" name="gridLayout_3">
<item row="0" column="0">
<widget class="QLabel" name="label_5">
<property name="text">
<string>Name</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="z_name"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_6">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="z_entry"/>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>DeviceLineEdit</class>
<extends>QLineEdit</extends>
<header>device_line_edit</header>
</customwidget>
<customwidget>
<class>ToggleSwitch</class>
<extends>QWidget</extends>
<header>toggle_switch</header>
</customwidget>
<customwidget>
<class>BECColorMapWidget</class>
<extends>QWidget</extends>
<header>bec_color_map_widget</header>
</customwidget>
</customwidgets>
<resources/>
<connections>
<connection>
<sender>x_name</sender>
<signal>textChanged(QString)</signal>
<receiver>x_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>156</x>
<y>123</y>
</hint>
<hint type="destinationlabel">
<x>158</x>
<y>157</y>
</hint>
</hints>
</connection>
<connection>
<sender>y_name</sender>
<signal>textChanged(QString)</signal>
<receiver>y_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>116</x>
<y>229</y>
</hint>
<hint type="destinationlabel">
<x>116</x>
<y>251</y>
</hint>
</hints>
</connection>
<connection>
<sender>z_name</sender>
<signal>textChanged(QString)</signal>
<receiver>z_entry</receiver>
<slot>clear()</slot>
<hints>
<hint type="sourcelabel">
<x>110</x>
<y>326</y>
</hint>
<hint type="destinationlabel">
<x>110</x>
<y>352</y>
</hint>
</hints>
</connection>
</connections>
</ui>

1
tests/unit_tests/client_mocks.py
View File

@ -219,6 +219,7 @@ def create_dummy_scan_item():
"""
dummy_live_data = {
"samx": {"samx": DummyData(val=[10, 20, 30], timestamps=[100, 200, 300])},
"samy": {"samy": DummyData(val=[5, 10, 15], timestamps=[100, 200, 300])},
"bpm4i": {"bpm4i": DummyData(val=[5, 6, 7], timestamps=[101, 201, 301])},
"async_device": {"async_device": DummyData(val=[1, 2, 3], timestamps=[11, 21, 31])},
}

18
tests/unit_tests/test_bec_dock.py
View File

@ -118,6 +118,15 @@ def test_toolbar_add_plot_waveform(bec_dock_area):
assert bec_dock_area.panels["Waveform_0"].widgets[0].config.widget_class == "Waveform"
def test_toolbar_add_plot_scatter_waveform(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["scatter_waveform"].trigger()
assert "ScatterWaveform_0" in bec_dock_area.panels
assert (
bec_dock_area.panels["ScatterWaveform_0"].widgets[0].config.widget_class
== "ScatterWaveform"
)
def test_toolbar_add_plot_image(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["image"].trigger()
assert "Image_0" in bec_dock_area.panels
@ -133,6 +142,15 @@ def test_toolbar_add_plot_motor_map(bec_dock_area):
)
def test_toolbar_add_multi_waveform(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_plots"].widgets["multi_waveform"].trigger()
assert "BECMultiWaveformWidget_0" in bec_dock_area.panels
assert (
bec_dock_area.panels["BECMultiWaveformWidget_0"].widgets[0].config.widget_class
== "BECMultiWaveformWidget"
)
def test_toolbar_add_device_positioner_box(bec_dock_area):
bec_dock_area.toolbar.widgets["menu_devices"].widgets["positioner_box"].trigger()
assert "PositionerBox_0" in bec_dock_area.panels

153
tests/unit_tests/test_scatter_waveform.py Normal file
View File

@ -0,0 +1,153 @@
import json
import numpy as np
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_curve import (
ScatterCurveConfig,
ScatterDeviceSignal,
)
from bec_widgets.widgets.plots_next_gen.scatter_waveform.scatter_waveform import ScatterWaveform
from tests.unit_tests.client_mocks import create_dummy_scan_item, mocked_client
from .conftest import create_widget
def test_waveform_initialization(qtbot, mocked_client):
"""
Test that a new Waveform widget initializes with the correct defaults.
"""
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
assert swf.objectName() == "ScatterWaveform"
# Inherited from PlotBase
assert swf.title == ""
assert swf.x_label == ""
assert swf.y_label == ""
# No crosshair or FPS monitor by default
assert swf.crosshair is None
assert swf.fps_monitor is None
assert swf.main_curve is not None
def test_scatter_waveform_plot(qtbot, mocked_client):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
curve = swf.plot("samx", "samy", "bpm4i")
assert curve is not None
assert isinstance(curve.config, ScatterCurveConfig)
assert curve.config.x_device == ScatterDeviceSignal(name="samx", entry="samx")
assert curve.config.label == "bpm4i-bpm4i"
def test_scatter_waveform_color_map(qtbot, mocked_client):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
assert swf.color_map == "magma"
swf.color_map = "plasma"
assert swf.color_map == "plasma"
def test_scatter_waveform_curve_json(qtbot, mocked_client):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
# Add a device-based scatter curve
swf.plot(x_name="samx", y_name="samy", z_name="bpm4i", label="test_curve")
json_str = swf.curve_json
data = json.loads(json_str)
assert isinstance(data, dict)
assert data["label"] == "test_curve"
assert data["x_device"]["name"] == "samx"
assert data["y_device"]["name"] == "samy"
assert data["z_device"]["name"] == "bpm4i"
# Clear and reload from JSON
swf.clear_all()
assert swf.main_curve.getData() == (None, None)
swf.curve_json = json_str
assert swf.main_curve.config.label == "test_curve"
def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeypatch):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
dummy_scan = create_dummy_scan_item()
mocked_client.history.get_by_scan_id.return_value = dummy_scan
mocked_client.history.__getitem__.return_value = dummy_scan
swf.plot("samx", "samy", "bpm4i", label="test_curve")
swf.update_with_scan_history(scan_id="dummy")
qtbot.wait(500)
assert swf.scan_item == dummy_scan
x_data, y_data = swf.main_curve.getData()
np.testing.assert_array_equal(x_data, [10, 20, 30])
np.testing.assert_array_equal(y_data, [5, 10, 15])
def test_scatter_waveform_live_update(qtbot, mocked_client, monkeypatch):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
dummy_scan = create_dummy_scan_item()
monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan)
swf.plot("samx", "samy", "bpm4i", label="live_curve")
# Simulate scan status indicating new scan start
msg = {"scan_id": "dummy"}
meta = {}
swf.on_scan_status(msg, meta)
assert swf.scan_id == "dummy"
assert swf.scan_item == dummy_scan
qtbot.wait(500)
x_data, y_data = swf.main_curve.getData()
np.testing.assert_array_equal(x_data, [10, 20, 30])
np.testing.assert_array_equal(y_data, [5, 10, 15])
def test_scatter_waveform_scan_progress(qtbot, mocked_client, monkeypatch):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
dummy_scan = create_dummy_scan_item()
monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan)
swf.plot("samx", "samy", "bpm4i")
# Simulate scan status indicating scan progress
swf.scan_id = "dummy"
swf.scan_item = dummy_scan
msg = {"progress": 50}
meta = {}
swf.on_scan_progress(msg, meta)
qtbot.wait(500)
# swf.update_sync_curves()
x_data, y_data = swf.main_curve.getData()
np.testing.assert_array_equal(x_data, [10, 20, 30])
np.testing.assert_array_equal(y_data, [5, 10, 15])
def test_scatter_waveform_settings_popup(qtbot, mocked_client):
"""
Test that the settings popup is created correctly.
"""
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
scatter_popup_action = swf.toolbar.widgets["scatter_waveform_settings"].action
assert not scatter_popup_action.isChecked(), "Should start unchecked"
swf.show_scatter_curve_settings()
assert swf.scatter_dialog is not None
assert swf.scatter_dialog.isVisible()
assert scatter_popup_action.isChecked()
swf.scatter_dialog.close()
assert swf.scatter_dialog is None
assert not scatter_popup_action.isChecked(), "Should be unchecked after closing dialog"

4
tests/unit_tests/test_waveform_next_gen.py
View File

@ -10,11 +10,11 @@ from bec_widgets.widgets.plots_next_gen.plot_base import UIMode
from bec_widgets.widgets.plots_next_gen.waveform.curve import DeviceSignal
from bec_widgets.widgets.plots_next_gen.waveform.waveform import Waveform
from tests.unit_tests.client_mocks import (
DummyData,
create_dummy_scan_item,
dap_plugin_message,
mocked_client,
mocked_client_with_dap,
create_dummy_scan_item,
DummyData,
)
from .conftest import create_widget