feat(scan_control): add QML and QWidget ports with a validated form model, per-row removal and a sticky footer

This commit is contained in:
Claude committed 2026-10-09 20:37:46 +00:00
1 parent 58fc7a07ac
commit 10bed0fa42
7 files changed
+2414

No files matched your search

+94
View File
@@ -14,6 +14,7 @@ from qtpy.QtWidgets import (
QFrame,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QSizePolicy,
QToolButton,
@@ -324,3 +325,96 @@ class PortedPropertiesMixin:
if cls is self.PORTED_FROM:
break
return objects
class ToggleSwitch(QPushButton):
"""Checkable switch painted like ``BecUi.SwitchField``; the label shows On or Off."""
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.setCheckable(True)
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setMinimumHeight(32)
self.setFixedWidth(84)
self._knob = 0.0
self._animation = QPropertyAnimation(self, b"knob", self)
self._animation.setDuration(120)
self._animation.setEasingCurve(QEasingCurve.Type.OutCubic)
self.toggled.connect(self._animate)
def _animate(self, checked: bool) -> None:
self._animation.stop()
self._animation.setStartValue(self._knob)
self._animation.setEndValue(1.0 if checked else 0.0)
self._animation.start()
def setChecked(self, checked: bool) -> None: # pylint: disable=invalid-name
"""Set the state without animating."""
self.blockSignals(True)
super().setChecked(checked)
self.blockSignals(False)
self._animation.stop()
self._knob = 1.0 if checked else 0.0
self.update()
def _get_knob(self) -> float:
return self._knob
def _set_knob(self, value: float) -> None:
self._knob = value
self.update()
knob = Property(float, _get_knob, _set_knob)
def paintEvent(self, _event): # pylint: disable=invalid-name
tokens = ThemeTokens()
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
track = QRectF(0, (self.height() - 22) / 2, 38, 22)
painter.setPen(tokens.primary if self.isChecked() else tokens.border)
painter.setBrush(tokens.primary if self.isChecked() else tokens.track)
painter.drawRoundedRect(track, 11, 11)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(tokens.on_primary if self.isChecked() else tokens.fg_muted)
painter.drawEllipse(QRectF(track.left() + 3 + self._knob * 16, track.top() + 3, 16, 16))
painter.setPen(tokens.fg)
font = self.font()
font.setPixelSize(13)
painter.setFont(font)
painter.drawText(
QRectF(track.right() + 8, 0, self.width() - track.right() - 8, self.height()),
Qt.AlignmentFlag.AlignVCenter,
"On" if self.isChecked() else "Off",
)
painter.end()
class SuffixLineEdit(QLineEdit):
"""Line edit that paints a unit suffix inside its right edge, like ``BecUi.InputField``."""
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._suffix = ""
def set_suffix(self, suffix: str) -> None:
"""Set the unit shown at the right edge."""
self._suffix = suffix or ""
width = self.fontMetrics().horizontalAdvance(self._suffix) + 10 if self._suffix else 0
self.setTextMargins(0, 0, width, 0)
self.update()
def paintEvent(self, event): # pylint: disable=invalid-name
super().paintEvent(event)
if not self._suffix:
return
painter = QPainter(self)
painter.setPen(ThemeTokens().fg_subtle)
font = self.font()
font.setPixelSize(12)
painter.setFont(font)
painter.drawText(
self.rect().adjusted(0, 0, -9, 0),
Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter,
self._suffix,
)
painter.end()
@@ -0,0 +1,347 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
import BecUi
// Scan submission form: scan picker, argument rows, parameter groups and a sticky footer.
// backend: ScanControlBackend (state, groups, devices and actions).
Rectangle {
id: root
required property QtObject backend
readonly property var s: backend.state
readonly property var errors: s.errors || ({})
color: theme.bg
implicitWidth: 420
implicitHeight: 520
// One labelled editor for a form field; the editor depends on the field kind.
component FieldEditor: ColumnLayout {
id: fieldRoot
required property var field
readonly property string error: root.errors[field.key] || ""
spacing: 3
RowLayout {
spacing: 4
Layout.fillWidth: true
Text {
text: fieldRoot.field.label
color: theme.fgMuted
font.pixelSize: 12
elide: Text.ElideRight
Layout.fillWidth: true
}
Icon {
visible: fieldRoot.field.tooltip !== ""
name: "info"
size: 13
color: theme.fgSubtle
HoverHandler { id: tipHover }
ToolTip.visible: tipHover.hovered
ToolTip.text: fieldRoot.field.tooltip
}
}
Loader {
Layout.fillWidth: true
sourceComponent: fieldRoot.field.kind === "bool" ? switchEditor
: fieldRoot.field.kind === "literal" ? literalEditor
: fieldRoot.field.kind === "device" ? deviceEditor
: textEditor
}
Text {
visible: fieldRoot.error !== ""
text: fieldRoot.error
color: theme.danger
font.pixelSize: 11
Layout.fillWidth: true
elide: Text.ElideRight
}
Component {
id: textEditor
InputField {
text: fieldRoot.field.value === undefined || fieldRoot.field.value === null ? "" : String(fieldRoot.field.value)
placeholderText: fieldRoot.field.placeholder
suffix: fieldRoot.field.units
invalid: fieldRoot.error !== ""
horizontalAlignment: fieldRoot.field.kind === "str" ? Text.AlignLeft : Text.AlignRight
inputMethodHints: fieldRoot.field.kind === "str" ? Qt.ImhNone : Qt.ImhFormattedNumbersOnly
onTextEdited: root.backend.setField(fieldRoot.field.key, text)
onAccepted: root.backend.run()
Accessible.name: fieldRoot.field.label
}
}
Component {
id: switchEditor
SwitchField {
checked: fieldRoot.field.value === true
text: checked ? "On" : "Off"
onToggled: root.backend.setField(fieldRoot.field.key, checked)
Accessible.name: fieldRoot.field.label
}
}
Component {
id: literalEditor
SelectField {
model: fieldRoot.field.options
currentIndex: Math.max(0, fieldRoot.field.options.indexOf(String(fieldRoot.field.value)))
displayText: currentText === "" ? "None" : currentText
onActivated: (index) => root.backend.setField(fieldRoot.field.key, textAt(index))
Accessible.name: fieldRoot.field.label
}
}
Component {
id: deviceEditor
SelectField {
editable: true
model: root.backend.devices
placeholder: "Device"
invalid: fieldRoot.error !== ""
Component.onCompleted: editText = fieldRoot.field.value || ""
onEditTextChanged: if (activeFocus || popup.visible) root.backend.setField(fieldRoot.field.key, editText)
onActivated: (index) => root.backend.setField(fieldRoot.field.key, textAt(index))
Accessible.name: fieldRoot.field.label
}
}
}
ColumnLayout {
anchors.fill: parent
anchors.margins: 8
spacing: 8
// ---- scan picker -------------------------------------------------------------------
ColumnLayout {
visible: root.s.showSelector === true
Layout.fillWidth: true
spacing: 4
RowLayout {
Layout.fillWidth: true
spacing: 4
SelectField {
id: scanPicker
Layout.fillWidth: true
editable: true
model: root.s.scans || []
placeholder: "Search scans"
invalid: editText !== "" && model.indexOf(editText) < 0 && find(editText, Qt.MatchFixedString) < 0
onActivated: (index) => root.backend.selectScan(textAt(index))
onAccepted: {
root.backend.selectScan(editText)
editText = root.s.current || ""
}
onActiveFocusChanged: if (!activeFocus) editText = root.s.current || ""
Connections {
target: root.backend
function onChanged() {
if (!scanPicker.activeFocus && scanPicker.editText !== (root.s.current || ""))
scanPicker.editText = root.s.current || ""
}
}
Accessible.name: "Scan"
}
IconButton {
iconName: "info"
tip: "Scan documentation"
enabled: (root.s.current || "") !== ""
onClicked: root.backend.openInfo()
}
IconButton {
visible: root.s.showFilter === true
iconName: "filter_list"
tip: "Choose scans shown in the list"
onClicked: root.backend.openFilter()
}
IconButton {
iconName: "history"
tip: "Restore the parameters of the last run of this scan"
enabled: (root.s.current || "") !== "" && root.s.restoreBusy !== true
onClicked: root.backend.restoreLast()
}
}
Text {
visible: text !== ""
text: root.s.summary || ""
color: theme.fgMuted
font.pixelSize: 12
wrapMode: Text.WordWrap
maximumLineCount: 2
elide: Text.ElideRight
Layout.fillWidth: true
}
}
// ---- parameter groups ------------------------------------------------------------------
ScrollView {
id: scroller
Layout.fillWidth: true
Layout.fillHeight: true
contentWidth: availableWidth
clip: true
ColumnLayout {
width: scroller.availableWidth
spacing: 8
Repeater {
model: root.backend.groups
delegate: Card {
id: groupCard
required property var modelData
readonly property bool isArgs: modelData.kind === "args"
visible: isArgs ? root.s.showArgs === true : root.s.showKwargs === true
Layout.fillWidth: true
title: modelData.title
subtitle: isArgs ? modelData.rows.length + (modelData.rows.length === 1 ? " row" : " rows") : ""
headerExtras: [
TextButton {
visible: groupCard.isArgs && root.s.showRowButtons === true
enabled: groupCard.modelData.canAdd
text: "Add row"
iconName: "add"
variant: "ghost"
implicitHeight: 28
onClicked: root.backend.addRow()
}
]
// argument rows
Repeater {
model: groupCard.isArgs ? groupCard.modelData.rows : []
delegate: RowLayout {
id: argRow
required property var modelData
required property int index
Layout.fillWidth: true
spacing: 8
Rectangle {
Layout.alignment: Qt.AlignTop
Layout.topMargin: 22
width: 22; height: 22; radius: 11
color: theme.track
Text {
anchors.centerIn: parent
text: argRow.index + 1
color: theme.fgMuted
font.pixelSize: 11
font.weight: Font.DemiBold
}
}
GridLayout {
Layout.fillWidth: true
columns: Math.max(1, Math.min(argRow.modelData.length, Math.floor((scroller.availableWidth - 80) / 75)))
columnSpacing: 8
rowSpacing: 6
Repeater {
model: argRow.modelData
delegate: FieldEditor {
required property var modelData
field: modelData
Layout.fillWidth: true
Layout.alignment: Qt.AlignTop
}
}
}
IconButton {
Layout.alignment: Qt.AlignTop
Layout.topMargin: 18
visible: root.s.showRowButtons === true
enabled: groupCard.modelData.canRemove
iconName: "close"
tip: "Remove row " + (argRow.index + 1)
onClicked: root.backend.removeRow(argRow.index)
}
}
}
// keyword fields
GridLayout {
visible: !groupCard.isArgs
Layout.fillWidth: true
columns: Math.max(1, Math.floor((scroller.availableWidth - 30) / 170))
columnSpacing: 10
rowSpacing: 8
Repeater {
model: groupCard.isArgs ? [] : groupCard.modelData.fields
delegate: FieldEditor {
required property var modelData
field: modelData
Layout.fillWidth: true
Layout.alignment: Qt.AlignTop
}
}
}
}
}
Text {
visible: (root.s.current || "") === ""
Layout.fillWidth: true
Layout.topMargin: 24
horizontalAlignment: Text.AlignHCenter
text: (root.s.scans || []).length === 0 ? "No scans are available. Check the scan filter or the BEC server." : "Choose a scan to set its parameters."
wrapMode: Text.WordWrap
color: theme.fgMuted
font.pixelSize: 13
}
Item { Layout.preferredHeight: 4 }
}
}
// ---- footer ------------------------------------------------------------------------------
Rectangle {
Layout.fillWidth: true
implicitHeight: footer.implicitHeight + 16
radius: 10
color: theme.card
border.color: theme.border
visible: root.s.showButtons === true || root.s.showMetadata === true
ColumnLayout {
id: footer
anchors.fill: parent
anchors.margins: 8
spacing: 6
Text {
visible: text !== ""
text: root.s.hint || ""
color: root.s.metadataValid === true ? theme.danger : theme.warning
font.pixelSize: 12
Layout.fillWidth: true
elide: Text.ElideRight
}
RowLayout {
Layout.fillWidth: true
spacing: 6
TextButton {
visible: root.s.showMetadata === true
text: "Metadata"
iconName: root.s.metadataValid === true ? "check_circle" : "error"
variant: "neutral"
tip: root.s.metadataValid === true ? "Metadata is complete" : "Metadata needs attention"
onClicked: root.backend.openMetadata()
}
Item { Layout.fillWidth: true; visible: root.s.showButtons !== true }
TextButton {
visible: root.s.showButtons === true
Layout.fillWidth: true
text: root.s.startLabel || "Start"
iconName: "play_arrow"
variant: "success"
enabled: root.s.canStart === true
onClicked: root.backend.run()
}
TextButton {
visible: root.s.showButtons === true
text: root.s.stopLabel || "Stop"
iconName: "stop"
variant: "danger"
onClicked: root.backend.stop()
}
}
}
}
}
}
@@ -0,0 +1,128 @@
"""Scan control rendered with Qt Quick (QML), same API as :class:`ScanControl`."""
from __future__ import annotations
from pathlib import Path
from qtpy.QtCore import Property, QObject, Signal, Slot
from qtpy.QtWidgets import QVBoxLayout
from bec_widgets.utils.quick import create_quick_widget, release_quick_widget
from bec_widgets.widgets.control.scan_control.scan_control_ux_common import ScanControlPortBase
QML_FILE = Path(__file__).parent / "qml" / "ScanControlView.qml"
class ScanControlBackend(QObject):
"""Bridge between :class:`ScanControlQML` and ``ScanControlView.qml``."""
changed = Signal()
structure_changed = Signal()
def __init__(self, control: "ScanControlQML"):
super().__init__(control)
self._control = control
self._state: dict = {}
self._groups: list = []
self._devices: list[str] = []
def update_from(self, control: ScanControlPortBase, structure: bool) -> None:
"""Copy the control state; the field structure only when it changed."""
self._state = control.view_state()
if structure:
self._groups = control.form.groups()
self._devices = control._device_names() # pylint: disable=protected-access
self.structure_changed.emit()
self.changed.emit()
@Slot(str)
def selectScan(self, name: str) -> None: # pylint: disable=invalid-name
"""Switch to another scan."""
self._control.on_scan_selection_changed(name)
@Slot(str, "QVariant")
def setField(self, key: str, value) -> None: # pylint: disable=invalid-name
"""Store an edited field value."""
self._control.set_field(key, value)
@Slot()
def addRow(self) -> None: # pylint: disable=invalid-name
"""Add an argument row."""
self._control.add_row()
@Slot(int)
def removeRow(self, index: int) -> None: # pylint: disable=invalid-name
"""Remove an argument row."""
self._control.remove_row(index)
@Slot()
def run(self) -> None:
"""Start the scan."""
self._control.run_scan()
@Slot()
def stop(self) -> None:
"""Stop the running scan."""
self._control.stop_scan()
@Slot()
def restoreLast(self) -> None: # pylint: disable=invalid-name
"""Load the parameters of the last run of this scan."""
self._control.request_last_executed_scan_parameters()
@Slot()
def openInfo(self) -> None: # pylint: disable=invalid-name
"""Show the scan documentation."""
self._control.show_selected_scan_info()
@Slot()
def openFilter(self) -> None: # pylint: disable=invalid-name
"""Choose the scans shown in the selector."""
self._control.show_scan_selector_settings()
@Slot()
def openMetadata(self) -> None: # pylint: disable=invalid-name
"""Edit the scan metadata."""
self._control.show_metadata_dialog()
state = Property("QVariantMap", lambda self: self._state, notify=changed)
groups = Property("QVariantList", lambda self: self._groups, notify=structure_changed)
devices = Property("QStringList", lambda self: self._devices, notify=structure_changed)
class ScanControlQML(ScanControlPortBase):
"""Scan control with the modernised UX rendered in QML."""
def _init_view(self) -> None:
self.backend = ScanControlBackend(self)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.view = create_quick_widget(self, QML_FILE, {"backend": self.backend})
self.view.setMinimumSize(320, 360)
layout.addWidget(self.view)
def _sync_view(self, structure: bool) -> None:
if hasattr(self, "backend"):
self.backend.update_from(self, structure)
def apply_theme(self, theme: str):
super().apply_theme(theme)
self.view.setClearColor(self.palette().window().color())
def cleanup(self):
release_quick_widget(self.view)
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
widget = ScanControlQML()
widget.show()
sys.exit(app.exec())
@@ -0,0 +1,410 @@
"""Scan control with the modernised UX in plain QWidgets, same API as :class:`ScanControl`.
This is the QWidget twin of :class:`ScanControlQML`; both render the same
:class:`ScanFormModel` and view state so they can be compared side by side.
"""
from __future__ import annotations
from qtpy.QtCore import Qt
from qtpy.QtWidgets import (
QComboBox,
QCompleter,
QFrame,
QGridLayout,
QHBoxLayout,
QLabel,
QScrollArea,
QSizePolicy,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.quick.host import ThemeTokens
from bec_widgets.utils.ux_kit import (
Card,
IconButton,
SuffixLineEdit,
TextButton,
ToggleSwitch,
field_qss,
refresh_kit_theme,
set_invalid,
)
from bec_widgets.widgets.control.scan_control.scan_control_ux_common import ScanControlPortBase
class FieldEditor(QWidget):
"""Label, editor and error line for one form field."""
def __init__(self, field: dict, control: "ScanControlQWidget", parent: QWidget):
super().__init__(parent)
self.key = field["key"]
self._control = control
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(3)
header = QHBoxLayout()
header.setSpacing(4)
self.label = QLabel(field["label"], self)
header.addWidget(self.label, 1)
if field["tooltip"]:
info = IconButton("info", field["tooltip"], self, size=13)
info.setFixedSize(16, 16)
info.setToolTip(field["tooltip"])
header.addWidget(info)
layout.addLayout(header)
kind = field["kind"]
value = field["value"]
if kind == "bool":
self.editor = ToggleSwitch(self)
self.editor.setChecked(bool(value))
self.editor.toggled.connect(lambda checked: control.set_field(self.key, checked))
elif kind == "literal":
self.editor = QComboBox(self)
for option in field["options"]:
self.editor.addItem(option or "None", option)
index = max(0, self.editor.findData(str(value) if value is not None else ""))
self.editor.setCurrentIndex(index)
self.editor.activated.connect(
lambda index: control.set_field(self.key, self.editor.itemData(index))
)
elif kind == "device":
self.editor = QComboBox(self)
self.editor.setEditable(True)
self.editor.setInsertPolicy(QComboBox.InsertPolicy.NoInsert)
self.editor.addItems(control._device_names()) # pylint: disable=protected-access
completer = QCompleter(self.editor.model(), self.editor)
completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive)
completer.setFilterMode(Qt.MatchFlag.MatchContains)
self.editor.setCompleter(completer)
self.editor.lineEdit().setPlaceholderText("Device")
self.editor.setCurrentText(value or "")
self.editor.currentTextChanged.connect(lambda text: control.set_field(self.key, text))
else:
self.editor = SuffixLineEdit(self)
self.editor.setText("" if value is None else str(value))
self.editor.setPlaceholderText(field["placeholder"])
self.editor.set_suffix(field["units"])
if kind != "str":
self.editor.setAlignment(
Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter
)
self.editor.textEdited.connect(lambda text: control.set_field(self.key, text))
self.editor.returnPressed.connect(control.run_scan)
self.editor.setAccessibleName(field["label"])
if field["tooltip"]:
self.editor.setToolTip(field["tooltip"])
layout.addWidget(self.editor)
self.error = QLabel(self)
self.error.setVisible(False)
layout.addWidget(self.error)
self.set_error("")
def set_error(self, message: str) -> None:
"""Show or clear the error of this field."""
self.error.setText(message)
self.error.setVisible(bool(message))
set_invalid(self.editor, bool(message))
def refresh_theme(self, tokens: ThemeTokens) -> None:
"""Apply the theme tokens."""
self.label.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 12px;")
self.error.setStyleSheet(f"color: {tokens.danger.name()}; font-size: 11px;")
class RowBadge(QLabel):
"""Round row number in front of an argument row."""
def __init__(self, number: int, parent: QWidget):
super().__init__(str(number), parent)
self.setFixedSize(22, 22)
self.setAlignment(Qt.AlignmentFlag.AlignCenter)
def refresh_theme(self, tokens: ThemeTokens) -> None:
"""Apply the theme tokens."""
self.setStyleSheet(
f"background: {tokens.track.name()}; color: {tokens.fg_muted.name()};"
" border-radius: 11px; font-size: 11px; font-weight: 600;"
)
class ScanControlQWidget(ScanControlPortBase):
"""Scan control with the modernised UX in plain QWidgets."""
def _init_view(self) -> None:
self.editors: dict[str, FieldEditor] = {}
self.group_cards: list[Card] = []
self._picker_typing = False
self._built_columns = (0, 0)
root = QVBoxLayout(self)
root.setContentsMargins(8, 8, 8, 8)
root.setSpacing(8)
# scan picker
self.header = QWidget(self)
header_layout = QVBoxLayout(self.header)
header_layout.setContentsMargins(0, 0, 0, 0)
header_layout.setSpacing(4)
picker_row = QHBoxLayout()
picker_row.setSpacing(4)
self.scan_picker = QComboBox(self.header)
self.scan_picker.setEditable(True)
self.scan_picker.setInsertPolicy(QComboBox.InsertPolicy.NoInsert)
self.scan_picker.setAccessibleName("Scan")
self.scan_picker.lineEdit().setPlaceholderText("Search scans")
completer = QCompleter(self.scan_picker.model(), self.scan_picker)
completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive)
completer.setFilterMode(Qt.MatchFlag.MatchContains)
self.scan_picker.setCompleter(completer)
self.scan_picker.activated.connect(
lambda index: self.on_scan_selection_changed(self.scan_picker.itemText(index))
)
self.scan_picker.lineEdit().editingFinished.connect(self._confirm_typed_scan)
self.scan_picker.lineEdit().textEdited.connect(self._on_scan_typed)
self.info_button = IconButton("info", "Scan documentation", self.header)
self.info_button.clicked.connect(self.show_selected_scan_info)
self.filter_button = IconButton(
"filter_list", "Choose scans shown in the list", self.header
)
self.filter_button.clicked.connect(self.show_scan_selector_settings)
self.restore_button = IconButton(
"history", "Restore the parameters of the last run of this scan", self.header
)
self.restore_button.clicked.connect(self.request_last_executed_scan_parameters)
picker_row.addWidget(self.scan_picker, 1)
picker_row.addWidget(self.info_button)
picker_row.addWidget(self.filter_button)
picker_row.addWidget(self.restore_button)
header_layout.addLayout(picker_row)
self.summary_label = QLabel(self.header)
self.summary_label.setWordWrap(True)
header_layout.addWidget(self.summary_label)
root.addWidget(self.header)
# groups
self.scroll = QScrollArea(self)
self.scroll.setWidgetResizable(True)
self.scroll.setFrameShape(QFrame.Shape.NoFrame)
self.groups_widget = QWidget(self.scroll)
self.groups_layout = QVBoxLayout(self.groups_widget)
self.groups_layout.setContentsMargins(0, 0, 0, 0)
self.groups_layout.setSpacing(8)
self.groups_layout.addStretch(1)
self.scroll.setWidget(self.groups_widget)
root.addWidget(self.scroll, 1)
self.empty_label = QLabel(self.groups_widget)
self.empty_label.setWordWrap(True)
self.empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.groups_layout.insertWidget(0, self.empty_label)
# footer
self.footer = Card(parent=self, padding=8)
self.footer.body.setSpacing(6)
self.hint_label = QLabel(self.footer)
self.footer.body.addWidget(self.hint_label)
buttons = QHBoxLayout()
buttons.setSpacing(6)
self.metadata_button = TextButton("Metadata", "neutral", "check_circle", self.footer)
self.metadata_button.clicked.connect(self.show_metadata_dialog)
self.start_button = TextButton("Start", "success", "play_arrow", self.footer)
self.start_button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
self.start_button.clicked.connect(self.run_scan)
self.stop_button = TextButton("Stop", "danger", "stop", self.footer)
self.stop_button.clicked.connect(self.stop_scan)
buttons.addWidget(self.metadata_button)
buttons.addWidget(self.start_button, 1)
buttons.addWidget(self.stop_button)
self.footer.body.addLayout(buttons)
root.addWidget(self.footer)
self._metadata_icon_valid: bool | None = None
self._apply_styles(ThemeTokens())
# ---- scan picker -------------------------------------------------------------------
def _on_scan_typed(self, text: str) -> None:
self._picker_typing = True
set_invalid(self.scan_picker, bool(text) and not self.is_valid_scan(text))
def _confirm_typed_scan(self) -> None:
self._picker_typing = False
text = self.scan_picker.currentText()
if self.is_valid_scan(text):
self.on_scan_selection_changed(text)
self.scan_picker.setCurrentText(self._selected_scan)
set_invalid(self.scan_picker, False)
# ---- layout --------------------------------------------------------------------------
def _columns(self) -> tuple[int, int]:
"""Columns for argument rows and for keyword fields at the current width."""
width = self.scroll.viewport().width()
return max(1, (width - 80) // 75), max(1, (width - 30) // 170)
def resizeEvent(self, event): # pylint: disable=invalid-name
super().resizeEvent(event)
if hasattr(self, "footer") and self._columns() != self._built_columns:
self._rebuild_groups()
self._sync_view(False)
# ---- rendering ---------------------------------------------------------------------
def _apply_styles(self, tokens: ThemeTokens) -> None:
self.setStyleSheet(field_qss(tokens))
self.summary_label.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 12px;")
self.empty_label.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 13px;")
self.scroll.setStyleSheet("QScrollArea { background: transparent; }")
self.groups_widget.setStyleSheet("background: transparent;")
self._hint_tokens = tokens
def _rebuild_groups(self) -> None:
for card in self.group_cards:
self.groups_layout.removeWidget(card)
card.hide()
card.deleteLater()
self.group_cards = []
self.editors = {}
self._built_columns = self._columns()
state = self.view_state()
for group in self.form.groups():
card = Card(group["title"], self.groups_widget)
if group["kind"] == "args":
count = len(group["rows"])
card.set_title(group["title"], f"{count} row{'s' if count != 1 else ''}")
add = TextButton("Add row", "ghost", "add", card)
add.setEnabled(group["canAdd"])
add.setVisible(state["showRowButtons"])
add.clicked.connect(self.add_row)
card.add_header_widget(add)
for index, row in enumerate(group["rows"]):
card.body.addLayout(self._build_row(index, row, group["canRemove"], card))
card.setVisible(state["showArgs"])
else:
grid = QGridLayout()
grid.setHorizontalSpacing(10)
grid.setVerticalSpacing(8)
columns = self._built_columns[1]
for index, field in enumerate(group["fields"]):
editor = FieldEditor(field, self, card)
self.editors[field["key"]] = editor
grid.addWidget(
editor, index // columns, index % columns, Qt.AlignmentFlag.AlignTop
)
for column in range(columns):
grid.setColumnStretch(column, 1)
card.body.addLayout(grid)
card.setVisible(state["showKwargs"])
self.groups_layout.insertWidget(self.groups_layout.count() - 1, card)
self.group_cards.append(card)
tokens = refresh_kit_theme(self.groups_widget)
self._apply_styles(tokens)
def _build_row(self, index: int, row: list[dict], can_remove: bool, card: Card) -> QHBoxLayout:
layout = QHBoxLayout()
layout.setSpacing(8)
badge = RowBadge(index + 1, card)
layout.addWidget(badge, 0, Qt.AlignmentFlag.AlignTop)
badge.setContentsMargins(0, 0, 0, 0)
layout.setAlignment(badge, Qt.AlignmentFlag.AlignTop)
grid = QGridLayout()
grid.setHorizontalSpacing(8)
grid.setVerticalSpacing(6)
columns = max(1, min(len(row), self._built_columns[0]))
for position, field in enumerate(row):
editor = FieldEditor(field, self, card)
self.editors[field["key"]] = editor
grid.addWidget(
editor, position // columns, position % columns, Qt.AlignmentFlag.AlignTop
)
for column in range(columns):
grid.setColumnStretch(column, 1)
layout.addLayout(grid, 1)
remove = IconButton("close", f"Remove row {index + 1}", card)
remove.setEnabled(can_remove)
remove.setVisible(not self._hide_add_remove_buttons)
remove.clicked.connect(lambda _checked=False, i=index: self.remove_row(i))
layout.addWidget(remove, 0, Qt.AlignmentFlag.AlignTop)
return layout
def _sync_view(self, structure: bool) -> None:
if not hasattr(self, "footer"):
return
state = self.view_state()
if structure:
names = state["scans"]
if [self.scan_picker.itemText(i) for i in range(self.scan_picker.count())] != names:
self.scan_picker.blockSignals(True)
self.scan_picker.clear()
self.scan_picker.addItems(names)
for index, name in enumerate(names):
self.scan_picker.setItemData(
index, self.scan_tooltip(name), Qt.ItemDataRole.ToolTipRole
)
self.scan_picker.blockSignals(False)
self._rebuild_groups()
if not self._picker_typing:
self.scan_picker.setCurrentText(state["current"])
self.scan_picker.setToolTip(self.scan_tooltip(state["current"]) if state["current"] else "")
self.header.setVisible(state["showSelector"])
self.filter_button.setVisible(state["showFilter"])
self.info_button.setEnabled(bool(state["current"]))
self.restore_button.setEnabled(bool(state["current"]) and not state["restoreBusy"])
self.summary_label.setText(state["summary"])
self.summary_label.setVisible(bool(state["summary"]))
if state["current"]:
self.empty_label.hide()
else:
self.empty_label.setText(
"No scans are available. Check the scan filter or the BEC server."
if not state["scans"]
else "Choose a scan to set its parameters."
)
self.empty_label.show()
for key, editor in self.editors.items():
editor.set_error(state["errors"].get(key, ""))
tokens = ThemeTokens()
self.hint_label.setText(state["hint"])
self.hint_label.setVisible(bool(state["hint"]))
hint_color = tokens.danger if state["metadataValid"] else tokens.warning
self.hint_label.setStyleSheet(f"color: {hint_color.name()}; font-size: 12px;")
self.footer.setVisible(state["showButtons"] or state["showMetadata"])
self.metadata_button.setVisible(state["showMetadata"])
if self._metadata_icon_valid != state["metadataValid"]:
self._metadata_icon_valid = state["metadataValid"]
# pylint: disable=protected-access
self.metadata_button._icon_name = "check_circle" if state["metadataValid"] else "error"
self.metadata_button.refresh_theme(tokens)
self.metadata_button.setToolTip(
"Metadata is complete" if state["metadataValid"] else "Metadata needs attention"
)
self.start_button.setVisible(state["showButtons"])
self.stop_button.setVisible(state["showButtons"])
self.start_button.setText(state["startLabel"])
self.start_button.setEnabled(state["canStart"])
self.stop_button.setText(state["stopLabel"])
def apply_theme(self, theme: str):
super().apply_theme(theme)
if hasattr(self, "footer"):
tokens = refresh_kit_theme(self)
self._apply_styles(tokens)
self._sync_view(False)
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
widget = ScanControlQWidget()
widget.show()
sys.exit(app.exec())
@@ -0,0 +1,660 @@
"""Behaviour shared by the QML and QWidget ports of :class:`ScanControl`.
The ports keep the public API of ``ScanControl`` (signals, Designer properties, ``run_scan``,
``get_scan_parameters``, scan filter, last-scan restore, ...). The form itself is a
:class:`ScanFormModel`; the renderers only draw :meth:`ScanControlPortBase.view_state` and call
back into the methods here.
UX changes compared to ``ScanControl``:
* every field shows its unit, default and a specific error message ("Must be ≥ 0");
* errors appear once a field was edited, or for all fields after an attempt to start;
* argument rows can be removed individually;
* Start names the scan and says what is missing; Stop turns into Emergency Stop for 2 s;
* metadata moves to a dialog opened from the footer, which shows whether it is complete.
"""
from __future__ import annotations
import threading
from functools import partial
from types import NoneType
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from qtpy.QtCore import QTimer, Signal
from qtpy.QtWidgets import QDialog, QDialogButtonBox, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ux_kit import PortedPropertiesMixin
from bec_widgets.widgets.control.scan_control.scan_control import (
ScanControl,
ScanControlConfig,
ScanParameterConfig,
)
from bec_widgets.widgets.control.scan_control.scan_docstring import render_scan_tooltip_html
from bec_widgets.widgets.control.scan_control.scan_form_model import ScanFormModel
from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter
from bec_widgets.widgets.control.scan_control.scan_info_dialog import ScanInfoDialog
from bec_widgets.widgets.control.scan_control.scan_selection_dialog import ScanSelectionDialog
from bec_widgets.widgets.editors.scan_metadata.scan_metadata import ScanMetadata
logger = bec_logger.logger
def doc_summary(docstring: str | None) -> str:
"""First paragraph of a scan docstring on one line."""
if not docstring:
return ""
lines = []
for line in docstring.strip().splitlines():
if not line.strip():
if lines:
break
continue
lines.append(line.strip())
return " ".join(lines)
class ScanControlPortBase(PortedPropertiesMixin, BECWidget, QWidget):
"""Shared implementation of the ScanControl ports.
Subclasses create the renderer in :meth:`_init_view` and redraw it in :meth:`_sync_view`;
``structure`` is True when fields were added or removed, False for value-only changes.
"""
PLUGIN = False
RPC = False
rpc_widget_class = "ScanControl"
USER_ACCESS = ["attach", "detach", "screenshot"]
ICON_NAME = "tune"
RECENT_SCAN_HISTORY_COUNT = ScanControl.RECENT_SCAN_HISTORY_COUNT
MAX_HISTORY_LOOKBACK = ScanControl.MAX_HISTORY_LOOKBACK
LAST_SCAN_FETCH_TIMEOUT_MS = ScanControl.LAST_SCAN_FETCH_TIMEOUT_MS
EMERGENCY_STOP_TIMEOUT_MS = 2000
scan_started = Signal()
scan_selected = Signal(str)
device_selected = Signal(str)
scan_args = Signal(list)
_last_scan_parameters_received = Signal(int, str, object)
# The last-scan lookup is reused unchanged from ScanControl.
_fetch_last_executed_scan_parameters = ScanControl._fetch_last_executed_scan_parameters
_lookup_last_scan_parameters = ScanControl._lookup_last_scan_parameters
_search_history_windows = ScanControl._search_history_windows
_read_history_window = ScanControl._read_history_window
_find_parameters_in_history_cache = ScanControl._find_parameters_in_history_cache
_parameters_from_request_inputs = staticmethod(ScanControl._parameters_from_request_inputs)
_find_last_scan_parameters = staticmethod(ScanControl._find_last_scan_parameters)
def __init__(
self,
parent=None,
client=None,
config: ScanControlConfig | dict | None = None,
gui_id: str | None = None,
allowed_scans: list | None = None,
default_scan: str | None = None,
**kwargs,
):
if config is None:
config = ScanControlConfig(
widget_class=self.__class__.__name__, allowed_scans=allowed_scans
)
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.get_bec_shortcuts()
if default_scan is not None:
self.config.default_scan = default_scan
if allowed_scans is not None:
self.config.allowed_scans = list(dict.fromkeys(allowed_scans)) or None
self.available_scans: dict = {}
self.previous_scan = None
self._selected_scan = ""
self._visible_scans: list[str] = []
self._touched: set[str] = set()
self._show_all_errors = False
self._hide_arg_box = False
self._hide_kwarg_boxes = False
self._hide_scan_control_buttons = False
self._hide_metadata = False
self._hide_scan_selection_combobox = False
self._hide_scan_selector_settings_button = False
self._hide_add_remove_buttons = False
self._scan_info_adapter = ScanInfoAdapter()
self._scan_info_dialog: ScanInfoDialog | None = None
self._device_name_cache: list[str] | None = None
self.form = ScanFormModel(self._device_names, self._device_object, self)
self.form.structure_changed.connect(lambda: self._sync_view(True))
self.form.values_changed.connect(lambda: self._sync_view(False))
self.form.device_selected.connect(self.emit_device_selected)
# metadata form lives in a dialog opened from the footer
self._scan_metadata: dict | None = None
self._metadata_form = ScanMetadata(parent=self)
self._metadata_form.hide()
self._metadata_dialog: QDialog | None = None
self._metadata_form.form_data_updated.connect(self.update_scan_metadata)
self._metadata_form.form_data_cleared.connect(self.update_scan_metadata)
self.scan_selected.connect(self._metadata_form.update_with_new_scan)
# two-stage stop like StopButton: abort first, then offer an emergency halt
self._emergency_stop = False
self._emergency_timer = QTimer(self)
self._emergency_timer.setSingleShot(True)
self._emergency_timer.timeout.connect(self._reset_emergency_stop)
# background fetch of the last executed parameters (see ScanControl)
self._restore_busy = False
self._last_scan_parameters_received.connect(self._apply_last_scan_parameters)
self._last_scan_lookup_memo: dict = {}
self._last_scan_lookup_lock = threading.Lock()
self._last_scan_fetch_generation = 0
self._last_scan_fetch_watchdog = QTimer(self)
self._last_scan_fetch_watchdog.setSingleShot(True)
self._last_scan_fetch_watchdog.setInterval(self.LAST_SCAN_FETCH_TIMEOUT_MS)
self._last_scan_fetch_watchdog.timeout.connect(self._on_last_scan_parameters_timeout)
self._init_view()
self.populate_scans()
if self.config.default_scan is not None:
self.current_scan = self.config.default_scan
self._metadata_form.update_with_new_scan(self._selected_scan)
self._metadata_form.validate_form()
self._sync_view(True)
# ---- devices ---------------------------------------------------------------------------
def _device_names(self) -> list[str]:
if self._device_name_cache is None:
try:
self._device_name_cache = sorted(self.dev.keys())
except Exception: # pylint: disable=broad-except
self._device_name_cache = []
return self._device_name_cache
def _device_object(self, name: str):
return self.dev[name]
# ---- scan list -----------------------------------------------------------------------
def populate_scans(self):
"""Load the available scans from BEC and apply the scan filter."""
message = self.client.connector.get(MessageEndpoints.available_scans())
self.available_scans = getattr(message, "resource", None) or {}
self._device_name_cache = None
self._update_scan_selector()
def _scan_docstring(self, scan_name: str) -> str | None:
scan_info = self.available_scans.get(scan_name, {})
docstring = scan_info.get("doc") if isinstance(scan_info, dict) else None
return docstring if isinstance(docstring, str) else None
def _supported_scan_names(self) -> list[str]:
return [
scan_name
for scan_name, scan_info in self.available_scans.items()
if self._scan_info_adapter.has_scan_ui_config(scan_info)
and not scan_name.startswith("_")
]
def _update_scan_selector(self) -> None:
if self._selected_scan:
self.save_current_scan_parameters()
allowed = self.config.allowed_scans
if allowed is None:
visible = self._supported_scan_names()
else:
visible = [scan for scan in allowed if scan in self.available_scans]
self._visible_scans = visible
if not visible:
if self._selected_scan:
self._selected_scan = ""
self.form.load(None)
self._sync_view(True)
return
if self._selected_scan not in visible:
self.on_scan_selection_changed(visible[0])
else:
self._sync_view(True)
def visible_scans(self) -> list[str]:
"""Scans offered by the selector, in order."""
return list(self._visible_scans)
def scan_tooltip(self, scan_name: str) -> str:
"""Rich tooltip with the documentation of a scan."""
return render_scan_tooltip_html(scan_name, self._scan_docstring(scan_name))
def is_valid_scan(self, scan_name: str) -> bool:
"""Whether ``scan_name`` is offered by the selector (case-insensitive)."""
return self._resolve_scan_name(scan_name) is not None
def _resolve_scan_name(self, scan_name: str) -> str | None:
if not scan_name:
return None
lowered = scan_name.lower()
for name in self._visible_scans:
if name.lower() == lowered:
return name
return None
def on_scan_selection_changed(self, scan_name: str):
"""Switch to ``scan_name``; unknown names are ignored."""
resolved = self._resolve_scan_name(scan_name)
if resolved is None or resolved == self._selected_scan:
return
if self._selected_scan:
self._save_scan_parameters(self._selected_scan)
self._selected_scan = resolved
self._touched.clear()
self._show_all_errors = False
gui_config = self._scan_info_adapter.build_scan_ui_config(
self.available_scans.get(resolved, {})
)
self.form.load(gui_config)
self.scan_selected.emit(resolved)
self.restore_scan_parameters(resolved)
self._sync_view(True)
@SafeProperty(str)
def current_scan(self) -> str:
"""Name of the selected scan."""
return self._selected_scan
@current_scan.setter
def current_scan(self, scan_name: str):
self.on_scan_selection_changed(scan_name)
@SafeSlot(str)
def set_current_scan(self, scan_name: str):
"""Slot for setting the current scan to the given scan name."""
self.current_scan = scan_name
@SafeProperty("QStringList")
def allowed_scans(self) -> list[str]:
"""Scans configured for the selector; see :attr:`ScanControl.allowed_scans`."""
allowed = getattr(self.config, "allowed_scans", None)
return self._supported_scan_names() if allowed is None else list(allowed)
@allowed_scans.setter
def allowed_scans(self, scan_names: list[str] | str | None):
if isinstance(scan_names, str):
scan_names = [scan_names]
if scan_names is not None:
scan_names = list(dict.fromkeys(scan_names))
if not scan_names or scan_names == self._supported_scan_names():
scan_names = None
self.config.allowed_scans = scan_names
self._update_scan_selector()
@SafeSlot()
@SafeSlot(bool)
def show_scan_selector_settings(self, *_):
"""Open the scan filter dialog and apply accepted changes."""
scan_names = self._supported_scan_names()
allowed = self.config.allowed_scans
if allowed is not None:
scan_names += [scan for scan in allowed if scan not in scan_names]
dialog = ScanSelectionDialog(
scan_names=scan_names,
selected_scans=scan_names if allowed is None else allowed,
scan_docs={name: self._scan_docstring(name) for name in scan_names},
parent=self,
)
try:
accepted = dialog.exec() == QDialog.DialogCode.Accepted
selected = dialog.selected_scans() if accepted else None
finally:
dialog.deleteLater()
if selected is not None:
self.allowed_scans = None if selected == scan_names else selected
@SafeSlot()
@SafeSlot(bool)
def show_selected_scan_info(self, *_args) -> None:
"""Show the documentation of the selected scan."""
self.show_scan_info(self._selected_scan)
@SafeSlot(str)
def show_scan_info(self, scan_name: str) -> None:
"""Show documentation for a specific scan."""
if self._scan_info_dialog is None:
self._scan_info_dialog = ScanInfoDialog(self)
self._scan_info_dialog.show_scan(scan_name, self._scan_docstring(scan_name))
# ---- form editing --------------------------------------------------------------------
def set_field(self, key: str, value) -> None:
"""Set a field from the renderer; marks it as edited so its error becomes visible."""
self._touched.add(key)
self.form.set_value(key, value)
def add_row(self) -> None:
"""Add an argument row."""
self.form.add_bundle()
def remove_row(self, index: int) -> None:
"""Remove argument row ``index``."""
self.form.remove_bundle(index)
self._touched = {key for key in self._touched if not key.startswith("args:")}
@SafeSlot(str)
def emit_device_selected(self, dev_names: str):
"""Forward the devices chosen in the argument rows."""
self._selected_devices = dev_names
self.device_selected.emit(dev_names)
# ---- parameters ----------------------------------------------------------------------
def get_scan_parameters(self, bec_object: bool = True):
"""Return ``(args, kwargs)`` for the selected scan, including metadata.
Args:
bec_object(bool): Return device objects for device arguments.
"""
args = self.form.args(bec_object)
kwargs = self.form.kwargs()
if self._scan_metadata is not None:
kwargs["metadata"] = self._scan_metadata
return args, kwargs
def save_current_scan_parameters(self):
"""Remember the parameters of the selected scan."""
self._save_scan_parameters(self._selected_scan)
def _save_scan_parameters(self, scan_name: str):
if not scan_name:
return
self.previous_scan = scan_name
args = self.form.args(bec_object=False)
kwargs = self.form.kwargs()
self.config.scans[scan_name] = ScanParameterConfig(name=scan_name, args=args, kwargs=kwargs)
def restore_scan_parameters(self, scan_name: str):
"""Restore remembered parameters of ``scan_name``, else shared ones of the last scan."""
params = self.config.scans.get(scan_name)
if params is None:
previous = self.config.scans.get(self.previous_scan) if self.previous_scan else None
if previous is not None and previous.kwargs:
self.form.set_kwargs(
{key: value for key, value in previous.kwargs.items() if value is not None}
)
return
if params.args:
self.form.set_args(params.args)
if params.kwargs:
self.form.set_kwargs(params.kwargs)
@SafeSlot()
@SafeSlot(bool)
def request_last_executed_scan_parameters(self, *_):
"""Fetch the parameters of the last run of the selected scan in the background."""
if self._restore_busy or not self._selected_scan:
return
self._set_restore_busy(True)
self._last_scan_fetch_generation += 1
generation = self._last_scan_fetch_generation
self._last_scan_fetch_watchdog.start()
try:
self.submit_task(
self._fetch_last_executed_scan_parameters,
generation,
self._selected_scan,
on_complete=partial(self._on_last_scan_parameters_finished, generation),
on_failed=lambda _msg, gen=generation: self._on_last_scan_parameters_finished(gen),
)
except Exception:
self._on_last_scan_parameters_finished(generation)
raise
@SafeSlot(int, str, object)
def _apply_last_scan_parameters(self, generation: int, scan_name: str, parameters):
if generation != self._last_scan_fetch_generation or scan_name != self._selected_scan:
return
args, kwargs = parameters
if args:
self.form.set_args(args)
if kwargs:
self.form.set_kwargs(kwargs)
@SafeSlot()
def _on_last_scan_parameters_finished(self, generation: int | None = None):
if generation is not None and generation != self._last_scan_fetch_generation:
return
self._last_scan_fetch_watchdog.stop()
self._set_restore_busy(False)
@SafeSlot()
def _on_last_scan_parameters_timeout(self):
logger.warning("Timed out while fetching the last executed scan parameters")
self._last_scan_fetch_generation += 1
self._set_restore_busy(False)
def _set_restore_busy(self, busy: bool) -> None:
self._restore_busy = busy
self._sync_view(False)
# ---- metadata ------------------------------------------------------------------------
@SafeSlot(dict)
@SafeSlot(NoneType)
def update_scan_metadata(self, md: dict | None):
"""Receive validated metadata (or None when the form is invalid)."""
self._scan_metadata = md
self._sync_view(False)
@SafeSlot()
def show_metadata_dialog(self) -> None:
"""Open the metadata form in a dialog."""
if self._metadata_dialog is None:
dialog = QDialog(self)
dialog.setWindowTitle(f"Scan metadata · {self._selected_scan}")
layout = QVBoxLayout(dialog)
self._metadata_form.setParent(dialog)
self._metadata_form.show()
layout.addWidget(self._metadata_form)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, dialog)
buttons.rejected.connect(dialog.close)
layout.addWidget(buttons)
dialog.resize(420, 360)
self._metadata_dialog = dialog
self._metadata_dialog.setWindowTitle(f"Scan metadata · {self._selected_scan}")
self._metadata_dialog.show()
self._metadata_dialog.raise_()
# ---- run / stop ----------------------------------------------------------------------
@SafeSlot(popup_error=True)
def run_scan(self):
"""Start the selected scan; if fields are invalid, reveal every error instead."""
scan_name = self._selected_scan
if not scan_name or self._scan_metadata is None:
return
if not self.form.is_valid():
self._show_all_errors = True
self._sync_view(False)
return
args, kwargs = self.get_scan_parameters()
self.scan_args.emit(args)
scan_function = getattr(self.scans, scan_name)
if callable(scan_function):
self.scan_started.emit()
scan_function(*args, **kwargs)
@SafeSlot()
def stop_scan(self) -> None:
"""Abort the running scan; a second press within 2 s halts it (emergency stop)."""
if self._emergency_stop:
self.queue.request_scan_halt()
else:
self.queue.request_scan_abortion()
self._emergency_stop = True
self._emergency_timer.start(self.EMERGENCY_STOP_TIMEOUT_MS)
self._sync_view(False)
def _reset_emergency_stop(self) -> None:
self._emergency_stop = False
self._sync_view(False)
# ---- view state ----------------------------------------------------------------------
def view_state(self) -> dict:
"""Display-ready state shared by both renderers (values excluded from ``groups``)."""
errors = self.form.errors()
visible_errors = {
key: message
for key, message in errors.items()
if self._show_all_errors or key in self._touched
}
scan = self._selected_scan
if not scan:
hint = "No scan available" if not self._visible_scans else "Choose a scan"
elif self._scan_metadata is None:
hint = "Complete the metadata to start"
elif errors and self._show_all_errors:
count = len(errors)
hint = (
f"{count} field{'s' if count != 1 else ''} need{'s' if count == 1 else ''} a value"
)
else:
hint = ""
return {
"scans": self._visible_scans,
"current": scan,
"summary": doc_summary(self._scan_docstring(scan)) if scan else "",
"errors": visible_errors,
"canStart": bool(scan) and self._scan_metadata is not None,
"startLabel": f"Start {scan}" if scan else "Start",
"hint": hint,
"metadataValid": self._scan_metadata is not None,
"restoreBusy": self._restore_busy,
"stopLabel": "Emergency Stop" if self._emergency_stop else "Stop",
"emergency": self._emergency_stop,
"showSelector": not self._hide_scan_selection_combobox,
"showFilter": not self._hide_scan_selector_settings_button,
"showArgs": not self._hide_arg_box,
"showKwargs": not self._hide_kwarg_boxes,
"showButtons": not self._hide_scan_control_buttons,
"showMetadata": not self._hide_metadata,
"showRowButtons": not self._hide_add_remove_buttons,
}
# ---- Designer properties (same as ScanControl) ------------------------------------------
def _set_flag(self, name: str, value: bool) -> None:
setattr(self, name, bool(value))
self._sync_view(True)
@SafeProperty(bool)
def hide_arg_box(self) -> bool:
"""Hide the argument rows."""
return self._hide_arg_box
@hide_arg_box.setter
def hide_arg_box(self, hide: bool):
self._set_flag("_hide_arg_box", hide)
@SafeProperty(bool)
def hide_kwarg_boxes(self) -> bool:
"""Hide the keyword parameter groups."""
return self._hide_kwarg_boxes
@hide_kwarg_boxes.setter
def hide_kwarg_boxes(self, hide: bool):
self._set_flag("_hide_kwarg_boxes", hide)
@SafeProperty(bool)
def hide_scan_control_buttons(self) -> bool:
"""Hide Start and Stop."""
return self._hide_scan_control_buttons
@hide_scan_control_buttons.setter
def hide_scan_control_buttons(self, hide: bool):
self._set_flag("_hide_scan_control_buttons", hide)
@SafeSlot(bool)
def show_scan_control_buttons(self, show: bool):
"""Shows or hides the scan control buttons."""
self.hide_scan_control_buttons = not show
@SafeProperty(bool)
def hide_metadata(self) -> bool:
"""Hide the metadata button."""
return self._hide_metadata
@hide_metadata.setter
def hide_metadata(self, hide: bool):
self._set_flag("_hide_metadata", hide)
@SafeProperty(bool)
def hide_optional_metadata(self) -> bool:
"""Hide optional metadata fields in the metadata form."""
return self._metadata_form.hide_optional_metadata
@hide_optional_metadata.setter
def hide_optional_metadata(self, hide: bool):
self._metadata_form.hide_optional_metadata = hide
@SafeProperty(bool)
def hide_scan_selection_combobox(self) -> bool:
"""Hide the scan selector."""
return self._hide_scan_selection_combobox
@hide_scan_selection_combobox.setter
def hide_scan_selection_combobox(self, hide: bool):
self._set_flag("_hide_scan_selection_combobox", hide)
@SafeSlot(bool)
def show_scan_selection_combobox(self, show: bool):
"""Shows or hides the scan selection combobox."""
self.hide_scan_selection_combobox = not show
@SafeProperty(bool)
def hide_scan_selector_settings_button(self) -> bool:
"""Hide the scan filter button."""
return self._hide_scan_selector_settings_button
@hide_scan_selector_settings_button.setter
def hide_scan_selector_settings_button(self, hide: bool):
self._set_flag("_hide_scan_selector_settings_button", hide)
@SafeProperty(bool)
def hide_add_remove_buttons(self) -> bool:
"""Hide the buttons that add and remove argument rows."""
return self._hide_add_remove_buttons
@hide_add_remove_buttons.setter
def hide_add_remove_buttons(self, hide: bool):
self._set_flag("_hide_add_remove_buttons", hide)
# ---- lifecycle -----------------------------------------------------------------------
def cleanup(self):
"""Stop timers, close dialogs and release the renderer."""
self._last_scan_fetch_generation += 1
self._last_scan_fetch_watchdog.stop()
self._emergency_timer.stop()
for dialog in (self._scan_info_dialog, self._metadata_dialog):
if dialog is not None:
dialog.close()
dialog.deleteLater()
self._scan_info_dialog = None
self._metadata_dialog = None
super().cleanup()
# ---- renderer hooks ------------------------------------------------------------------
def _init_view(self) -> None:
"""Create the renderer."""
raise NotImplementedError
def _sync_view(self, structure: bool) -> None:
"""Redraw the renderer; ``structure`` is True when fields were added or removed."""
raise NotImplementedError
ScanControlPortBase.PORTED_FROM = ScanControlPortBase
@@ -0,0 +1,448 @@
"""Toolkit-independent model of the scan parameter form.
The QML and QWidget ports of :class:`ScanControl` both render this model. It turns the scan UI
configuration produced by :class:`ScanInfoAdapter` into groups of typed fields, keeps the entered
text of every field, validates it and converts it into scan ``args`` and ``kwargs``.
Field values are stored as the text the user typed (booleans as ``bool``), so a half-typed
number never loses keystrokes and validation can explain what is wrong.
"""
from __future__ import annotations
import math
from typing import Any, Callable
from qtpy.QtCore import QObject, Signal
from bec_widgets.utils.scan_arg_metadata import device_units
FIELD_TYPES = {"device": "device", "DeviceBase": "device", "float": "float", "int": "int"}
FIELD_TYPES.update({"bool": "bool", "str": "str"})
def field_type(item_type: Any) -> str:
"""Map a scan input type to a form field type."""
if isinstance(item_type, dict) and "Literal" in item_type:
return "literal"
return FIELD_TYPES.get(item_type, "str") if isinstance(item_type, str) else "str"
def literal_options(item_type: Any) -> list[str]:
"""Options of a ``Literal`` input as strings; ``None`` becomes an empty first option."""
values = list(dict.fromkeys(item_type.get("Literal", [])))
options = [""] if None in values else []
options += [str(value) for value in values if value is not None]
return options
def _format_number(value: Any) -> str:
if value is None or value == "_empty":
return ""
if isinstance(value, float) and value.is_integer() and abs(value) < 1e15:
return str(int(value))
return str(value)
class ScanFormModel(QObject):
"""Fields, values and validation of the scan parameter form.
Args:
device_names(Callable[[], list[str]]): Returns the names of selectable devices.
device_lookup(Callable[[str], Any]): Returns the device object for a name.
parent: Parent QObject.
"""
structure_changed = Signal()
values_changed = Signal()
device_selected = Signal(str)
def __init__(
self,
device_names: Callable[[], list[str]],
device_lookup: Callable[[str], Any],
parent: QObject | None = None,
):
super().__init__(parent)
self._device_names = device_names
self._device_lookup = device_lookup
self._arg_group: dict | None = None
self._kwarg_groups: list[dict] = []
self._bundles = 0
self._values: dict[str, Any] = {}
self._last_devices = ""
# ---- structure -----------------------------------------------------------------------
def load(self, gui_config: dict | None) -> None:
"""Build the fields of a scan from its UI configuration (empty config clears the form)."""
gui_config = gui_config or {}
arg_group = gui_config.get("arg_group")
self._arg_group = arg_group if arg_group and arg_group.get("inputs") else None
self._kwarg_groups = [
group for group in gui_config.get("kwarg_groups", []) if group.get("inputs")
]
self._values = {}
self._bundles = 0
if self._arg_group is not None:
for _ in range(max(1, int(self._arg_group.get("min") or 1))):
self._append_bundle()
for group in self._kwarg_groups:
for item in group["inputs"]:
self._values[self.kwarg_key(item["name"])] = self._initial_value(item)
self.structure_changed.emit()
self._values_updated()
@staticmethod
def arg_key(bundle: int, name: str) -> str:
"""Key of an argument field in bundle ``bundle``."""
return f"args:{bundle}:{name}"
@staticmethod
def kwarg_key(name: str) -> str:
"""Key of a keyword argument field."""
return f"kwargs:{name}"
@staticmethod
def _initial_value(item: dict) -> Any:
kind = field_type(item.get("type"))
default = item.get("default")
if default == "_empty":
default = None
if kind == "bool":
return bool(default)
if kind == "literal":
options = literal_options(item["type"])
text = "" if default is None else str(default)
if text in options:
return text
return options[0] if options else ""
return (
_format_number(default)
if kind in ("float", "int")
else ("" if default is None else str(default))
)
def _append_bundle(self) -> None:
for item in self._arg_group["inputs"]:
self._values[self.arg_key(self._bundles, item["name"])] = self._initial_value(item)
self._bundles += 1
@property
def has_args(self) -> bool:
"""Whether the scan has a positional argument group."""
return self._arg_group is not None
@property
def bundle_count(self) -> int:
"""Number of rows in the argument group."""
return self._bundles
def can_add_bundle(self) -> bool:
"""Whether another argument row may be added."""
if self._arg_group is None:
return False
maximum = self._arg_group.get("max")
return maximum is None or self._bundles < maximum
def can_remove_bundle(self) -> bool:
"""Whether an argument row may be removed."""
if self._arg_group is None:
return False
return self._bundles > max(1, int(self._arg_group.get("min") or 1))
def add_bundle(self) -> bool:
"""Append an argument row. Returns False when the maximum is reached."""
if not self.can_add_bundle():
return False
self._append_bundle()
self.structure_changed.emit()
self._values_updated()
return True
def remove_bundle(self, index: int | None = None) -> bool:
"""Remove an argument row (the last one by default)."""
if not self.can_remove_bundle():
return False
index = self._bundles - 1 if index is None else index
if not 0 <= index < self._bundles:
return False
names = [item["name"] for item in self._arg_group["inputs"]]
for bundle in range(index, self._bundles - 1):
for name in names:
self._values[self.arg_key(bundle, name)] = self._values[
self.arg_key(bundle + 1, name)
]
for name in names:
self._values.pop(self.arg_key(self._bundles - 1, name), None)
self._bundles -= 1
self.structure_changed.emit()
self._values_updated()
return True
def _items(self) -> dict[str, dict]:
items = {}
if self._arg_group is not None:
for bundle in range(self._bundles):
for item in self._arg_group["inputs"]:
items[self.arg_key(bundle, item["name"])] = item
for group in self._kwarg_groups:
for item in group["inputs"]:
items[self.kwarg_key(item["name"])] = item
return items
# ---- values --------------------------------------------------------------------------
def value(self, key: str) -> Any:
"""Raw value of a field (text, or bool for switches)."""
return self._values.get(key)
def set_value(self, key: str, value: Any) -> None:
"""Set the raw value of a field."""
if key not in self._values:
return
if isinstance(value, float) and math.isnan(value):
value = ""
if self._values[key] == value:
return
self._values[key] = value
self._values_updated()
def _values_updated(self) -> None:
self.values_changed.emit()
devices = " ".join(self.selected_devices())
if devices != self._last_devices:
self._last_devices = devices
self.device_selected.emit(devices)
def selected_devices(self) -> list[str]:
"""Valid devices chosen in the argument rows, in order."""
if self._arg_group is None:
return []
known = set(self._device_names())
devices = []
for bundle in range(self._bundles):
for item in self._arg_group["inputs"]:
if field_type(item["type"]) != "device":
continue
name = str(self._values.get(self.arg_key(bundle, item["name"]), "")).strip()
if name in known:
devices.append(name)
return devices
# ---- validation and conversion --------------------------------------------------------
def _parse(self, key: str, item: dict) -> tuple[Any, str]:
"""Return ``(value, error)`` for a field."""
kind = field_type(item.get("type"))
raw = self._values.get(key)
is_arg = key.startswith("args:")
if kind == "bool":
return bool(raw), ""
text = "" if raw is None else str(raw).strip()
if kind == "device":
if not text:
return None, "Select a device" if is_arg else ""
if text not in self._device_names():
return None, f"Unknown device '{text}'"
return text, ""
if kind == "literal":
return (text or None), ""
if kind == "str":
return text, ""
if not text:
default = item.get("default")
if not is_arg and default not in (None, "_empty"):
return default, ""
return None, "Required"
try:
number = int(text) if kind == "int" else float(text)
except ValueError:
return None, "Enter a whole number" if kind == "int" else "Enter a number"
for bound, check, message in (
("ge", lambda v, b: v >= b, "≥"),
("gt", lambda v, b: v > b, ">"),
("le", lambda v, b: v <= b, "≤"),
("lt", lambda v, b: v < b, "<"),
):
limit = item.get(bound)
if limit is not None and not check(number, limit):
return None, f"Must be {message} {_format_number(limit)}"
return number, ""
def errors(self) -> dict[str, str]:
"""Error message per field key, only for fields with a problem."""
errors = {}
for key, item in self._items().items():
_value, error = self._parse(key, item)
if error:
errors[key] = error
return errors
def is_valid(self) -> bool:
"""Whether every field holds a usable value."""
return not self.errors()
def args(self, bec_object: bool = True) -> list:
"""Positional arguments, flattened over the rows.
Args:
bec_object(bool): Return device objects instead of device names.
"""
if self._arg_group is None:
return []
args = []
for bundle in range(self._bundles):
for item in self._arg_group["inputs"]:
key = self.arg_key(bundle, item["name"])
value, error = self._parse(key, item)
if field_type(item["type"]) == "device":
if error:
value = str(self._values.get(key, ""))
elif bec_object and value:
value = self._device_lookup(value)
args.append(value)
return args
def kwargs(self) -> dict:
"""Keyword arguments; device fields are passed by name."""
kwargs = {}
for group in self._kwarg_groups:
for item in group["inputs"]:
key = self.kwarg_key(item["name"])
value, _error = self._parse(key, item)
kwargs[item["name"]] = value
return kwargs
def set_args(self, args: list) -> None:
"""Fill the argument rows from a flat list, adding rows as needed."""
if self._arg_group is None or not args:
return
names = [item["name"] for item in self._arg_group["inputs"]]
if not names:
return
bundles = max(-(-len(args) // len(names)), int(self._arg_group.get("min") or 1))
maximum = self._arg_group.get("max")
if maximum is not None:
bundles = min(bundles, maximum)
for key in [key for key in self._values if key.startswith("args:")]:
del self._values[key]
self._bundles = 0
for _ in range(bundles):
self._append_bundle()
for index, value in enumerate(args[: bundles * len(names)]):
bundle, position = divmod(index, len(names))
item = self._arg_group["inputs"][position]
self._values[self.arg_key(bundle, item["name"])] = self._to_raw(item, value)
self.structure_changed.emit()
self._values_updated()
def set_kwargs(self, kwargs: dict) -> None:
"""Fill keyword fields from a dict; unknown keys are ignored."""
changed = False
for group in self._kwarg_groups:
for item in group["inputs"]:
if item["name"] not in kwargs:
continue
key = self.kwarg_key(item["name"])
raw = self._to_raw(item, kwargs[item["name"]])
if self._values.get(key) != raw:
self._values[key] = raw
changed = True
if changed:
self.structure_changed.emit()
self._values_updated()
@staticmethod
def _to_raw(item: dict, value: Any) -> Any:
kind = field_type(item.get("type"))
if kind == "bool":
return bool(value)
if kind == "device":
return getattr(value, "name", "" if value is None else str(value))
if value is None:
return ""
return _format_number(value) if kind in ("float", "int") else str(value)
# ---- presentation ---------------------------------------------------------------------
def _field(self, key: str, item: dict, units_for: Callable[[dict], str]) -> dict:
kind = field_type(item.get("type"))
units = units_for(item)
default = item.get("default")
placeholder = ""
if kind in ("float", "int") and default not in (None, "_empty"):
placeholder = f"default {_format_number(default)}"
return {
"key": key,
"name": item["name"],
"label": item.get("display_name") or item["name"],
"kind": kind,
"value": self._values.get(key),
"options": literal_options(item["type"]) if kind == "literal" else [],
"tooltip": item.get("tooltip") or "",
"units": units,
"placeholder": placeholder,
}
def _units_resolver(self, bundle: int | None) -> Callable[[dict], str]:
def units_for(item: dict) -> str:
if item.get("units"):
return str(item["units"])
reference = item.get("reference_units")
if not reference:
return ""
candidates = []
if bundle is not None:
candidates.append(self.arg_key(bundle, reference))
candidates.append(self.kwarg_key(reference))
candidates.append(self.arg_key(0, reference))
for key in candidates:
name = self._values.get(key)
if name and name in self._device_names():
return device_units(self._device_lookup(name)) or ""
return ""
return units_for
def groups(self) -> list[dict]:
"""Description of all groups and fields for the renderers."""
groups = []
if self._arg_group is not None:
rows = []
for bundle in range(self._bundles):
units_for = self._units_resolver(bundle)
rows.append(
[
self._field(self.arg_key(bundle, item["name"]), item, units_for)
for item in self._arg_group["inputs"]
]
)
groups.append(
{
"id": "args",
"title": self._arg_group.get("name") or "Scan arguments",
"kind": "args",
"rows": rows,
"fields": [],
"canAdd": self.can_add_bundle(),
"canRemove": self.can_remove_bundle(),
}
)
units_for = self._units_resolver(None)
for index, group in enumerate(self._kwarg_groups):
groups.append(
{
"id": f"kwargs{index}",
"title": group.get("name") or "Parameters",
"kind": "kwargs",
"rows": [],
"fields": [
self._field(self.kwarg_key(item["name"]), item, units_for)
for item in group["inputs"]
],
"canAdd": False,
"canRemove": False,
}
)
return groups
+327
View File
@@ -0,0 +1,327 @@
"""Tests for the scan form model and the QML and QWidget ports of ScanControl."""
# pylint: disable=redefined-outer-name, protected-access, missing-function-docstring
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from bec_lib.endpoints import MessageEndpoints
from qtpy.QtQuickWidgets import QQuickWidget
from bec_widgets.widgets.control.scan_control.scan_control_qml import ScanControlQML
from bec_widgets.widgets.control.scan_control.scan_control_qwidget import ScanControlQWidget
from bec_widgets.widgets.control.scan_control.scan_control_ux_common import doc_summary
from bec_widgets.widgets.control.scan_control.scan_form_model import (
ScanFormModel,
field_type,
literal_options,
)
from .conftest import create_widget
from .test_scan_control import available_scans_message, scan_history
GRID = available_scans_message.resource["grid_scan"]["gui_config"]
LINE = available_scans_message.resource["line_scan"]["gui_config"]
DEVICES = {"samx": SimpleNamespace(name="samx"), "samy": SimpleNamespace(name="samy")}
@pytest.fixture
def form():
return ScanFormModel(lambda: sorted(DEVICES), DEVICES.__getitem__)
@pytest.fixture(params=[ScanControlQML, ScanControlQWidget])
def control(request, qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
mocked_client.connector.xadd(
topic=MessageEndpoints.scan_history(), msg_dict={"data": scan_history}
)
widget = create_widget(qtbot, request.param, client=mocked_client)
widget.resize(460, 640)
yield widget
def _fill_line_scan(control, device="samx"):
control.current_scan = "line_scan"
control.set_field("args:0:device", device)
control.set_field("args:0:start", "0")
control.set_field("args:0:stop", "2.5")
control.set_field("kwargs:steps", "10")
# ---- form model ------------------------------------------------------------------------
def test_field_types_and_literals():
assert field_type("DeviceBase") == "device"
assert field_type({"Literal": ("a", None)}) == "literal"
assert field_type("whatever") == "str"
assert literal_options({"Literal": ("a", None, "b", "a")}) == ["", "a", "b"]
assert doc_summary("Run a scan.\n More.\n\nArgs:\n x") == "Run a scan. More."
def test_form_loads_minimum_rows_and_defaults(form):
form.load(GRID)
assert form.bundle_count == 2
assert not form.can_remove_bundle()
assert form.can_add_bundle()
assert form.value("kwargs:exp_time") == "0"
assert form.value("kwargs:relative") is False
assert form.value("kwargs:optim_trajectory") == ""
groups = form.groups()
assert [group["kind"] for group in groups] == ["args", "kwargs"]
assert len(groups[0]["rows"]) == 2
assert groups[1]["fields"][4]["options"] == ["", "option1", "option2", "option3"]
def test_form_validation_messages(form):
form.load(LINE)
errors = form.errors()
assert errors["args:0:device"] == "Select a device"
assert errors["args:0:start"] == "Required"
assert errors["kwargs:steps"] == "Required"
form.set_value("args:0:device", "nope")
form.set_value("args:0:start", "abc")
form.set_value("kwargs:steps", "1.5")
errors = form.errors()
assert errors["args:0:device"] == "Unknown device 'nope'"
assert errors["args:0:start"] == "Enter a number"
assert errors["kwargs:steps"] == "Enter a whole number"
def test_form_numeric_limits(form):
config = {
"kwarg_groups": [
{"name": "G", "inputs": [{"name": "n", "type": "int", "default": 1, "ge": 1}]}
]
}
form.load(config)
form.set_value("kwargs:n", "0")
assert form.errors() == {"kwargs:n": "Must be ≥ 1"}
form.set_value("kwargs:n", "")
assert form.is_valid()
assert form.kwargs() == {"n": 1}
def test_form_args_kwargs_conversion(form):
form.load(LINE)
form.set_value("args:0:device", "samx")
form.set_value("args:0:start", "0")
form.set_value("args:0:stop", "2")
form.set_value("kwargs:steps", "10")
assert form.is_valid()
assert form.args(bec_object=False) == ["samx", 0.0, 2.0]
assert form.args()[0] is DEVICES["samx"]
kwargs = form.kwargs()
assert kwargs["steps"] == 10 and kwargs["relative"] is False and kwargs["exp_time"] == 0
def test_form_rows_add_remove_and_restore(form):
form.load(LINE)
assert form.add_bundle()
form.set_value("args:1:device", "samy")
assert form.selected_devices() == ["samy"]
assert form.remove_bundle(0)
assert form.bundle_count == 1
assert form.value("args:0:device") == "samy"
form.set_args(["samx", 0, 1, "samy", 2, 3])
assert form.bundle_count == 2
assert form.args(bec_object=False) == ["samx", 0.0, 1.0, "samy", 2.0, 3.0]
form.set_kwargs({"steps": 5, "unknown": 1, "relative": True})
assert form.value("kwargs:steps") == "5"
assert form.value("kwargs:relative") is True
def test_form_device_selected_signal(form, qtbot):
form.load(LINE)
with qtbot.waitSignal(form.device_selected) as blocker:
form.set_value("args:0:device", "samx")
assert blocker.args == ["samx"]
def test_form_units_from_reference(form):
config = {
"arg_group": {
"name": "A",
"min": 1,
"max": None,
"inputs": [
{"name": "device", "type": "device"},
{"name": "start", "type": "float", "reference_units": "device"},
{"name": "width", "type": "float", "units": "mm"},
],
}
}
form.load(config)
form.set_value("args:0:device", "samx")
with patch(
"bec_widgets.widgets.control.scan_control.scan_form_model.device_units", return_value="deg"
):
fields = form.groups()[0]["rows"][0]
assert fields[1]["units"] == "deg"
assert fields[2]["units"] == "mm"
# ---- ports -----------------------------------------------------------------------------
def test_ports_populate_and_select(control):
assert control.visible_scans() == ["line_scan", "grid_scan"]
assert control.current_scan == "line_scan"
control.current_scan = "GRID_SCAN"
assert control.current_scan == "grid_scan"
control.current_scan = "unknown"
assert control.current_scan == "grid_scan"
assert control.view_state()["summary"] == "Run a grid scan over one or more devices."
def test_ports_allowed_scans_filter(control):
control.allowed_scans = ["grid_scan"]
assert control.visible_scans() == ["grid_scan"]
assert control.current_scan == "grid_scan"
control.allowed_scans = []
assert control.config.allowed_scans is None
assert control.allowed_scans == ["line_scan", "grid_scan"]
def test_ports_errors_appear_after_edit_or_start(control):
assert control.view_state()["errors"] == {}
control.set_field("args:0:start", "x")
assert control.view_state()["errors"] == {"args:0:start": "Enter a number"}
scans = SimpleNamespace(line_scan=MagicMock())
with patch.object(control, "scans", scans):
control.run_scan()
scans.line_scan.assert_not_called()
state = control.view_state()
assert "args:0:device" in state["errors"]
assert state["hint"].endswith("need a value")
def test_ports_run_scan(control, qtbot):
_fill_line_scan(control)
scans = SimpleNamespace(line_scan=MagicMock())
with (
patch.object(control, "scans", scans),
qtbot.waitSignal(control.scan_started),
qtbot.waitSignal(control.scan_args) as args_signal,
):
control.run_scan()
args, kwargs = scans.line_scan.call_args
assert args[0] is control.dev["samx"]
assert args[1:] == (0.0, 2.5)
assert kwargs["steps"] == 10
assert "metadata" in kwargs
assert args_signal.args[0][1:] == [0.0, 2.5]
def test_ports_device_selected_signal(control, qtbot):
control.current_scan = "line_scan"
with qtbot.waitSignal(control.device_selected) as blocker:
control.set_field("args:0:device", "samx")
assert blocker.args == ["samx"]
def test_ports_remember_parameters_per_scan(control):
_fill_line_scan(control)
control.current_scan = "grid_scan"
control.current_scan = "line_scan"
args, kwargs = control.get_scan_parameters(bec_object=False)
assert args == ["samx", 0.0, 2.5]
assert kwargs["steps"] == 10
def test_ports_restore_last_executed(control, qtbot):
control.current_scan = "line_scan"
control.request_last_executed_scan_parameters()
qtbot.waitUntil(lambda: not control.view_state()["restoreBusy"], timeout=5000)
qtbot.waitUntil(lambda: control.form.value("args:0:device") == "samx", timeout=5000)
args, kwargs = control.get_scan_parameters(bec_object=False)
assert args == ["samx", 0.0, 2.0]
assert kwargs["steps"] == 10
assert kwargs["exp_time"] == 2
def test_ports_stop_is_two_stage(control):
queue = MagicMock()
with patch.object(control, "queue", queue):
control.stop_scan()
queue.request_scan_abortion.assert_called_once()
assert control.view_state()["stopLabel"] == "Emergency Stop"
control.stop_scan()
queue.request_scan_halt.assert_called_once()
control._reset_emergency_stop()
assert control.view_state()["stopLabel"] == "Stop"
def test_ports_hide_flags(control):
for prop, key in [
("hide_arg_box", "showArgs"),
("hide_kwarg_boxes", "showKwargs"),
("hide_scan_control_buttons", "showButtons"),
("hide_metadata", "showMetadata"),
("hide_scan_selection_combobox", "showSelector"),
("hide_scan_selector_settings_button", "showFilter"),
("hide_add_remove_buttons", "showRowButtons"),
]:
setattr(control, prop, True)
assert getattr(control, prop)
assert control.view_state()[key] is False
setattr(control, prop, False)
assert control.view_state()[key] is True
def test_ports_metadata_dialog(control):
control.show_metadata_dialog()
assert control._metadata_dialog.isVisible()
assert control._metadata_form.parent() is control._metadata_dialog
control._metadata_dialog.close()
def test_ports_export_properties(control):
names = set(control._get_bec_meta_objects())
assert {"current_scan", "allowed_scans", "hide_arg_box", "hide_metadata"} <= names
def test_qml_view_and_backend(qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
widget = create_widget(qtbot, ScanControlQML, client=mocked_client)
assert widget.view.status() == QQuickWidget.Status.Ready
assert widget.view.errors() == []
backend = widget.backend
assert [group["kind"] for group in backend.groups] == ["args", "kwargs", "kwargs"]
assert "samx" in backend.devices
backend.selectScan("grid_scan")
assert backend.state["current"] == "grid_scan"
backend.addRow()
assert len(backend.groups[0]["rows"]) == 3
backend.removeRow(0)
assert len(backend.groups[0]["rows"]) == 2
backend.setField("kwargs:exp_time", "x")
assert backend.state["errors"] == {"kwargs:exp_time": "Enter a number"}
def test_qwidget_editors(qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
widget = create_widget(qtbot, ScanControlQWidget, client=mocked_client)
widget.resize(460, 640)
widget.show()
qtbot.waitExposed(widget)
editor = widget.editors["args:0:start"]
editor.editor.setText("abc")
editor.editor.textEdited.emit("abc")
assert editor.error.text() == "Enter a number"
assert editor.editor.property("invalid")
widget.editors["args:0:device"].editor.setCurrentText("samx")
assert widget.form.value("args:0:device") == "samx"
widget.current_scan = "grid_scan"
switch = widget.editors["kwargs:relative"].editor
switch.click()
assert widget.form.value("kwargs:relative") is True
assert widget.start_button.text() == "Start grid_scan"
assert len(widget.group_cards) == 2