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

refactor(waveform): plot can be prompted without specifying kwargs

Browse Source
This commit is contained in:
wyzula_j
2024-07-11 22:09:46 +02:00
parent e4e1a905d1
commit 48911e9348
4 changed files with 97 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
bec_widgets/cli/client.py
View File

@ -468,8 +468,9 @@ class BECFigure(RPCBase):
@rpc_call
def plot(
self,
x: "list | np.ndarray | None" = None,
arg1: "list | np.ndarray | str | None" = None,
y: "list | np.ndarray | None" = None,
x: "list | np.ndarray | None" = None,
x_name: "str | None" = None,
y_name: "str | None" = None,
z_name: "str | None" = None,
@ -491,8 +492,9 @@ class BECFigure(RPCBase):
Add a 1D waveform plot to the figure. Always access the first waveform widget in the figure.
Args:
x(list | np.ndarray): Custom x data to plot.
arg1(list | np.ndarray | str | None): First argument which can be x data, y data, or y_name.
y(list | np.ndarray): Custom y data to plot.
x(list | np.ndarray): Custom x data to plot.
x_name(str): The name of the device for the x-axis.
y_name(str): The name of the device for the y-axis.
z_name(str): The name of the device for the z-axis.
@ -1497,8 +1499,9 @@ class BECWaveform(RPCBase):
@rpc_call
def plot(
self,
x: "list | np.ndarray | None" = None,
arg1: "list | np.ndarray | str | None" = None,
y: "list | np.ndarray | None" = None,
x: "list | np.ndarray | None" = None,
x_name: "str | None" = None,
y_name: "str | None" = None,
z_name: "str | None" = None,
@ -1516,8 +1519,9 @@ class BECWaveform(RPCBase):
Plot a curve to the plot widget.
Args:
x(list | np.ndarray): Custom x data to plot.
arg1(list | np.ndarray | str | None): First argument which can be x data, y data, or y_name.
y(list | np.ndarray): Custom y data to plot.
x(list | np.ndarray): Custom y data to plot.
x_name(str): The name of the device for the x-axis.
y_name(str): The name of the device for the y-axis.
z_name(str): The name of the device for the z-axis.
@ -1759,6 +1763,12 @@ class BECWaveform(RPCBase):
Remove the plot widget from the figure.
"""
@rpc_call
def clear_all(self):
"""
None
"""
@rpc_call
def set_legend_label_size(self, size: "int" = None):
"""

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

@ -4,8 +4,15 @@ import numpy as np
import pyqtgraph as pg
from qtpy.QtCore import QSize
from qtpy.QtGui import QIcon
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
from qtpy.QtWidgets import QHBoxLayout, QSplitter, QTabWidget, QGroupBox
from qtpy.QtWidgets import (
QApplication,
QGroupBox,
QHBoxLayout,
QSplitter,
QTabWidget,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils import BECDispatcher
from bec_widgets.utils.colors import apply_theme

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

@ -227,88 +227,12 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
"""
self._widgets = value
def _init_waveform(
self,
waveform,
x_name: str = None,
y_name: str = None,
z_name: str = None,
x_entry: str = None,
y_entry: str = None,
z_entry: str = None,
x: list | np.ndarray = None,
y: list | np.ndarray = None,
color: str | None = None,
color_map_z: str | None = "plasma",
label: str | None = None,
validate: bool = True,
dap: str | None = None,
) -> BECWaveform:
"""
Configure the waveform based on the provided parameters.
Args:
waveform (BECWaveform): The waveform to configure.
x (list | np.ndarray): Custom x data to plot.
y (list | np.ndarray): Custom y data to plot.
x_name (str): The name of the device for the x-axis.
y_name (str): The name of the device for the y-axis.
z_name (str): The name of the device for the z-axis.
x_entry (str): The name of the entry for the x-axis.
y_entry (str): The name of the entry for the y-axis.
z_entry (str): The name of the entry for the z-axis.
color (str): The color of the curve.
color_map_z (str): The color map to use for the z-axis.
label (str): The label of the curve.
validate (bool): If True, validate the device names and entries.
dap (str): The DAP model to use for the curve.
"""
if x is not None and y is None:
if isinstance(x, np.ndarray):
if x.ndim == 1:
y = np.arange(x.size)
waveform.add_curve_custom(x=np.arange(x.size), y=x, color=color, label=label)
return waveform
if x.ndim == 2:
waveform.add_curve_custom(x=x[:, 0], y=x[:, 1], color=color, label=label)
return waveform
elif isinstance(x, list):
y = np.arange(len(x))
waveform.add_curve_custom(x=np.arange(len(x)), y=x, color=color, label=label)
return waveform
else:
raise ValueError(
"Invalid input. Provide either device names (x_name, y_name) or custom data."
)
if x is not None and y is not None:
waveform.add_curve_custom(x=x, y=y, color=color, label=label)
return waveform
if y_name is not None:
waveform.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=color,
color_map_z=color_map_z,
label=label,
validate=validate,
dap=dap,
)
# User wants to add custom curve
elif x is not None and y is not None and x_name is None and y_name is None:
waveform.add_curve_custom(x=x, y=y, color=color, label=label)
return waveform
@typechecked
def plot(
self,
x: list | np.ndarray | None = None,
arg1: list | np.ndarray | str | None = None,
y: list | np.ndarray | None = None,
x: list | np.ndarray | None = None,
x_name: str | None = None,
y_name: str | None = None,
z_name: str | None = None,
@ -330,8 +254,9 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Add a 1D waveform plot to the figure. Always access the first waveform widget in the figure.
Args:
x(list | np.ndarray): Custom x data to plot.
arg1(list | np.ndarray | str | None): First argument which can be x data, y data, or y_name.
y(list | np.ndarray): Custom y data to plot.
x(list | np.ndarray): Custom x data to plot.
x_name(str): The name of the device for the x-axis.
y_name(str): The name of the device for the y-axis.
z_name(str): The name of the device for the z-axis.
@ -358,23 +283,23 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
if config is not None:
return waveform
# Passing args to init_waveform
waveform = self._init_waveform(
waveform=waveform,
x=x,
y=y,
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=color,
color_map_z=color_map_z,
label=label,
validate=validate,
dap=dap,
)
if arg1 is not None or y_name is not None or (y is not None and x is not None):
waveform.plot(
arg1=arg1,
y=y,
x=x,
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=color,
color_map_z=color_map_z,
label=label,
validate=validate,
dap=dap,
)
return waveform
def _init_image(

77
bec_widgets/widgets/figure/plots/waveform/waveform.py
View File

@ -65,6 +65,7 @@ class BECWaveform(BECPlotBase):
"set_grid",
"lock_aspect_ratio",
"remove",
"clear_all",
"set_legend_label_size",
]
scan_signal_update = pyqtSignal()
@ -250,8 +251,9 @@ class BECWaveform(BECPlotBase):
def plot(
self,
x: list | np.ndarray | None = None,
arg1: list | np.ndarray | str | None = None,
y: list | np.ndarray | None = None,
x: list | np.ndarray | None = None,
x_name: str | None = None,
y_name: str | None = None,
z_name: str | None = None,
@ -269,8 +271,9 @@ class BECWaveform(BECPlotBase):
Plot a curve to the plot widget.
Args:
x(list | np.ndarray): Custom x data to plot.
arg1(list | np.ndarray | str | None): First argument which can be x data, y data, or y_name.
y(list | np.ndarray): Custom y data to plot.
x(list | np.ndarray): Custom y data to plot.
x_name(str): The name of the device for the x-axis.
y_name(str): The name of the device for the y-axis.
z_name(str): The name of the device for the z-axis.
@ -286,28 +289,45 @@ class BECWaveform(BECPlotBase):
Returns:
BECCurve: The curve object.
"""
if x is not None and y is not None:
return self.add_curve_custom(x=x, y=y, label=label, color=color, **kwargs)
else:
if dap:
self.add_dap(x_name=x_name, y_name=y_name, dap=dap)
curve = self.add_curve_bec(
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=color,
color_map_z=color_map_z,
label=label,
validate_bec=validate,
**kwargs,
)
self.scan_signal_update.emit()
self.async_signal_update.emit()
return curve
if isinstance(arg1, str):
y_name = arg1
elif isinstance(arg1, list):
if isinstance(y, list):
return self.add_curve_custom(x=arg1, y=y, label=label, color=color, **kwargs)
if y is None:
x = np.arange(len(arg1))
return self.add_curve_custom(x=x, y=arg1, label=label, color=color, **kwargs)
elif isinstance(arg1, np.ndarray) and y is None:
if arg1.ndim == 1:
x = np.arange(arg1.size)
return self.add_curve_custom(x=x, y=arg1, label=label, color=color, **kwargs)
if arg1.ndim == 2:
x = arg1[:, 0]
y = arg1[:, 1]
return self.add_curve_custom(x=x, y=y, label=label, color=color, **kwargs)
if y_name is None:
raise ValueError("y_name must be provided.")
if dap:
self.add_dap(x_name=x_name, y_name=y_name, dap=dap)
curve = self.add_curve_bec(
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=color,
color_map_z=color_map_z,
label=label,
validate_bec=validate,
**kwargs,
)
self.scan_signal_update.emit()
self.async_signal_update.emit()
return curve
def change_x_axis(self, x_name: str, x_entry: str | None = None):
"""
@ -666,7 +686,9 @@ class BECWaveform(BECPlotBase):
self._x_axis_mode["entry"] = x_entry
if readout_priority_y == "async":
raise ValueError(
f"Async devices '{y_name}' cannot be used with custom x signal '{x_name}-{x_entry}'."
f"Async devices '{y_name}' cannot be used with custom x signal '{x_name}-{x_entry}'.\n"
f"Please use mode 'best_effort', 'timestamp', or 'index' signal for x axis."
f"You can change the x axis mode with '.change_x_axis(mode)'"
)
if auto_switch is True:
@ -984,7 +1006,6 @@ class BECWaveform(BECPlotBase):
msg(dict): Message with the async data.
metadata(dict): Metadata of the message.
"""
print(msg)
instruction = metadata.get("async_update")
for curve_id, curve in self._curves_data["async"].items():
y_name = curve.config.signals.y.name
@ -1234,6 +1255,14 @@ class BECWaveform(BECPlotBase):
]
return data2float
def clear_all(self):
curves_data = self._curves_data
sources = list(curves_data.keys())
for source in sources:
curve_ids_to_remove = list(curves_data[source].keys())
for curve_id in curve_ids_to_remove:
self.remove_curve(curve_id)
def cleanup(self):
"""Cleanup the widget connection from BECDispatcher."""
self.bec_dispatcher.disconnect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())