From 58fc7a07ac722b206d029c5e07e9b53bbdeebb35 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 20:29:20 +0000 Subject: [PATCH] feat(positioner_box): add QML and QWidget ports with status pill, limit bar and validated moves --- .../utils/quick/qml/BecUi/SelectField.qml | 3 + .../positioner_box/positioner_box_qml.py | 113 +++++ .../positioner_box/positioner_box_qwidget.py | 354 +++++++++++++++ .../positioner_box/positioner_ux_common.py | 413 ++++++++++++++++++ .../positioner_box/qml/PositionerBoxView.qml | 231 ++++++++++ tests/unit_tests/test_positioner_box_ports.py | 173 ++++++++ 6 files changed, 1287 insertions(+) create mode 100644 bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qml.py create mode 100644 bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qwidget.py create mode 100644 bec_widgets/widgets/control/device_control/positioner_box/positioner_ux_common.py create mode 100644 bec_widgets/widgets/control/device_control/positioner_box/qml/PositionerBoxView.qml create mode 100644 tests/unit_tests/test_positioner_box_ports.py diff --git a/bec_widgets/utils/quick/qml/BecUi/SelectField.qml b/bec_widgets/utils/quick/qml/BecUi/SelectField.qml index d2149dce..2d4f814b 100644 --- a/bec_widgets/utils/quick/qml/BecUi/SelectField.qml +++ b/bec_widgets/utils/quick/qml/BecUi/SelectField.qml @@ -5,6 +5,7 @@ import QtQuick.Controls.Basic ComboBox { id: root property bool invalid: false + property string placeholder: "" implicitHeight: 32 implicitWidth: 140 @@ -22,6 +23,8 @@ ComboBox { rightPadding: 4 text: root.editable ? root.editText : root.displayText readOnly: !root.editable + placeholderText: root.placeholder + placeholderTextColor: theme.fgSubtle enabled: root.enabled color: theme.fg selectionColor: theme.primary diff --git a/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qml.py b/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qml.py new file mode 100644 index 00000000..21ddb767 --- /dev/null +++ b/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qml.py @@ -0,0 +1,113 @@ +"""Positioner box rendered with Qt Quick (QML), same API as :class:`PositionerBox`.""" + +from __future__ import annotations + +from pathlib import Path + +from qtpy.QtCore import Property, QObject, Signal, Slot + +from bec_widgets.utils.quick import create_quick_widget, release_quick_widget +from bec_widgets.widgets.control.device_control.positioner_box.positioner_ux_common import ( + PositionerBoxPortBase, +) + +QML_FILE = Path(__file__).parent / "qml" / "PositionerBoxView.qml" + + +class PositionerBackend(QObject): + """Bridge between :class:`PositionerBoxQML` and ``PositionerBoxView.qml``.""" + + changed = Signal() + positioners_changed = Signal() + + def __init__(self, box: "PositionerBoxQML"): + super().__init__(box) + self._box = box + self._state: dict = {} + self._positioners: list[str] = [] + + def update_from(self, state: dict) -> None: + """Take a new view state from the box.""" + self._state = state + self.changed.emit() + + @Slot() + def refreshPositioners(self) -> None: # pylint: disable=invalid-name + """Reload the positioner list, e.g. when the selector opens.""" + names = self._box.positioner_names() + if names != self._positioners: + self._positioners = names + self.positioners_changed.emit() + + @Slot(str) + def selectDevice(self, name: str) -> None: # pylint: disable=invalid-name + """Switch to another positioner.""" + self._box.set_positioner(name) + + @Slot(str, result=str) + def validate(self, text: str) -> str: + """Return an error message for a move target or an empty string.""" + return self._box.validate_target(text) + + @Slot(str, result=bool) + def moveTo(self, text: str) -> bool: # pylint: disable=invalid-name + """Request an absolute move.""" + return self._box.move_to(text) + + @Slot(int) + def tweak(self, direction: int) -> None: + """Tweak by one step; direction is -1 or 1.""" + if direction < 0: + self._box.on_tweak_left() + else: + self._box.on_tweak_right() + + @Slot(float) + def setStep(self, step: float) -> None: # pylint: disable=invalid-name + """Change the tweak step.""" + self._box.step_size = step + + @Slot() + def stop(self) -> None: + """Stop the positioner.""" + self._box.on_stop() + + state = Property("QVariantMap", lambda self: self._state, notify=changed) + positioners = Property( + "QStringList", lambda self: self._positioners, notify=positioners_changed + ) + + +class PositionerBoxQML(PositionerBoxPortBase): + """Positioner box with the modernised UX rendered in QML.""" + + def _init_view(self) -> None: + self.backend = PositionerBackend(self) + self.view = create_quick_widget(self, QML_FILE, {"backend": self.backend}) + self.view.setMinimumSize(260, 250) + self.main_layout.addWidget(self.view) + + def _sync_view(self) -> None: + self.backend.update_from(self.view_state()) + + 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 = PositionerBoxQML(device="samx") + widget.show() + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qwidget.py b/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qwidget.py new file mode 100644 index 00000000..b9bd5424 --- /dev/null +++ b/bec_widgets/widgets/control/device_control/positioner_box/positioner_box_qwidget.py @@ -0,0 +1,354 @@ +"""Positioner box with the modernised UX in plain QWidgets, same API as :class:`PositionerBox`. + +This is the QWidget twin of :class:`PositionerBoxQML`, built from the same state so both +rendering stacks can be compared side by side. +""" + +from __future__ import annotations + +from qtpy.QtCore import QEasingCurve, QRectF, Qt, QVariantAnimation +from qtpy.QtGui import QColor, QDoubleValidator, QFont, QPainter +from qtpy.QtWidgets import ( + QComboBox, + QCompleter, + QHBoxLayout, + QLabel, + QLineEdit, + QSizePolicy, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.quick.host import ThemeTokens +from bec_widgets.utils.ux_kit import ( + Card, + StatusPill, + TextButton, + field_qss, + refresh_kit_theme, + set_invalid, +) +from bec_widgets.widgets.control.device_control.positioner_box.positioner_ux_common import ( + PositionerBoxPortBase, +) + + +class LimitBar(QWidget): + """Horizontal limit track with the current position and the target marker.""" + + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + self.setFixedHeight(14) + self._position = -1.0 + self._shown_position = -1.0 + self._target = -1.0 + self._show_target = False + self._at_limit = False + self._animation = QVariantAnimation(self) + self._animation.setDuration(180) + self._animation.setEasingCurve(QEasingCurve.Type.OutCubic) + self._animation.valueChanged.connect(self._on_step) + + def set_values(self, position: float, target: float, show_target: bool, at_limit: bool): + """Update the markers; the position marker glides to its new place.""" + self._target = target + self._show_target = show_target + self._at_limit = at_limit + if position != self._position: + start = self._shown_position if self._shown_position >= 0 else position + self._position = position + self._animation.stop() + if position >= 0 and start >= 0: + self._animation.setStartValue(float(start)) + self._animation.setEndValue(float(position)) + self._animation.start() + else: + self._shown_position = position + self.update() + + def _on_step(self, value: float) -> None: + self._shown_position = float(value) + self.update() + + def stop_animation(self) -> None: + """Stop the marker animation.""" + self._animation.stop() + + def paintEvent(self, _event): # pylint: disable=invalid-name + tokens = ThemeTokens() + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + painter.setPen(Qt.PenStyle.NoPen) + mid = self.height() / 2 + painter.setBrush(tokens.track) + painter.drawRoundedRect(QRectF(0, mid - 2, self.width(), 4), 2, 2) + if self._show_target and self._target >= 0: + painter.setBrush(tokens.fg_muted) + painter.drawRect(QRectF(self._target * (self.width() - 2), 0, 2, self.height())) + if self._shown_position >= 0: + x = self._shown_position * (self.width() - 12) + painter.setBrush(tokens.card) + painter.drawEllipse(QRectF(x - 1, mid - 7, 14, 14)) + painter.setBrush(tokens.warning if self._at_limit else tokens.primary) + painter.drawEllipse(QRectF(x + 1, mid - 5, 10, 10)) + painter.end() + + +class PositionerBoxQWidget(PositionerBoxPortBase): + """Positioner box with the modernised UX in plain QWidgets.""" + + def _init_view(self) -> None: + self.setMinimumSize(260, 250) + self.card = Card(parent=self) + self.main_layout.setContentsMargins(8, 8, 8, 8) + self.main_layout.addWidget(self.card) + body = self.card.body + body.setSpacing(10) + + # header + header = QHBoxLayout() + header.setSpacing(8) + self.selector = QComboBox(self.card) + self.selector.setEditable(True) + self.selector.setInsertPolicy(QComboBox.InsertPolicy.NoInsert) + self.selector.setAccessibleName("Positioner") + self.selector.lineEdit().setPlaceholderText("Select positioner") + completer = QCompleter(self.selector.model(), self.selector) + completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive) + completer.setFilterMode(Qt.MatchFlag.MatchContains) + self.selector.setCompleter(completer) + self.selector.activated.connect( + lambda index: self.set_positioner(self.selector.itemText(index)) + ) + self.selector.lineEdit().returnPressed.connect( + lambda: self.set_positioner(self.selector.currentText()) + ) + self.device_label = QLabel(self.card) + self.status_pill = StatusPill(self.card) + header.addWidget(self.selector, 1) + header.addWidget(self.device_label, 1) + header.addWidget(self.status_pill) + body.addLayout(header) + + # readback + self.readback_block = QWidget(self.card) + readback_layout = QVBoxLayout(self.readback_block) + readback_layout.setContentsMargins(0, 0, 0, 0) + readback_layout.setSpacing(0) + value_row = QHBoxLayout() + value_row.setSpacing(6) + self.readback_label = QLabel(self.readback_block) + self.readback_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + self.units_label = QLabel(self.readback_block) + value_row.addWidget(self.readback_label) + value_row.addWidget(self.units_label, 0, Qt.AlignmentFlag.AlignBaseline) + value_row.addStretch(1) + readback_layout.addLayout(value_row) + self.target_label = QLabel(self.readback_block) + readback_layout.addWidget(self.target_label) + body.addWidget(self.readback_block) + + # limits + self.limits_block = QWidget(self.card) + limits_layout = QVBoxLayout(self.limits_block) + limits_layout.setContentsMargins(0, 0, 0, 0) + limits_layout.setSpacing(3) + self.limit_bar = LimitBar(self.limits_block) + limits_layout.addWidget(self.limit_bar) + limit_labels = QHBoxLayout() + self.limit_low = QLabel(self.limits_block) + self.limit_high = QLabel(self.limits_block) + limit_labels.addWidget(self.limit_low) + limit_labels.addStretch(1) + limit_labels.addWidget(self.limit_high) + limits_layout.addLayout(limit_labels) + body.addWidget(self.limits_block) + + # absolute move + self.move_block = QWidget(self.card) + move_layout = QVBoxLayout(self.move_block) + move_layout.setContentsMargins(0, 0, 0, 0) + move_layout.setSpacing(4) + move_row = QHBoxLayout() + move_row.setSpacing(6) + self.target_input = QLineEdit(self.move_block) + self.target_input.setPlaceholderText("Move to…") + self.target_input.setAccessibleName("Target position") + self.target_input.textChanged.connect(self._on_target_edited) + self.target_input.returnPressed.connect(self._submit_target) + self.go_button = TextButton("Go", "primary", parent=self.move_block) + self.go_button.clicked.connect(self._submit_target) + move_row.addWidget(self.target_input, 1) + move_row.addWidget(self.go_button) + move_layout.addLayout(move_row) + self.error_label = QLabel(self.move_block) + self.error_label.setVisible(False) + move_layout.addWidget(self.error_label) + body.addWidget(self.move_block) + + # tweak + self.tweak_block = QWidget(self.card) + tweak_row = QHBoxLayout(self.tweak_block) + tweak_row.setContentsMargins(0, 0, 0, 0) + tweak_row.setSpacing(6) + self.tweak_left = TextButton("", "neutral", "chevron_left", self.tweak_block) + self.tweak_left.setToolTip("Tweak down by the step") + self.tweak_left.clicked.connect(self.on_tweak_left) + self.step_input = QLineEdit(self.tweak_block) + self.step_input.setFixedWidth(78) + self.step_input.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.step_input.setToolTip("Step size") + self.step_input.setAccessibleName("Step size") + step_validator = QDoubleValidator(0.0, float("inf"), 12, self.step_input) + step_validator.setNotation(QDoubleValidator.Notation.ScientificNotation) + self.step_input.setValidator(step_validator) + self.step_input.editingFinished.connect(self._on_step_edited) + self.tweak_right = TextButton("", "neutral", "chevron_right", self.tweak_block) + self.tweak_right.setToolTip("Tweak up by the step") + self.tweak_right.clicked.connect(self.on_tweak_right) + for button in (self.tweak_left, self.tweak_right): + button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + tweak_row.addWidget(self.tweak_left, 1) + tweak_row.addWidget(self.step_input) + tweak_row.addWidget(self.tweak_right, 1) + body.addWidget(self.tweak_block) + + self.stop_button = TextButton("Stop", "danger", "stop", self.card) + self.stop_button.clicked.connect(self.on_stop) + body.addWidget(self.stop_button) + + self.empty_label = QLabel( + "Choose a positioner above to see its position and move it.", self.card + ) + self.empty_label.setWordWrap(True) + self.empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + body.addWidget(self.empty_label) + body.addStretch(1) + + self._apply_styles(ThemeTokens()) + + # ---- rendering ----------------------------------------------------------------------- + + def _apply_styles(self, tokens: ThemeTokens) -> None: + self.card.setStyleSheet(self.card.styleSheet() + field_qss(tokens)) + font = QFont("monospace") + font.setStyleHint(QFont.StyleHint.Monospace) + font.setPixelSize(26) + font.setWeight(QFont.Weight.DemiBold) + self.readback_label.setFont(font) + plain = "background: none; border: none;" + self.readback_label.setStyleSheet(f"color: {tokens.fg.name()}; {plain}") + self.units_label.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 14px; {plain}") + self.target_label.setStyleSheet( + f"color: {tokens.fg_muted.name()}; font-size: 12px; font-family: monospace; {plain}" + ) + for label in (self.limit_low, self.limit_high): + label.setStyleSheet(f"color: {tokens.fg_subtle.name()}; font-size: 11px; {plain}") + self.error_label.setStyleSheet(f"color: {tokens.danger.name()}; font-size: 11px; {plain}") + self.device_label.setStyleSheet( + f"color: {tokens.fg.name()}; font-size: 14px; font-weight: 600; {plain}" + ) + self.empty_label.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 12px; {plain}") + for block in (self.readback_block, self.limits_block, self.move_block, self.tweak_block): + block.setStyleSheet("background: none;") + + def _tone(self, tone: str, tokens: ThemeTokens) -> QColor: + return { + "primary": tokens.primary, + "success": tokens.success, + "warning": tokens.warning, + "danger": tokens.danger, + }.get(tone, tokens.fg_muted) + + def _sync_view(self) -> None: + state = self.view_state() + tokens = ThemeTokens() + has_device = state["hasDevice"] + + self.selector.setVisible(state["selectable"]) + self.device_label.setVisible(not state["selectable"]) + self.device_label.setText(state["device"] or "No positioner") + if not self.selector.lineEdit().hasFocus(): + names = self.positioner_names() + if [self.selector.itemText(i) for i in range(self.selector.count())] != names: + self.selector.blockSignals(True) + self.selector.clear() + self.selector.addItems(names) + self.selector.blockSignals(False) + self.selector.setCurrentText(state["device"]) + self.status_pill.set_status( + state["status"], self._tone(state["statusTone"], tokens), state["moving"] + ) + + for block in (self.readback_block, self.move_block, self.tweak_block, self.stop_button): + block.setVisible(has_device) + self.empty_label.setVisible(not has_device) + self.readback_label.setText(state["readbackText"]) + self.units_label.setText(state["units"]) + self.units_label.setVisible(bool(state["units"])) + self.target_label.setText(f"target {state['setpointText']}") + self.target_label.setVisible(state["showTarget"]) + + self.limits_block.setVisible(state["hasLimits"]) + self.limit_low.setText(state["limitLowText"]) + self.limit_high.setText(state["limitHighText"]) + self.limit_bar.set_values( + state["position"], state["target"], state["showTarget"], state["atLimit"] + ) + + self.tweak_left.setText(f"−{state['stepText']}") + self.tweak_right.setText(f"+{state['stepText']}") + if not self.step_input.hasFocus(): + self.step_input.setText(state["stepText"]) + self._show_target_problem(state["error"]) + + def _on_target_edited(self, text: str) -> None: + self._show_target_problem(self._error) + + def _show_target_problem(self, fallback: str) -> None: + text = self.target_input.text() + problem = self.validate_target(text) + set_invalid(self.target_input, bool(problem)) + message = problem or fallback + self.error_label.setText(message) + self.error_label.setVisible(bool(message)) + self.go_button.setEnabled(bool(text.strip()) and not problem) + + def _submit_target(self) -> None: + if self.move_to(self.target_input.text()): + self.target_input.clear() + self.target_input.clearFocus() + + def _on_step_edited(self) -> None: + text = self.step_input.text() + try: + value = float(text) + except ValueError: + return + if value > 0: + self.step_size = value + + def apply_theme(self, theme: str): + super().apply_theme(theme) + tokens = refresh_kit_theme(self) + self._apply_styles(tokens) + self._sync_view() + + def cleanup(self): + self.limit_bar.stop_animation() + self.status_pill.cleanup() + 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 = PositionerBoxQWidget(device="samx") + widget.show() + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/control/device_control/positioner_box/positioner_ux_common.py b/bec_widgets/widgets/control/device_control/positioner_box/positioner_ux_common.py new file mode 100644 index 00000000..42238bb8 --- /dev/null +++ b/bec_widgets/widgets/control/device_control/positioner_box/positioner_ux_common.py @@ -0,0 +1,413 @@ +"""Logic shared by the QML and QWidget ports of the positioner box. + +The ports keep the public API of :class:`PositionerBox` (``set_positioner``, ``device``, +``hide_device_selection``, ``device_changed``, ``position_update``, ...) but hold the device state +in a plain dictionary that both renderers draw from. Readback parsing, move validation, tweak +and stop are implemented once here. +""" + +from __future__ import annotations + +import math + +from bec_lib.device import Positioner +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from qtpy.QtCore import Signal + +from bec_widgets.utils.error_popups import SafeProperty, SafeSlot +from bec_widgets.utils.ux_kit import PortedPropertiesMixin +from bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base import ( + PositionerBoxBase, +) + +logger = bec_logger.logger + +DEFAULT_PRECISION = 8 + + +def device_precision(device_obj) -> int: + """Precision of a device, falling back to 8 like :class:`PositionerBox`.""" + precision = getattr(device_obj, "precision", DEFAULT_PRECISION) + try: + return int(precision) + except (TypeError, ValueError): + return DEFAULT_PRECISION + + +def parse_readback(device: str, hints: list[str], signals: dict) -> dict: + """Extract readback, setpoint and motion state from a device readback message. + + Mirrors :meth:`PositionerBoxBase._on_device_readback`. + + Returns: + dict: ``readback``, ``setpoint`` and ``moving`` entries; each may be ``None`` when the + message does not contain the value. + """ + readback = None + if len(hints) == 1: + readback = signals.get(hints[0], {}).get("value") + setpoint = None + for setpoint_signal in ("setpoint", "user_setpoint"): + setpoint = signals.get(f"{device}_{setpoint_signal}", {}).get("value") + if setpoint is not None: + break + if f"{device}_motor_done_move" in signals: + moving = not signals[f"{device}_motor_done_move"].get("value") + elif f"{device}_motor_is_moving" in signals: + moving = bool(signals[f"{device}_motor_is_moving"].get("value")) + else: + moving = None + return {"readback": readback, "setpoint": setpoint, "moving": moving} + + +def format_value(value: float | None, precision: int) -> str: + """Format a position with the device precision, or an em dash when unknown.""" + if value is None: + return "—" + return f"{value:.{precision}f}" + + +def format_step(step: float) -> str: + """Compact representation of a step size for button labels (``0.01``, ``1e-06``).""" + return f"{step:g}" + + +class PositionerBoxPortBase(PortedPropertiesMixin, PositionerBoxBase): + """Shared implementation of the positioner box ports. + + Subclasses build the renderer in :meth:`_init_view` and redraw it from :attr:`state` in + :meth:`_sync_view`. + """ + + PLUGIN = False + RPC = False + rpc_widget_class = "PositionerBox" + USER_ACCESS = ["set_positioner", "attach", "detach", "screenshot"] + + device_changed = Signal(str, str) + position_update = Signal(float) + + def __init__(self, parent=None, device: Positioner | str | None = None, **kwargs): + super().__init__(parent=parent, **kwargs) + self._device = "" + self._limits = None + self._hide_device_selection = False + self._error = "" + self.state: dict = self._empty_state() + self.device_changed.connect(self.on_device_change) + self._init_view() + self._sync_view() + if isinstance(device, Positioner): + device = device.name + self.device = device + + # ---- state --------------------------------------------------------------------------- + + @staticmethod + def _empty_state() -> dict: + return { + "device": "", + "readback": None, + "setpoint": None, + "moving": None, + "limits": None, + "precision": DEFAULT_PRECISION, + "units": "", + "step": 10**-DEFAULT_PRECISION * 10, + } + + def view_state(self) -> dict: + """Display-ready state shared by both renderers.""" + state = self.state + precision = state["precision"] + limits = state["limits"] + has_limits = limits is not None and limits[0] != limits[1] + readback = state["readback"] + setpoint = state["setpoint"] + + def fraction(value): + if not has_limits or value is None: + return -1.0 + return max(0.0, min(1.0, (value - limits[0]) / (limits[1] - limits[0]))) + + at_limit = False + if has_limits and readback is not None: + span = abs(limits[1] - limits[0]) + tolerance = max(span * 1e-6, 10**-precision) + at_limit = ( + abs(readback - limits[0]) <= tolerance or abs(readback - limits[1]) <= tolerance + ) + + moving = state["moving"] + if not state["device"]: + status, tone = "No device", "muted" + elif moving is None: + status, tone = "Unknown", "muted" + elif moving: + status, tone = "Moving", "primary" + elif at_limit: + status, tone = "At limit", "warning" + else: + status, tone = "Idle", "success" + + target_differs = ( + setpoint is not None + and readback is not None + and not math.isclose(setpoint, readback, abs_tol=10**-precision) + ) + step = state["step"] + return { + "device": state["device"], + "hasDevice": bool(state["device"]), + "readbackText": format_value(readback, precision), + "setpointText": format_value(setpoint, precision), + "showTarget": bool(moving) or target_differs, + "units": state["units"], + "status": status, + "statusTone": tone, + "moving": bool(moving), + "hasLimits": has_limits, + "limitLowText": format_value(limits[0], min(precision, 3)) if has_limits else "", + "limitHighText": format_value(limits[1], min(precision, 3)) if has_limits else "", + "position": fraction(readback), + "target": fraction(setpoint), + "atLimit": at_limit, + "step": step, + "stepText": format_step(step), + "decimals": precision, + "error": self._error, + "selectable": not self._hide_device_selection, + } + + def positioner_names(self) -> list[str]: + """Names of all positioners known to the device manager, sorted.""" + try: + devices = self.dev + return sorted(name for name in devices.keys() if isinstance(devices[name], Positioner)) + except Exception: # pylint: disable=broad-except + return [] + + def _set_state(self, **values) -> None: + self.state.update(values) + self._sync_view() + + def _set_error(self, message: str) -> None: + self._error = message + self._sync_view() + + # ---- device handling ----------------------------------------------------------------- + + @SafeProperty(str) + def device(self) -> str: + """Name of the controlled positioner.""" + return self._device + + @device.setter + def device(self, value: str): + if not value or not isinstance(value, str): + return + if not self._check_device_is_valid(value): + return + old_device = self._device + self._device = value + self.device_changed.emit(old_device, value) + + @SafeProperty(bool) + def hide_device_selection(self) -> bool: + """Hide the device selector; the device name is still shown.""" + return self._hide_device_selection + + @hide_device_selection.setter + def hide_device_selection(self, value: bool): + self._hide_device_selection = bool(value) + self._sync_view() + + @SafeSlot(bool) + def show_device_selection(self, value: bool): + """Show the device selection. + + Args: + value (bool): Show the device selection + """ + self.hide_device_selection = not value + + @SafeSlot(str) + def set_positioner(self, positioner: str | Positioner): + """Set the device. + + Args: + positioner (Positioner | str) : Positioner to set, accepts str or the device + """ + if isinstance(positioner, Positioner): + positioner = positioner.name + self.device = positioner + + @SafeSlot(str, str) + def on_device_change(self, old_device: str, new_device: str): + """Reset the state for the new device and move the readback subscription. + + Args: + old_device (str): The old device name. + new_device (str): The new device name. + """ + if not self._check_device_is_valid(new_device): + return + device_obj = self.dev[new_device] + precision = device_precision(device_obj) + try: + units = str(device_obj.egu() or "") + except Exception: # pylint: disable=broad-except + units = "" + self._limits = None + self._error = "" + self.state = self._empty_state() + self.state.update( + device=new_device, precision=precision, units=units, step=10**-precision * 10 + ) + if old_device: + self._swap_readback_signal_connection(self.on_device_readback, old_device, new_device) + else: + self.bec_dispatcher.connect_slot( + self.on_device_readback, MessageEndpoints.device_readback(new_device) + ) + self.force_update_readback() + + def force_update_readback(self): + """Read the cached device values and refresh the view.""" + if not self._device or not self._check_device_is_valid(self._device): + return + data = self.dev[self._device].read(cached=True) + self.on_device_readback({"signals": data}, {}) + + @SafeSlot(dict, dict) + def on_device_readback(self, msg_content: dict, metadata: dict): + """Callback for device readback. + + Args: + msg_content (dict): The message content. + metadata (dict): The message metadata. + """ + device = self._device + if not device: + return + # pylint: disable=protected-access + parsed = parse_readback(device, self.dev[device]._hints, msg_content.get("signals", {})) + self.state.update({key: value for key, value in parsed.items() if value is not None}) + self.update_limits(self.dev[device].limits) + if parsed["readback"] is not None: + self.position_update.emit(parsed["readback"]) + self._sync_view() + + def update_limits(self, limits: tuple | list | None): + """Update limits. + + Args: + limits (tuple): Limits of the positioner + """ + if limits == self._limits: + return + self._limits = limits + self.state["limits"] = tuple(limits) if limits is not None else None + self._sync_view() + + # ---- actions ------------------------------------------------------------------------- + + @property + def step_size(self) -> float: + """Step size for tweak.""" + return self.state["step"] + + @step_size.setter + def step_size(self, value: float): + value = abs(float(value)) + if value > 0: + self._set_state(step=value) + + def validate_target(self, text: str) -> str: + """Return an error message for a move target, or an empty string when it is valid.""" + text = text.strip() + if not text: + return "" + try: + value = float(text) + except ValueError: + return "Enter a number" + limits = self.state["limits"] + if limits is not None and limits[0] != limits[1]: + low, high = sorted(limits) + if not low <= value <= high: + precision = min(self.state["precision"], 3) + return ( + f"Outside limits {format_value(low, precision)} … " + f"{format_value(high, precision)}" + ) + return "" + + def move_to(self, text: str) -> bool: + """Move to the absolute position in ``text`` if it is valid. + + Returns: + bool: True if the move was requested. + """ + error = self.validate_target(text) + if error or not text.strip() or not self._device: + self._set_error(error or ("Enter a target position" if self._device else "")) + return False + self._set_error("") + self.scans.mv(self.dev[self._device], float(text), relative=False) + return True + + @SafeSlot() + def on_stop(self): + """Stop the device.""" + self._stop_device(self.device) + + @SafeSlot() + def on_tweak_right(self): + """Tweak motor right.""" + self._tweak(1) + + @SafeSlot() + def on_tweak_left(self): + """Tweak motor left.""" + self._tweak(-1) + + def _tweak(self, direction: int) -> None: + if not self._device: + return + step = direction * self.step_size + setpoint = self._get_setpoint() + if setpoint is None: + self.scans.mv(self.dev[self._device], step, relative=True) + return + self.scans.mv(self.dev[self._device], setpoint + step, relative=False) + + def _get_setpoint(self) -> float | None: + """Get the setpoint of the motor.""" + setpoint = getattr(self.dev[self._device], "setpoint", None) + if not setpoint: + setpoint = getattr(self.dev[self._device], "user_setpoint", None) + if not setpoint: + return None + try: + return float(setpoint.get()) + except Exception: # pylint: disable=broad-except + return None + + # The ports do not use the .ui based helpers of PositionerBoxBase. + def _device_ui_components(self, device: str): + raise NotImplementedError + + # ---- renderer hooks ------------------------------------------------------------------ + + def _init_view(self) -> None: + """Create the renderer and add it to ``self.main_layout``.""" + raise NotImplementedError + + def _sync_view(self) -> None: + """Redraw the renderer from :meth:`view_state`.""" + raise NotImplementedError + + +PositionerBoxPortBase.PORTED_FROM = PositionerBoxPortBase diff --git a/bec_widgets/widgets/control/device_control/positioner_box/qml/PositionerBoxView.qml b/bec_widgets/widgets/control/device_control/positioner_box/qml/PositionerBoxView.qml new file mode 100644 index 00000000..a494bfd8 --- /dev/null +++ b/bec_widgets/widgets/control/device_control/positioner_box/qml/PositionerBoxView.qml @@ -0,0 +1,231 @@ +import QtQuick +import QtQuick.Controls.Basic +import QtQuick.Layouts +import BecUi + +// Positioner control card: live readback, limits, absolute move, tweak and stop. +// backend: PositionerBackend (state map, positioners list, actions). +Rectangle { + id: root + required property QtObject backend + readonly property var s: backend.state + + function toneColor(tone) { + return tone === "primary" ? theme.primary : tone === "success" ? theme.success + : tone === "warning" ? theme.warning : tone === "danger" ? theme.danger : theme.fgMuted + } + + color: theme.bg + implicitWidth: 280 + implicitHeight: card.implicitHeight + 16 + + Card { + id: card + anchors.fill: parent + anchors.margins: 8 + spacing: 10 + + // ---- header: device selector and status ------------------------------------------ + RowLayout { + Layout.fillWidth: true + spacing: 8 + SelectField { + id: selector + visible: root.s.selectable === true + Layout.fillWidth: true + editable: true + model: root.backend.positioners + placeholder: "Select positioner" + Component.onCompleted: root.backend.refreshPositioners() + onActivated: (index) => root.backend.selectDevice(textAt(index)) + onAccepted: root.backend.selectDevice(editText) + popup.onAboutToShow: root.backend.refreshPositioners() + Connections { + target: root.backend + function onChanged() { + if (!selector.activeFocus) + selector.editText = root.s.device || "" + } + } + Accessible.name: "Positioner" + } + Text { + visible: root.s.selectable !== true + text: root.s.device || "No positioner" + color: theme.fg + font.pixelSize: 14 + font.weight: Font.DemiBold + elide: Text.ElideRight + Layout.fillWidth: true + } + StatusPill { + text: root.s.status || "" + tone: root.toneColor(root.s.statusTone) + pulse: root.s.moving === true + } + } + + // ---- readback ---------------------------------------------------------------------- + ColumnLayout { + visible: root.s.hasDevice === true + Layout.fillWidth: true + spacing: 0 + RowLayout { + spacing: 6 + Text { + text: root.s.readbackText || "—" + color: theme.fg + font.pixelSize: 26 + font.weight: Font.DemiBold + font.family: "monospace" + Layout.maximumWidth: card.width - 70 + elide: Text.ElideRight + } + Text { + text: root.s.units || "" + visible: text !== "" + color: theme.fgMuted + font.pixelSize: 14 + Layout.alignment: Qt.AlignBaseline + } + } + Text { + text: "target " + (root.s.setpointText || "") + visible: root.s.showTarget === true + color: theme.fgMuted + font.pixelSize: 12 + font.family: "monospace" + } + } + + // ---- limits bar ---------------------------------------------------------------------- + ColumnLayout { + visible: root.s.hasLimits === true + Layout.fillWidth: true + spacing: 3 + Item { + Layout.fillWidth: true + implicitHeight: 14 + Rectangle { + id: track + anchors.verticalCenter: parent.verticalCenter + width: parent.width + height: 4 + radius: 2 + color: theme.track + } + Rectangle { + visible: root.s.target >= 0 && root.s.showTarget === true + x: root.s.target * (track.width - 2) + width: 2; height: 14 + color: theme.fgMuted + } + Rectangle { + visible: root.s.position >= 0 + width: 12; height: 12; radius: 6 + x: root.s.position * (track.width - width) + anchors.verticalCenter: parent.verticalCenter + color: root.s.atLimit ? theme.warning : theme.primary + border.color: theme.card + border.width: 2 + Behavior on x { NumberAnimation { duration: 180; easing.type: Easing.OutCubic } } + } + } + RowLayout { + Layout.fillWidth: true + Text { text: root.s.limitLowText || ""; color: theme.fgSubtle; font.pixelSize: 11 } + Item { Layout.fillWidth: true } + Text { text: root.s.limitHighText || ""; color: theme.fgSubtle; font.pixelSize: 11 } + } + } + + // ---- absolute move ------------------------------------------------------------------- + ColumnLayout { + visible: root.s.hasDevice === true + Layout.fillWidth: true + spacing: 4 + RowLayout { + Layout.fillWidth: true + spacing: 6 + InputField { + id: target + Layout.fillWidth: true + placeholderText: "Move to…" + suffix: root.s.units || "" + readonly property string problem: root.backend.validate(text) + invalid: problem !== "" + inputMethodHints: Qt.ImhFormattedNumbersOnly + onAccepted: if (root.backend.moveTo(text)) { text = ""; focus = false } + Accessible.name: "Target position" + } + TextButton { + text: "Go" + variant: "primary" + enabled: target.text !== "" && !target.invalid + onClicked: if (root.backend.moveTo(target.text)) target.text = "" + } + } + Text { + readonly property string message: target.problem !== "" ? target.problem : (root.s.error || "") + visible: message !== "" + text: message + color: theme.danger + font.pixelSize: 11 + } + } + + // ---- tweak --------------------------------------------------------------------------- + RowLayout { + visible: root.s.hasDevice === true + Layout.fillWidth: true + spacing: 6 + TextButton { + text: "−" + (root.s.stepText || "") + iconName: "chevron_left" + tip: "Tweak down by the step" + Layout.fillWidth: true + onClicked: root.backend.tweak(-1) + } + InputField { + id: step + Layout.preferredWidth: 78 + horizontalAlignment: Text.AlignHCenter + text: root.s.stepText || "" + validator: DoubleValidator { bottom: 0; notation: DoubleValidator.ScientificNotation } + onEditingFinished: if (acceptableInput && Number(text) > 0) root.backend.setStep(Number(text)) + ToolTip.visible: hovered + ToolTip.text: "Step size" + Accessible.name: "Step size" + } + TextButton { + text: "+" + (root.s.stepText || "") + iconName: "chevron_right" + tip: "Tweak up by the step" + Layout.fillWidth: true + onClicked: root.backend.tweak(1) + } + } + + TextButton { + visible: root.s.hasDevice === true + Layout.fillWidth: true + text: "Stop" + iconName: "stop" + variant: "danger" + onClicked: root.backend.stop() + } + + // ---- empty state ----------------------------------------------------------------------- + Text { + visible: root.s.hasDevice !== true + Layout.fillWidth: true + Layout.topMargin: 12 + horizontalAlignment: Text.AlignHCenter + wrapMode: Text.WordWrap + text: "Choose a positioner above to see its position and move it." + color: theme.fgMuted + font.pixelSize: 12 + } + Item { Layout.fillHeight: true } + } +} diff --git a/tests/unit_tests/test_positioner_box_ports.py b/tests/unit_tests/test_positioner_box_ports.py new file mode 100644 index 00000000..9d6c2279 --- /dev/null +++ b/tests/unit_tests/test_positioner_box_ports.py @@ -0,0 +1,173 @@ +"""Tests for the QML and QWidget ports of the positioner box.""" + +# pylint: disable=redefined-outer-name, protected-access, missing-function-docstring +from unittest import mock + +import pytest +from bec_lib.endpoints import MessageEndpoints +from bec_lib.messages import VariableMessage +from qtpy.QtQuickWidgets import QQuickWidget + +from bec_widgets.widgets.control.device_control.positioner_box.positioner_box_qml import ( + PositionerBoxQML, +) +from bec_widgets.widgets.control.device_control.positioner_box.positioner_box_qwidget import ( + PositionerBoxQWidget, +) +from bec_widgets.widgets.control.device_control.positioner_box.positioner_ux_common import ( + format_step, + parse_readback, +) + +from .conftest import create_widget + +MOVING = { + "signals": { + "samx": {"value": 2.5}, + "samx_setpoint": {"value": 6.0}, + "samx_motor_is_moving": {"value": 1}, + } +} + + +@pytest.fixture(params=[PositionerBoxQML, PositionerBoxQWidget]) +def box(request, qtbot, mocked_client): + mocked_client.device_manager.devices["samx"].limits = [-10, 10] + widget = create_widget(qtbot, request.param, device="samx", client=mocked_client) + yield widget + + +def test_parse_readback(): + parsed = parse_readback("samx", ["samx"], MOVING["signals"]) + assert parsed == {"readback": 2.5, "setpoint": 6.0, "moving": True} + parsed = parse_readback("samx", ["samx"], {"samx_motor_done_move": {"value": 1}}) + assert parsed == {"readback": None, "setpoint": None, "moving": False} + assert format_step(0.01) == "0.01" + + +def test_init_reads_cached_values(box): + data = box.dev["samx"].read(cached=True) + precision = box.dev["samx"].precision + state = box.view_state() + assert box.device == "samx" + assert state["setpointText"] == f"{data['samx_setpoint']['value']:.{precision}f}" + assert box.step_size == pytest.approx(10**-precision * 10) + assert state["hasLimits"] + assert state["limitLowText"] == "-10.000" + + +def test_readback_updates_state_and_emits(box, qtbot): + with qtbot.waitSignal(box.position_update) as blocker: + box.on_device_readback(MOVING, {}) + assert blocker.args == [2.5] + state = box.view_state() + assert state["status"] == "Moving" + assert state["moving"] + assert state["showTarget"] + assert state["position"] == pytest.approx(0.625) + assert state["target"] == pytest.approx(0.8) + + +def test_status_idle_and_at_limit(box): + box.on_device_readback( + {"signals": {"samx": {"value": 10.0}, "samx_motor_is_moving": {"value": 0}}}, {} + ) + assert box.view_state()["status"] == "At limit" + box.on_device_readback({"signals": {"samx": {"value": 1.0}}}, {}) + assert box.view_state()["status"] == "Idle" + + +def test_validate_and_move(box): + assert box.validate_target("abc") == "Enter a number" + assert box.validate_target("100").startswith("Outside limits") + assert box.validate_target("5") == "" + with mock.patch.object(box.scans, "mv") as mock_mv: + assert not box.move_to("100") + mock_mv.assert_not_called() + assert box.view_state()["error"].startswith("Outside limits") + assert box.move_to("5") + mock_mv.assert_called_once_with(box.dev["samx"], 5.0, relative=False) + assert box.view_state()["error"] == "" + + +def test_tweak_uses_setpoint_or_relative(box): + box.step_size = 0.1 + with mock.patch.object(box.scans, "mv") as mock_mv: + box.on_tweak_right() + mock_mv.assert_called_once_with(box.dev["samx"], 0.1, relative=True) + with mock.patch.object(box.scans, "mv") as mock_mv: + box.on_tweak_left() + mock_mv.assert_called_once_with(box.dev["samx"], -0.1, relative=True) + with ( + mock.patch.object(box, "_get_setpoint", return_value=2.0), + mock.patch.object(box.scans, "mv") as mock_mv, + ): + box.on_tweak_right() + mock_mv.assert_called_once_with(box.dev["samx"], pytest.approx(2.1), relative=False) + + +def test_stop(box): + with mock.patch.object(box.client.connector, "send") as mock_send: + box.on_stop() + mock_send.assert_called_once_with( + MessageEndpoints.stop_devices(), VariableMessage(value=["samx"]) + ) + + +def test_switch_device_and_hide_selection(box, qtbot): + with qtbot.waitSignal(box.device_changed) as blocker: + box.set_positioner("samy") + assert blocker.args == ["samx", "samy"] + assert box.view_state()["device"] == "samy" + assert "samy" in box.positioner_names() + box.hide_device_selection = True + assert not box.view_state()["selectable"] + box.show_device_selection(True) + assert box.view_state()["selectable"] + + +def test_empty_box(qtbot, mocked_client): + for cls in (PositionerBoxQML, PositionerBoxQWidget): + widget = create_widget(qtbot, cls, client=mocked_client) + state = widget.view_state() + assert not state["hasDevice"] + assert state["status"] == "No device" + assert not widget.move_to("1") + + +def test_qml_view_loads_and_backend(qtbot, mocked_client): + widget = create_widget(qtbot, PositionerBoxQML, device="samx", client=mocked_client) + assert widget.view.status() == QQuickWidget.Status.Ready + assert widget.view.errors() == [] + backend = widget.backend + assert backend.state["device"] == "samx" + backend.refreshPositioners() + assert "samx" in backend.positioners + assert backend.validate("x") == "Enter a number" + backend.setStep(0.5) + assert widget.step_size == 0.5 + with mock.patch.object(widget, "on_tweak_left") as tweak_left: + backend.tweak(-1) + tweak_left.assert_called_once() + + +def test_qwidget_controls(qtbot, mocked_client): + mocked_client.device_manager.devices["samx"].limits = [-10, 10] + widget = create_widget(qtbot, PositionerBoxQWidget, device="samx", client=mocked_client) + widget.target_input.setText("100") + assert widget.target_input.property("invalid") + assert not widget.go_button.isEnabled() + assert widget.error_label.text().startswith("Outside limits") + widget.target_input.setText("3") + assert widget.go_button.isEnabled() + with mock.patch.object(widget.scans, "mv") as mock_mv: + widget.go_button.click() + mock_mv.assert_called_once_with(widget.dev["samx"], 3.0, relative=False) + assert widget.target_input.text() == "" + widget.step_input.setText("0.5") + widget.step_input.editingFinished.emit() + assert widget.step_size == 0.5 + assert widget.tweak_right.text() == "+0.5" + widget.on_device_readback(MOVING, {}) + assert widget.status_pill.text == "Moving" + assert widget.status_pill.pulsing