1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 16:02:51 +01:00

refactor: improve device-manager-view

This commit is contained in:
2025-10-03 16:28:32 +02:00
parent 9ff0db4831
commit 797a5046d1
6 changed files with 479 additions and 108 deletions

View File

@@ -195,6 +195,7 @@ if __name__ == "__main__": # pragma: no cover
app = QApplication([sys.argv[0], *qt_args])
apply_theme("dark")
w = BECMainApp(show_examples=args.examples)
w.resize(1920, 1200)
w.show()
sys.exit(app.exec())

View File

@@ -14,10 +14,23 @@ from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
from bec_qthemes import apply_theme
from PySide6QtAds import CDockManager, CDockWidget
from qtpy.QtCore import Qt, QThreadPool, QTimer
from qtpy.QtWidgets import QFileDialog, QMessageBox, QSplitter, QVBoxLayout, QWidget
from qtpy.QtWidgets import (
QDialog,
QFileDialog,
QHBoxLayout,
QLabel,
QMessageBox,
QPushButton,
QSizePolicy,
QSplitter,
QTextEdit,
QVBoxLayout,
QWidget,
)
from bec_widgets import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.help_inspector.help_inspector import HelpInspector
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
@@ -83,6 +96,53 @@ def set_splitter_weights(splitter: QSplitter, weights: List[float]) -> None:
QTimer.singleShot(0, apply)
class ConfigChoiceDialog(QDialog):
REPLACE = 1
ADD = 2
CANCEL = 0
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("Load Config")
layout = QVBoxLayout(self)
label = QLabel("Do you want to replace the current config or add to it?")
label.setWordWrap(True)
layout.addWidget(label)
# Buttons: equal size, stacked vertically
self.replace_btn = QPushButton("Replace")
self.add_btn = QPushButton("Add")
self.cancel_btn = QPushButton("Cancel")
btn_layout = QHBoxLayout()
for btn in (self.replace_btn, self.add_btn, self.cancel_btn):
btn.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred)
btn_layout.addWidget(btn)
layout.addLayout(btn_layout)
# Connect signals to explicit slots
self.replace_btn.clicked.connect(self.accept_replace)
self.add_btn.clicked.connect(self.accept_add)
self.cancel_btn.clicked.connect(self.reject_cancel)
self._result = self.CANCEL
def accept_replace(self):
self._result = self.REPLACE
self.accept()
def accept_add(self):
self._result = self.ADD
self.accept()
def reject_cancel(self):
self._result = self.CANCEL
self.reject()
def result(self):
return self._result
class DeviceManagerView(BECWidget, QWidget):
def __init__(self, parent=None, *args, **kwargs):
@@ -99,14 +159,14 @@ class DeviceManagerView(BECWidget, QWidget):
self.dock_manager.setStyleSheet("")
self._root_layout.addWidget(self.dock_manager)
# Available Resources Widget
self.available_devices = AvailableDeviceResources(
self, shared_selection_signal=self._shared_selection
)
self.available_devices_dock = QtAds.CDockWidget(
self.dock_manager, "Available Devices", self
)
self.available_devices_dock.setWidget(self.available_devices)
# # Available Resources Widget
# self.available_devices = AvailableDeviceResources(
# self, shared_selection_signal=self._shared_selection
# )
# self.available_devices_dock = QtAds.CDockWidget(
# self.dock_manager, "Available Devices", self
# )
# self.available_devices_dock.setWidget(self.available_devices)
# Device Table View widget
self.device_table_view = DeviceTableView(
@@ -130,29 +190,62 @@ class DeviceManagerView(BECWidget, QWidget):
self.ophyd_test_dock_view = QtAds.CDockWidget(self.dock_manager, "Ophyd Test View", self)
self.ophyd_test_dock_view.setWidget(self.ophyd_test_view)
# Arrange widgets within the QtAds dock manager
# Help Inspector
widget = QWidget(self)
layout = QVBoxLayout(widget)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self.help_inspector = HelpInspector(self)
layout.addWidget(self.help_inspector)
text_box = QTextEdit(self)
text_box.setReadOnly(False)
text_box.setPlaceholderText("Help text will appear here...")
layout.addWidget(text_box)
self.help_inspector_dock = QtAds.CDockWidget(self.dock_manager, "Help Inspector", self)
self.help_inspector_dock.setWidget(widget)
# # Hook inspector signals
# def _class_cb(text: str):
# print(text)
# # Register callback
self.help_inspector.bec_widget_help.connect(text_box.setMarkdown)
# Error Logs View
self.error_logs_view = QTextEdit(self)
self.error_logs_view.setReadOnly(True)
self.error_logs_view.setPlaceholderText("Error logs will appear here...")
self.error_logs_dock = QtAds.CDockWidget(self.dock_manager, "Error Logs", self)
self.error_logs_dock.setWidget(self.error_logs_view)
self.ophyd_test_view.validation_msg_md.connect(self.error_logs_view.setMarkdown)
# Arrange widgets within the QtAds dock manager
# Central widget area
self.central_dock_area = self.dock_manager.setCentralWidget(self.device_table_view_dock)
# Right area - should be pushed into view if something is active
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea,
self.dm_docs_view_dock,
QtAds.DockWidgetArea.RightDockWidgetArea,
self.ophyd_test_dock_view,
self.central_dock_area,
)
# Left Area
self.left_dock_area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.LeftDockWidgetArea, self.available_devices_dock
)
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea, self.dm_config_view_dock, self.left_dock_area
# create bottom area (2-arg -> area)
self.bottom_dock_area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea, self.dm_docs_view_dock
)
# Right area
# YAML view left of docstrings (docks relative to bottom area)
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.RightDockWidgetArea, self.ophyd_test_dock_view
QtAds.DockWidgetArea.LeftDockWidgetArea, self.dm_config_view_dock, self.bottom_dock_area
)
# Error/help area right of docstrings (dock relative to bottom area)
area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.RightDockWidgetArea,
self.help_inspector_dock,
self.bottom_dock_area,
)
self.dock_manager.addDockWidgetTabToArea(self.error_logs_dock, area)
for dock in self.dock_manager.dockWidgets():
# dock.setFeature(CDockWidget.DockWidgetDeleteOnClose, True)#TODO implement according to MonacoDock or AdvancedDockArea
# dock.setFeature(CDockWidget.CustomCloseHandling, True) #TODO same
@@ -160,13 +253,16 @@ class DeviceManagerView(BECWidget, QWidget):
dock.setFeature(CDockWidget.DockWidgetFloatable, False)
dock.setFeature(CDockWidget.DockWidgetMovable, False)
# TODO decide if we like to hide the title bars..
# Fetch all dock areas of the dock widgets (on our case always one dock area)
for dock in self.dock_manager.dockWidgets():
area = dock.dockAreaWidget()
area.titleBar().setVisible(False)
# for dock in self.dock_manager.dockWidgets():
# if dock.objectName() in ["Help Inspector", "Error Logs"]:
# continue
# area = dock.dockAreaWidget()
# area.titleBar().setVisible(False)
# Apply stretch after the layout is done
self.set_default_view([2, 8, 2], [3, 1])
self.set_default_view([2, 8, 2], [7, 3])
# self.set_default_view([2, 8, 2], [2, 2, 4])
# Connect slots
@@ -175,29 +271,29 @@ class DeviceManagerView(BECWidget, QWidget):
self.device_table_view.selected_devices,
(self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
),
(
self.available_devices.selected_devices,
(self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
),
# (
# self.available_devices.selected_devices,
# (self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
# ),
(
self.ophyd_test_view.device_validated,
(self.device_table_view.update_device_validation,),
),
(
self.device_table_view.device_configs_changed,
(
self.ophyd_test_view.change_device_configs,
self.available_devices.mark_devices_used,
),
),
(
self.available_devices.add_selected_devices,
(self.device_table_view.add_device_configs,),
),
(
self.available_devices.del_selected_devices,
(self.device_table_view.remove_device_configs,),
),
# (
# self.device_table_view.device_configs_changed,
# (
# self.ophyd_test_view.change_device_configs,
# self.available_devices.mark_devices_used,
# ),
# ),
# (
# self.available_devices.add_selected_devices,
# (self.device_table_view.add_device_configs,),
# ),
# (
# self.available_devices.del_selected_devices,
# (self.device_table_view.remove_device_configs,),
# ),
]:
for slot in slots:
signal.connect(slot)
@@ -217,7 +313,9 @@ class DeviceManagerView(BECWidget, QWidget):
# Create IO bundle
io_bundle = ToolbarBundle("IO", self.toolbar.components)
# Load from disk
load = MaterialIconAction(
text_position="under",
icon_name="file_open",
parent=self,
tooltip="Load configuration file from disk",
@@ -229,6 +327,7 @@ class DeviceManagerView(BECWidget, QWidget):
# Add safe to disk
safe_to_disk = MaterialIconAction(
text_position="under",
icon_name="file_save",
parent=self,
tooltip="Save config to disk",
@@ -240,10 +339,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Add load config from redis
load_redis = MaterialIconAction(
text_position="under",
icon_name="cached",
parent=self,
tooltip="Load current config from Redis",
label_text="Reload Config",
label_text="Get Current Config",
)
load_redis.action.triggered.connect(self._load_redis_action)
self.toolbar.components.add_safe("load_redis", load_redis)
@@ -251,11 +351,13 @@ class DeviceManagerView(BECWidget, QWidget):
# Update config action
update_config_redis = MaterialIconAction(
text_position="under",
icon_name="cloud_upload",
parent=self,
tooltip="Update current config in Redis",
label_text="Update Config",
)
update_config_redis.action.setEnabled(False)
update_config_redis.action.triggered.connect(self._update_redis_action)
self.toolbar.components.add_safe("update_config_redis", update_config_redis)
io_bundle.add_action("update_config_redis")
@@ -270,6 +372,7 @@ class DeviceManagerView(BECWidget, QWidget):
# Reset composed view
reset_composed = MaterialIconAction(
text_position="under",
icon_name="delete_sweep",
parent=self,
tooltip="Reset current composed config view",
@@ -281,7 +384,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Add device
add_device = MaterialIconAction(
icon_name="add", parent=self, tooltip="Add new device", label_text="Add Device"
text_position="under",
icon_name="add",
parent=self,
tooltip="Add new device",
label_text="Add Device",
)
add_device.action.triggered.connect(self._add_device_action)
self.toolbar.components.add_safe("add_device", add_device)
@@ -289,7 +396,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Remove device
remove_device = MaterialIconAction(
icon_name="remove", parent=self, tooltip="Remove device", label_text="Remove Device"
text_position="under",
icon_name="remove",
parent=self,
tooltip="Remove device",
label_text="Remove Device",
)
remove_device.action.triggered.connect(self._remove_device_action)
self.toolbar.components.add_safe("remove_device", remove_device)
@@ -297,10 +408,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Rerun validation
rerun_validation = MaterialIconAction(
text_position="under",
icon_name="checklist",
parent=self,
tooltip="Run device validation with 'connect' on selected devices",
label_text="Rerun Validation",
label_text="Validate Connection",
)
rerun_validation.action.triggered.connect(self._rerun_validation_action)
self.toolbar.components.add_safe("rerun_validation", rerun_validation)
@@ -346,15 +458,26 @@ class DeviceManagerView(BECWidget, QWidget):
file_path, _ = QFileDialog.getOpenFileName(
self, caption="Select Config File", dir=start_dir
)
if file_path:
try:
config = [{"name": k, **v} for k, v in yaml_load(file_path).items()]
except Exception as e:
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
return
self.device_table_view.set_device_config(
config
) # TODO ADD QDialog with 'replace', 'add' & 'cancel'
self._load_config_from_file(file_path)
def _load_config_from_file(self, file_path: str):
"""
Load device config from a given file path and update the device table view.
Args:
file_path (str): Path to the configuration file.
"""
try:
config = [{"name": k, **v} for k, v in yaml_load(file_path).items()]
except Exception as e:
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
return
dialog = ConfigChoiceDialog(self)
if dialog.exec():
if dialog.result() == ConfigChoiceDialog.REPLACE:
self.device_table_view.set_device_config(config)
elif dialog.result() == ConfigChoiceDialog.ADD:
self.device_table_view.add_device_configs(config)
# TODO would we ever like to add the current config to an existing composition
@SafeSlot()

View File

@@ -11,6 +11,7 @@ from qtpy import QtCore, QtWidgets
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import AccentColors, get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.widget_io import WidgetHierarchy
logger = bec_logger.logger
@@ -100,7 +101,7 @@ class HelpInspector(BECWidget, QtWidgets.QWidget):
self._button.setChecked(False)
QtWidgets.QApplication.restoreOverrideCursor()
def eventFilter(self, obj, event):
def eventFilter(self, obj: QtWidgets.QWidget, event: QtCore.QEvent) -> bool:
"""
Filter events to capture Key_Escape event, and mouse clicks
if event filter is active. Any click event on a widget is suppressed, if
@@ -111,25 +112,32 @@ class HelpInspector(BECWidget, QtWidgets.QWidget):
obj (QObject): The object that received the event.
event (QEvent): The event to filter.
"""
if (
event.type() == QtCore.QEvent.KeyPress
and event.key() == QtCore.Qt.Key_Escape
and self._active
):
# If not active, return immediately
if not self._active:
return super().eventFilter(obj, event)
# If active, handle escape key
if event.type() == QtCore.QEvent.KeyPress and event.key() == QtCore.Qt.Key_Escape:
self._toggle_mode(False)
return super().eventFilter(obj, event)
if self._active and event.type() == QtCore.QEvent.MouseButtonPress:
# If active, and left mouse button pressed, handle click
if event.type() == QtCore.QEvent.MouseButtonPress:
if event.button() == QtCore.Qt.LeftButton:
widget = self._app.widgetAt(event.globalPos())
if widget is None:
return super().eventFilter(obj, event)
# Get BECWidget ancestor
# TODO check what happens if the HELP Inspector itself is embedded in another BECWidget
# I suppose we would like to get the first ancestor that is a BECWidget, not the topmost one
widget = WidgetHierarchy._get_becwidget_ancestor(widget)
if widget:
if widget is self or self.isAncestorOf(widget):
if widget is self:
self._toggle_mode(False)
return True
for cb in self._callbacks.values():
try:
cb(widget)
except Exception as e:
print(f"Error occurred in callback {cb}: {e}")
logger.error(f"Error occurred in callback {cb}: {e}")
return True
return super().eventFilter(obj, event)

View File

@@ -6,3 +6,58 @@ MIME_DEVICE_CONFIG: Final[str] = "application/x-bec_device_config"
# Custom user roles
SORT_KEY_ROLE: Final[int] = 117
CONFIG_DATA_ROLE: Final[int] = 118
# TODO keep in sync with header...
HEADERS_HELP_MD: dict[str, str] = {
"status": (
"## Status"
"\n"
"The current status of the device. Can be one of the following values: \n"
"### **LOADED** \n The device with the specified configuration is loaded in the current config.\n"
"### **CONNECT_READY** \n The device config is valid and the connection has been validated. It has not yet been loaded to the current config.\n"
"### **CONNECT_FAILED** \n The device config is valid, but the connection could not be established.\n"
"### **VALID** \n The device config is valid, but the connection has not yet been validated.\n"
"### **INVALID** \n The device config is invalid and can not be loaded to the current config.\n"
),
"name": ("## Name " "\n" "The name of the device."),
"deviceClass": (
"## Device Class"
"\n"
"The device class specifies the type of the device. It will be used to create the instance."
),
"readoutPriority": (
"## Readout Priority"
"\n"
"The readout priority of the device. Can be one of the following values: \n"
"### **monitored** \n The monitored priority is used for devices that are read out during the scan (i.e. at every step) and whose value may change during the scan.\n"
"### **baseline** \n The baseline priority is used for devices that are read out at the beginning of the scan and whose value does not change during the scan.\n"
"### **async** \n The async priority is used for devices that are asynchronous to the monitored devices, and send their data independently.\n"
"### **continuous** \n The continuous priority is used for devices that are read out continuously during the scan.\n"
"### **on_request** \n The on_request priority is used for devices that should not be read out during the scan, yet are configured to be read out manually.\n"
),
"deviceTags": (
"## Device Tags"
"\n"
"A list of tags associated with the device. Tags can be used to group devices and filter them in the device manager."
),
"enabled": (
"## Enabled"
"\n"
"Indicator whether the device is enabled or disabled. Disabled devices can not be used."
),
"readOnly": ("## Read Only" "\n" "Indicator that a device is read-only or can be modified."),
"onFailure": (
"## On Failure"
"\n"
"Specifies the behavior of the device in case of a failure. Can be one of the following values: \n"
"### **buffer** \n The device readback will fall back to the last known value.\n"
"### **retry** \n The device readback will be retried once, and raises an error if it fails again.\n"
"### **raise** \n The device readback will raise immediately.\n"
),
"softwareTrigger": (
"## Software Trigger"
"\n"
"Indicator whether the device receives a software trigger from BEC during a scan."
),
"description": ("## Description" "\n" "A short description of the device."),
}

View File

@@ -4,11 +4,13 @@ from __future__ import annotations
import copy
import json
import textwrap
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Iterable, List
from uuid import uuid4
from bec_lib.atlas_models import Device
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
@@ -21,7 +23,10 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
from bec_widgets.widgets.control.device_manager.components.constants import MIME_DEVICE_CONFIG
from bec_widgets.widgets.control.device_manager.components.constants import (
HEADERS_HELP_MD,
MIME_DEVICE_CONFIG,
)
from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import ValidationStatus
if TYPE_CHECKING: # pragma: no cover
@@ -75,6 +80,111 @@ class CustomDisplayDelegate(DictToolTipDelegate):
painter.restore()
class WrappingTextDelegate(CustomDisplayDelegate):
"""A lightweight delegate that wraps text without expensive size recalculation."""
def __init__(self, parent=None, max_width=300, margin=6):
super().__init__(parent)
self._parent = parent
self.max_width = max_width
self.margin = margin
self._cache = {} # cache text metrics for performance
def _do_custom_paint(self, painter, option, index, value: str):
text = str(value)
if not text:
return
painter.save()
painter.setClipRect(option.rect)
# Use cached layout if available
cache_key = (text, option.rect.width())
layout = self._cache.get(cache_key)
if layout is None:
layout = QtGui.QTextLayout(text, option.font)
layout.beginLayout()
height = 0
while True:
line = layout.createLine()
if not line.isValid():
break
line.setLineWidth(option.rect.width() - self.margin)
line.setPosition(QtCore.QPointF(self.margin / 2, height))
height += line.height()
layout.endLayout()
self._cache[cache_key] = layout
# # Draw background if selected
# if option.state & QtWidgets.QStyle.State_Selected:
# painter.fillRect(option.rect, option.palette.highlight())
# Draw text
painter.setPen(option.palette.text().color())
layout.draw(painter, option.rect.topLeft())
painter.restore()
def sizeHint(self, option, index):
"""Return a cached or approximate height; avoids costly recomputation."""
text = str(index.data(QtCore.Qt.DisplayRole) or "")
view = self._parent
view.initViewItemOption(option)
if view.isColumnHidden(index.column()) or not view.isVisible() or not text:
return QtCore.QSize(0, option.fontMetrics.height() + 2 * self.margin)
# Use cache for consistent size computation
cache_key = (text, self.max_width)
if cache_key in self._cache:
layout = self._cache[cache_key]
height = 0
for i in range(layout.lineCount()):
height += layout.lineAt(i).height()
return QtCore.QSize(self.max_width, int(height + self.margin))
# Approximate without layout (fast path)
metrics = option.fontMetrics
pixel_width = max(self._parent.columnWidth(index.column()), 100)
if pixel_width > 2000: # safeguard against uninitialized columns, may return large values
pixel_width = 100
char_per_line = self.estimate_chars_per_line(text, option, pixel_width - 2 * self.margin)
wrapped_lines = textwrap.wrap(text, width=char_per_line)
lines = len(wrapped_lines)
return QtCore.QSize(pixel_width, lines * (metrics.height()) + 2 * self.margin)
def estimate_chars_per_line(self, text: str, option, column_width: int) -> int:
"""Estimate number of characters that fit in a line for given width."""
metrics = option.fontMetrics
elided = metrics.elidedText(text, Qt.ElideRight, column_width)
return len(elided.rstrip(""))
@SafeSlot(int, int, int)
def _on_section_resized(self, logical_index, old_size=None, new_size=None):
"""Only update rows if a wrapped column was resized."""
self._cache.clear()
self._update_row_heights()
def _update_row_heights(self):
"""Efficiently adjust row heights based on wrapped columns."""
view = self._parent
proxy = view.model()
model = proxy.sourceModel()
option = QtWidgets.QStyleOptionViewItem()
view.initViewItemOption(option)
# wrapping delegates
wrap_delegate_columns = []
for row in range(proxy.rowCount()):
max_height = 18
for column in [5, 6]: # TODO don't hardcode columns.. to be improved
index = proxy.index(row, column)
# model_index = proxy.mapToSource(index)
# delegate = view.itemDelegateForColumn(model_index) or view.itemDelegate()
delegate = view.itemDelegateForColumn(column)
hint = delegate.sizeHint(option, index)
max_height = max(max_height, hint.height())
if view.rowHeight(row) != max_height:
view.setRowHeight(row, max_height)
class CenterCheckBoxDelegate(CustomDisplayDelegate):
"""Custom checkbox delegate to center checkboxes in table cells."""
@@ -89,8 +199,9 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
def apply_theme(self, theme: str | None = None):
colors = get_accent_colors()
self._icon_checked.setColor(colors.default)
self._icon_unchecked.setColor(colors.default)
_icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
self._icon_checked = _icon("check_box")
self._icon_unchecked = _icon("check_box_outline_blank")
def _do_custom_paint(self, painter, option, index, value):
pixmap = self._icon_checked if value == Qt.CheckState.Checked else self._icon_unchecked
@@ -123,8 +234,12 @@ class DeviceValidatedDelegate(CustomDisplayDelegate):
def apply_theme(self, theme: str | None = None):
colors = get_accent_colors()
for status, icon in self._icons.items():
icon.setColor(colors[status])
_icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
self._icons = {
ValidationStatus.PENDING: _icon(color=colors.default),
ValidationStatus.VALID: _icon(color=colors.success),
ValidationStatus.FAILED: _icon(color=colors.emergency),
}
def _do_custom_paint(self, painter, option, index, value):
if pixmap := self._icons.get(value):
@@ -148,15 +263,19 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
self._device_config: list[dict[str, Any]] = []
self._validation_status: dict[str, ValidationStatus] = {}
self.headers = [
"",
"status",
"name",
"deviceClass",
"readoutPriority",
"onFailure",
"deviceTags",
"description",
"enabled",
"readOnly",
"softwareTrigger",
]
self._checkable_columns_enabled = {"enabled": True, "readOnly": True}
self._device_model_schema = Device.model_json_schema()
###############################################
########## Override custom Qt methods #########
@@ -172,6 +291,8 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
def headerData(self, section, orientation, role=int(Qt.ItemDataRole.DisplayRole)):
if role == Qt.ItemDataRole.DisplayRole and orientation == Qt.Orientation.Horizontal:
if section == 9: # softwareTrigger
return "softTrig"
return self.headers[section]
return None
@@ -192,20 +313,24 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
return self._validation_status.get(dev_name, ValidationStatus.PENDING)
key = self.headers[col]
value = self._device_config[row].get(key)
value = self._device_config[row].get(key, None)
if value is None:
value = (
self._device_model_schema.get("properties", {}).get(key, {}).get("default", None)
)
if role == Qt.ItemDataRole.DisplayRole:
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
return bool(value)
if key == "deviceTags":
return ", ".join(str(tag) for tag in value) if value else ""
if key == "deviceClass":
return str(value).split(".")[-1]
return str(value) if value is not None else ""
if role == USER_CHECK_DATA_ROLE and key in ("enabled", "readOnly"):
if role == USER_CHECK_DATA_ROLE and key in ("enabled", "readOnly", "softwareTrigger"):
return Qt.CheckState.Checked if value else Qt.CheckState.Unchecked
if role == Qt.ItemDataRole.TextAlignmentRole:
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
return Qt.AlignmentFlag.AlignCenter
return Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter
if role == Qt.ItemDataRole.FontRole:
@@ -223,7 +348,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable | Qt.ItemFlag.ItemIsDropEnabled
)
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
if self._checkable_columns_enabled.get(key, True):
return base_flags | Qt.ItemFlag.ItemIsUserCheckable
else:
@@ -245,7 +370,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
if not index.isValid():
return False
key = self.headers[index.column()]
if key in ("enabled", "readOnly") and role == USER_CHECK_DATA_ROLE:
if key in ("enabled", "readOnly", "softwareTrigger") and role == USER_CHECK_DATA_ROLE:
if not self._checkable_columns_enabled.get(key, True):
return False # ignore changes if column is disabled
self._device_config[index.row()][key] = value == Qt.CheckState.Checked
@@ -301,6 +426,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
if self._name_exists_in_config(name := cfg.get("name", "<not found>"), True):
logger.warning(f"Device {name} already exists in the model.")
already_in_list.append(name)
# TODO add a warning that some devices were already in the list, how is this handled...
continue
row = len(self._device_config)
self.beginInsertRows(QtCore.QModelIndex(), row, row)
@@ -489,6 +615,12 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
self._filter_text = ""
self._enable_fuzzy = True
self._filter_columns = [1, 2] # name and deviceClass for search
# TODO refactor if enums are changed!!
self._status_order = {
ValidationStatus.VALID: 0,
ValidationStatus.PENDING: 1,
ValidationStatus.FAILED: 2,
}
def get_row_data(self, rows: Iterable[QModelIndex]) -> Iterable[dict[str, Any]]:
return (self.sourceModel().get_row_data(self.mapToSource(idx)) for idx in rows)
@@ -506,6 +638,14 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
self._hidden_rows.update(row_indices)
self.invalidateFilter()
def lessThan(self, left, right):
"""Add custom sorting for the status column"""
if left.column() != 0 or right.column() != 0:
return super().lessThan(left, right)
left_data = self.sourceModel().data(left, Qt.ItemDataRole.DisplayRole)
right_data = self.sourceModel().data(right, Qt.ItemDataRole.DisplayRole)
return self._status_order.get(left_data, 99) < self._status_order.get(right_data, 99)
def show_rows(self, row_indices: list[int]):
"""
Show specific rows in the model.
@@ -602,6 +742,21 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
# Connect signals
self._model.configs_changed.connect(self.device_configs_changed.emit)
def get_help_md(self) -> str:
"""
Generate Markdown help for a cell or header.
"""
pos = self.table.mapFromGlobal(QtGui.QCursor.pos())
model: DeviceTableModel = self._model # access underlying model
index = self.table.indexAt(pos)
if index.isValid():
column = index.column()
label = model.headerData(column, QtCore.Qt.Horizontal, QtCore.Qt.DisplayRole)
if label == "softTrig":
label = "softwareTrigger"
return HEADERS_HELP_MD.get(label, "")
return ""
def _setup_search(self):
"""Create components related to the search functionality"""
@@ -653,15 +808,20 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
self.checkbox_delegate = CenterCheckBoxDelegate(self.table, colors=colors)
self.tool_tip_delegate = DictToolTipDelegate(self.table)
self.validated_delegate = DeviceValidatedDelegate(self.table, colors=colors)
self.table.setItemDelegateForColumn(0, self.validated_delegate) # ValidationStatus
self.wrapped_delegate = WrappingTextDelegate(self.table, max_width=300)
# Add resize handling for wrapped delegate
header = self.table.horizontalHeader()
self.table.setItemDelegateForColumn(0, self.validated_delegate) # status
self.table.setItemDelegateForColumn(1, self.tool_tip_delegate) # name
self.table.setItemDelegateForColumn(2, self.tool_tip_delegate) # deviceClass
self.table.setItemDelegateForColumn(3, self.tool_tip_delegate) # readoutPriority
self.table.setItemDelegateForColumn(
4, self.tool_tip_delegate
) # deviceTags (was wrap_delegate)
self.table.setItemDelegateForColumn(5, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(6, self.checkbox_delegate) # readOnly
self.table.setItemDelegateForColumn(4, self.tool_tip_delegate) # onFailure
self.table.setItemDelegateForColumn(5, self.wrapped_delegate) # deviceTags
self.table.setItemDelegateForColumn(6, self.wrapped_delegate) # description
self.table.setItemDelegateForColumn(7, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(8, self.checkbox_delegate) # readOnly
self.table.setItemDelegateForColumn(9, self.checkbox_delegate) # softwareTrigger
# Disable wrapping, use eliding, and smooth scrolling
self.table.setWordWrap(False)
@@ -675,19 +835,35 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
header.setSectionResizeMode(1, QHeaderView.ResizeMode.Interactive) # name
header.setSectionResizeMode(2, QHeaderView.ResizeMode.Interactive) # deviceClass
header.setSectionResizeMode(3, QHeaderView.ResizeMode.Interactive) # readoutPriority
header.setSectionResizeMode(4, QHeaderView.ResizeMode.Stretch) # deviceTags: expand to fill
header.setSectionResizeMode(5, QHeaderView.ResizeMode.Fixed) # enabled
header.setSectionResizeMode(6, QHeaderView.ResizeMode.Fixed) # readOnly
header.setSectionResizeMode(4, QHeaderView.ResizeMode.Interactive) # onFailure
header.setSectionResizeMode(
5, QHeaderView.ResizeMode.Interactive
) # deviceTags: expand to fill
header.setSectionResizeMode(6, QHeaderView.ResizeMode.Stretch) # descript: expand to fill
header.setSectionResizeMode(7, QHeaderView.ResizeMode.Fixed) # enabled
header.setSectionResizeMode(8, QHeaderView.ResizeMode.Fixed) # readOnly
header.setSectionResizeMode(9, QHeaderView.ResizeMode.Fixed) # softwareTrigger
self.table.setColumnWidth(0, 25)
self.table.setColumnWidth(5, 70)
self.table.setColumnWidth(6, 70)
self.table.setColumnWidth(0, 70)
self.table.setColumnWidth(5, 200)
self.table.setColumnWidth(6, 200)
self.table.setColumnWidth(7, 70)
self.table.setColumnWidth(8, 70)
self.table.setColumnWidth(9, 70)
# Ensure column widths stay fixed
header.setMinimumSectionSize(25)
header.setDefaultSectionSize(90)
header.setStretchLastSection(False)
# Resize policy for wrapped text delegate
self._resize_proxy = BECSignalProxy(
header.sectionResized,
rateLimit=25,
slot=self.wrapped_delegate._on_section_resized,
timeout=1.0,
)
# Selection behavior
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)

View File

@@ -322,28 +322,36 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
self.validation_msg_md.emit("")
def _format_markdown_text(self, device_name: str, raw_msg: str) -> str:
"""Simple HTML formatting for validation messages, wrapping text naturally."""
if not raw_msg.strip():
return f"### Validation in progress for {device_name}... \n\n"
if raw_msg == "Validation in progress...":
"""
Simple HTML formatting for validation messages, wrapping text naturally.
Args:
device_name (str): The name of the device.
raw_msg (str): The raw validation message.
"""
if not raw_msg.strip() or raw_msg.strip() == "Validation in progress...":
return f"### Validation in progress for {device_name}... \n\n"
m = re.search(r"ERROR:\s*([^\s]+)\s+is not valid:\s*(.+?errors?)", raw_msg)
device, summary = m.group(1), m.group(2)
lines = [f"## Error for '{device}'", f"'{device}' is not valid: {summary}"]
# Find each field block: \n<field>\n Field required ...
field_pat = re.compile(
r"\n(?P<field>\w+)\n\s+(?P<rest>Field required.*?(?=\n\w+\n|$))", re.DOTALL
# Regex to capture repeated ERROR patterns
pat = re.compile(
r"ERROR:\s*(?P<device>[^\s]+)\s+"
r"(?P<status>is not valid|is not connectable|failed):\s*"
r"(?P<detail>.*?)(?=ERROR:|$)",
re.DOTALL,
)
blocks = []
for m in pat.finditer(raw_msg):
dev = m.group("device")
status = m.group("status")
detail = m.group("detail").strip()
lines = [f"## Error for {dev}", f"**{dev} {status}**", f"```\n{detail}\n```"]
blocks.append("\n\n".join(lines))
for m in field_pat.finditer(raw_msg):
field = m.group("field")
rest = m.group("rest").rstrip()
lines.append(f"### {field}")
lines.append(rest)
# Fallback: If no patterns matched, return the raw message
if not blocks:
return f"## Error for {device_name}\n```\n{raw_msg.strip()}\n```"
return "\n".join(lines)
return "\n\n---\n\n".join(blocks)
def validation_running(self):
return self._device_list_items != {}
@@ -386,7 +394,7 @@ if __name__ == "__main__":
layout.setSpacing(0)
device_manager_ophyd_test = DMOphydTest()
try:
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/endstation.yaml"
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/first_light.yaml"
config = [{"name": k, **v} for k, v in yaml_load(config_path).items()]
except Exception as e:
logger.error(f"Error loading config: {e}")