From 10bed0fa42a6bc6cd21f1c7959c744fdf26e9924 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 20:37:46 +0000 Subject: [PATCH] feat(scan_control): add QML and QWidget ports with a validated form model, per-row removal and a sticky footer --- bec_widgets/utils/ux_kit.py | 94 +++ .../scan_control/qml/ScanControlView.qml | 347 +++++++++ .../control/scan_control/scan_control_qml.py | 128 ++++ .../scan_control/scan_control_qwidget.py | 410 +++++++++++ .../scan_control/scan_control_ux_common.py | 660 ++++++++++++++++++ .../control/scan_control/scan_form_model.py | 448 ++++++++++++ tests/unit_tests/test_scan_control_ports.py | 327 +++++++++ 7 files changed, 2414 insertions(+) create mode 100644 bec_widgets/widgets/control/scan_control/qml/ScanControlView.qml create mode 100644 bec_widgets/widgets/control/scan_control/scan_control_qml.py create mode 100644 bec_widgets/widgets/control/scan_control/scan_control_qwidget.py create mode 100644 bec_widgets/widgets/control/scan_control/scan_control_ux_common.py create mode 100644 bec_widgets/widgets/control/scan_control/scan_form_model.py create mode 100644 tests/unit_tests/test_scan_control_ports.py diff --git a/bec_widgets/utils/ux_kit.py b/bec_widgets/utils/ux_kit.py index 11719f28..c8497afc 100644 --- a/bec_widgets/utils/ux_kit.py +++ b/bec_widgets/utils/ux_kit.py @@ -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() diff --git a/bec_widgets/widgets/control/scan_control/qml/ScanControlView.qml b/bec_widgets/widgets/control/scan_control/qml/ScanControlView.qml new file mode 100644 index 00000000..87fa56b1 --- /dev/null +++ b/bec_widgets/widgets/control/scan_control/qml/ScanControlView.qml @@ -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() + } + } + } + } + } +} diff --git a/bec_widgets/widgets/control/scan_control/scan_control_qml.py b/bec_widgets/widgets/control/scan_control/scan_control_qml.py new file mode 100644 index 00000000..a757d5ce --- /dev/null +++ b/bec_widgets/widgets/control/scan_control/scan_control_qml.py @@ -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()) diff --git a/bec_widgets/widgets/control/scan_control/scan_control_qwidget.py b/bec_widgets/widgets/control/scan_control/scan_control_qwidget.py new file mode 100644 index 00000000..e1b637c8 --- /dev/null +++ b/bec_widgets/widgets/control/scan_control/scan_control_qwidget.py @@ -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()) diff --git a/bec_widgets/widgets/control/scan_control/scan_control_ux_common.py b/bec_widgets/widgets/control/scan_control/scan_control_ux_common.py new file mode 100644 index 00000000..4dc03758 --- /dev/null +++ b/bec_widgets/widgets/control/scan_control/scan_control_ux_common.py @@ -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 diff --git a/bec_widgets/widgets/control/scan_control/scan_form_model.py b/bec_widgets/widgets/control/scan_control/scan_form_model.py new file mode 100644 index 00000000..dedf9279 --- /dev/null +++ b/bec_widgets/widgets/control/scan_control/scan_form_model.py @@ -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 diff --git a/tests/unit_tests/test_scan_control_ports.py b/tests/unit_tests/test_scan_control_ports.py new file mode 100644 index 00000000..deec8b96 --- /dev/null +++ b/tests/unit_tests/test_scan_control_ports.py @@ -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