bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-09-13 03:32:39 +02:00
Code Activity

fix(signal_info_label): various fixes to implementation of signal info in plotting widgets

Browse Source
This commit is contained in:
wyzula_j
2026-09-07 15:11:42 +02:00
committed by Jan Wyzula
parent bc5dca2915
commit 870eb910fa
15 changed files with 438 additions and 239 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/cli/client.py
+84
View File
@@ -1832,6 +1832,20 @@ class Heatmap(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
@@ -2540,6 +2554,20 @@ class Image(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
@@ -3713,6 +3741,20 @@ class MotorMap(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
@@ -4185,6 +4227,20 @@ class MultiWaveform(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
@@ -5446,6 +5502,20 @@ class ScatterWaveform(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
@@ -6097,6 +6167,20 @@ class Waveform(RPCBase):
Enable the FPS monitor.
"""
@property
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@show_info_label.setter
@rpc_call
def show_info_label(self) -> "bool":
"""
Whether the plot info label is shown.
"""
@rpc_call
def set(self, **kwargs):
"""
bec_widgets/widgets/plots/heatmap/heatmap.py
+4 -40
View File
@@ -276,7 +276,6 @@ class Heatmap(ImageBase):
# Scan ID the widget is pinned to when plotting from history; None means live mode.
self._history_scan_id: str | None = None
self._selected_history_scan_id: str | None = None
self.config_label = self.info_label
self.show_info_label = self._image_config.show_config_label
self.reload = False
self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
@@ -309,15 +308,6 @@ class Heatmap(ImageBase):
# Widget Specific GUI interactions
################################################################################
@SafeSlot(str)
def apply_theme(self, theme: str):
"""
Apply the current theme to the heatmap widget.
"""
super().apply_theme(theme)
if hasattr(self, "info_label"):
self.redraw_config_label()
@SafeSlot(popup_error=True)
def plot(
self,
@@ -380,7 +370,7 @@ class Heatmap(ImageBase):
lock_aspect_ratio = self._image_config.lock_aspect_ratio
if show_config_label is None:
show_config_label = self._image_config.show_config_label
show_config_label = self.show_info_label
def _device_key(device: HeatmapDeviceSignal | None) -> tuple[str | None, str | None]:
return (device.device if device else None, device.signal if device else None)
@@ -409,6 +399,7 @@ class Heatmap(ImageBase):
lock_aspect_ratio=lock_aspect_ratio,
show_config_label=show_config_label,
)
self.show_info_label = show_config_label
self.color_map = color_map
self.reload = reload or config_changed
if config_changed:
@@ -637,28 +628,6 @@ class Heatmap(ImageBase):
self._selected_history_scan_id = None
self.toolbar.components.get_action("scan_history").action.setChecked(False)
@SafeProperty(bool, auto_emit=True)
def show_config_label(self) -> bool:
"""Whether to show heatmap configuration rows in the plot info label."""
return self._image_config.show_config_label
@show_config_label.setter
def show_config_label(self, value: bool) -> None:
self._image_config.show_config_label = bool(value)
self.show_info_label = bool(value)
self.redraw_config_label()
@SafeSlot()
def toggle_info_label(self):
"""Toggle the heatmap information label and keep heatmap config in sync."""
self.show_config_label = not self.show_config_label
def toggle_interpolation_info(self):
"""
Toggle the visibility of the heatmap information label.
"""
self.toggle_info_label()
def _heatmap_dialog_closed(self):
"""
Slot for when the heatmap settings dialog is closed.
@@ -785,9 +754,7 @@ class Heatmap(ImageBase):
logger.warning("Scan message is None; skipping update.")
return
self.status_message = scan_msg
if self._image_config.show_config_label:
self.redraw_config_label()
self.redraw_config_label()
if self._is_grid_scan_supported(scan_msg):
img, transform = self.get_grid_scan_image(z_data, scan_msg)
@@ -951,12 +918,9 @@ class Heatmap(ImageBase):
self._latest_interpolation_version = -1
def redraw_config_label(self):
scan_msg = getattr(self, "status_message", None)
scan_msg = self.status_message
if scan_msg is None:
return
if not self._image_config.show_config_label:
self.show_info_label = False
return
# Indicate whether the widget follows the live acquisition or is pinned to a history scan
mode = "history" if self._history_scan_id is not None else "live"
bec_widgets/widgets/plots/image/image.py
+1
View File
@@ -159,6 +159,7 @@ class Image(ImageBase):
crosshair_bundle = self.toolbar.get_bundle("image_crosshair")
crosshair_bundle.add_action("image_autorange")
crosshair_bundle.add_action("image_colorbar_switch")
self.remove_info_label_action()
self.toolbar.show_bundles(
[
bec_widgets/widgets/plots/motor_map/motor_map.py
+1 -1
View File
@@ -167,7 +167,7 @@ class MotorMap(PlotBase):
)
self.toolbar.components.get_action("reset_legend").action.setVisible(False)
self.toolbar.get_bundle("axis_popup").remove_action("plot_info_label")
self.remove_info_label_action()
reset_legend = MaterialIconAction(
icon_name="history",
bec_widgets/widgets/plots/multi_waveform/multi_waveform.py
+4 -2
View File
@@ -361,8 +361,10 @@ class MultiWaveform(PlotBase):
current_scan_id = metadata.get("scan_id", None)
if current_scan_id != self.scan_id:
scan_item = self.update_scan_info_from_scan_id(current_scan_id, mode="live")
self.scan_id = scan_item.scan_id if scan_item is not None else current_scan_id
self.scan_id = current_scan_id
self.update_scan_info_from_source(
self.queue.scan_storage.find_scan_by_ID(current_scan_id)
)
self.clear_curves()
self.curves.clear()
if self.crosshair:
bec_widgets/widgets/plots/plot_base.py
+64 -84
View File
@@ -1,11 +1,12 @@
from __future__ import annotations
from collections.abc import Iterable, Mapping
from enum import Enum
from typing import TypeAlias
from typing import Literal, TypeAlias
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger, messages
from bec_lib import bec_logger
from bec_lib.scan_data_container import ScanDataContainer
from bec_lib.scan_items import ScanItem
from qtpy.QtCore import QPoint, QPointF, Qt, Signal
@@ -44,6 +45,7 @@ from bec_widgets.widgets.plots.toolbar_components.roi import RoiConnection, roi_
logger = bec_logger.logger
ScanInfoSource: TypeAlias = ScanItem | ScanDataContainer | None
ScanInfoMode: TypeAlias = Literal["live", "history"]
class BECViewBox(pg.ViewBox):
@@ -79,6 +81,8 @@ class PlotBase(BECWidget, QWidget):
"enable_side_panel.setter",
"enable_fps_monitor",
"enable_fps_monitor.setter",
"show_info_label",
"show_info_label.setter",
"set",
"title",
"title.setter",
@@ -171,6 +175,8 @@ class PlotBase(BECWidget, QWidget):
self.plot_item.addLegend()
self.info_label = PlotInfoLabel()
self.info_label.setParentItem(self.plot_item.vb)
self._scan_info_rows: list[tuple[str, object | None]] = []
self._custom_info_rows: list[tuple[str, object | None]] = []
self.crosshair = None
# Holds a pin that outlived its crosshair (e.g. crosshair toggled off) so it
# can be re-adopted when the crosshair is hooked again.
@@ -290,7 +296,7 @@ class PlotBase(BECWidget, QWidget):
@SafeProperty(bool, auto_emit=True, doc="Show the plot info label.")
def show_info_label(self) -> bool:
"""Return whether the plot info label is enabled."""
"""Whether the plot info label is shown."""
return self._show_info_label
@show_info_label.setter
@@ -309,111 +315,85 @@ class PlotBase(BECWidget, QWidget):
self.show_info_label = not self.show_info_label
def set_info_label_rows(
self, rows: dict[str, object | None] | list[tuple[str, object | None]]
self, rows: Mapping[str, object | None] | Iterable[tuple[str, object | None]]
) -> None:
"""Submit arbitrary rows for the plot info label."""
self.info_label.set_rows(rows)
self._sync_info_label_visibility()
self._sync_info_label_action()
"""Show caller-defined rows in the info label, below the scan rows.
The rows replace previously submitted custom rows and survive scan updates.
"""
self._custom_info_rows = list(rows.items() if isinstance(rows, Mapping) else rows)
self._render_info_label()
def clear_info_label(self) -> None:
"""Clear all rows from the plot info label."""
self.info_label.clear_rows()
"""Clear the scan rows and the custom rows of the info label."""
self._scan_info_rows = []
self._custom_info_rows = []
self._render_info_label()
def _render_info_label(self) -> None:
self.info_label.set_rows([*self._scan_info_rows, *self._custom_info_rows])
self._sync_info_label_visibility()
self._sync_info_label_action()
def set_scan_info(
self,
*,
scan_id: str | None = None,
scan_number: int | str | None = None,
scan_name: str | None = None,
mode: str | None = None,
extra_rows: dict[str, object | None] | list[tuple[str, object | None]] | None = None,
mode: ScanInfoMode | None = None,
extra_rows: Iterable[tuple[str, object | None]] | None = None,
) -> None:
"""Submit common scan metadata rows to the plot info label."""
"""Submit the common scan rows (scan number with mode, scan name) to the info label."""
rows: list[tuple[str, object | None]] = []
if scan_number is not None:
scan_value = f"{scan_number} ({mode})" if mode else scan_number
rows.append(("Scan", scan_value))
elif scan_id is not None and mode != "history":
scan_value = f"{scan_id} ({mode})" if mode else scan_id
rows.append(("Scan ID", scan_value))
rows.append(("Scan", f"{scan_number} ({mode})" if mode else scan_number))
if scan_name is not None:
rows.append(("Scan Name", scan_name))
if extra_rows:
rows.extend(extra_rows.items() if isinstance(extra_rows, dict) else extra_rows)
self.set_info_label_rows(rows)
rows.extend(extra_rows)
self._scan_info_rows = rows
self._render_info_label()
def update_scan_info_from_source(
self, scan_source: ScanInfoSource, *, mode: str | None = None
) -> None:
"""Extract common scan metadata from a scan item and submit it."""
scan_info = self._extract_scan_info(scan_source)
if scan_info is None:
return
scan_id, scan_number, scan_name = scan_info
self.set_scan_info(scan_id=scan_id, scan_number=scan_number, scan_name=scan_name, mode=mode)
def update_scan_info_from_source(self, scan_source: ScanInfoSource) -> None:
"""
Show the scan rows of a live ScanItem or a history ScanDataContainer.
def update_scan_info_from_scan_id(
self, scan_id: str | None, *, mode: str | None = None
) -> ScanItem | None:
"""Fetch a live scan item by ID and submit common scan metadata."""
if scan_id is None:
return None
scan_item = self.queue.scan_storage.find_scan_by_ID(scan_id)
if not isinstance(scan_item, ScanItem):
return None
self.update_scan_info_from_source(scan_item, mode=mode)
return scan_item
Args:
scan_source(ScanInfoSource): The source of the scan information, which can be a ScanItem, ScanDataContainer, or None.
"""
if isinstance(scan_source, ScanItem) and scan_source.status_message is not None:
status_message = scan_source.status_message
self.set_scan_info(
scan_number=status_message.scan_number,
scan_name=status_message.scan_name,
mode="live",
)
elif isinstance(scan_source, ScanDataContainer):
metadata = scan_source.metadata.get("bec", {})
self.set_scan_info(
scan_number=metadata.get("scan_number"),
scan_name=metadata.get("scan_name"),
mode="history",
)
else:
self.set_scan_info()
def _extract_scan_info(
self, scan_source: ScanInfoSource
) -> tuple[str | None, int | str | None, str | None] | None:
if scan_source is None:
return None
if isinstance(scan_source, ScanItem):
return self._scan_info_from_status_message(scan_source.status_message)
if isinstance(scan_source, ScanDataContainer):
metadata = scan_source.metadata
scan_info = self._scan_info_from_metadata(metadata.get("bec", metadata))
if scan_info is not None:
return scan_info
return None
@staticmethod
def _scan_info_from_status_message(
status_message: messages.ScanStatusMessage | None,
) -> tuple[str | None, int | str | None, str | None] | None:
if status_message is None:
return None
scan_info = (status_message.scan_id, status_message.scan_number, status_message.scan_name)
return scan_info if any(value is not None for value in scan_info) else None
@staticmethod
def _scan_info_from_metadata(
metadata: dict | None,
) -> tuple[str | None, int | str | None, str | None] | None:
if not isinstance(metadata, dict):
return None
scan_id = metadata.get("scan_id")
scan_number = metadata.get("scan_number")
scan_name = metadata.get("scan_name")
scan_info = (scan_id, scan_number, scan_name)
return scan_info if any(value is not None for value in scan_info) else None
def remove_info_label_action(self) -> None:
"""Remove the info label toolbar button, for widgets that never populate the label."""
component = self.toolbar.components.get_action("plot_info_label")
component.action.triggered.disconnect(self.toggle_info_label)
self.toolbar.components.remove_action("plot_info_label")
component.action.deleteLater()
def _sync_info_label_visibility(self) -> None:
self.info_label.setVisible(self._show_info_label and bool(self.info_label.rows))
def _sync_info_label_action(self) -> None:
try:
action = self.toolbar.components.get_action("plot_info_label").action
except KeyError:
return
action.setEnabled(True)
action.setChecked(self._show_info_label)
if not self.toolbar.components.exists("plot_info_label"):
return # the widget removed the button with remove_info_label_action()
self.toolbar.components.get_action("plot_info_label").action.setChecked(
self._show_info_label
)
################################################################################
# Toggle UI Elements
bec_widgets/widgets/plots/plot_info_label.py
+22 -42
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from collections.abc import Iterable, Mapping
from collections.abc import Iterable
import pyqtgraph as pg
from qtpy.QtCore import QRectF
@@ -32,45 +32,36 @@ class TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample):
class PlotInfoLabel(pg.LegendItem):
"""Paint-safe text overlay for plot metadata.
Plot widgets submit arbitrary rows; this class owns the pyqtgraph legend
mechanics and keeps the display independent from any specific metadata
source such as scans.
Plot widgets submit (label, value) rows; this class owns the pyqtgraph
legend mechanics and keeps the display independent from any specific
metadata source such as scans.
"""
def __init__(self, offset: tuple[int, int] = (-30, 1), theme: str = "light"):
super().__init__(offset=offset, horSpacing=0)
self._rows: list[tuple[str, object | None]] = []
self._rows: list[tuple[str, str | None]] = []
self._offset = offset
self._has_drawn_rows = False
self.set_theme(theme)
self.setVisible(False)
@property
def rows(self) -> list[tuple[str, object | None]]:
"""Return a copy of the currently submitted rows."""
def rows(self) -> list[tuple[str, str | None]]:
"""Return a copy of the currently displayed rows."""
return list(self._rows)
def set_rows(
self, rows: Mapping[str, object | None] | Iterable[tuple[str, object | None]]
) -> None:
"""Replace all displayed rows."""
if isinstance(rows, Mapping):
rows = rows.items()
self._rows = [(str(label), value) for label, value in rows]
self.redraw()
def set_rows(self, rows: Iterable[tuple[str, object | None]]) -> None:
"""
Replace all displayed rows. Unchanged rows are not re-rendered.
def add_row(self, label: str, value: object | None = None) -> None:
"""Append one displayed row."""
self._rows.append((str(label), value))
Args:
rows(Iterable[tuple[str, object | None]]): An iterable of (label, value) pairs to display. The label is a string and the value can be any object or None.
"""
rows = [(str(label), None if value is None else str(value)) for label, value in rows]
if rows == self._rows:
return
self._rows = rows
self.redraw()
def clear_rows(self) -> None:
"""Remove all displayed rows and hide the label."""
self._rows = []
self._has_drawn_rows = False
self.clear()
self.setVisible(False)
def reset_position(self) -> None:
"""Reset the label to its default anchored position."""
self.setOffset(self._offset)
@@ -88,24 +79,13 @@ class PlotInfoLabel(pg.LegendItem):
self.redraw()
def redraw(self) -> None:
"""Refresh the pyqtgraph legend rows."""
previous_pos = self.pos()
preserve_pos = self._has_drawn_rows and self.isVisible()
row_texts = [self._format_row(label, value) for label, value in self._rows]
if len(row_texts) == len(self.items):
for row_text, (_, label_item) in zip(row_texts, self.items, strict=True):
label_item.setText(row_text)
self.updateSize()
else:
self.clear()
for row_text in row_texts:
self.addItem(TextOnlyLegendSample(), row_text)
if preserve_pos:
self.setPos(previous_pos)
self._has_drawn_rows = bool(self._rows)
"""Rebuild the pyqtgraph legend rows from scratch."""
self.clear()
for label, value in self._rows:
self.addItem(TextOnlyLegendSample(), self._format_row(label, value))
@staticmethod
def _format_row(label: str, value: object | None) -> str:
def _format_row(label: str, value: str | None) -> str:
if value is None:
return label
return f"{label}: {value}"
bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py
+4 -4
View File
@@ -318,7 +318,7 @@ class ScatterWaveform(PlotBase):
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)
self.update_scan_info_from_source(self.scan_item, mode="live")
self.update_scan_info_from_source(self.scan_item)
# First trigger to update the scan curves
self.sync_signal_update.emit()
@@ -704,7 +704,7 @@ class ScatterWaveform(PlotBase):
if scan_index is None:
self.scan_id = scan_id
self.scan_item = self.client.history.get_by_scan_id(scan_id)
self.update_scan_info_from_source(self.scan_item, mode="history")
self.update_scan_info_from_source(self.scan_item)
self.sync_signal_update.emit()
return
@@ -716,7 +716,7 @@ class ScatterWaveform(PlotBase):
return
self.scan_item = scan_item
self.scan_id = scan_item.scan_id
self.update_scan_info_from_source(scan_item, mode="live")
self.update_scan_info_from_source(scan_item)
self.sync_signal_update.emit()
return
@@ -727,7 +727,7 @@ class ScatterWaveform(PlotBase):
self.scan_item = self.client.history[scan_index]
metadata = self.scan_item.metadata
self.scan_id = metadata["bec"]["scan_id"]
self.update_scan_info_from_source(self.scan_item, mode="history")
self.update_scan_info_from_source(self.scan_item)
self.sync_signal_update.emit()
bec_widgets/widgets/plots/waveform/waveform.py
+2 -2
View File
@@ -1594,7 +1594,7 @@ class Waveform(PlotBase):
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) # live scan
self.update_scan_info_from_source(self.scan_item, mode="live")
self.update_scan_info_from_source(self.scan_item)
self._slice_index = None # Reset the slice index
self._update_curve_visibility()
self._mode = self._categorise_device_curves()
@@ -2496,7 +2496,7 @@ class Waveform(PlotBase):
else:
self.scan_id = self.scan_item.scan_id
self.update_scan_info_from_source(self.scan_item, mode="history")
self.update_scan_info_from_source(self.scan_item)
self._emit_signal_update()
def _emit_signal_update(self):
tests/unit_tests/client_mocks.py
+19 -3
View File
@@ -188,9 +188,16 @@ class DummyData:
return default
def create_dummy_scan_item():
def create_dummy_scan_item(
scan_id: str = "dummy", scan_number: int | None = None, scan_name: str | None = None
):
"""
Helper to create a dummy scan item with both live_data and metadata/status_message info.
Args:
scan_id (str): Scan ID used in the metadata and the status message.
scan_number (int | None): Scan number of the status message and the metadata.
scan_name (str | None): Scan name of the status message and the metadata.
"""
dummy_live_data = {
"samx": {"samx": DummyData(val=[10, 20, 30], timestamps=[100, 200, 300])},
@@ -198,17 +205,26 @@ def create_dummy_scan_item():
"bpm4i": {"bpm4i": DummyData(val=[5, 6, 7], timestamps=[101, 201, 301])},
"async_device": {"async_device": DummyData(val=[1, 2, 3], timestamps=[11, 21, 31])},
}
# spec=ScanItem makes isinstance checks pass but hides the instance attributes that
# ScanItem.__init__ assigns, so the ones the widgets read are set explicitly.
dummy_scan = MagicMock(spec=ScanItem)
dummy_scan.scan_id = scan_id
dummy_scan.scan_number = scan_number
dummy_scan.status = "open"
dummy_scan.live_data = dummy_live_data
dummy_scan.metadata = {
"bec": {
"scan_id": "dummy",
"scan_id": scan_id,
"scan_number": scan_number,
"scan_name": scan_name,
"scan_report_devices": ["samx"],
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
}
}
dummy_scan.status_message = messages.ScanStatusMessage(
scan_id="dummy",
scan_id=scan_id,
scan_number=scan_number,
scan_name=scan_name,
status="open",
info={
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
tests/unit_tests/test_heatmap_widget.py
+63 -6
View File
@@ -1044,20 +1044,77 @@ def test_heatmap_config_label_shows_live_or_history(heatmap_widget):
scan_msg.scan_number = 5
scan_msg.scan_name = "line_scan"
heatmap_widget.status_message = scan_msg
heatmap_widget._image_config.show_config_label = True
heatmap_widget.show_info_label = True
heatmap_widget.redraw_config_label()
labels = [label.text for _, label in heatmap_widget.config_label.items]
assert heatmap_widget.config_label is heatmap_widget.info_label
labels = [label.text for _, label in heatmap_widget.info_label.items]
assert "Scan: 5 (live)" in labels
assert "Interpolation: linear" in labels
heatmap_widget._history_scan_id = "scan-1"
heatmap_widget.redraw_config_label()
labels = [label.text for _, label in heatmap_widget.config_label.items]
labels = [label.text for _, label in heatmap_widget.info_label.items]
assert "Scan: 5 (history)" in labels
def test_heatmap_plot_show_config_label_reenables_hidden_label(heatmap_widget):
"""Regression: plot(show_config_label=True) after the toolbar hid the label kept it hidden."""
scan_msg = mock.MagicMock()
scan_msg.scan_number = 5
scan_msg.scan_name = "grid_scan"
heatmap_widget.status_message = scan_msg
heatmap_widget.redraw_config_label()
assert heatmap_widget.info_label.isVisible()
action = heatmap_widget.toolbar.components.get_action("plot_info_label").action
action.trigger()
assert not heatmap_widget.show_info_label
assert not heatmap_widget.info_label.isVisible()
heatmap_widget.plot(
"samx",
"samy",
"bpm4i",
signal_x="samx",
signal_y="samy",
signal_z="bpm4i",
validate_bec=False,
show_config_label=True,
)
heatmap_widget.redraw_config_label()
assert heatmap_widget.show_info_label
assert heatmap_widget._image_config.show_config_label
assert heatmap_widget.info_label.isVisible()
assert action.isChecked()
action.trigger()
assert not heatmap_widget.info_label.isVisible()
def test_heatmap_show_info_label_is_the_only_visibility_flag(heatmap_widget):
"""Regression: writes to the inherited show_info_label property were reverted on redraw."""
scan_msg = mock.MagicMock()
scan_msg.scan_number = 5
scan_msg.scan_name = "grid_scan"
heatmap_widget.status_message = scan_msg
action = heatmap_widget.toolbar.components.get_action("plot_info_label").action
# what a Designer .ui file / QUiLoader does
heatmap_widget.setProperty("show_info_label", False)
heatmap_widget.redraw_config_label()
assert not heatmap_widget.info_label.isVisible()
assert not action.isChecked()
action.trigger()
assert heatmap_widget.info_label.isVisible()
heatmap_widget.apply_theme("dark")
heatmap_widget.redraw_config_label()
assert heatmap_widget.show_info_label
assert heatmap_widget.info_label.isVisible()
def test_heatmap_config_label_paints_without_error(heatmap_widget):
"""The config label rows must survive a real paint pass.
@@ -1068,10 +1125,10 @@ def test_heatmap_config_label_paints_without_error(heatmap_widget):
scan_msg.scan_number = 5
scan_msg.scan_name = "line_scan"
heatmap_widget.status_message = scan_msg
heatmap_widget._image_config.show_config_label = True
heatmap_widget.show_info_label = True
heatmap_widget.redraw_config_label()
samples = [sample for sample, _ in heatmap_widget.config_label.items]
samples = [sample for sample, _ in heatmap_widget.info_label.items]
assert samples
assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples)
tests/unit_tests/test_multi_waveform_next_gen.py
+24
View File
@@ -106,6 +106,30 @@ def test_multiwaveform_scan_update_populates_info_label(qtbot, mocked_client, mo
]
def test_multiwaveform_clears_info_label_when_scan_id_is_unknown(qtbot, mocked_client, monkeypatch):
"""Regression: monitor data without a known scan_id kept the previous scan's rows."""
mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
mw.show_info_label = True
scan_item = ScanItem(queue_id="queue-1", scan_number=1, scan_id="scan_1", status="open")
scan_item.status_message = messages.ScanStatusMessage(
scan_id="scan_1", scan_number=1, scan_name="line_scan", status="open", info={}
)
storage = {"scan_1": scan_item}
monkeypatch.setattr(
mw.queue.scan_storage, "find_scan_by_ID", lambda scan_id: storage.get(scan_id)
)
mw.on_monitor_1d_update({"data": np.array([1, 2, 3])}, metadata={"scan_id": "scan_1"})
assert mw.info_label.isVisible()
# e.g. the device was triggered outside a scan: no scan_id in the metadata
mw.on_monitor_1d_update({"data": np.array([4, 5, 6])}, metadata={})
assert mw.scan_id is None
assert mw.info_label.rows == []
assert not mw.info_label.isVisible()
def test_multiwaveform_curve_limit_flush(qtbot, mocked_client):
"""Check that limiting the number of curves with flush removes older ones."""
mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
tests/unit_tests/test_plot_base_next_gen.py
+78 -14
View File
@@ -154,7 +154,7 @@ def test_update_scan_info_from_scan_item_status_message(qtbot, mocked_client):
)
scan_item.status_message = status_message
pb.update_scan_info_from_source(scan_item, mode="live")
pb.update_scan_info_from_source(scan_item)
labels = [label.text for _, label in pb.info_label.items]
assert labels == ["Scan: 7 (live)", "Scan Name: line_scan"]
@@ -169,28 +169,55 @@ def test_update_scan_info_from_history_scan_item_uses_scan_number(
)
scan_item = ScanDataContainer(file_path=history_message.file_path, msg=history_message)
pb.update_scan_info_from_source(scan_item, mode="history")
pb.update_scan_info_from_source(scan_item)
labels = [label.text for _, label in pb.info_label.items]
assert labels == ["Scan: 42 (history)", "Scan Name: line_scan"]
def test_history_scan_info_does_not_fall_back_to_scan_id(qtbot, mocked_client):
@pytest.mark.parametrize(
"scan_source",
[None, ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open")],
ids=["no_scan_item", "scan_item_without_status_message"],
)
def test_update_scan_info_clears_rows_when_source_has_no_scan_info(
qtbot, mocked_client, scan_source
):
"""Regression: a missing scan item left the previous scan's rows on the label."""
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.show_info_label = True
pb.set_scan_info(scan_number=6, scan_name="line_scan", mode="live")
assert pb.info_label.isVisible()
pb.set_scan_info(scan_id="history-id", mode="history")
pb.update_scan_info_from_source(scan_source)
assert pb.info_label.rows == []
assert not pb.info_label.isVisible()
def test_update_scan_info_noops_when_source_has_no_scan_info(qtbot, mocked_client):
def test_custom_info_rows_survive_scan_info_updates(qtbot, mocked_client):
"""Regression: set_info_label_rows rows were replaced by the next set_scan_info call."""
pb = create_widget(qtbot, PlotBase, client=mocked_client)
scan_item = ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open")
pb.set_scan_info(scan_id="scan-7", mode="live")
pb.update_scan_info_from_source(scan_item, mode="live")
pb.set_info_label_rows({"Sample": "Si"})
pb.set_scan_info(scan_number=5, scan_name="line_scan", mode="live")
assert [label.text for _, label in pb.info_label.items] == [
"Scan: 5 (live)",
"Scan Name: line_scan",
"Sample: Si",
]
assert [label.text for _, label in pb.info_label.items] == ["Scan ID: scan-7 (live)"]
pb.set_scan_info(scan_number=6, scan_name="grid_scan", mode="live")
pb.set_info_label_rows({"Sample": "Ge"})
assert [label.text for _, label in pb.info_label.items] == [
"Scan: 6 (live)",
"Scan Name: grid_scan",
"Sample: Ge",
]
pb.clear_info_label()
assert pb.info_label.rows == []
assert not pb.info_label.isVisible()
def test_info_label_paints_without_error(qtbot, mocked_client):
@@ -219,6 +246,41 @@ def test_info_label_redraw_preserves_user_position(qtbot, mocked_client):
assert pb.info_label.pos() == position
def test_info_label_rows_accept_values_without_boolean_equality(qtbot, mocked_client):
"""Regression: comparing raw row values raised for numpy arrays on the second submit."""
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.set_info_label_rows({"Positions": np.array([1, 2]), "Points": 3})
pb.set_info_label_rows({"Positions": np.array([1, 2]), "Points": 3})
assert pb.info_label.rows == [("Positions", "[1 2]"), ("Points", "3")]
assert [label.text for _, label in pb.info_label.items] == ["Positions: [1 2]", "Points: 3"]
def test_info_label_box_shrinks_when_row_text_gets_shorter(qtbot, mocked_client):
"""Regression: the same-row-count redraw used to keep the widest frame ever drawn."""
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.show_info_label = True
pb.set_info_label_rows({"Scan": "1 (live)", "Scan Name": "line_scan"})
short_width = pb.info_label.geometry().width()
pb.set_info_label_rows({"Scan": "2 (live)", "Scan Name": "fermat_scan_with_a_much_longer_name"})
assert pb.info_label.geometry().width() > short_width + 50
pb.set_info_label_rows({"Scan": "3 (live)", "Scan Name": "line_scan"})
assert pb.info_label.geometry().width() == pytest.approx(short_width, abs=1)
def test_info_label_set_rows_skips_redraw_for_unchanged_rows(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.set_info_label_rows({"Scan": "1 (live)"})
items_before = [label for _, label in pb.info_label.items]
pb.set_info_label_rows({"Scan": "1 (live)"})
assert [label for _, label in pb.info_label.items] == items_before
def test_info_label_toggle_resets_position(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.set_info_label_rows({"Scan": "5 (live)"})
@@ -240,10 +302,7 @@ def test_info_label_toggle_resets_position(qtbot, mocked_client):
def test_info_label_is_available_on_plot_base_subclasses(qtbot, mocked_client, widget_cls):
widget = create_widget(qtbot, widget_cls, client=mocked_client)
if hasattr(widget, "show_config_label"):
widget.show_config_label = False
else:
widget.show_info_label = False
widget.show_info_label = False
widget.set_info_label_rows([("Widget", widget_cls.__name__), ("Mode", "custom")])
assert [label.text for _, label in widget.info_label.items] == [
@@ -252,8 +311,13 @@ def test_info_label_is_available_on_plot_base_subclasses(qtbot, mocked_client, w
]
assert not widget.info_label.isVisible()
if isinstance(widget, MotorMap):
if isinstance(widget, (Image, MotorMap)):
# these widgets never publish scan info and remove the toolbar button
assert not widget.toolbar.components.exists("plot_info_label")
assert "plot_info_label" not in widget.toolbar.get_bundle("axis_popup").bundle_actions
# the property itself keeps working without the button
widget.show_info_label = True
assert widget.info_label.isVisible()
return
action = widget.toolbar.components.get_action("plot_info_label").action
tests/unit_tests/test_scatter_waveform.py
+20 -29
View File
@@ -1,7 +1,6 @@
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import numpy as np
from bec_lib import messages
from bec_lib.scan_data_container import ScanDataContainer
from bec_widgets.widgets.plots.scatter_waveform.scatter_curve import (
@@ -12,7 +11,12 @@ from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterW
from bec_widgets.widgets.plots.scatter_waveform.settings.scatter_curve_setting import (
ScatterCurveSettings,
)
from tests.unit_tests.client_mocks import DummyData, create_dummy_scan_item, mocked_client
from tests.unit_tests.client_mocks import (
DummyData,
create_dummy_scan_item,
inject_scan_history,
mocked_client,
)
from .conftest import create_widget
@@ -51,27 +55,24 @@ def test_scatter_waveform_color_map(qtbot, mocked_client):
assert swf.color_map == "plasma"
def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeypatch):
def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, scan_history_factory):
"""History scans are read from the HDF5-backed ScanDataContainer, not from live data."""
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
dummy_scan = create_dummy_scan_item()
history_scan = ScanDataContainer()
history_scan.live_data = dummy_scan.live_data
history_scan.metadata = {
"bec": {"scan_id": "dummy", "scan_number": 24, "scan_name": "line_scan"}
}
mocked_client.history = MagicMock()
mocked_client.history.get_by_scan_id.return_value = history_scan
mocked_client.history.__getitem__.return_value = history_scan
inject_scan_history(swf, scan_history_factory, ("dummy", 24))
swf.plot("samx", "samy", "bpm4i", label="test_curve")
swf.update_with_scan_history(scan_id="dummy")
qtbot.waitUntil(lambda: swf.scan_item == history_scan, timeout=500)
qtbot.wait(200)
qtbot.waitUntil(lambda: isinstance(swf.scan_item, ScanDataContainer), timeout=500)
def _has_points():
x_data = swf.main_curve.getData()[0]
return x_data is not None and len(x_data) > 0
qtbot.waitUntil(_has_points, timeout=2000)
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])
assert len(x_data) == len(y_data) > 0
assert swf.scan_id == "dummy"
assert [label.text for _, label in swf.info_label.items] == [
"Scan: 24 (history)",
"Scan Name: line_scan",
@@ -81,17 +82,7 @@ def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeyp
def test_scatter_waveform_live_update(qtbot, mocked_client, monkeypatch):
swf = create_widget(qtbot, ScatterWaveform, client=mocked_client)
dummy_scan = create_dummy_scan_item()
dummy_scan.status_message = messages.ScanStatusMessage(
scan_id="dummy",
scan_number=13,
scan_name="line_scan",
status="open",
info={
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
"scan_report_devices": ["samx"],
},
)
dummy_scan = create_dummy_scan_item(scan_number=13, scan_name="line_scan")
monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan)
swf.plot("samx", "samy", "bpm4i", label="live_curve")
tests/unit_tests/test_waveform.py
+48 -12
View File
@@ -9,6 +9,7 @@ import numpy as np
import pyqtgraph as pg
import pytest
from bec_lib import messages
from bec_lib.scan_items import ScanItem
from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox
@@ -458,18 +459,7 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls):
wf.plot(arg1="async_device")
# We mock out the scan_item, pretending we found a new scan.
dummy_scan = create_dummy_scan_item()
dummy_scan.metadata["bec"]["scan_id"] = "1234"
dummy_scan.status_message = messages.ScanStatusMessage(
scan_id="1234",
scan_number=12,
scan_name="line_scan",
status="open",
info={
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
"scan_report_devices": ["samx"],
},
)
dummy_scan = create_dummy_scan_item(scan_id="1234", scan_number=12, scan_name="line_scan")
monkeypatch.setattr(wf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan)
# We'll track calls to sync_signal_update and async_signal_update
@@ -1665,6 +1655,52 @@ def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_fa
]
def test_on_scan_status_clears_info_label_when_scan_item_is_missing(
qtbot, mocked_client, monkeypatch
):
"""Regression: a scan whose ScanItem is not in scan storage kept the previous scan's rows."""
wf = create_widget(qtbot, Waveform, client=mocked_client)
wf.show_info_label = True
scan_item = ScanItem(queue_id="queue-1", scan_number=1, scan_id="scan-1", status="open")
scan_item.status_message = messages.ScanStatusMessage(
scan_id="scan-1", scan_number=1, scan_name="line_scan", status="open", info={}
)
storage = {"scan-1": scan_item}
monkeypatch.setattr(
wf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: storage.get(scan_id)
)
wf.on_scan_status({"scan_id": "scan-1", "status": "open"}, {})
assert [label.text for _, label in wf.info_label.items] == [
"Scan: 1 (live)",
"Scan Name: line_scan",
]
wf.on_scan_status({"scan_id": "scan-2", "status": "closed"}, {})
assert wf.scan_id == "scan-2"
assert wf.info_label.rows == []
assert not wf.info_label.isVisible()
def test_update_with_scan_history_labels_running_scan_as_live(qtbot, mocked_client):
"""Regression: a Waveform created during a running scan labelled the live scan '(history)'."""
wf = create_widget(qtbot, Waveform, client=mocked_client)
scan_item = ScanItem(queue_id="queue-1", scan_number=9, scan_id="scan-9", status="open")
scan_item.status_message = messages.ScanStatusMessage(
scan_id="scan-9", scan_number=9, scan_name="line_scan", status="open", info={}
)
mocked_client.queue.scan_storage.current_scan = scan_item
wf.update_with_scan_history(-1)
assert wf.scan_item is scan_item
assert [label.text for _, label in wf.info_label.items] == [
"Scan: 9 (live)",
"Scan Name: line_scan",
]
@pytest.mark.parametrize("mode", ["auto", "timestamp", "index", "samx"])
def test_history_curve_x_modes_pre_plot(qtbot, mocked_client, scan_history_factory, mode):
"""