mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-03-05 00:12:49 +01:00
refactor: cleanup, add compact popup view for scan_history_browser and update tests
This commit is contained in:
Jan Wyzula
@@ -1,56 +1,66 @@
|
||||
"""Module for displaying scan history devices in a viewer widget."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ScanHistoryMessage
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy import QtCore, QtGui, QtWidgets
|
||||
from qtpy import QtCore, QtWidgets
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_lib.messages import _StoredDataInfo
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
# TODO check cleanup
|
||||
# Custom model
|
||||
class DeviceModel(QtCore.QAbstractListModel):
|
||||
def __init__(self, devices=None):
|
||||
super().__init__()
|
||||
if devices is None:
|
||||
devices = {}
|
||||
self._devices = sorted(devices.items(), key=lambda x: -x[1])
|
||||
class SignalModel(QtCore.QAbstractListModel):
|
||||
"""Custom model for displaying scan history signals in a combo box."""
|
||||
|
||||
def __init__(self, parent=None, signals: dict[str, _StoredDataInfo] = None):
|
||||
super().__init__(parent)
|
||||
if signals is None:
|
||||
signals = {}
|
||||
self._signals: list[tuple[str, _StoredDataInfo]] = sorted(
|
||||
signals.items(), key=lambda x: -x[1].shape[0]
|
||||
)
|
||||
|
||||
@property
|
||||
def devices(self):
|
||||
def signals(self) -> list[tuple[str, _StoredDataInfo]]:
|
||||
"""Return the list of devices."""
|
||||
return self._devices
|
||||
return self._signals
|
||||
|
||||
@devices.setter
|
||||
def devices(self, value: dict[str, int]):
|
||||
@signals.setter
|
||||
def signals(self, value: dict[str, _StoredDataInfo]):
|
||||
self.beginResetModel()
|
||||
self._devices = sorted(value.items(), key=lambda x: -x[1])
|
||||
self._signals = sorted(value.items(), key=lambda x: -x[1].shape[0])
|
||||
self.endResetModel()
|
||||
|
||||
def rowCount(self, parent=QtCore.QModelIndex()):
|
||||
return len(self.devices)
|
||||
return len(self._signals)
|
||||
|
||||
def data(self, index, role=QtCore.Qt.DisplayRole):
|
||||
if not index.isValid():
|
||||
return None
|
||||
name, num_points = self.devices[index.row()]
|
||||
name, info = self.signals[index.row()]
|
||||
if role == QtCore.Qt.DisplayRole:
|
||||
return f"{name} ({num_points})" # fallback display
|
||||
return f"{name} {info.shape}" # fallback display
|
||||
elif role == QtCore.Qt.UserRole:
|
||||
return name
|
||||
elif role == QtCore.Qt.UserRole + 1:
|
||||
return num_points
|
||||
return info.shape
|
||||
return None
|
||||
|
||||
|
||||
# Custom delegate for better formatting
|
||||
class DeviceDelegate(QtWidgets.QStyledItemDelegate):
|
||||
class SignalDelegate(QtWidgets.QStyledItemDelegate):
|
||||
"""Custom delegate for displaying device names and points in the combo box."""
|
||||
|
||||
def paint(self, painter, option, index):
|
||||
name = index.data(QtCore.Qt.UserRole)
|
||||
points = index.data(QtCore.Qt.UserRole + 1)
|
||||
@@ -76,7 +86,7 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
|
||||
RPC = False
|
||||
PLUGIN = False
|
||||
|
||||
request_history_plot = QtCore.Signal(str, dict) # (str, ScanHistoryMessage.model_dump())
|
||||
request_history_plot = QtCore.Signal(str, str, str) # (scan_id, device_name, signal_name)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -97,54 +107,93 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
|
||||
)
|
||||
# Current scan history message
|
||||
self.scan_history_msg: ScanHistoryMessage | None = None
|
||||
self._selected_device: str = ""
|
||||
self._last_device_name: str | None = None
|
||||
self._last_signal_name: str | None = None
|
||||
# Init layout
|
||||
layout = QtWidgets.QHBoxLayout(self)
|
||||
layout = QtWidgets.QVBoxLayout(self)
|
||||
self.setLayout(layout)
|
||||
# Init ComboBox
|
||||
self.device_combo = QtWidgets.QComboBox(self)
|
||||
# Init widgets
|
||||
self.device_combo = QtWidgets.QComboBox(parent=self)
|
||||
self.signal_combo = QtWidgets.QComboBox(parent=self)
|
||||
colors = get_accent_colors()
|
||||
self.request_plotting_button = QtWidgets.QPushButton(
|
||||
material_icon("play_arrow", size=(24, 24), color=colors.success),
|
||||
"Request Plotting",
|
||||
self,
|
||||
)
|
||||
self.device_model = DeviceModel({})
|
||||
self.device_combo.setModel(self.device_model)
|
||||
layout.addWidget(self.device_combo)
|
||||
layout.addWidget(self.request_plotting_button)
|
||||
self.device_combo.setItemDelegate(DeviceDelegate())
|
||||
self.signal_model = SignalModel(parent=self.signal_combo)
|
||||
self.signal_combo.setModel(self.signal_model)
|
||||
self.signal_combo.setItemDelegate(SignalDelegate())
|
||||
self._init_layout()
|
||||
# Connect signals
|
||||
self.request_plotting_button.clicked.connect(self._on_request_plotting_clicked)
|
||||
self.device_combo.currentTextChanged.connect(self._signal_combo_update)
|
||||
|
||||
@SafeProperty(str)
|
||||
def device(self) -> str:
|
||||
"""Get the currently selected device name."""
|
||||
return self._selected_device
|
||||
def _init_layout(self):
|
||||
"""Initialize the layout for the device viewer."""
|
||||
main_layout = self.layout()
|
||||
main_layout.setContentsMargins(0, 0, 0, 0)
|
||||
main_layout.setSpacing(0)
|
||||
# horzizontal layout for device combo and signal combo boxes
|
||||
widget = QtWidgets.QWidget(self)
|
||||
hor_layout = QtWidgets.QHBoxLayout()
|
||||
hor_layout.setContentsMargins(0, 0, 0, 0)
|
||||
hor_layout.setSpacing(0)
|
||||
widget.setLayout(hor_layout)
|
||||
hor_layout.addWidget(self.device_combo)
|
||||
hor_layout.addWidget(self.signal_combo)
|
||||
main_layout.addWidget(widget)
|
||||
main_layout.addWidget(self.request_plotting_button)
|
||||
|
||||
@device.setter
|
||||
def device(self, value: str):
|
||||
"""Set the currently selected device name."""
|
||||
if not isinstance(value, str):
|
||||
logger.info(f"Device name must be a string {value}.")
|
||||
if value not in self.scan_history_msg.device_data_info:
|
||||
logger.info(f"Device name must in the list of selected devices {value}.")
|
||||
self._selected_device = value
|
||||
@SafeSlot(dict, dict)
|
||||
def update_devices_from_scan_history(self, msg: dict, metadata: dict | None = None) -> None:
|
||||
"""Update the device combo box with the scan history message.
|
||||
|
||||
@SafeSlot()
|
||||
def update_devices_from_scan_history(self, msg: ScanHistoryMessage) -> None:
|
||||
"""Update the device combo box with the scan history message."""
|
||||
if not isinstance(msg, ScanHistoryMessage):
|
||||
logger.info(f"Received message of type {type(msg)} instead of ScanHistoryMessage.")
|
||||
return
|
||||
Args:
|
||||
msg (ScanHistoryMessage): The scan history message containing device data.
|
||||
"""
|
||||
msg = ScanHistoryMessage(**msg)
|
||||
if metadata is not None:
|
||||
msg.metadata = metadata
|
||||
# Keep track of current device name
|
||||
self._last_device_name = self.device_combo.currentText()
|
||||
|
||||
current_signal_index = self.signal_combo.currentIndex()
|
||||
self._last_signal_name = self.signal_combo.model().data(
|
||||
self.signal_combo.model().index(current_signal_index, 0), QtCore.Qt.UserRole
|
||||
)
|
||||
# Update the scan history message
|
||||
self.scan_history_msg = msg
|
||||
self.device_model.devices = msg.device_data_info
|
||||
self.device_combo.clear()
|
||||
self.device_combo.addItems(msg.stored_data_info.keys())
|
||||
index = self.device_combo.findData(self._last_device_name, role=QtCore.Qt.DisplayRole)
|
||||
if index != -1:
|
||||
self.device_combo.setCurrentIndex(index)
|
||||
|
||||
@SafeSlot(str)
|
||||
def _signal_combo_update(self, device_name: str) -> None:
|
||||
"""Update the signal combo box based on the selected device."""
|
||||
if not self.scan_history_msg:
|
||||
logger.info("No scan history message available to update signals.")
|
||||
return
|
||||
if not device_name:
|
||||
return
|
||||
signal_data = self.scan_history_msg.stored_data_info.get(device_name, None)
|
||||
if signal_data is None:
|
||||
logger.info(f"No signal data found for device {device_name}.")
|
||||
return
|
||||
self.signal_model.signals = signal_data
|
||||
if self._last_signal_name is not None:
|
||||
# Try to restore the last selected signal
|
||||
index = self.signal_combo.findData(self._last_signal_name, role=QtCore.Qt.UserRole)
|
||||
if index != -1:
|
||||
self.signal_combo.setCurrentIndex(index)
|
||||
|
||||
@SafeSlot()
|
||||
def clear_view(self, msg: ScanHistoryMessage | None = None) -> None:
|
||||
def clear_view(self) -> None:
|
||||
"""Clear the device combo box."""
|
||||
self.scan_history_msg = None
|
||||
self.device_model.devices = {}
|
||||
self.signal_model.signals = {}
|
||||
self.device_combo.clear()
|
||||
|
||||
@SafeSlot()
|
||||
@@ -153,20 +202,31 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
|
||||
if self.scan_history_msg is None:
|
||||
logger.info("No scan history message available for plotting.")
|
||||
return
|
||||
current_index = self.device_combo.currentIndex()
|
||||
device_name = self.device_combo.model().data(
|
||||
self.device_combo.model().index(current_index, 0), QtCore.Qt.UserRole
|
||||
device_name = self.device_combo.currentText()
|
||||
|
||||
signal_index = self.signal_combo.currentIndex()
|
||||
signal_name = self.signal_combo.model().data(
|
||||
self.device_combo.model().index(signal_index, 0), QtCore.Qt.UserRole
|
||||
)
|
||||
logger.info(
|
||||
f"Requesting plotting for device: {device_name} with {self.scan_history_msg} points."
|
||||
f"Requesting plotting clicked: Scan ID:{self.scan_history_msg.scan_id}, device name: {device_name} with signal name: {signal_name}."
|
||||
)
|
||||
self.request_history_plot.emit((device_name, self.scan_history_msg))
|
||||
self.request_history_plot.emit(self.scan_history_msg.scan_id, device_name, signal_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import sys
|
||||
|
||||
app = QtWidgets.QApplication(sys.argv)
|
||||
|
||||
main_window = QtWidgets.QMainWindow()
|
||||
central_widget = QtWidgets.QWidget()
|
||||
main_window.setCentralWidget(central_widget)
|
||||
ly = QtWidgets.QVBoxLayout(central_widget)
|
||||
ly.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
viewer = ScanHistoryDeviceViewer()
|
||||
viewer.show()
|
||||
sys.exit(app.exec_())
|
||||
ly.addWidget(viewer)
|
||||
main_window.show()
|
||||
app.exec_()
|
||||
app.exec_()
|
||||
|
||||
@@ -1,19 +1,16 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ScanHistoryMessage
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy import QtCore, QtGui, QtWidgets
|
||||
from qtpy import QtGui, QtWidgets
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig
|
||||
from bec_widgets.utils.colors import get_theme_palette
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_lib.messages import ScanHistoryMessage
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
@@ -25,13 +22,24 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent: QtWidgets.QWidget | None = None,
|
||||
client=None,
|
||||
config: ConnectionConfig | None = None,
|
||||
gui_id: str | None = None,
|
||||
theme_update: bool = True,
|
||||
scan_history_msg: ScanHistoryMessage | None = None,
|
||||
parent=None,
|
||||
):
|
||||
"""
|
||||
Initialize the ScanHistoryMetadataViewer widget.
|
||||
|
||||
Args:
|
||||
parent (QtWidgets.QWidget, optional): The parent widget.
|
||||
client: The BEC client.
|
||||
config (ConnectionConfig, optional): The connection configuration.
|
||||
gui_id (str, optional): The GUI ID.
|
||||
theme_update (bool, optional): Whether to subscribe to theme updates. Defaults to True.
|
||||
scan_history_msg (ScanHistoryMessage, optional): The scan history message to display. Defaults
|
||||
"""
|
||||
super().__init__(
|
||||
parent=parent, client=client, config=config, gui_id=gui_id, theme_update=theme_update
|
||||
)
|
||||
@@ -47,14 +55,12 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
"num_points": "Nr of Points",
|
||||
}
|
||||
self.setTitle("No Scan Selected")
|
||||
# self.setMinimumHeight(100)
|
||||
# self.setSizePolicy(QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Expanding)
|
||||
layout = QtWidgets.QGridLayout()
|
||||
self.setLayout(layout)
|
||||
self._init_grid_layout()
|
||||
self.scan_history_msg = scan_history_msg
|
||||
if scan_history_msg is not None:
|
||||
self.update_view(self.scan_history_msg)
|
||||
self.update_view(self.scan_history_msg.content, self.scan_history_msg.metadata)
|
||||
self.apply_theme()
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
@@ -81,14 +87,13 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
v = self._scan_history_msg_labels[k]
|
||||
# Label for the key
|
||||
label = QtWidgets.QLabel(f"{v}:")
|
||||
# label.setSizePolicy(QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Preferred)
|
||||
layout.addWidget(label, row, 0)
|
||||
# Value field
|
||||
value_field = QtWidgets.QLabel("")
|
||||
value_field.setSizePolicy(
|
||||
QtWidgets.QSizePolicy.Ignored, QtWidgets.QSizePolicy.Preferred
|
||||
)
|
||||
layout.addWidget(value_field, row, 1) # Placeholder for value
|
||||
layout.addWidget(value_field, row, 1)
|
||||
# Copy button
|
||||
copy_button = QtWidgets.QToolButton()
|
||||
copy_button.setSizePolicy(QtWidgets.QSizePolicy.Fixed, QtWidgets.QSizePolicy.Minimum)
|
||||
@@ -97,24 +102,27 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
copy_button.setIcon(material_icon(icon_name="content_copy", size=(16, 16)))
|
||||
copy_button.setToolTip("Copy to clipboard")
|
||||
copy_button.setVisible(False)
|
||||
copy_button.setEnabled(False) # Initially disabled
|
||||
copy_button.setEnabled(False)
|
||||
copy_button.clicked.connect(
|
||||
lambda _, field=value_field: QtWidgets.QApplication.clipboard().setText(
|
||||
field.text()
|
||||
)
|
||||
)
|
||||
layout.addWidget(copy_button, row, 2) # Placeholder for copy button
|
||||
layout.addWidget(copy_button, row, 2)
|
||||
|
||||
@SafeSlot()
|
||||
def update_view(self, msg: ScanHistoryMessage):
|
||||
@SafeSlot(dict, dict)
|
||||
def update_view(self, msg: dict, metadata: dict | None = None) -> None:
|
||||
"""
|
||||
Update the view with the given ScanHistoryMessage.
|
||||
|
||||
Args:
|
||||
msg (ScanHistoryMessage): The message containing scan metadata.
|
||||
"""
|
||||
msg = ScanHistoryMessage(**msg)
|
||||
if metadata is not None:
|
||||
msg.metadata = metadata
|
||||
if msg == self.scan_history_msg:
|
||||
return # Same message, no update needed
|
||||
return
|
||||
self.scan_history_msg = msg
|
||||
layout = self.layout()
|
||||
if layout.count() == 0:
|
||||
@@ -139,7 +147,7 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
logger.warning(f"ScanHistoryMessage missing value for {k} and msg {msg}.")
|
||||
continue
|
||||
layout.itemAtPosition(row, 1).widget().setText(str(value))
|
||||
if k in ["file_path", "scan_id"]:
|
||||
if k in ["file_path", "scan_id"]: # Enable copy for file path and scan ID
|
||||
layout.itemAtPosition(row, 2).widget().setVisible(True)
|
||||
layout.itemAtPosition(row, 2).widget().setEnabled(True)
|
||||
else:
|
||||
@@ -147,12 +155,10 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
layout.itemAtPosition(row, 2).widget().setToolTip("")
|
||||
|
||||
@SafeSlot()
|
||||
def clear_view(self, msg: ScanHistoryMessage | None = None):
|
||||
def clear_view(self):
|
||||
"""
|
||||
Clear the view by resetting the labels and values.
|
||||
"""
|
||||
if self.scan_history_msg is not None and msg != self.scan_history_msg:
|
||||
return
|
||||
layout = self.layout()
|
||||
lauout_counts = layout.count()
|
||||
for i in range(lauout_counts):
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from bec_lib.callback_handler import EventType
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ScanHistoryMessage
|
||||
from qtpy import QtCore, QtGui, QtWidgets
|
||||
@@ -8,19 +10,57 @@ from qtpy import QtCore, QtGui, QtWidgets
|
||||
from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from bec_lib.client import BECClient
|
||||
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class BECHistoryManager(QtCore.QObject):
|
||||
"""History manager for scan history operations. This class
|
||||
is responsible for emitting signals when the scan history is updated.
|
||||
"""
|
||||
|
||||
# ScanHistoryMessage.model_dump() (dict)
|
||||
scan_history_updated = QtCore.Signal(dict)
|
||||
|
||||
def __init__(self, parent, client: BECClient):
|
||||
super().__init__(parent)
|
||||
self.client = client
|
||||
self._cb_id = self.client.callbacks.register(
|
||||
event_type=EventType.SCAN_HISTORY_UPDATE, callback=self._on_scan_history_update
|
||||
)
|
||||
|
||||
def refresh_scan_history(self) -> None:
|
||||
"""Refresh the scan history from the client."""
|
||||
for scan_id in self.client.history._scan_ids: # pylint: disable=protected-access
|
||||
history_msg = self.client.history._scan_data.get(scan_id, None)
|
||||
if history_msg is None:
|
||||
logger.info(f"Scan history message for scan_id {scan_id} not found.")
|
||||
continue
|
||||
self.scan_history_updated.emit(history_msg.model_dump())
|
||||
|
||||
def _on_scan_history_update(self, history_msg: ScanHistoryMessage) -> None:
|
||||
"""Handle scan history updates from the client."""
|
||||
self.scan_history_updated.emit(history_msg.model_dump())
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Clean up the manager by disconnecting callbacks."""
|
||||
self.client.callbacks.remove(self._cb_id)
|
||||
self.scan_history_updated.disconnect()
|
||||
|
||||
|
||||
class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
"""ScanHistoryTree is a widget that displays the scan history in a tree format."""
|
||||
|
||||
RPC = False
|
||||
PLUGIN = False
|
||||
|
||||
scan_selected = QtCore.Signal(object)
|
||||
scan_removed = QtCore.Signal(object)
|
||||
# ScanHistoryMessage.content, ScanHistoryMessage.metadata
|
||||
scan_selected = QtCore.Signal(dict, dict)
|
||||
no_scan_selected = QtCore.Signal()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -50,13 +90,17 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
self.column_header = ["Scan Nr", "Scan Name", "Status"]
|
||||
self.scan_history: list[ScanHistoryMessage] = [] # newest at index 0
|
||||
self.max_length = max_length # Maximum number of scan history entries to keep
|
||||
self.bec_scan_history_manager = BECHistoryManager(parent=self, client=self.client)
|
||||
self._set_policies()
|
||||
self.apply_theme()
|
||||
self._start_subscription()
|
||||
self.itemClicked.connect(self._on_item_clicked)
|
||||
self.currentItemChanged.connect(self._current_item_changed)
|
||||
header = self.header()
|
||||
header.setToolTip(f"Last {self.max_length} scans in history.")
|
||||
self.bec_scan_history_manager.scan_history_updated.connect(self.update_history)
|
||||
self.refresh()
|
||||
|
||||
def _set_policies(self):
|
||||
"""Set the policies for the tree widget."""
|
||||
self.setColumnCount(len(self.column_header))
|
||||
self.setHeaderLabels(self.column_header)
|
||||
self.setRootIsDecorated(False) # allow expand arrow for per‑scan details
|
||||
@@ -74,6 +118,7 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
header.setSectionResizeMode(column, QtWidgets.QHeaderView.ResizeMode.Stretch)
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
"""Apply the theme to the widget."""
|
||||
colors = get_accent_colors()
|
||||
self.status_colors = {
|
||||
"closed": colors.success,
|
||||
@@ -87,45 +132,35 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
):
|
||||
"""
|
||||
Handle current item change events in the tree widget.
|
||||
Emits a signal with the selected scan message when the current item changes.
|
||||
|
||||
Args:
|
||||
current (QtWidgets.QTreeWidgetItem): The currently selected item.
|
||||
previous (QtWidgets.QTreeWidgetItem): The previously selected item.
|
||||
"""
|
||||
if not current:
|
||||
return
|
||||
self._on_item_clicked(current, self.currentColumn())
|
||||
|
||||
def _on_item_clicked(self, item: QtWidgets.QTreeWidgetItem, column: int):
|
||||
"""
|
||||
Handle item click events in the tree widget.
|
||||
Emits a signal with the selected scan message when an item is clicked.
|
||||
"""
|
||||
if not item:
|
||||
return
|
||||
index = self.indexOfTopLevelItem(item)
|
||||
self.scan_selected.emit(self.scan_history[index])
|
||||
|
||||
def _start_subscription(self):
|
||||
"""
|
||||
Subscribe to scan history updates.
|
||||
"""
|
||||
self.bec_dispatcher.connect_slot(
|
||||
slot=self.update_history, topics=MessageEndpoints.scan_history(), from_start=True
|
||||
)
|
||||
index = self.indexOfTopLevelItem(current)
|
||||
self.scan_selected.emit(self.scan_history[index].content, self.scan_history[index].metadata)
|
||||
|
||||
@SafeSlot()
|
||||
def update_history(self, msg_content: dict, metdata: dict):
|
||||
"""
|
||||
This method is called whenever a new scan history is available.
|
||||
"""
|
||||
# TODO directly receive ScanHistoryMessage through dispatcher
|
||||
msg = ScanHistoryMessage(**msg_content)
|
||||
msg.metadata = metdata
|
||||
def refresh(self):
|
||||
"""Refresh the scan history view."""
|
||||
while len(self.scan_history) > 0:
|
||||
self.remove_scan(index=0)
|
||||
self.bec_scan_history_manager.refresh_scan_history()
|
||||
|
||||
@SafeSlot(dict)
|
||||
def update_history(self, msg_dump: dict):
|
||||
"""Update the scan history with new scan data."""
|
||||
msg = ScanHistoryMessage(**msg_dump)
|
||||
self.add_scan(msg)
|
||||
self.ensure_history_max_length()
|
||||
|
||||
def ensure_history_max_length(self) -> None:
|
||||
"""
|
||||
Clean up the scan history by clearing the list.
|
||||
This method can be called when the widget is closed or no longer needed.
|
||||
Method to ensure the scan history does not exceed the maximum length.
|
||||
If the length exceeds the maximum, it removes the oldest entry.
|
||||
This is called after adding a new scan to the history.
|
||||
"""
|
||||
while len(self.scan_history) > self.max_length:
|
||||
logger.warning(
|
||||
@@ -136,9 +171,18 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
def add_scan(self, msg: ScanHistoryMessage):
|
||||
"""
|
||||
Add a scan entry to the tree widget.
|
||||
|
||||
Args:
|
||||
msg (ScanHistoryMessage): The scan history message containing scan details.
|
||||
"""
|
||||
if msg.stored_data_info is None:
|
||||
logger.info(
|
||||
f"Old scan history entry fo scan {msg.scan_id} without stored_data_info, skipping."
|
||||
)
|
||||
return
|
||||
if msg in self.scan_history:
|
||||
logger.info(f"Scan {msg.scan_id} already in history, skipping.")
|
||||
return
|
||||
self.scan_history.insert(0, msg)
|
||||
tree_item = QtWidgets.QTreeWidgetItem([str(msg.scan_number), msg.scan_name, ""])
|
||||
color = QtGui.QColor(self.status_colors.get(msg.exit_status, "#b0bec5"))
|
||||
@@ -158,7 +202,9 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
|
||||
def remove_scan(self, index: int):
|
||||
"""
|
||||
Remove a scan entry from the tree widget. We supoprt negative indexing where -1, -2, etc. refer to the last, second last, etc. entry.
|
||||
Remove a scan entry from the tree widget.
|
||||
We supoprt negative indexing where -1, -2, etc.
|
||||
|
||||
Args:
|
||||
index (int): The index of the scan entry to remove.
|
||||
"""
|
||||
@@ -166,12 +212,17 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
|
||||
index = len(self.scan_history) + index
|
||||
try:
|
||||
msg = self.scan_history.pop(index)
|
||||
self.scan_removed.emit(msg)
|
||||
self.no_scan_selected.emit()
|
||||
except IndexError:
|
||||
logger.warning(f"Invalid index {index} for removing scan entry from history.")
|
||||
return
|
||||
self.takeTopLevelItem(index)
|
||||
|
||||
def cleanup(self):
|
||||
"""Cleanup the widget"""
|
||||
self.bec_scan_history_manager.cleanup()
|
||||
super().cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
# pylint: disable=import-outside-toplevel
|
||||
@@ -203,8 +254,8 @@ if __name__ == "__main__": # pragma: no cover
|
||||
layout.addWidget(device_viewer)
|
||||
browser.scan_selected.connect(view.update_view)
|
||||
browser.scan_selected.connect(device_viewer.update_devices_from_scan_history)
|
||||
browser.scan_removed.connect(view.clear_view)
|
||||
browser.scan_removed.connect(device_viewer.clear_view)
|
||||
browser.no_scan_selected.connect(view.clear_view)
|
||||
browser.no_scan_selected.connect(device_viewer.clear_view)
|
||||
|
||||
main_window.show()
|
||||
app.exec_()
|
||||
|
||||
@@ -9,6 +9,11 @@ from bec_widgets.widgets.services.scan_history_browser.components import (
|
||||
|
||||
|
||||
class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
|
||||
"""
|
||||
ScanHistoryBrowser is a widget combining the scan history view, metadata viewer, and device viewer.
|
||||
|
||||
Target is to provide a popup view for the Waveform Widget to browse the scan history.
|
||||
"""
|
||||
|
||||
RPC = False
|
||||
PLUGIN = False
|
||||
@@ -22,6 +27,16 @@ class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
|
||||
theme_update: bool = False,
|
||||
**kwargs,
|
||||
):
|
||||
"""
|
||||
Initialize the ScanHistoryBrowser widget.
|
||||
|
||||
Args:
|
||||
parent (QtWidgets.QWidget, optional): The parent widget.
|
||||
client: The BEC client.
|
||||
config (ConnectionConfig, optional): The connection configuration.
|
||||
gui_id (str, optional): The GUI ID.
|
||||
theme_update (bool, optional): Whether to subscribe to theme updates. Defaults to False.
|
||||
"""
|
||||
super().__init__(
|
||||
parent=parent,
|
||||
client=client,
|
||||
@@ -31,8 +46,8 @@ class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
|
||||
**kwargs,
|
||||
)
|
||||
layout = QtWidgets.QHBoxLayout()
|
||||
|
||||
self.setLayout(layout)
|
||||
|
||||
self.scan_history_view = ScanHistoryView(
|
||||
parent=self, client=client, config=config, gui_id=gui_id, theme_update=theme_update
|
||||
)
|
||||
@@ -45,10 +60,9 @@ class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
self.init_layout()
|
||||
self.connect_signals()
|
||||
QtCore.QTimer.singleShot(0, self.select_first_history_entry)
|
||||
|
||||
def init_layout(self):
|
||||
"""Initialize the layout of the widget."""
|
||||
"""Initialize compact layout for the widget."""
|
||||
# Add Scan history view
|
||||
layout: QtWidgets.QHBoxLayout = self.layout()
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
@@ -66,19 +80,15 @@ class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
|
||||
layout.addWidget(widget)
|
||||
|
||||
def connect_signals(self):
|
||||
"""Connect signals to the appropriate slots."""
|
||||
"""Connect signals from scan history components."""
|
||||
self.scan_history_view.scan_selected.connect(self.scan_history_metadata_viewer.update_view)
|
||||
self.scan_history_view.scan_selected.connect(
|
||||
self.scan_history_device_viewer.update_devices_from_scan_history
|
||||
)
|
||||
self.scan_history_view.scan_removed.connect(self.scan_history_metadata_viewer.clear_view)
|
||||
self.scan_history_view.scan_removed.connect(self.scan_history_device_viewer.clear_view)
|
||||
|
||||
def select_first_history_entry(self):
|
||||
"""Select the first entry in the scan history view."""
|
||||
if self.scan_history_view.topLevelItemCount() > 0:
|
||||
self.scan_history_view.setCurrentItem(self.scan_history_view.topLevelItem(0))
|
||||
self.scan_history_view.itemActivated.emit(self.scan_history_view.topLevelItem(0), 0)
|
||||
self.scan_history_view.no_scan_selected.connect(
|
||||
self.scan_history_metadata_viewer.clear_view
|
||||
)
|
||||
self.scan_history_view.no_scan_selected.connect(self.scan_history_device_viewer.clear_view)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
@@ -100,6 +110,6 @@ if __name__ == "__main__": # pragma: no cover
|
||||
layout.addWidget(button)
|
||||
layout.addWidget(browser)
|
||||
main_window.setWindowTitle("Scan History Browser")
|
||||
main_window.resize(800, 600)
|
||||
main_window.resize(800, 400)
|
||||
main_window.show()
|
||||
app.exec_()
|
||||
|
||||
380
tests/unit_tests/test_scan_history_browser.py
Normal file
380
tests/unit_tests/test_scan_history_browser.py
Normal file
@@ -0,0 +1,380 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_lib.messages import ScanHistoryMessage, _StoredDataInfo
|
||||
from pytestqt import qtbot
|
||||
from qtpy import QtCore
|
||||
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.widgets.services.scan_history_browser.components import (
|
||||
ScanHistoryDeviceViewer,
|
||||
ScanHistoryMetadataViewer,
|
||||
ScanHistoryView,
|
||||
)
|
||||
from bec_widgets.widgets.services.scan_history_browser.scan_history_browser import (
|
||||
ScanHistoryBrowser,
|
||||
)
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_msg():
|
||||
"""Fixture to create a mock ScanHistoryMessage."""
|
||||
yield ScanHistoryMessage(
|
||||
scan_id="test_scan",
|
||||
dataset_number=1,
|
||||
scan_number=1,
|
||||
scan_name="Test Scan",
|
||||
file_path="/path/to/scan",
|
||||
start_time=1751957906.3310962,
|
||||
end_time=1751957907.3310962, # 1s later
|
||||
exit_status="closed",
|
||||
num_points=10,
|
||||
request_inputs={"some_input": "value"},
|
||||
stored_data_info={
|
||||
"device2": {
|
||||
"device2_signal1": _StoredDataInfo(shape=(10,)),
|
||||
"device2_signal2": _StoredDataInfo(shape=(20,)),
|
||||
"device2_signal3": _StoredDataInfo(shape=(25,)),
|
||||
},
|
||||
"device3": {"device3_signal1": _StoredDataInfo(shape=(1,))},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_msg_2():
|
||||
"""Fixture to create a second mock ScanHistoryMessage."""
|
||||
yield ScanHistoryMessage(
|
||||
scan_id="test_scan_2",
|
||||
dataset_number=2,
|
||||
scan_number=2,
|
||||
scan_name="Test Scan 2",
|
||||
file_path="/path/to/scan_2",
|
||||
start_time=1751957908.3310962,
|
||||
end_time=1751957909.3310962, # 1s later
|
||||
exit_status="closed",
|
||||
num_points=5,
|
||||
request_inputs={"some_input": "new_value"},
|
||||
stored_data_info={
|
||||
"device0": {
|
||||
"device0_signal1": _StoredDataInfo(shape=(15,)),
|
||||
"device0_signal2": _StoredDataInfo(shape=(25,)),
|
||||
"device0_signal3": _StoredDataInfo(shape=(3,)),
|
||||
"device0_signal4": _StoredDataInfo(shape=(20,)),
|
||||
},
|
||||
"device2": {
|
||||
"device2_signal1": _StoredDataInfo(shape=(10,)),
|
||||
"device2_signal2": _StoredDataInfo(shape=(20,)),
|
||||
"device2_signal3": _StoredDataInfo(shape=(25,)),
|
||||
"device2_signal4": _StoredDataInfo(shape=(30,)),
|
||||
},
|
||||
"device1": {"device1_signal1": _StoredDataInfo(shape=(25,))},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_device_viewer(qtbot, mocked_client):
|
||||
widget = ScanHistoryDeviceViewer(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_metadata_viewer(qtbot, mocked_client):
|
||||
widget = ScanHistoryMetadataViewer(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_view(qtbot, mocked_client):
|
||||
widget = ScanHistoryView(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_history_browser(qtbot, mocked_client):
|
||||
"""Fixture to create a ScanHistoryBrowser widget."""
|
||||
widget = ScanHistoryBrowser(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
def test_scan_history_device_viewer_receive_msg(
|
||||
qtbot, scan_history_device_viewer, scan_history_msg, scan_history_msg_2
|
||||
):
|
||||
"""Test updating devices from scan history."""
|
||||
# Update with first scan history message
|
||||
assert scan_history_device_viewer.scan_history_msg is None
|
||||
assert scan_history_device_viewer.signal_model.signals == []
|
||||
assert scan_history_device_viewer.signal_model.rowCount() == 0
|
||||
scan_history_device_viewer.update_devices_from_scan_history(
|
||||
scan_history_msg.content, scan_history_msg.metadata
|
||||
)
|
||||
assert scan_history_device_viewer.scan_history_msg == scan_history_msg
|
||||
assert scan_history_device_viewer.device_combo.currentText() == "device2"
|
||||
assert scan_history_device_viewer.signal_model.signals == [
|
||||
("device2_signal3", _StoredDataInfo(shape=(25,))),
|
||||
("device2_signal2", _StoredDataInfo(shape=(20,))),
|
||||
("device2_signal1", _StoredDataInfo(shape=(10,))),
|
||||
]
|
||||
current_index = scan_history_device_viewer.signal_combo.currentIndex()
|
||||
assert current_index == 0
|
||||
signal_name = scan_history_device_viewer.signal_combo.model().data(
|
||||
scan_history_device_viewer.signal_combo.model().index(current_index, 0), QtCore.Qt.UserRole
|
||||
)
|
||||
assert signal_name == "device2_signal3"
|
||||
|
||||
## Update of second message should not change the device if still available
|
||||
new_msg = scan_history_msg_2
|
||||
scan_history_device_viewer.update_devices_from_scan_history(new_msg.content, new_msg.metadata)
|
||||
assert scan_history_device_viewer.scan_history_msg == new_msg
|
||||
assert scan_history_device_viewer.signal_model.signals == [
|
||||
("device2_signal4", _StoredDataInfo(shape=(30,))),
|
||||
("device2_signal3", _StoredDataInfo(shape=(25,))),
|
||||
("device2_signal2", _StoredDataInfo(shape=(20,))),
|
||||
("device2_signal1", _StoredDataInfo(shape=(10,))),
|
||||
]
|
||||
assert scan_history_device_viewer.device_combo.currentText() == "device2"
|
||||
current_index = scan_history_device_viewer.signal_combo.currentIndex()
|
||||
assert current_index == 1
|
||||
signal_name = scan_history_device_viewer.signal_combo.model().data(
|
||||
scan_history_device_viewer.signal_combo.model().index(current_index, 0), QtCore.Qt.UserRole
|
||||
)
|
||||
assert signal_name == "device2_signal3"
|
||||
|
||||
|
||||
def test_scan_history_device_viewer_clear_view(qtbot, scan_history_device_viewer, scan_history_msg):
|
||||
"""Test clearing the device viewer."""
|
||||
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg.content)
|
||||
assert scan_history_device_viewer.scan_history_msg == scan_history_msg
|
||||
scan_history_device_viewer.clear_view()
|
||||
assert scan_history_device_viewer.scan_history_msg is None
|
||||
assert scan_history_device_viewer.device_combo.model().rowCount() == 0
|
||||
|
||||
|
||||
def test_scan_history_device_viewer_on_request_plotting_clicked(
|
||||
qtbot, scan_history_device_viewer, scan_history_msg
|
||||
):
|
||||
"""Test the request plotting button click."""
|
||||
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg.content)
|
||||
|
||||
plotting_callback_args = []
|
||||
|
||||
def plotting_callback(device_name, signal_name, msg):
|
||||
"""Callback to check if the request plotting signal is emitted."""
|
||||
plotting_callback_args.append((device_name, signal_name, msg))
|
||||
|
||||
scan_history_device_viewer.request_history_plot.connect(plotting_callback)
|
||||
qtbot.mouseClick(scan_history_device_viewer.request_plotting_button, QtCore.Qt.LeftButton)
|
||||
qtbot.waitUntil(lambda: len(plotting_callback_args) > 0, timeout=5000)
|
||||
# scan_id
|
||||
assert plotting_callback_args[0][0] == scan_history_msg.scan_id
|
||||
# device_name
|
||||
assert plotting_callback_args[0][1] in scan_history_msg.stored_data_info.keys()
|
||||
# signal_name
|
||||
assert (
|
||||
plotting_callback_args[0][2]
|
||||
in scan_history_msg.stored_data_info[plotting_callback_args[0][1]].keys()
|
||||
)
|
||||
|
||||
|
||||
def test_scan_history_metadata_viewer_receive_msg(
|
||||
qtbot, scan_history_metadata_viewer, scan_history_msg
|
||||
):
|
||||
"""Test the initialization of ScanHistoryMetadataViewer."""
|
||||
assert scan_history_metadata_viewer.scan_history_msg is None
|
||||
assert scan_history_metadata_viewer.title() == "No Scan Selected"
|
||||
scan_history_metadata_viewer.update_view(scan_history_msg.content)
|
||||
assert scan_history_metadata_viewer.scan_history_msg == scan_history_msg
|
||||
assert scan_history_metadata_viewer.title() == f"Metadata - Scan {scan_history_msg.scan_number}"
|
||||
for row, k in enumerate(scan_history_metadata_viewer._scan_history_msg_labels.keys()):
|
||||
if k == "elapsed_time":
|
||||
scan_history_metadata_viewer.layout().itemAtPosition(row, 1).widget().text() == "1.000s"
|
||||
if k == "scan_name":
|
||||
scan_history_metadata_viewer.layout().itemAtPosition(
|
||||
row, 1
|
||||
).widget().text() == "Test Scan"
|
||||
|
||||
|
||||
def test_scan_history_metadata_viewer_clear_view(
|
||||
qtbot, scan_history_metadata_viewer, scan_history_msg
|
||||
):
|
||||
"""Test clearing the metadata viewer."""
|
||||
scan_history_metadata_viewer.update_view(scan_history_msg.content)
|
||||
assert scan_history_metadata_viewer.scan_history_msg == scan_history_msg
|
||||
scan_history_metadata_viewer.clear_view()
|
||||
assert scan_history_metadata_viewer.scan_history_msg is None
|
||||
assert scan_history_metadata_viewer.title() == "No Scan Selected"
|
||||
|
||||
|
||||
def test_scan_history_view(qtbot, scan_history_view, scan_history_msg):
|
||||
"""Test the initialization of ScanHistoryView."""
|
||||
assert scan_history_view.scan_history == []
|
||||
assert scan_history_view.topLevelItemCount() == 0
|
||||
header = scan_history_view.headerItem()
|
||||
assert [header.text(i) for i in range(header.columnCount())] == [
|
||||
"Scan Nr",
|
||||
"Scan Name",
|
||||
"Status",
|
||||
]
|
||||
|
||||
|
||||
def test_scan_history_view_add_remove_scan(qtbot, scan_history_view, scan_history_msg):
|
||||
"""Test adding a scan to the ScanHistoryView."""
|
||||
scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
assert len(scan_history_view.scan_history) == 1
|
||||
assert scan_history_view.scan_history[0] == scan_history_msg
|
||||
assert scan_history_view.topLevelItemCount() == 1
|
||||
tree_item = scan_history_view.topLevelItem(0)
|
||||
tree_item.text(0) == str(scan_history_msg.scan_number)
|
||||
tree_item.text(1) == scan_history_msg.scan_name
|
||||
tree_item.text(2) == ""
|
||||
|
||||
# remove scan
|
||||
def remove_callback(msg):
|
||||
"""Callback to check if the no_scan_selected signal is emitted."""
|
||||
assert msg == scan_history_msg
|
||||
|
||||
scan_history_view.remove_scan(0)
|
||||
assert len(scan_history_view.scan_history) == 0
|
||||
assert scan_history_view.topLevelItemCount() == 0
|
||||
|
||||
|
||||
def test_scan_history_view_current_scan_item_changed(
|
||||
qtbot, scan_history_view, scan_history_msg, scan_history_device_viewer
|
||||
):
|
||||
"""Test the current scan item changed signal."""
|
||||
scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
scan_history_msg.scan_id = "test_scan_2"
|
||||
scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
scan_history_msg.scan_id = "test_scan_3"
|
||||
scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
assert len(scan_history_view.scan_history) == 3
|
||||
|
||||
def scan_selected_callback(msg):
|
||||
"""Callback to check if the scan_selected signal is emitted."""
|
||||
return msg == scan_history_msg
|
||||
|
||||
scan_history_view.scan_selected.connect(scan_selected_callback)
|
||||
|
||||
qtbot.mouseClick(
|
||||
scan_history_view.viewport(),
|
||||
QtCore.Qt.LeftButton,
|
||||
pos=scan_history_view.visualItemRect(scan_history_view.topLevelItem(0)).center(),
|
||||
)
|
||||
|
||||
|
||||
def test_scan_history_view_refresh(qtbot, scan_history_view, scan_history_msg, scan_history_msg_2):
|
||||
"""Test the refresh method of ScanHistoryView."""
|
||||
scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
scan_history_view.update_history(scan_history_msg_2.model_dump())
|
||||
assert len(scan_history_view.scan_history) == 2
|
||||
with mock.patch.object(
|
||||
scan_history_view.bec_scan_history_manager, "refresh_scan_history"
|
||||
) as mock_refresh:
|
||||
scan_history_view.refresh()
|
||||
mock_refresh.assert_called_once()
|
||||
assert len(scan_history_view.scan_history) == 0
|
||||
assert scan_history_view.topLevelItemCount() == 0
|
||||
|
||||
|
||||
def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, scan_history_msg_2):
|
||||
"""Test the initialization of ScanHistoryBrowser."""
|
||||
assert isinstance(scan_history_browser.scan_history_view, ScanHistoryView)
|
||||
assert isinstance(scan_history_browser.scan_history_metadata_viewer, ScanHistoryMetadataViewer)
|
||||
assert isinstance(scan_history_browser.scan_history_device_viewer, ScanHistoryDeviceViewer)
|
||||
|
||||
# Add 2 scans to the history browser, new item will be added to the top
|
||||
scan_history_browser.scan_history_view.update_history(scan_history_msg.model_dump())
|
||||
scan_history_browser.scan_history_view.update_history(scan_history_msg_2.model_dump())
|
||||
|
||||
assert len(scan_history_browser.scan_history_view.scan_history) == 2
|
||||
# Click on first scan item history to select it
|
||||
qtbot.mouseClick(
|
||||
scan_history_browser.scan_history_view.viewport(),
|
||||
QtCore.Qt.LeftButton,
|
||||
pos=scan_history_browser.scan_history_view.visualItemRect(
|
||||
scan_history_browser.scan_history_view.topLevelItem(0)
|
||||
).center(),
|
||||
)
|
||||
assert scan_history_browser.scan_history_view.currentIndex().row() == 0
|
||||
|
||||
# Both metadata and device viewers should be updated with the first scan
|
||||
qtbot.waitUntil(
|
||||
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
|
||||
== scan_history_msg_2,
|
||||
timeout=2000,
|
||||
)
|
||||
qtbot.waitUntil(
|
||||
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
|
||||
== scan_history_msg_2,
|
||||
timeout=2000,
|
||||
)
|
||||
|
||||
# TODO #771 ; Multiple clicks to the QTreeView item fail, but only in the CI, not locally.
|
||||
# Click on second scan item history to select it
|
||||
# qtbot.mouseClick(
|
||||
# scan_history_browser.scan_history_view.viewport(),
|
||||
# QtCore.Qt.LeftButton,
|
||||
# pos=scan_history_browser.scan_history_view.visualItemRect(
|
||||
# scan_history_browser.scan_history_view.topLevelItem(1)
|
||||
# ).center(),
|
||||
# )
|
||||
# assert scan_history_browser.scan_history_view.currentIndex().row() == 1
|
||||
|
||||
# # Both metadata and device viewers should be updated with the first scan
|
||||
# qtbot.waitUntil(
|
||||
# lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
|
||||
# == scan_history_msg,
|
||||
# timeout=2000,
|
||||
# )
|
||||
# qtbot.waitUntil(
|
||||
# lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
|
||||
# == scan_history_msg,
|
||||
# timeout=2000,
|
||||
# )
|
||||
|
||||
callback_args = []
|
||||
|
||||
def plotting_callback(device_name, signal_name, msg):
|
||||
"""Callback to check if the request plotting signal is emitted."""
|
||||
# device_name should be the first device
|
||||
callback_args.append((device_name, signal_name, msg))
|
||||
|
||||
scan_history_browser.scan_history_device_viewer.request_history_plot.connect(plotting_callback)
|
||||
# Test emit plotting request
|
||||
qtbot.mouseClick(
|
||||
scan_history_browser.scan_history_device_viewer.request_plotting_button,
|
||||
QtCore.Qt.LeftButton,
|
||||
)
|
||||
qtbot.waitUntil(lambda: len(callback_args) > 0, timeout=5000)
|
||||
assert callback_args[0][0] == scan_history_msg_2.scan_id
|
||||
device_name = callback_args[0][1]
|
||||
signal_name = callback_args[0][2]
|
||||
assert device_name in scan_history_msg_2.stored_data_info.keys()
|
||||
assert signal_name in scan_history_msg_2.stored_data_info[device_name].keys()
|
||||
|
||||
# Test clearing the view, removing both scans
|
||||
scan_history_browser.scan_history_view.remove_scan(-1)
|
||||
scan_history_browser.scan_history_view.remove_scan(-1)
|
||||
|
||||
assert len(scan_history_browser.scan_history_view.scan_history) == 0
|
||||
assert scan_history_browser.scan_history_view.topLevelItemCount() == 0
|
||||
|
||||
qtbot.waitUntil(
|
||||
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg is None,
|
||||
timeout=2000,
|
||||
)
|
||||
qtbot.waitUntil(
|
||||
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg is None,
|
||||
timeout=2000,
|
||||
)
|
||||
Reference in New Issue
Block a user