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

fix(waveform): x axis switching logic fixed when axis are not compatible

Browse Source
This commit is contained in:
wyzula_j
2024-07-11 17:12:50 +02:00
parent fc935d9fc8
commit e4e1a905d1
2 changed files with 65 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
bec_widgets/cli/client.py
View File

@ -1514,6 +1514,7 @@ class BECWaveform(RPCBase):
) -> "BECCurve":
"""
Plot a curve to the plot widget.
Args:
x(list | np.ndarray): Custom x data to plot.
y(list | np.ndarray): Custom y data to plot.
@ -1527,7 +1528,7 @@ class BECWaveform(RPCBase):
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 not specified, none will be added.
dap(str): The dap model to use for the curve, only available for sync devices. If not specified, none will be added.
Returns:
BECCurve: The curve object.
@ -1536,12 +1537,13 @@ class BECWaveform(RPCBase):
@rpc_call
def add_dap(
self,
x_name: "str",
y_name: "str",
x_name: "str | None" = None,
y_name: "str | None" = None,
x_entry: "Optional[str]" = None,
y_entry: "Optional[str]" = None,
color: "Optional[str]" = None,
dap: "str" = "GaussianModel",
validate_bec: "bool" = True,
**kwargs,
) -> "BECCurve":
"""
@ -1556,12 +1558,23 @@ class BECWaveform(RPCBase):
color_map_z(str): The color map to use for the z-axis.
label(str, optional): Label of the curve. Defaults to None.
dap(str): The dap model to use for the curve.
validate_bec(bool, optional): If True, validate the signal with BEC. Defaults to True.
**kwargs: Additional keyword arguments for the curve configuration.
Returns:
BECCurve: The curve object.
"""
@rpc_call
def change_x_axis(self, x_name: "str", x_entry: "str | None" = None):
"""
Change the x axis of the plot widget.
Args:
x_name(str): Name of the x signal.
x_entry(str): Entry of the x signal.
"""
@rpc_call
def get_dap_params(self) -> "dict":
"""

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

@ -89,7 +89,12 @@ class BECWaveform(BECPlotBase):
self.old_scan_id = None
self.scan_id = None
self.scan_item = None
self._x_axis_mode = {"name": None, "entry": None, "readout_priority": None}
self._x_axis_mode = {
"name": None,
"entry": None,
"readout_priority": None,
"label_suffix": "",
}
# Scan segment update proxy
self.proxy_update_plot = pg.SignalProxy(
@ -105,7 +110,8 @@ class BECWaveform(BECPlotBase):
# Connect dispatcher signals
self.bec_dispatcher.connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())
self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
# TODO disabled -> scan_status is SET_AND_PUBLISH -> do not work in combination with autoupdate from CLI
# self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
self.entry_validator = EntryValidator(self.dev)
@ -330,8 +336,15 @@ class BECWaveform(BECPlotBase):
}
if len(self.curves) > 0:
# validate all curves
for curve in self.curves:
self._validate_x_axis_behaviour(curve.config.signals.y.name, x_name, x_entry, False)
self._switch_x_axis_item(
f"{x_name}-{x_entry}"
if x_name not in ["best_effort", "timestamp", "index"]
else x_name
)
for curve_id, curve_config in zip(curve_ids, curve_configs):
self._validate_x_axis_behaviour(curve_config.signals.y.name, x_name, x_entry)
if curve_config.signals.x:
curve_config.signals.x.name = x_name
curve_config.signals.x.entry = x_entry
@ -610,7 +623,7 @@ class BECWaveform(BECPlotBase):
return source
def _validate_x_axis_behaviour(
self, y_name: str, x_name: str | None = None, x_entry: str | None = None
self, y_name: str, x_name: str | None = None, x_entry: str | None = None, auto_switch=True
) -> None:
"""
Validate the x axis behaviour and consistency for the plot item.
@ -636,7 +649,7 @@ class BECWaveform(BECPlotBase):
f"All curves must have the same x axis.\n"
f" Current valid x axis: '{self._x_axis_mode['name']}'\n"
f" Attempted to add curve with x axis: '{x_name}'\n"
f"If you want to change the x-axis of the curve, please remove previous curves."
f"If you want to change the x-axis of the curve, please remove previous curves or change the x axis of the plot widget with '.change_x_axis({x_name})'."
)
# If x_axis_mode["name"] is None, determine the mode based on x_name
@ -656,10 +669,13 @@ class BECWaveform(BECPlotBase):
f"Async devices '{y_name}' cannot be used with custom x signal '{x_name}-{x_entry}'."
)
# Switch the x axis mode accordingly
self._switch_x_axis_item(
f"{x_name}-{x_entry}" if x_name not in ["best_effort", "timestamp", "index"] else x_name
)
if auto_switch is True:
# Switch the x axis mode accordingly
self._switch_x_axis_item(
f"{x_name}-{x_entry}"
if x_name not in ["best_effort", "timestamp", "index"]
else x_name
)
def _get_device_readout_priority(self, name: str):
"""
@ -688,16 +704,17 @@ class BECWaveform(BECPlotBase):
current_label = "" if self.config.axis.x_label is None else self.config.axis.x_label
date_axis = pg.graphicsItems.DateAxisItem.DateAxisItem(orientation="bottom")
default_axis = pg.AxisItem(orientation="bottom")
self._x_axis_mode["label_suffix"] = f" [{mode}]"
if mode == "timestamp":
self.plot_item.setAxisItems({"bottom": date_axis})
self.plot_item.setLabel("bottom", f"{current_label} [timestamp]")
self.plot_item.setLabel("bottom", f"{current_label}{self._x_axis_mode['label_suffix']}")
elif mode == "index":
self.plot_item.setAxisItems({"bottom": default_axis})
self.plot_item.setLabel("bottom", f"{current_label} [index]")
self.plot_item.setLabel("bottom", f"{current_label}{self._x_axis_mode['label_suffix']}")
else:
self.plot_item.setAxisItems({"bottom": default_axis})
self.plot_item.setLabel("bottom", f"{current_label} [{mode}]")
self.plot_item.setLabel("bottom", f"{current_label}{self._x_axis_mode['label_suffix']}")
def _validate_signal_entries(
self,
@ -848,9 +865,22 @@ class BECWaveform(BECPlotBase):
msg (dict): Message received with scan data.
metadata (dict): Metadata of the scan.
"""
self.on_scan_status(msg)
self.scan_signal_update.emit()
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.
"""
super().set_x_label(label, size)
current_label = "" if self.config.axis.x_label is None else self.config.axis.x_label
self.plot_item.setLabel("bottom", f"{current_label}{self._x_axis_mode['label_suffix']}")
def setup_dap(self, old_scan_id: str | None, new_scan_id: str | None):
"""
Setup DAP for the new scan.
@ -1082,10 +1112,13 @@ class BECWaveform(BECPlotBase):
except TypeError:
x_data = []
else:
x_name = self.scan_item.status_message.info["scan_report_devices"][0]
x_entry = self.entry_validator.validate_signal(x_name, None)
x_data = self.scan_item.data[x_name][x_entry].val
self.set_x_label(f"[auto: {x_name}-{x_entry}]")
if len(self._curves_data["async"]) > 0:
x_data = None
else:
x_name = self.scan_item.status_message.info["scan_report_devices"][0]
x_entry = self.entry_validator.validate_signal(x_name, None)
x_data = self.scan_item.data[x_name][x_entry].val
self.set_x_label(f"[auto: {x_name}-{x_entry}]")
return x_data