mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-13 11:11:49 +02:00
feat(scan-history-browser): Add history browser and history metadata viewer
This commit is contained in:
@ -0,0 +1,198 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ScanHistoryMessage
|
||||
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
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class ScanHistoryBrowser(BECWidget, QtWidgets.QTreeWidget):
|
||||
"""ScanHistoryTree is a widget that displays the scan history in a tree format."""
|
||||
|
||||
scan_selected = QtCore.Signal(object)
|
||||
scan_removed = QtCore.Signal(object)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent: QtWidgets.QWidget = None,
|
||||
client=None,
|
||||
config: ConnectionConfig = None,
|
||||
gui_id: str = None,
|
||||
max_length: int = 100,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(
|
||||
parent=parent, client=client, config=config, gui_id=gui_id, theme_update=True, **kwargs
|
||||
)
|
||||
colors = get_accent_colors()
|
||||
self.status_colors = {
|
||||
"closed": colors.success,
|
||||
"halted": colors.warning,
|
||||
"aborted": colors.emergency,
|
||||
}
|
||||
# self.status_colors = {"closed": "#00e676", "halted": "#ffca28", "aborted": "#ff5252"}
|
||||
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._set_policies()
|
||||
self.apply_theme()
|
||||
self._start_subscription()
|
||||
self.itemActivated.connect(self._on_item_clicked)
|
||||
self.itemClicked.connect(self._on_item_clicked)
|
||||
|
||||
def keyPressEvent(self, event):
|
||||
"""Handle Enter key press to activate emitting the selected item."""
|
||||
if event.key() in (QtCore.Qt.Key_Enter, QtCore.Qt.Key_Return):
|
||||
item = self.currentItem()
|
||||
if item:
|
||||
self.itemActivated.emit(item, self.currentColumn())
|
||||
else:
|
||||
super().keyPressEvent(event)
|
||||
|
||||
def _set_policies(self):
|
||||
self.setColumnCount(len(self.column_header))
|
||||
self.setHeaderLabels(self.column_header)
|
||||
self.setRootIsDecorated(False) # allow expand arrow for per‑scan details
|
||||
self.setUniformRowHeights(True)
|
||||
self.setFocusPolicy(QtCore.Qt.StrongFocus)
|
||||
self.setAlternatingRowColors(True)
|
||||
self.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection)
|
||||
self.setIndentation(12)
|
||||
self.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)
|
||||
self.setAnimated(True)
|
||||
|
||||
header = self.header()
|
||||
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeMode.ResizeToContents)
|
||||
for column in range(1, self.columnCount()):
|
||||
header.setSectionResizeMode(column, QtWidgets.QHeaderView.ResizeMode.Stretch)
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
colors = get_accent_colors()
|
||||
self.status_colors = {
|
||||
"closed": colors.success,
|
||||
"halted": colors.warning,
|
||||
"aborted": colors.emergency,
|
||||
}
|
||||
self.repaint()
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
@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
|
||||
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.
|
||||
"""
|
||||
while len(self.scan_history) > self.max_length:
|
||||
logger.warning(
|
||||
f"Removing oldest scan history entry to maintain max length of {self.max_length}."
|
||||
)
|
||||
self.remove_scan(index=-1)
|
||||
|
||||
def add_scan(self, msg: ScanHistoryMessage):
|
||||
"""
|
||||
Add a scan entry to the tree widget.
|
||||
Args:
|
||||
msg (ScanHistoryMessage): The scan history message containing scan details.
|
||||
"""
|
||||
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"))
|
||||
pix = QtGui.QPixmap(10, 10)
|
||||
pix.fill(QtCore.Qt.transparent)
|
||||
with QtGui.QPainter(pix) as p:
|
||||
p.setRenderHint(QtGui.QPainter.Antialiasing)
|
||||
p.setPen(QtCore.Qt.NoPen)
|
||||
p.setBrush(color)
|
||||
p.drawEllipse(0, 0, 10, 10)
|
||||
tree_item.setIcon(2, QtGui.QIcon(pix))
|
||||
tree_item.setForeground(2, QtGui.QBrush(color))
|
||||
for col in range(tree_item.columnCount()):
|
||||
tree_item.setToolTip(col, f"Status: {msg.exit_status}")
|
||||
self.insertTopLevelItem(0, tree_item)
|
||||
tree_item.setExpanded(False)
|
||||
|
||||
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.
|
||||
Args:
|
||||
index (int): The index of the scan entry to remove.
|
||||
"""
|
||||
if index < 0:
|
||||
index = len(self.scan_history) + index
|
||||
try:
|
||||
msg = self.scan_history.pop(index)
|
||||
self.scan_removed.emit(msg)
|
||||
except IndexError:
|
||||
logger.warning(f"Invalid index {index} for removing scan entry from history.")
|
||||
return
|
||||
self.takeTopLevelItem(index)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
# pylint: disable=import-outside-toplevel
|
||||
|
||||
from bec_widgets.widgets.services.scan_history_browser.scan_history_metadata_viewer import (
|
||||
ScanHistoryMetadataViewer,
|
||||
)
|
||||
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
|
||||
|
||||
app = QtWidgets.QApplication([])
|
||||
|
||||
main_window = QtWidgets.QMainWindow()
|
||||
central_widget = QtWidgets.QWidget()
|
||||
button = DarkModeButton()
|
||||
layout = QtWidgets.QVBoxLayout(central_widget)
|
||||
main_window.setCentralWidget(central_widget)
|
||||
|
||||
# Create a ScanHistoryBrowser instance
|
||||
browser = ScanHistoryBrowser()
|
||||
|
||||
# Create a ScanHistoryView instance
|
||||
view = ScanHistoryMetadataViewer()
|
||||
|
||||
layout.addWidget(button)
|
||||
layout.addWidget(browser)
|
||||
layout.addWidget(view)
|
||||
browser.scan_selected.connect(view.update_view)
|
||||
browser.scan_removed.connect(view.clear_view)
|
||||
|
||||
main_window.show()
|
||||
app.exec_()
|
||||
@ -0,0 +1,161 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy import QtCore, 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
|
||||
|
||||
|
||||
class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
|
||||
"""ScanHistoryView is a widget to display the metadata of a ScanHistoryMessage in a structured format."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client=None,
|
||||
config: ConnectionConfig | None = None,
|
||||
gui_id: str | None = None,
|
||||
theme_update: bool = True,
|
||||
scan_history_msg: ScanHistoryMessage | None = None,
|
||||
parent=None,
|
||||
):
|
||||
super().__init__(
|
||||
parent=parent, client=client, config=config, gui_id=gui_id, theme_update=theme_update
|
||||
)
|
||||
self._scan_history_msg_labels = {
|
||||
"scan_id": "Scan ID",
|
||||
"dataset_number": "Dataset Nr",
|
||||
"file_path": "File Path",
|
||||
"start_time": "Start Time",
|
||||
"end_time": "End Time",
|
||||
"elapsed_time": "Elapsed Time",
|
||||
"exit_status": "Status",
|
||||
"scan_name": "Scan Name",
|
||||
"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.apply_theme()
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
"""Apply the theme to the widget."""
|
||||
colors = get_theme_palette()
|
||||
palette = QtGui.QPalette()
|
||||
palette.setColor(self.backgroundRole(), colors.midlight().color())
|
||||
self.setPalette(palette)
|
||||
|
||||
def _init_grid_layout(self):
|
||||
"""Initialize the layout of the widget."""
|
||||
layout: QtWidgets.QGridLayout = self.layout()
|
||||
layout.setContentsMargins(10, 10, 10, 10)
|
||||
layout.setHorizontalSpacing(0)
|
||||
layout.setVerticalSpacing(0)
|
||||
layout.setColumnStretch(0, 0)
|
||||
layout.setColumnStretch(1, 1)
|
||||
layout.setColumnStretch(2, 0)
|
||||
|
||||
def setup_content_widget_label(self) -> None:
|
||||
"""Setup the labels for the content widget for the scan history view."""
|
||||
layout = self.layout()
|
||||
for row, k in enumerate(self._scan_history_msg_labels.keys()):
|
||||
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
|
||||
# Copy button
|
||||
copy_button = QtWidgets.QToolButton()
|
||||
copy_button.setSizePolicy(QtWidgets.QSizePolicy.Fixed, QtWidgets.QSizePolicy.Minimum)
|
||||
copy_button.setContentsMargins(0, 0, 0, 0)
|
||||
copy_button.setStyleSheet("padding: 0px; margin: 0px; border: none;")
|
||||
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.clicked.connect(
|
||||
lambda _, field=value_field: QtWidgets.QApplication.clipboard().setText(
|
||||
field.text()
|
||||
)
|
||||
)
|
||||
layout.addWidget(copy_button, row, 2) # Placeholder for copy button
|
||||
|
||||
@SafeSlot()
|
||||
def update_view(self, msg: ScanHistoryMessage):
|
||||
"""
|
||||
Update the view with the given ScanHistoryMessage.
|
||||
|
||||
Args:
|
||||
msg (ScanHistoryMessage): The message containing scan metadata.
|
||||
"""
|
||||
if msg == self.scan_history_msg:
|
||||
return # Same message, no update needed
|
||||
self.scan_history_msg = msg
|
||||
layout = self.layout()
|
||||
if layout.count() == 0:
|
||||
self.setup_content_widget_label()
|
||||
self.setTitle(f"Metadata - Scan {msg.scan_number}")
|
||||
for row, k in enumerate(self._scan_history_msg_labels.keys()):
|
||||
if k == "elapsed_time":
|
||||
value = (
|
||||
f"{(msg.end_time - msg.start_time):.3f}s"
|
||||
if msg.start_time and msg.end_time
|
||||
else None
|
||||
)
|
||||
else:
|
||||
value = getattr(msg, k, None)
|
||||
if k in ["start_time", "end_time"]:
|
||||
value = (
|
||||
datetime.fromtimestamp(value).strftime("%a %b %d %H:%M:%S %Y")
|
||||
if value
|
||||
else None
|
||||
)
|
||||
if value is None:
|
||||
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"]:
|
||||
layout.itemAtPosition(row, 2).widget().setVisible(True)
|
||||
layout.itemAtPosition(row, 2).widget().setEnabled(True)
|
||||
else:
|
||||
layout.itemAtPosition(row, 2).widget().setText("")
|
||||
layout.itemAtPosition(row, 2).widget().setToolTip("")
|
||||
|
||||
@SafeSlot()
|
||||
def clear_view(self, msg: ScanHistoryMessage | None = None):
|
||||
"""
|
||||
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):
|
||||
item = layout.itemAt(i)
|
||||
if item.widget():
|
||||
item.widget().close()
|
||||
item.widget().deleteLater()
|
||||
self.scan_history_msg = None
|
||||
self.setTitle("No Scan Selected")
|
||||
Reference in New Issue
Block a user