feat: generalized scan info label

This commit is contained in:
2026-09-02 11:37:20 +02:00
parent 96ce9ab891
commit bc7897be2d
14 changed files with 586 additions and 88 deletions
+37 -73
View File
@@ -8,10 +8,10 @@ import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger, messages
from bec_lib.endpoints import MessageEndpoints
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from bec_lib.utils.import_utils import lazy_import_from
from bec_qthemes import material_icon
from pydantic import BaseModel, Field, field_validator
from qtpy.QtCore import QObject, QRectF, Qt, QThread, QTimer, Signal
from qtpy.QtCore import QObject, Qt, QThread, QTimer, Signal
from qtpy.QtGui import QTransform
from qtpy.QtWidgets import QDialog, QPushButton, QVBoxLayout
from toolz import partition
@@ -93,30 +93,6 @@ class HeatmapConfig(ConnectionConfig):
_validate_color_palette = field_validator("color_map")(Colors.validate_color_map)
class _TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample):
"""Zero-size legend sample for text-only rows in the config label.
The stock ItemSample expects a plottable item with an ``opts`` dict; since
PySide 6.10 an exception raised inside its paint() override propagates out
of the C++ paint loop and crashes the application, so the config label rows
must not carry a real sample item.
"""
def __init__(self):
super().__init__(item=None)
self.setFixedWidth(0)
self.setFixedHeight(0)
def boundingRect(self):
return QRectF(0, 0, 0, 0)
def paint(self, p, *args):
pass
def mouseClickEvent(self, event):
event.ignore()
@dataclass
class _InterpolationRequest:
"""Immutable payload describing an interpolation request for the worker thread.
@@ -300,12 +276,8 @@ 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
bg_color = pg.mkColor((240, 240, 240, 150))
self.config_label = pg.LegendItem(
labelTextColor=(0, 0, 0), offset=(-30, 1), brush=pg.mkBrush(bg_color), horSpacing=0
)
self.config_label.setParentItem(self.plot_item.vb)
self.config_label.setVisible(False)
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())
self.bec_dispatcher.connect_slot(self.on_scan_progress, MessageEndpoints.scan_progress())
@@ -325,7 +297,6 @@ class Heatmap(ImageBase):
"image_colorbar",
"image_processing",
"axis_popup",
"interpolation_info",
]
)
@@ -344,15 +315,7 @@ class Heatmap(ImageBase):
Apply the current theme to the heatmap widget.
"""
super().apply_theme(theme)
if theme == "dark":
brush = pg.mkBrush(pg.mkColor(50, 50, 50, 150))
color = pg.mkColor(255, 255, 255)
else:
brush = pg.mkBrush(pg.mkColor(240, 240, 240, 150))
color = pg.mkColor(0, 0, 0)
if hasattr(self, "config_label"):
self.config_label.setBrush(brush)
self.config_label.setLabelTextColor(color)
if hasattr(self, "info_label"):
self.redraw_config_label()
@SafeSlot(popup_error=True)
@@ -588,19 +551,6 @@ class Heatmap(ImageBase):
if name not in ["image_processing_fft", "image_processing_log"]:
action().action.setVisible(False)
self.toolbar.add_action(
"interpolation_info",
MaterialIconAction(
icon_name="info", tooltip="Show Interpolation Info", checkable=True, parent=self
),
)
self.toolbar.components.get_action("interpolation_info").action.triggered.connect(
self.toggle_interpolation_info
)
self.toolbar.components.get_action("interpolation_info").action.setChecked(
self._image_config.show_config_label
)
def show_heatmap_settings(self):
"""
Show the heatmap settings dialog.
@@ -687,15 +637,27 @@ 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 interpolation info label.
Toggle the visibility of the heatmap information label.
"""
self._image_config.show_config_label = not self._image_config.show_config_label
self.toolbar.components.get_action("interpolation_info").action.setChecked(
self._image_config.show_config_label
)
self.redraw_config_label()
self.toggle_info_label()
def _heatmap_dialog_closed(self):
"""
@@ -989,27 +951,29 @@ class Heatmap(ImageBase):
self._latest_interpolation_version = -1
def redraw_config_label(self):
scan_msg = self.status_message
scan_msg = getattr(self, "status_message", None)
if scan_msg is None:
return
if not self._image_config.show_config_label:
self.config_label.setVisible(False)
self.show_info_label = False
return
self.config_label.setOffset((-30, 1))
self.config_label.setVisible(True)
self.config_label.clear()
# 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"
self.config_label.addItem(_TextOnlyLegendSample(), f"Scan: {scan_msg.scan_number} ({mode})")
self.config_label.addItem(_TextOnlyLegendSample(), f"Scan Name: {scan_msg.scan_name}")
extra_rows = []
if scan_msg.scan_name != "grid_scan" or self._image_config.enforce_interpolation:
self.config_label.addItem(
_TextOnlyLegendSample(), f"Interpolation: {self._image_config.interpolation}"
)
self.config_label.addItem(
_TextOnlyLegendSample(), f"Oversampling: {self._image_config.oversampling_factor}x"
extra_rows.extend(
[
("Interpolation", self._image_config.interpolation),
("Oversampling", f"{self._image_config.oversampling_factor}x"),
]
)
self.set_scan_info(
scan_number=scan_msg.scan_number,
scan_name=scan_msg.scan_name,
mode=mode,
extra_rows=extra_rows,
)
def get_image_data(
self,
@@ -167,6 +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")
reset_legend = MaterialIconAction(
icon_name="history",
@@ -1,7 +1,7 @@
from __future__ import annotations
from collections import deque
from typing import TYPE_CHECKING, cast
from typing import cast
import pyqtgraph as pg
from bec_lib.endpoints import MessageEndpoints
@@ -361,7 +361,8 @@ class MultiWaveform(PlotBase):
current_scan_id = metadata.get("scan_id", None)
if current_scan_id != self.scan_id:
self.scan_id = current_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.clear_curves()
self.curves.clear()
if self.crosshair:
+155 -1
View File
@@ -1,10 +1,13 @@
from __future__ import annotations
from enum import Enum
from typing import TypeAlias
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger
from bec_lib import bec_logger, messages
from bec_lib.scan_data_container import ScanDataContainer
from bec_lib.scan_items import ScanItem
from qtpy.QtCore import QPoint, QPointF, Qt, Signal
from qtpy.QtWidgets import QHBoxLayout, QLabel, QMainWindow, QVBoxLayout, QWidget
@@ -17,10 +20,12 @@ from bec_widgets.utils.fps_counter import FPSCounter
from bec_widgets.utils.plot_indicator_items import BECArrowItem, BECTickItem
from bec_widgets.utils.round_frame import RoundedFrame
from bec_widgets.utils.side_panel import SidePanel
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.utils.toolbars.performance import PerformanceConnection, performance_bundle
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.utils.widget_state_manager import WidgetStateManager
from bec_widgets.widgets.containers.layout_manager.layout_manager import LayoutManagerWidget
from bec_widgets.widgets.plots.plot_info_label import PlotInfoLabel
from bec_widgets.widgets.plots.setting_menus.axis_settings import AxisSettings
from bec_widgets.widgets.plots.toolbar_components.axis_settings_popup import (
AxisSettingsPopupConnection,
@@ -38,6 +43,8 @@ from bec_widgets.widgets.plots.toolbar_components.roi import RoiConnection, roi_
logger = bec_logger.logger
ScanInfoSource: TypeAlias = ScanItem | ScanDataContainer | None
class BECViewBox(pg.ViewBox):
sigPaint = Signal()
@@ -162,6 +169,8 @@ class PlotBase(BECWidget, QWidget):
# PlotItem Addons
self.plot_item.addLegend()
self.info_label = PlotInfoLabel()
self.info_label.setParentItem(self.plot_item.vb)
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.
@@ -183,6 +192,7 @@ class PlotBase(BECWidget, QWidget):
# Visibility States
self._toolbar_visible = True
self._enable_fps_monitor = False
self._show_info_label = False
self._outer_axes_visible = self.plot_item.getAxis("top").isVisible()
self._inner_axes_visible = self.plot_item.getAxis("bottom").isVisible()
@@ -195,6 +205,8 @@ class PlotBase(BECWidget, QWidget):
def apply_theme(self, theme: str):
self.round_plot_widget.apply_theme(theme)
if hasattr(self, "info_label"):
self.info_label.set_theme(theme)
def _init_ui(self):
self.layout.addWidget(self.layout_manager)
@@ -219,6 +231,16 @@ class PlotBase(BECWidget, QWidget):
self.toolbar.add_bundle(mouse_interaction_bundle(self.toolbar.components))
self.toolbar.add_bundle(roi_bundle(self.toolbar.components))
self.toolbar.add_bundle(axis_popup_bundle(self.toolbar.components))
self.toolbar.components.add_safe(
"plot_info_label",
MaterialIconAction(
icon_name="info", tooltip="Show Plot Info", checkable=True, parent=self
),
)
self.toolbar.get_bundle("axis_popup").add_action("plot_info_label")
self.toolbar.components.get_action("plot_info_label").action.triggered.connect(
self.toggle_info_label
)
self.toolbar.connect_bundle(
"plot_base", PlotExportConnection(self.toolbar.components, self)
@@ -236,6 +258,7 @@ class PlotBase(BECWidget, QWidget):
# hide some options by default
self.toolbar.toggle_action_visibility("fps_monitor", False)
self._sync_info_label_action()
# Get default viewbox state
self.toolbar.show_bundles(
@@ -261,6 +284,137 @@ class PlotBase(BECWidget, QWidget):
"""In the case that the legend is not visible, reset it to be visible to top left corner"""
self.plot_item.legend.autoAnchor(50)
################################################################################
# Plot Info Label
################################################################################
@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."""
return self._show_info_label
@show_info_label.setter
def show_info_label(self, value: bool) -> None:
"""Show or hide the plot info label."""
visible = bool(value)
if visible and not self._show_info_label:
self.info_label.reset_position()
self._show_info_label = visible
self._sync_info_label_visibility()
self._sync_info_label_action()
@SafeSlot()
def toggle_info_label(self) -> None:
"""Toggle the plot info label visibility."""
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]]
) -> 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()
def clear_info_label(self) -> None:
"""Clear all rows from the plot info label."""
self.info_label.clear_rows()
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,
) -> None:
"""Submit common scan metadata rows to the plot 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))
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)
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_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
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 _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)
################################################################################
# Toggle UI Elements
################################################################################
@@ -0,0 +1,111 @@
from __future__ import annotations
from collections.abc import Iterable, Mapping
import pyqtgraph as pg
from qtpy.QtCore import QRectF
class TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample):
"""Zero-size legend sample for text-only rows.
The stock ItemSample expects a plottable item with an ``opts`` dict; since
PySide 6.10 an exception raised inside its paint() override propagates out
of the C++ paint loop and crashes the application.
"""
def __init__(self):
super().__init__(item=None)
self.setFixedWidth(0)
self.setFixedHeight(0)
def boundingRect(self):
return QRectF(0, 0, 0, 0)
def paint(self, p, *args):
pass
def mouseClickEvent(self, event):
event.ignore()
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.
"""
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._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."""
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 add_row(self, label: str, value: object | None = None) -> None:
"""Append one displayed row."""
self._rows.append((str(label), value))
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)
def set_theme(self, theme: str) -> None:
"""Update the label colors for the active application theme."""
if theme == "dark":
brush = pg.mkBrush(pg.mkColor(50, 50, 50, 150))
color = pg.mkColor(255, 255, 255)
else:
brush = pg.mkBrush(pg.mkColor(240, 240, 240, 150))
color = pg.mkColor(0, 0, 0)
self.setBrush(brush)
self.setLabelTextColor(color)
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)
@staticmethod
def _format_row(label: str, value: object | None) -> str:
if value is None:
return label
return f"{label}: {value}"
@@ -318,6 +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")
# First trigger to update the scan curves
self.sync_signal_update.emit()
@@ -703,6 +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.sync_signal_update.emit()
return
@@ -714,6 +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.sync_signal_update.emit()
return
@@ -724,6 +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.sync_signal_update.emit()
@@ -1594,6 +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._slice_index = None # Reset the slice index
self._update_curve_visibility()
self._mode = self._categorise_device_curves()
@@ -2495,6 +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._emit_signal_update()
def _emit_signal_update(self):
+10 -5
View File
@@ -7,6 +7,7 @@ import pytest
from bec_lib.bec_service import messages
from bec_lib.endpoints import MessageEndpoints
from bec_lib.scan_history import ScanHistory
from bec_lib.scan_items import ScanItem
from bec_widgets.tests.utils import FakePositioner, Positioner
@@ -197,7 +198,7 @@ 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])},
}
dummy_scan = MagicMock()
dummy_scan = MagicMock(spec=ScanItem)
dummy_scan.live_data = dummy_live_data
dummy_scan.metadata = {
"bec": {
@@ -206,10 +207,14 @@ def create_dummy_scan_item():
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
}
}
dummy_scan.status_message.info = {
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
"scan_report_devices": ["samx"],
}
dummy_scan.status_message = messages.ScanStatusMessage(
scan_id="dummy",
status="open",
info={
"readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]},
"scan_report_devices": ["samx"],
},
)
return dummy_scan
+4 -2
View File
@@ -13,8 +13,8 @@ from bec_widgets.widgets.plots.heatmap.heatmap import (
HeatmapDeviceSignal,
_InterpolationRequest,
_StepInterpolationWorker,
_TextOnlyLegendSample,
)
from bec_widgets.widgets.plots.plot_info_label import TextOnlyLegendSample
# pytest: disable=unused-import
from tests.unit_tests.client_mocks import mocked_client
@@ -1048,7 +1048,9 @@ def test_heatmap_config_label_shows_live_or_history(heatmap_widget):
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
assert "Scan: 5 (live)" in labels
assert "Interpolation: linear" in labels
heatmap_widget._history_scan_id = "scan-1"
heatmap_widget.redraw_config_label()
@@ -1071,7 +1073,7 @@ def test_heatmap_config_label_paints_without_error(heatmap_widget):
samples = [sample for sample, _ in heatmap_widget.config_label.items]
assert samples
assert all(isinstance(sample, _TextOnlyLegendSample) for sample in samples)
assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples)
pixmap = heatmap_widget.grab()
assert not pixmap.isNull()
@@ -674,6 +674,7 @@ def test_handle_scan_change_clears_buffers_and_resets_crosshair(qtbot, mocked_cl
assert view.old_scan_id == "scan_1"
assert view.scan_id == "scan_2"
assert view.info_label.rows == []
assert clear_called == [True]
assert view.main_image.buffer == []
assert view.main_image.max_len == 0
@@ -1,4 +1,6 @@
import numpy as np
from bec_lib import messages
from bec_lib.scan_items import ScanItem
from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
from tests.unit_tests.client_mocks import mocked_client
@@ -88,6 +90,22 @@ def test_multiwaveform_curve_limit_no_flush(qtbot, mocked_client):
assert len(visible_curves) == 3
def test_multiwaveform_scan_update_populates_info_label(qtbot, mocked_client, monkeypatch):
mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
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={}
)
monkeypatch.setattr(mw.queue.scan_storage, "find_scan_by_ID", lambda scan_id: scan_item)
mw.on_monitor_1d_update({"data": np.array([1, 2, 3])}, metadata={"scan_id": "scan_1"})
assert [label.text for _, label in mw.info_label.items] == [
"Scan: 1 (live)",
"Scan Name: line_scan",
]
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)
+190
View File
@@ -1,6 +1,17 @@
import numpy as np
import pytest
from bec_lib import messages
from bec_lib.scan_data_container import ScanDataContainer
from bec_lib.scan_items import ScanItem
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
from bec_widgets.widgets.plots.plot_base import PlotBase, UIMode
from bec_widgets.widgets.plots.plot_info_label import TextOnlyLegendSample
from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from .client_mocks import mocked_client
from .conftest import create_widget
@@ -73,6 +84,185 @@ def test_set_y_label_emits_signal(qtbot, mocked_client):
assert pb.plot_item.getAxis("left").labelText == "Current [A]"
def test_info_label_rows_are_submitted_without_forcing_visibility(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
action = pb.toolbar.components.get_action("plot_info_label").action
assert action.isVisible()
assert action.isEnabled()
assert not action.isChecked()
pb.set_info_label_rows([("Scan", "5 (live)"), ("Mode", "alignment")])
labels = [label.text for _, label in pb.info_label.items]
assert labels == ["Scan: 5 (live)", "Mode: alignment"]
assert pb.info_label.rows == [("Scan", "5 (live)"), ("Mode", "alignment")]
assert not pb.info_label.isVisible()
assert action.isVisible()
assert action.isEnabled()
assert not action.isChecked()
def test_info_label_can_be_enabled_before_rows_are_available(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
action = pb.toolbar.components.get_action("plot_info_label").action
action.trigger()
assert pb.show_info_label
assert action.isChecked()
assert not pb.info_label.isVisible()
pb.set_info_label_rows({"Scan": "5 (live)"})
assert pb.info_label.isVisible()
assert action.isChecked()
def test_info_label_visibility_property_and_toolbar_action(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.set_info_label_rows({"Scan": "5 (live)"})
with qtbot.waitSignal(pb.property_changed, timeout=500) as signal:
pb.show_info_label = True
assert signal.args == ["show_info_label", True]
assert pb.info_label.isVisible()
assert pb.toolbar.components.get_action("plot_info_label").action.isChecked()
pb.toolbar.components.get_action("plot_info_label").action.trigger()
assert not pb.show_info_label
assert not pb.info_label.isVisible()
def test_set_scan_info_formats_common_scan_rows(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.show_info_label = True
pb.set_scan_info(
scan_number=7, scan_name="grid_scan", mode="history", extra_rows=[("Detector", "eiger")]
)
labels = [label.text for _, label in pb.info_label.items]
assert labels == ["Scan: 7 (history)", "Scan Name: grid_scan", "Detector: eiger"]
def test_update_scan_info_from_scan_item_status_message(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
scan_item = ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open")
status_message = messages.ScanStatusMessage(
scan_id="scan-7", scan_number=7, scan_name="line_scan", status="open", info={}
)
scan_item.status_message = status_message
pb.update_scan_info_from_source(scan_item, mode="live")
labels = [label.text for _, label in pb.info_label.items]
assert labels == ["Scan: 7 (live)", "Scan Name: line_scan"]
def test_update_scan_info_from_history_scan_item_uses_scan_number(
qtbot, mocked_client, scan_history_factory
):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
history_message = scan_history_factory(
scan_id="history-id", scan_number=42, scan_name="line_scan"
)
scan_item = ScanDataContainer(file_path=history_message.file_path, msg=history_message)
pb.update_scan_info_from_source(scan_item, mode="history")
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):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.set_scan_info(scan_id="history-id", mode="history")
assert pb.info_label.rows == []
def test_update_scan_info_noops_when_source_has_no_scan_info(qtbot, mocked_client):
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")
assert [label.text for _, label in pb.info_label.items] == ["Scan ID: scan-7 (live)"]
def test_info_label_paints_without_error(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.show_info_label = True
pb.set_info_label_rows({"Scan": "5 (live)"})
samples = [sample for sample, _ in pb.info_label.items]
assert samples
assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples)
pixmap = pb.grab()
assert not pixmap.isNull()
def test_info_label_redraw_preserves_user_position(qtbot, mocked_client):
pb = create_widget(qtbot, PlotBase, client=mocked_client)
pb.show_info_label = True
pb.set_info_label_rows({"Scan": "5 (live)"})
pb.info_label.autoAnchor((42, 24), relative=False)
position = pb.info_label.pos()
pb.set_info_label_rows({"Scan": "6 (live)"})
assert pb.info_label.pos() == position
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)"})
pb.show_info_label = True
default_position = pb.info_label.pos()
pb.info_label.autoAnchor((42, 24), relative=False)
assert pb.info_label.pos() != default_position
pb.show_info_label = False
pb.show_info_label = True
assert pb.info_label.pos() == default_position
@pytest.mark.parametrize(
"widget_cls", [PlotBase, Waveform, ScatterWaveform, MultiWaveform, MotorMap, Image, Heatmap]
)
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.set_info_label_rows([("Widget", widget_cls.__name__), ("Mode", "custom")])
assert [label.text for _, label in widget.info_label.items] == [
f"Widget: {widget_cls.__name__}",
"Mode: custom",
]
assert not widget.info_label.isVisible()
if isinstance(widget, MotorMap):
assert "plot_info_label" not in widget.toolbar.get_bundle("axis_popup").bundle_actions
return
action = widget.toolbar.components.get_action("plot_info_label").action
action.trigger()
assert widget.show_info_label
assert widget.info_label.isVisible()
assert action.isChecked()
def test_set_x_min_max(qtbot, mocked_client):
"""
Test setting x_min, x_max changes the actual X-range of the plot
+28 -5
View File
@@ -1,6 +1,8 @@
from unittest.mock import MagicMock, 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 (
ScatterCurveConfig,
@@ -53,26 +55,43 @@ def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeyp
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()
# .get_by_scan_id() typically returns historical data, but we abuse it here
# to return mock live data
mocked_client.history.get_by_scan_id.return_value = dummy_scan
mocked_client.history.__getitem__.return_value = dummy_scan
mocked_client.history.get_by_scan_id.return_value = history_scan
mocked_client.history.__getitem__.return_value = history_scan
swf.plot("samx", "samy", "bpm4i", label="test_curve")
swf.update_with_scan_history(scan_id="dummy")
qtbot.waitUntil(lambda: swf.scan_item == dummy_scan, timeout=500)
qtbot.waitUntil(lambda: swf.scan_item == history_scan, timeout=500)
qtbot.wait(200)
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 [label.text for _, label in swf.info_label.items] == [
"Scan: 24 (history)",
"Scan Name: line_scan",
]
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"],
},
)
monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan)
swf.plot("samx", "samy", "bpm4i", label="live_curve")
@@ -84,6 +103,10 @@ def test_scatter_waveform_live_update(qtbot, mocked_client, monkeypatch):
assert swf.scan_id == "dummy"
assert swf.scan_item == dummy_scan
assert [label.text for _, label in swf.info_label.items] == [
"Scan: 13 (live)",
"Scan Name: line_scan",
]
qtbot.wait(500)
+22
View File
@@ -8,6 +8,7 @@ from unittest.mock import MagicMock
import numpy as np
import pyqtgraph as pg
import pytest
from bec_lib import messages
from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox
@@ -459,6 +460,16 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls):
# 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"],
},
)
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
@@ -475,6 +486,10 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls):
assert wf.scan_id == "1234"
assert wf.scan_item == dummy_scan
assert wf._mode == mode
assert [label.text for _, label in wf.info_label.items] == [
"Scan: 12 (live)",
"Scan Name: line_scan",
]
assert sync_spy.call_count == calls[0], "sync_signal_update should be called exactly once"
assert async_spy.call_count == calls[1], "async_signal_update should be called exactly once"
@@ -1642,6 +1657,13 @@ def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_fa
assert c2.config.scan_number == 2
assert c2.name() == "bpm4i-bpm4i-scan-2"
wf.update_with_scan_history(scan_id="hist1")
assert [label.text for _, label in wf.info_label.items] == [
"Scan: 1 (history)",
"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):