mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-27 16:01:27 +01:00
feat(positioner_box): add QML and QWidget ports with status pill, limit bar and validated moves
This commit is contained in:
6 files changed
+1287
No files matched your search
@@ -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
|
||||
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -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
|
||||
@@ -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 }
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
Reference in new issue
Block a user