bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 03:01:50 +02:00
Code Activity

refactor(scan-history): update after message refactoring

Browse Source
This commit is contained in:
appel_c
2025-07-10 07:39:54 +02:00
parent 9644c75e95
commit 2ae1c0e71c
1 changed files with 97 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
bec_widgets/widgets/services/scan_history_browser/components/scan_history_device_viewer.py
View File

@ -2,6 +2,8 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from bec_lib.logger import bec_logger
from bec_lib.messages import ScanHistoryMessage
from bec_qthemes import material_icon
@ -9,54 +11,56 @@ 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):
class SignalModel(QtCore.QAbstractListModel):
"""Custom model for displaying scan history signals in a combo box."""
def __init__(self, parent=None, devices: dict[str, tuple[int,]] = None):
def __init__(self, parent=None, signals: dict[str, _StoredDataInfo] = None):
super().__init__(parent)
if devices is None:
devices = {}
self._devices: list[tuple] = sorted(devices.items(), key=lambda x: -x[1])
if signals is None:
signals = {}
self._signals: list[tuple[str, _StoredDataInfo]] = sorted(
signals.items(), key=lambda x: -x[1].shape[0]
)
# @SafeProperty(list) # TODO if SafeProperty, how to pass the type list[tuple[str, int]]?
@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, tuple[int,]]):
@signals.setter
def signals(self, value: dict[str, _StoredDataInfo]):
self.beginResetModel()
self._devices = sorted(value.items(), key=lambda x: -x[1][0])
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()]
# TODO how to display shape here nicely, mostly it's number of points, just occasionally it is a tuple
if len(num_points) == 1:
num_points = num_points[0] # Assuming num_points is a tuple, take the first element
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)
@ -82,7 +86,9 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
RPC = False
PLUGIN = False
request_history_plot = QtCore.Signal(str, object) # (str, ScanHistoryMessage.model_dump())
request_history_plot = QtCore.Signal(
str, str, object
) # (device_name, signal_name, ScanHistoryMessage)
def __init__(
self,
@ -103,24 +109,43 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
)
# Current scan history message
self.scan_history_msg: ScanHistoryMessage | None = None
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
# 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(parent=self.device_combo)
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)
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)
@SafeSlot()
def update_devices_from_scan_history(self, msg: ScanHistoryMessage) -> None:
@ -132,21 +157,45 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
if not isinstance(msg, ScanHistoryMessage):
logger.info(f"Received message of type {type(msg)} instead of ScanHistoryMessage.")
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
# 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 = self.scan_history_msg.device_data_info
index = self.device_combo.findData(device_name, QtCore.Qt.UserRole)
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.warning(f"No signal data found for device {device_name}.")
return
self.signal_model.signals = self.scan_history_msg.stored_data_info[device_name]
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:
"""Clear the device combo box."""
self.scan_history_msg = None
self.device_model.devices = {}
self.signal_model.signals = {}
self.device_combo.clear()
@SafeSlot()
@ -155,17 +204,19 @@ 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 for device: {device_name}, {signal_name} with {self.scan_history_msg} points."
)
self.request_history_plot.emit(device_name, self.scan_history_msg)
self.request_history_plot.emit(device_name, signal_name, self.scan_history_msg)
if __name__ == "__main__":
if __name__ == "__main__": # pragma: no cover
import sys
app = QtWidgets.QApplication(sys.argv)
@ -173,10 +224,11 @@ if __name__ == "__main__":
main_window = QtWidgets.QMainWindow()
central_widget = QtWidgets.QWidget()
main_window.setCentralWidget(central_widget)
layout = QtWidgets.QVBoxLayout(central_widget)
layout.setContentsMargins(0, 0, 0, 0)
ly = QtWidgets.QVBoxLayout(central_widget)
ly.setContentsMargins(0, 0, 0, 0)
viewer = ScanHistoryDeviceViewer()
layout.addWidget(viewer)
ly.addWidget(viewer)
main_window.show()
app.exec_()
app.exec_()