mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-28 00:01:27 +01:00
feat(beamline_states): add QML and QWidget ports with status chips, search, interlock switch and tripped banner
This commit is contained in:
6 files changed
+2412
No files matched your search
@@ -10,6 +10,8 @@ Rectangle {
|
||||
property alias headerExtras: extras.data
|
||||
property int padding: 12
|
||||
property alias spacing: column.spacing
|
||||
// Let the content column take the remaining height, e.g. for a list that fills the card.
|
||||
property bool fillContent: false
|
||||
|
||||
color: theme.card
|
||||
radius: 10
|
||||
@@ -47,6 +49,7 @@ Rectangle {
|
||||
ColumnLayout {
|
||||
id: column
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: root.fillContent
|
||||
spacing: 8
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
"""Beamline state manager rendered with Qt Quick (QML), same API as :class:`BeamlineStateManager`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from qtpy.QtCore import Property, QObject, Signal, Slot
|
||||
from qtpy.QtWidgets import QVBoxLayout
|
||||
|
||||
from bec_widgets.utils.quick import DictListModel, create_quick_widget, release_quick_widget
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_states_ux_common import (
|
||||
BeamlineStatesPortBase,
|
||||
)
|
||||
|
||||
QML_FILE = Path(__file__).parent / "qml" / "BeamlineStatesView.qml"
|
||||
|
||||
ROW_ROLES = [
|
||||
"kind",
|
||||
"key",
|
||||
"name",
|
||||
"label",
|
||||
"status",
|
||||
"statusText",
|
||||
"tone",
|
||||
"icon",
|
||||
"device",
|
||||
"stateType",
|
||||
"params",
|
||||
"watched",
|
||||
"tripped",
|
||||
"armed",
|
||||
"acceptedText",
|
||||
"tripOnWarning",
|
||||
"expanded",
|
||||
"title",
|
||||
"count",
|
||||
]
|
||||
|
||||
|
||||
class BeamlineStatesBackend(QObject):
|
||||
"""Bridge between :class:`BeamlineStatesQML` and ``BeamlineStatesView.qml``."""
|
||||
|
||||
changed = Signal()
|
||||
|
||||
def __init__(self, manager: "BeamlineStatesQML"):
|
||||
super().__init__(manager)
|
||||
self._manager = manager
|
||||
self._state: dict = {}
|
||||
self._rows = DictListModel(ROW_ROLES, self)
|
||||
|
||||
def update_from(self, state: dict) -> None:
|
||||
"""Take a new view state from the manager."""
|
||||
self._rows.set_items(state.pop("rows"))
|
||||
self._state = state
|
||||
self.changed.emit()
|
||||
|
||||
# pylint: disable=invalid-name,missing-function-docstring
|
||||
@Slot(str)
|
||||
def toggleStatus(self, status: str) -> None:
|
||||
self._manager.toggle_status_filter(status)
|
||||
|
||||
@Slot(str)
|
||||
def setFilterText(self, text: str) -> None:
|
||||
self._manager.set_filter_text(text)
|
||||
|
||||
@Slot()
|
||||
def clearFilters(self) -> None:
|
||||
self._manager.clear_filters()
|
||||
|
||||
@Slot()
|
||||
def addState(self) -> None:
|
||||
self._manager.open_add_state_dialog()
|
||||
|
||||
@Slot(str)
|
||||
def toggleExpanded(self, name: str) -> None:
|
||||
self._manager.toggle_expanded(name)
|
||||
|
||||
@Slot(str)
|
||||
def toggleInterlockFor(self, name: str) -> None:
|
||||
self._manager.toggle_interlock_for(name)
|
||||
|
||||
@Slot(str, bool)
|
||||
def setTripOnWarning(self, name: str, trip: bool) -> None:
|
||||
self._manager.set_trip_on_warning(name, trip)
|
||||
|
||||
@Slot(bool)
|
||||
def setInterlockEnabled(self, enabled: bool) -> None:
|
||||
self._manager.set_interlock_enabled(enabled)
|
||||
|
||||
@Slot(str)
|
||||
def editState(self, name: str) -> None:
|
||||
self._manager.edit_state(name)
|
||||
|
||||
@Slot(str)
|
||||
def removeState(self, name: str) -> None:
|
||||
self._manager.remove_state(name)
|
||||
|
||||
@Slot(bool)
|
||||
def setShowHidden(self, show: bool) -> None:
|
||||
self._manager.set_show_hidden(show)
|
||||
|
||||
state = Property("QVariantMap", lambda self: self._state, notify=changed)
|
||||
rows = Property(QObject, lambda self: self._rows, constant=True)
|
||||
|
||||
|
||||
class BeamlineStatesQML(BeamlineStatesPortBase):
|
||||
"""Beamline state manager with the modernised UX rendered in QML."""
|
||||
|
||||
def _init_view(self) -> None:
|
||||
self.backend = BeamlineStatesBackend(self)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.view = create_quick_widget(self, QML_FILE, {"backend": self.backend})
|
||||
self.view.setMinimumSize(320, 300)
|
||||
layout.addWidget(self.view)
|
||||
|
||||
def _sync_view(self) -> None:
|
||||
if hasattr(self, "backend"):
|
||||
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 = BeamlineStatesQML()
|
||||
widget.resize(460, 560)
|
||||
widget.show()
|
||||
sys.exit(app.exec())
|
||||
@@ -0,0 +1,713 @@
|
||||
"""Beamline state manager with the modernised UX in plain QWidgets.
|
||||
|
||||
This is the QWidget twin of :class:`BeamlineStatesQML`, built from the same view state so both
|
||||
rendering stacks can be compared side by side. It keeps the public API of
|
||||
:class:`BeamlineStateManager`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QEasingCurve, QRectF, QSize, Qt, QVariantAnimation, Signal
|
||||
from qtpy.QtGui import QColor, QPainter, QPen
|
||||
from qtpy.QtWidgets import (
|
||||
QFrame,
|
||||
QGridLayout,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QPushButton,
|
||||
QScrollArea,
|
||||
QSizePolicy,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.eliding_label import ElidingLabel
|
||||
from bec_widgets.utils.quick.host import ThemeTokens
|
||||
from bec_widgets.utils.ux_kit import (
|
||||
Card,
|
||||
IconButton,
|
||||
StatusPill,
|
||||
TextButton,
|
||||
ToggleSwitch,
|
||||
field_qss,
|
||||
refresh_kit_theme,
|
||||
rgba,
|
||||
)
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_states_ux_common import (
|
||||
BeamlineStatesPortBase,
|
||||
)
|
||||
|
||||
PLAIN = "background: none; border: none;"
|
||||
|
||||
|
||||
def tone_color(tone: str, tokens: ThemeTokens) -> QColor:
|
||||
"""Colour of a tone name used in the view state."""
|
||||
return {
|
||||
"primary": tokens.primary,
|
||||
"success": tokens.success,
|
||||
"warning": tokens.warning,
|
||||
"danger": tokens.danger,
|
||||
}.get(tone, tokens.fg_muted)
|
||||
|
||||
|
||||
def alpha(color: QColor, value: float) -> QColor:
|
||||
"""Copy of ``color`` with alpha ``value`` (0..1)."""
|
||||
result = QColor(color)
|
||||
result.setAlphaF(max(0.0, min(1.0, value)))
|
||||
return result
|
||||
|
||||
|
||||
def icon_label(parent: QWidget, size: int = 18) -> QLabel:
|
||||
"""Transparent label sized for a Material icon pixmap."""
|
||||
label = QLabel(parent)
|
||||
label.setFixedSize(size, size)
|
||||
label.setStyleSheet(PLAIN)
|
||||
label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
|
||||
return label
|
||||
|
||||
|
||||
def set_icon(label: QLabel, name: str, color: QColor, size: int = 18, filled: bool = False):
|
||||
"""Show a tinted Material icon in ``label``."""
|
||||
label.setPixmap(material_icon(name, size=(size, size), color=color, filled=filled))
|
||||
|
||||
|
||||
class StatusChip(QPushButton):
|
||||
"""Status count that toggles a status filter, matching the QML chip."""
|
||||
|
||||
def __init__(self, parent: QWidget | None = None):
|
||||
super().__init__(parent)
|
||||
self.setCursor(Qt.CursorShape.PointingHandCursor)
|
||||
self.setFixedHeight(24)
|
||||
self._tone = QColor("#808080")
|
||||
self._active = False
|
||||
self._fg = QColor("#808080")
|
||||
self.status = ""
|
||||
|
||||
def set_chip(self, chip: dict, tokens: ThemeTokens) -> None:
|
||||
"""Update text, colour and the active state from a view-state chip."""
|
||||
self.status = chip["status"]
|
||||
self._tone = tone_color(chip["tone"], tokens)
|
||||
self._active = chip["active"]
|
||||
self.setText(chip["text"])
|
||||
self.setAccessibleName(f"Show only {self.status} states")
|
||||
self.setToolTip(
|
||||
"Click to stop filtering by this status"
|
||||
if self._active
|
||||
else "Click to show only this status"
|
||||
)
|
||||
font = self.font()
|
||||
font.setPixelSize(12)
|
||||
font.setBold(self._active)
|
||||
self.setFont(font)
|
||||
self.setFixedWidth(self.fontMetrics().horizontalAdvance(chip["text"]) + 30)
|
||||
self._fg = tokens.fg if self._active else tokens.fg_muted
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, _event): # pylint: disable=invalid-name
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
rect = QRectF(self.rect()).adjusted(0.5, 0.5, -0.5, -0.5)
|
||||
fill = (
|
||||
alpha(self._tone, 0.28)
|
||||
if self._active
|
||||
else alpha(self._tone, 0.16) if self.underMouse() else QColor(0, 0, 0, 0)
|
||||
)
|
||||
painter.setPen(QPen(self._tone if self._active else alpha(self._tone, 0.45), 1))
|
||||
painter.setBrush(fill)
|
||||
painter.drawRoundedRect(rect, rect.height() / 2, rect.height() / 2)
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
painter.setBrush(self._tone)
|
||||
painter.drawEllipse(QRectF(10, rect.height() / 2 - 3, 7, 7))
|
||||
painter.setPen(self._fg)
|
||||
painter.setFont(self.font())
|
||||
painter.drawText(rect.adjusted(22, 0, 0, 0), Qt.AlignmentFlag.AlignVCenter, self.text())
|
||||
painter.end()
|
||||
|
||||
def enterEvent(self, event): # pylint: disable=invalid-name
|
||||
super().enterEvent(event)
|
||||
self.update()
|
||||
|
||||
def leaveEvent(self, event): # pylint: disable=invalid-name
|
||||
super().leaveEvent(event)
|
||||
self.update()
|
||||
|
||||
|
||||
class SectionHeader(QWidget):
|
||||
"""Lock icon, section title with count and a rule line."""
|
||||
|
||||
def __init__(self, parent: QWidget | None = None):
|
||||
super().__init__(parent)
|
||||
self.setFixedHeight(26)
|
||||
layout = QHBoxLayout(self)
|
||||
layout.setContentsMargins(4, 0, 0, 0)
|
||||
layout.setSpacing(6)
|
||||
self.icon = icon_label(self, 14)
|
||||
self.title = QLabel(self)
|
||||
self.rule = QFrame(self)
|
||||
self.rule.setFixedHeight(1)
|
||||
self.rule.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
layout.addWidget(self.icon)
|
||||
layout.addWidget(self.title)
|
||||
layout.addWidget(self.rule, 1, Qt.AlignmentFlag.AlignVCenter)
|
||||
|
||||
def set_row(self, row: dict, tokens: ThemeTokens) -> None:
|
||||
"""Update from a header row of the view state."""
|
||||
armed = bool(row["armed"])
|
||||
color = tokens.fg if armed else tokens.fg_muted
|
||||
set_icon(
|
||||
self.icon, "lock" if row["name"] == "interlock" else "lock_open_right", color, 14, armed
|
||||
)
|
||||
self.title.setText(f"{row['title']} · {row['count']}")
|
||||
self.title.setStyleSheet(
|
||||
f"color: {color.name()}; font-size: 11px; font-weight: 700; {PLAIN}"
|
||||
)
|
||||
self.rule.setStyleSheet(f"background: {tokens.border.name()}; border: none;")
|
||||
|
||||
|
||||
class _ClickableHeader(QWidget):
|
||||
clicked = Signal()
|
||||
|
||||
def __init__(self, parent: QWidget | None = None):
|
||||
super().__init__(parent)
|
||||
self.setCursor(Qt.CursorShape.PointingHandCursor)
|
||||
self.setAttribute(Qt.WidgetAttribute.WA_Hover, True)
|
||||
|
||||
def mousePressEvent(self, event): # pylint: disable=invalid-name
|
||||
if event.button() == Qt.MouseButton.LeftButton:
|
||||
self.clicked.emit()
|
||||
event.accept()
|
||||
return
|
||||
super().mousePressEvent(event)
|
||||
|
||||
def enterEvent(self, event): # pylint: disable=invalid-name
|
||||
super().enterEvent(event)
|
||||
self.parentWidget().update()
|
||||
|
||||
def leaveEvent(self, event): # pylint: disable=invalid-name
|
||||
super().leaveEvent(event)
|
||||
self.parentWidget().update()
|
||||
|
||||
|
||||
class StateRow(QFrame):
|
||||
"""One beamline state: status, label, interlock lock and expandable details."""
|
||||
|
||||
def __init__(self, manager: "BeamlineStatesQWidget", parent: QWidget | None = None):
|
||||
super().__init__(parent)
|
||||
self._manager = manager
|
||||
self.row: dict = {}
|
||||
self._flash = 0.0
|
||||
self._flash_animation = QVariantAnimation(self)
|
||||
self._flash_animation.setDuration(1400)
|
||||
self._flash_animation.setKeyValueAt(0.0, 0.0)
|
||||
self._flash_animation.setKeyValueAt(0.5, 1.0)
|
||||
self._flash_animation.setKeyValueAt(1.0, 0.0)
|
||||
self._flash_animation.setEasingCurve(QEasingCurve.Type.InOutSine)
|
||||
self._flash_animation.setLoopCount(-1)
|
||||
self._flash_animation.valueChanged.connect(self._on_flash)
|
||||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
self.header = _ClickableHeader(self)
|
||||
self.header.setFixedHeight(52)
|
||||
self.header.clicked.connect(lambda: self._manager.toggle_expanded(self.row["name"]))
|
||||
header_layout = QHBoxLayout(self.header)
|
||||
header_layout.setContentsMargins(10, 0, 4, 0)
|
||||
header_layout.setSpacing(10)
|
||||
self.icon = icon_label(self.header, 30)
|
||||
self.icon.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
text_layout = QVBoxLayout()
|
||||
text_layout.setSpacing(1)
|
||||
self.name_label = ElidingLabel(self.header)
|
||||
self.detail_label = ElidingLabel(self.header)
|
||||
for label in (self.name_label, self.detail_label):
|
||||
label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
|
||||
label.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred)
|
||||
text_layout.addStretch(1)
|
||||
text_layout.addWidget(self.name_label)
|
||||
text_layout.addWidget(self.detail_label)
|
||||
text_layout.addStretch(1)
|
||||
self.status_pill = StatusPill(self.header)
|
||||
self.lock_button = IconButton("lock_open_right", "", self.header)
|
||||
self.lock_button.clicked.connect(
|
||||
lambda: self._manager.toggle_interlock_for(self.row["name"])
|
||||
)
|
||||
self.expand_icon = icon_label(self.header)
|
||||
header_layout.addWidget(self.icon)
|
||||
header_layout.addLayout(text_layout, 1)
|
||||
header_layout.addWidget(self.status_pill)
|
||||
header_layout.addWidget(self.lock_button)
|
||||
header_layout.addWidget(self.expand_icon)
|
||||
layout.addWidget(self.header)
|
||||
|
||||
self.details = QWidget(self)
|
||||
self.details.setStyleSheet("background: none;")
|
||||
details_layout = QVBoxLayout(self.details)
|
||||
details_layout.setContentsMargins(50, 0, 12, 12)
|
||||
details_layout.setSpacing(8)
|
||||
self.params = QGridLayout()
|
||||
self.params.setHorizontalSpacing(14)
|
||||
self.params.setVerticalSpacing(3)
|
||||
details_layout.addLayout(self.params)
|
||||
warning_row = QHBoxLayout()
|
||||
warning_row.setSpacing(0)
|
||||
self.trip_switch = ToggleSwitch(self.details)
|
||||
self.trip_switch.setFixedWidth(46)
|
||||
self.trip_switch.setAccessibleName("WARNING trips the scan interlock")
|
||||
self.trip_switch.clicked.connect(
|
||||
lambda checked: self._manager.set_trip_on_warning(self.row["name"], checked)
|
||||
)
|
||||
self.trip_label = QLabel("WARNING trips the scan interlock", self.details)
|
||||
warning_row.addWidget(self.trip_switch)
|
||||
warning_row.addWidget(self.trip_label)
|
||||
warning_row.addStretch(1)
|
||||
details_layout.addLayout(warning_row)
|
||||
buttons = QHBoxLayout()
|
||||
buttons.setSpacing(6)
|
||||
self.edit_button = TextButton("Edit…", "neutral", "edit", self.details)
|
||||
self.edit_button.clicked.connect(lambda: self._manager.edit_state(self.row["name"]))
|
||||
self.remove_button = TextButton("Remove", "ghost", "delete", self.details)
|
||||
self.remove_button.clicked.connect(lambda: self._manager.remove_state(self.row["name"]))
|
||||
buttons.addWidget(self.edit_button)
|
||||
buttons.addWidget(self.remove_button)
|
||||
buttons.addStretch(1)
|
||||
details_layout.addLayout(buttons)
|
||||
layout.addWidget(self.details)
|
||||
self._param_labels: list[QLabel] = []
|
||||
|
||||
def set_row(self, row: dict, tokens: ThemeTokens) -> None:
|
||||
"""Update from a state row of the view state."""
|
||||
previous = self.row
|
||||
self.row = row
|
||||
tone = tone_color(row["tone"], tokens)
|
||||
if (previous.get("icon"), previous.get("tone"), previous.get("_theme")) != (
|
||||
row["icon"],
|
||||
row["tone"],
|
||||
tokens.name,
|
||||
):
|
||||
set_icon(self.icon, row["icon"], tone, 18, True)
|
||||
self.icon.setStyleSheet(
|
||||
f"background: {rgba(tone, 0.18)}; border: none; border-radius: 15px;"
|
||||
)
|
||||
row["_theme"] = tokens.name
|
||||
self.name_label.setText(row["name"])
|
||||
self.detail_label.setText(row["label"])
|
||||
self.setToolTip(row["label"])
|
||||
self.name_label.setStyleSheet(
|
||||
f"color: {tokens.fg.name()}; font-size: 13px; font-weight: 600; {PLAIN}"
|
||||
)
|
||||
self.detail_label.setStyleSheet(
|
||||
f"color: {tokens.fg_muted.name()}; font-size: 11px; {PLAIN}"
|
||||
)
|
||||
self.status_pill.set_status(row["statusText"], tone)
|
||||
|
||||
watched = row["watched"]
|
||||
self.lock_button.set_icon_name("lock" if watched else "lock_open_right")
|
||||
lock_color = tokens.danger if row["tripped"] else tokens.fg if watched else tokens.fg_subtle
|
||||
self.lock_button.setIcon(
|
||||
material_icon(
|
||||
"lock" if watched else "lock_open_right",
|
||||
size=(48, 48),
|
||||
color=lock_color,
|
||||
convert_to_pixmap=False,
|
||||
)
|
||||
)
|
||||
tip = (
|
||||
f"Watched by the scan interlock (accepts {row['acceptedText']}). "
|
||||
"Click to stop watching."
|
||||
if watched
|
||||
else "Not watched by the scan interlock. Click to watch this state."
|
||||
)
|
||||
self.lock_button.setToolTip(tip)
|
||||
self.lock_button.setAccessibleName(tip)
|
||||
set_icon(
|
||||
self.expand_icon, "expand_less" if row["expanded"] else "expand_more", tokens.fg_muted
|
||||
)
|
||||
|
||||
self.details.setVisible(row["expanded"])
|
||||
if row["expanded"]:
|
||||
self._set_params(row, tokens)
|
||||
self.trip_switch.setChecked(row["tripOnWarning"])
|
||||
self.trip_label.setStyleSheet(f"color: {tokens.fg.name()}; font-size: 13px; {PLAIN}")
|
||||
|
||||
if row["tripped"]:
|
||||
if self._flash_animation.state() != QVariantAnimation.State.Running:
|
||||
self._flash_animation.start()
|
||||
else:
|
||||
self._flash_animation.stop()
|
||||
self._flash = 0.0
|
||||
self.update()
|
||||
|
||||
def _set_params(self, row: dict, tokens: ThemeTokens) -> None:
|
||||
entries = [("type", row["stateType"])] + [(p["key"], p["value"]) for p in row["params"]]
|
||||
while len(self._param_labels) < 2 * len(entries):
|
||||
label = QLabel(self.details)
|
||||
label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
|
||||
self._param_labels.append(label)
|
||||
for index, label in enumerate(self._param_labels):
|
||||
entry = index // 2
|
||||
if entry >= len(entries):
|
||||
label.hide()
|
||||
continue
|
||||
is_value = index % 2 == 1
|
||||
label.setText(entries[entry][1 if is_value else 0])
|
||||
color = tokens.fg if is_value else tokens.fg_subtle
|
||||
family = "font-family: monospace;" if is_value and entry > 0 else ""
|
||||
label.setStyleSheet(f"color: {color.name()}; font-size: 12px; {family} {PLAIN}")
|
||||
self.params.addWidget(label, entry, 1 if is_value else 0)
|
||||
label.show()
|
||||
self.params.setColumnStretch(1, 1)
|
||||
|
||||
def _on_flash(self, value) -> None:
|
||||
self._flash = float(value)
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, _event): # pylint: disable=invalid-name
|
||||
if not self.row:
|
||||
return
|
||||
tokens = ThemeTokens()
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
tripped = self.row.get("tripped")
|
||||
expanded = self.row.get("expanded")
|
||||
if tripped:
|
||||
fill = alpha(tokens.danger, 0.08 + 0.10 * self._flash)
|
||||
pen = QPen(alpha(tokens.danger, 0.5 + 0.5 * self._flash), 2)
|
||||
elif expanded:
|
||||
fill, pen = tokens.field, QPen(tokens.border, 1)
|
||||
else:
|
||||
hovered = self.header.underMouse()
|
||||
fill = tokens.hover if hovered else QColor(0, 0, 0, 0)
|
||||
pen = QPen(QColor(0, 0, 0, 0), 1)
|
||||
inset = pen.widthF() / 2
|
||||
painter.setPen(pen)
|
||||
painter.setBrush(fill)
|
||||
painter.drawRoundedRect(QRectF(self.rect()).adjusted(inset, inset, -inset, -inset), 8, 8)
|
||||
painter.end()
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Stop animations."""
|
||||
self._flash_animation.stop()
|
||||
self.status_pill.cleanup()
|
||||
|
||||
|
||||
class BeamlineStatesQWidget(BeamlineStatesPortBase):
|
||||
"""Beamline state manager with the modernised UX in plain QWidgets."""
|
||||
|
||||
def _init_view(self) -> None:
|
||||
self.setMinimumSize(320, 300)
|
||||
self._rows: dict[str, StateRow] = {}
|
||||
self._headers: dict[str, SectionHeader] = {}
|
||||
outer = QVBoxLayout(self)
|
||||
outer.setContentsMargins(8, 8, 8, 8)
|
||||
self.card = Card("Beamline states", parent=self)
|
||||
outer.addWidget(self.card)
|
||||
body = self.card.body
|
||||
body.setSpacing(8)
|
||||
|
||||
# scan interlock
|
||||
self.interlock_panel = QFrame(self.card)
|
||||
self.interlock_panel.setObjectName("interlockPanel")
|
||||
panel = QHBoxLayout(self.interlock_panel)
|
||||
panel.setContentsMargins(10, 6, 6, 6)
|
||||
panel.setSpacing(8)
|
||||
self.interlock_icon = icon_label(self.interlock_panel)
|
||||
titles = QVBoxLayout()
|
||||
titles.setSpacing(0)
|
||||
self.interlock_title = QLabel("Scan interlock", self.interlock_panel)
|
||||
self.interlock_sub = QLabel(self.interlock_panel)
|
||||
titles.addWidget(self.interlock_title)
|
||||
titles.addWidget(self.interlock_sub)
|
||||
self.interlock_pill = StatusPill(self.interlock_panel)
|
||||
self.interlock_switch = ToggleSwitch(self.interlock_panel)
|
||||
self.interlock_switch.setFixedWidth(46)
|
||||
self.interlock_switch.setAccessibleName("Arm scan interlock")
|
||||
self.interlock_switch.clicked.connect(self.set_interlock_enabled)
|
||||
panel.addWidget(self.interlock_icon)
|
||||
panel.addLayout(titles, 1)
|
||||
panel.addWidget(self.interlock_pill)
|
||||
panel.addWidget(self.interlock_switch)
|
||||
body.addWidget(self.interlock_panel)
|
||||
|
||||
# tripped banner
|
||||
self.banner = QFrame(self.card)
|
||||
self.banner.setObjectName("trippedBanner")
|
||||
banner_layout = QHBoxLayout(self.banner)
|
||||
banner_layout.setContentsMargins(10, 7, 10, 7)
|
||||
banner_layout.setSpacing(8)
|
||||
self.banner_icon = icon_label(self.banner, 16)
|
||||
self.banner_text = QLabel(self.banner)
|
||||
self.banner_text.setWordWrap(True)
|
||||
banner_layout.addWidget(self.banner_icon, 0, Qt.AlignmentFlag.AlignTop)
|
||||
banner_layout.addWidget(self.banner_text, 1)
|
||||
body.addWidget(self.banner)
|
||||
|
||||
# search, clear and add
|
||||
self.tools = QWidget(self.card)
|
||||
tools = QHBoxLayout(self.tools)
|
||||
tools.setContentsMargins(0, 0, 0, 0)
|
||||
tools.setSpacing(6)
|
||||
self.search = QLineEdit(self.tools)
|
||||
self.search.setPlaceholderText("Filter by name or device")
|
||||
self.search.setAccessibleName("Filter states")
|
||||
self.search.setClearButtonEnabled(True)
|
||||
self.search_action = self.search.addAction(
|
||||
material_icon("search", convert_to_pixmap=False),
|
||||
QLineEdit.ActionPosition.LeadingPosition,
|
||||
)
|
||||
self.search.textEdited.connect(self.set_filter_text)
|
||||
self.clear_button = IconButton("filter_alt_off", "Clear filters", self.tools)
|
||||
self.clear_button.clicked.connect(self.clear_filters)
|
||||
self.add_button = TextButton("Add", "primary", "add", self.tools)
|
||||
self.add_button.clicked.connect(self.open_add_state_dialog)
|
||||
tools.addWidget(self.search, 1)
|
||||
tools.addWidget(self.clear_button)
|
||||
tools.addWidget(self.add_button)
|
||||
body.addWidget(self.tools)
|
||||
|
||||
# status chips
|
||||
self.chips_box = QWidget(self.card)
|
||||
self.chips_layout = QHBoxLayout(self.chips_box)
|
||||
self.chips_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.chips_layout.setSpacing(6)
|
||||
self.chips_layout.addStretch(1)
|
||||
self._chips: list[StatusChip] = []
|
||||
body.addWidget(self.chips_box)
|
||||
|
||||
# list
|
||||
self.scroll = QScrollArea(self.card)
|
||||
self.scroll.setWidgetResizable(True)
|
||||
self.scroll.setFrameShape(QFrame.Shape.NoFrame)
|
||||
self.scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
self.list_widget = QWidget(self.scroll)
|
||||
self.list_layout = QVBoxLayout(self.list_widget)
|
||||
self.list_layout.setContentsMargins(0, 0, 2, 0)
|
||||
self.list_layout.setSpacing(4)
|
||||
self.list_layout.addStretch(1)
|
||||
self.scroll.setWidget(self.list_widget)
|
||||
body.addWidget(self.scroll, 1)
|
||||
|
||||
# no matches
|
||||
self.no_match = QWidget(self.card)
|
||||
no_match = QVBoxLayout(self.no_match)
|
||||
no_match.addStretch(1)
|
||||
self.no_match_label = QLabel("No states match the filters.", self.no_match)
|
||||
self.no_match_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.no_match_button = TextButton(
|
||||
"Clear filters", "neutral", "filter_alt_off", self.no_match
|
||||
)
|
||||
self.no_match_button.clicked.connect(self.clear_filters)
|
||||
no_match.addWidget(self.no_match_label)
|
||||
no_match.addWidget(self.no_match_button, 0, Qt.AlignmentFlag.AlignHCenter)
|
||||
no_match.addStretch(1)
|
||||
body.addWidget(self.no_match, 1)
|
||||
|
||||
# hidden by filters
|
||||
self.hidden_button = QPushButton(self.card)
|
||||
self.hidden_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||||
self.hidden_button.setFixedHeight(28)
|
||||
self.hidden_button.clicked.connect(lambda: self.set_show_hidden(not self._show_hidden))
|
||||
body.addWidget(self.hidden_button)
|
||||
|
||||
# empty state
|
||||
self.empty = QWidget(self.card)
|
||||
empty = QVBoxLayout(self.empty)
|
||||
empty.setSpacing(8)
|
||||
empty.addStretch(1)
|
||||
self.empty_icon = icon_label(self.empty, 36)
|
||||
self.empty_title = QLabel("No beamline states yet", self.empty)
|
||||
self.empty_text = QLabel(
|
||||
"A state watches a device, for example whether a motor is within limits, "
|
||||
"and can block scans through the scan interlock.",
|
||||
self.empty,
|
||||
)
|
||||
self.empty_text.setWordWrap(True)
|
||||
for label in (self.empty_title, self.empty_text):
|
||||
label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.empty_button = TextButton("Add state", "primary", "add", self.empty)
|
||||
self.empty_button.clicked.connect(self.open_add_state_dialog)
|
||||
empty.addWidget(self.empty_icon, 0, Qt.AlignmentFlag.AlignHCenter)
|
||||
empty.addWidget(self.empty_title)
|
||||
empty.addWidget(self.empty_text)
|
||||
empty.addWidget(self.empty_button, 0, Qt.AlignmentFlag.AlignHCenter)
|
||||
empty.addStretch(1)
|
||||
body.addWidget(self.empty, 1)
|
||||
|
||||
self._apply_styles(ThemeTokens())
|
||||
|
||||
# ---- rendering -----------------------------------------------------------------------
|
||||
|
||||
def _apply_styles(self, tokens: ThemeTokens) -> None:
|
||||
self.card.setStyleSheet(self.card.styleSheet() + field_qss(tokens))
|
||||
self.search.setStyleSheet("QLineEdit { padding-left: 2px; }")
|
||||
self.search_action.setIcon(
|
||||
material_icon("search", color=tokens.fg_subtle, convert_to_pixmap=False)
|
||||
)
|
||||
for widget in (self.tools, self.chips_box, self.no_match, self.empty):
|
||||
widget.setStyleSheet("background: none;")
|
||||
self.scroll.setStyleSheet(
|
||||
f"QScrollArea {{ background: {tokens.card.name()}; border: none; }}"
|
||||
f"QScrollArea > QWidget > QWidget {{ background: {tokens.card.name()}; }}"
|
||||
)
|
||||
self.interlock_title.setStyleSheet(
|
||||
f"color: {tokens.fg.name()}; font-size: 13px; font-weight: 600; {PLAIN}"
|
||||
)
|
||||
self.interlock_sub.setStyleSheet(
|
||||
f"color: {tokens.fg_muted.name()}; font-size: 11px; {PLAIN}"
|
||||
)
|
||||
self.banner.setStyleSheet(
|
||||
f"QFrame#trippedBanner {{ background: {rgba(tokens.danger, 0.14)};"
|
||||
f" border: 1px solid {rgba(tokens.danger, 0.5)}; border-radius: 8px; }}"
|
||||
)
|
||||
self.banner_text.setStyleSheet(f"color: {tokens.fg.name()}; font-size: 12px; {PLAIN}")
|
||||
set_icon(self.banner_icon, "block", tokens.danger, 16)
|
||||
self.no_match_label.setStyleSheet(
|
||||
f"color: {tokens.fg_muted.name()}; font-size: 13px; {PLAIN}"
|
||||
)
|
||||
self.hidden_button.setStyleSheet(
|
||||
f"QPushButton {{ color: {tokens.fg_muted.name()}; font-size: 12px; text-align: left;"
|
||||
f" border: none; border-radius: 6px; padding: 0 6px; background: transparent; }}"
|
||||
f"QPushButton:hover {{ background: {tokens.hover.name()}; }}"
|
||||
)
|
||||
set_icon(self.empty_icon, "rule", tokens.fg_subtle, 36)
|
||||
self.empty_title.setStyleSheet(
|
||||
f"color: {tokens.fg.name()}; font-size: 14px; font-weight: 600; {PLAIN}"
|
||||
)
|
||||
self.empty_text.setStyleSheet(f"color: {tokens.fg_muted.name()}; font-size: 12px; {PLAIN}")
|
||||
|
||||
def _sync_view(self) -> None:
|
||||
if not hasattr(self, "card"):
|
||||
return
|
||||
state = self.view_state()
|
||||
tokens = ThemeTokens()
|
||||
total = state["total"]
|
||||
self.card.set_title(
|
||||
"Beamline states", f"{total} state{'s' if total != 1 else ''}" if total else ""
|
||||
)
|
||||
|
||||
# interlock
|
||||
enabled = state["interlockEnabled"]
|
||||
tone = tone_color(state["interlockTone"], tokens)
|
||||
if enabled:
|
||||
panel_bg, panel_border = rgba(tone, 0.10), rgba(tone, 0.45)
|
||||
else:
|
||||
panel_bg, panel_border = tokens.field.name(), tokens.border.name()
|
||||
self.interlock_panel.setStyleSheet(
|
||||
f"QFrame#interlockPanel {{ background: {panel_bg}; border: 1px solid {panel_border};"
|
||||
f" border-radius: 8px; }}"
|
||||
)
|
||||
set_icon(self.interlock_icon, "lock" if enabled else "no_encryption", tone, 18, True)
|
||||
count = state["interlockCount"]
|
||||
self.interlock_sub.setText(
|
||||
"No states watched"
|
||||
if count == 0
|
||||
else f"{count} state{' watched' if count == 1 else 's watched'}"
|
||||
)
|
||||
self.interlock_pill.set_status(state["interlockText"], tone, bool(state["banner"]))
|
||||
self.interlock_switch.setChecked(enabled)
|
||||
self.interlock_switch.setToolTip(
|
||||
"Disarm the scan interlock" if enabled else "Arm the scan interlock"
|
||||
)
|
||||
self.banner.setVisible(bool(state["banner"]))
|
||||
self.banner_text.setText(state["banner"])
|
||||
|
||||
# tools and chips
|
||||
empty = state["empty"]
|
||||
self.tools.setVisible(not empty)
|
||||
self.chips_box.setVisible(not empty)
|
||||
if not self.search.hasFocus() and self.search.text() != state["filterText"]:
|
||||
self.search.setText(state["filterText"])
|
||||
self.clear_button.setVisible(state["filtersActive"])
|
||||
self._sync_chips(state["chips"], tokens)
|
||||
|
||||
# list
|
||||
show_list = not empty and not state["noMatches"]
|
||||
self.scroll.setVisible(show_list)
|
||||
self._sync_rows(state["rows"], tokens)
|
||||
self.no_match.setVisible(state["noMatches"])
|
||||
self.hidden_button.setVisible(state["hiddenCount"] > 0 and not state["noMatches"])
|
||||
self.hidden_button.setText(
|
||||
state["hiddenText"] + (" · Hide them" if state["showHidden"] else " · Show them")
|
||||
)
|
||||
self.hidden_button.setIcon(
|
||||
material_icon(
|
||||
"visibility_off" if state["showHidden"] else "visibility",
|
||||
color=tokens.fg_muted,
|
||||
convert_to_pixmap=False,
|
||||
)
|
||||
)
|
||||
self.hidden_button.setIconSize(QSize(16, 16))
|
||||
self.empty.setVisible(empty)
|
||||
|
||||
def _sync_chips(self, chips: list[dict], tokens: ThemeTokens) -> None:
|
||||
while len(self._chips) < len(chips):
|
||||
chip = StatusChip(self.chips_box)
|
||||
chip.clicked.connect(lambda _=False, c=chip: self.toggle_status_filter(c.status))
|
||||
self.chips_layout.insertWidget(len(self._chips), chip)
|
||||
self._chips.append(chip)
|
||||
for index, chip in enumerate(self._chips):
|
||||
if index < len(chips):
|
||||
chip.set_chip(chips[index], tokens)
|
||||
chip.show()
|
||||
else:
|
||||
chip.hide()
|
||||
|
||||
def _sync_rows(self, rows: list[dict], tokens: ThemeTokens) -> None:
|
||||
keys = {row["key"] for row in rows}
|
||||
for key in [key for key in self._rows if key not in keys]:
|
||||
widget = self._rows.pop(key)
|
||||
widget.cleanup()
|
||||
widget.deleteLater()
|
||||
for key in [key for key in self._headers if key not in keys]:
|
||||
self._headers.pop(key).deleteLater()
|
||||
for position, row in enumerate(rows):
|
||||
if row["kind"] == "header":
|
||||
widget = self._headers.get(row["key"])
|
||||
if widget is None:
|
||||
widget = SectionHeader(self.list_widget)
|
||||
self._headers[row["key"]] = widget
|
||||
else:
|
||||
widget = self._rows.get(row["key"])
|
||||
if widget is None:
|
||||
widget = StateRow(self, self.list_widget)
|
||||
self._rows[row["key"]] = widget
|
||||
widget.set_row(row, tokens)
|
||||
if self.list_layout.indexOf(widget) != position:
|
||||
self.list_layout.insertWidget(position, widget)
|
||||
widget.show()
|
||||
|
||||
@property
|
||||
def state_rows(self) -> dict[str, StateRow]:
|
||||
"""Row widgets by state name, for tests and inspection."""
|
||||
return {row.row["name"]: row for row in self._rows.values()}
|
||||
|
||||
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):
|
||||
for row in self._rows.values():
|
||||
row.cleanup()
|
||||
self.interlock_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 = BeamlineStatesQWidget()
|
||||
widget.resize(460, 560)
|
||||
widget.show()
|
||||
sys.exit(app.exec())
|
||||
@@ -0,0 +1,705 @@
|
||||
"""Logic shared by the QML and QWidget ports of the beamline state manager.
|
||||
|
||||
The ports keep the public API of :class:`BeamlineStateManager` (``clear_filters``,
|
||||
``collapse_all``, ``state_summary``, ``refresh_states``, ``update_available_states``, the add and
|
||||
filter dialogs and ``scan_interlock_enabled_changed``) but render from one display-ready state
|
||||
instead of one ``BeamlineStatePill`` widget per state. The manager subscribes to every state
|
||||
endpoint itself, so a state costs one row of data rather than a widget with its own connector
|
||||
subscription.
|
||||
|
||||
Differences in the UX compared to the original manager:
|
||||
|
||||
* status counts double as quick status filters, and a search field filters by name or device;
|
||||
* the scan interlock is a labelled switch with a banner naming the states that trip it;
|
||||
* an expanded row shows the state parameters read-only, with Edit (a form dialog) and Remove.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from bec_lib import bl_states, messages
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from qtpy.QtCore import Signal
|
||||
from qtpy.QtWidgets import (
|
||||
QDialog,
|
||||
QDialogButtonBox,
|
||||
QFormLayout,
|
||||
QLabel,
|
||||
QMessageBox,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.bec_connector import ConnectionConfig
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.utils.forms_from_types.pydantic_widget_form import PydanticWidgetForm
|
||||
from bec_widgets.utils.ux_kit import PortedPropertiesMixin
|
||||
from bec_widgets.widgets.services.beamline_states.dialogs import (
|
||||
BEAMLINE_STATE_STATUS_LABELS,
|
||||
SUPPORTED_BEAMLINE_STATES,
|
||||
AddBeamlineStateDialog,
|
||||
DeviceFilterDialog,
|
||||
StatusFilterDialog,
|
||||
)
|
||||
|
||||
STATUS_ORDER = ("invalid", "warning", "unknown", "valid")
|
||||
STATUS_TONES = {"valid": "success", "invalid": "danger", "warning": "warning", "unknown": "muted"}
|
||||
STATUS_ICONS = {
|
||||
"valid": "check_circle",
|
||||
"invalid": "cancel",
|
||||
"warning": "warning",
|
||||
"unknown": "help",
|
||||
}
|
||||
STATUS_WORDS = {"valid": "valid", "invalid": "invalid", "warning": "warning", "unknown": "unknown"}
|
||||
DEFAULT_INTERLOCK_STATUSES = ["valid", "warning"]
|
||||
|
||||
|
||||
def state_device(config: messages.BeamlineStateConfig | None) -> str | None:
|
||||
"""Device a state watches, if its parameters name one."""
|
||||
device = config.parameters.get("device") if config is not None else None
|
||||
return str(device) if device else None
|
||||
|
||||
|
||||
def format_parameter(value: Any) -> str:
|
||||
"""Compact text for a state parameter value."""
|
||||
if isinstance(value, float):
|
||||
return f"{value:g}"
|
||||
if value is None:
|
||||
return "—"
|
||||
return str(value)
|
||||
|
||||
|
||||
def config_class_for(state_type: str) -> type[bl_states.BeamlineStateConfig] | None:
|
||||
"""Pydantic config class of a supported state type, or ``None``."""
|
||||
for state_class in SUPPORTED_BEAMLINE_STATES:
|
||||
if state_type in {state_class.__name__, state_class.CONFIG_CLASS.state_type}:
|
||||
return state_class.CONFIG_CLASS
|
||||
return None
|
||||
|
||||
|
||||
class EditBeamlineStateDialog(QDialog):
|
||||
"""Form dialog that edits the parameters of one beamline state.
|
||||
|
||||
Args:
|
||||
config(BeamlineStateConfig): Current configuration of the state.
|
||||
parent(QWidget): Parent widget.
|
||||
client: BEC client handed to the form (device pickers).
|
||||
"""
|
||||
|
||||
def __init__(self, config: messages.BeamlineStateConfig, parent=None, client=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(f"Edit {config.name}")
|
||||
self._config: bl_states.BeamlineStateConfig | None = None
|
||||
config_class = config_class_for(config.state_type)
|
||||
if config_class is None:
|
||||
raise ValueError(f"Unsupported beamline state type '{config.state_type}'.")
|
||||
self.form = PydanticWidgetForm(
|
||||
config_class, parent=self, client=client, read_only_fields={"name"}
|
||||
)
|
||||
data = {"name": config.name}
|
||||
data.update(
|
||||
{
|
||||
key: value
|
||||
for key, value in config.parameters.items()
|
||||
if key in config_class.model_fields
|
||||
}
|
||||
)
|
||||
self.form.set_partial_data(data)
|
||||
self.form.mark_clean()
|
||||
|
||||
header = QFormLayout()
|
||||
header.addRow("Type", QLabel(config.state_type, self))
|
||||
buttons = QDialogButtonBox(
|
||||
QDialogButtonBox.StandardButton.Save | QDialogButtonBox.StandardButton.Cancel, self
|
||||
)
|
||||
buttons.accepted.connect(self.accept)
|
||||
buttons.rejected.connect(self.reject)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.addLayout(header)
|
||||
layout.addWidget(self.form)
|
||||
layout.addWidget(buttons)
|
||||
|
||||
def accept(self) -> None:
|
||||
try:
|
||||
self._config = self.form.model_instance() # type: ignore[assignment]
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Invalid Beamline State", str(exc))
|
||||
return
|
||||
super().accept()
|
||||
|
||||
@property
|
||||
def config_result(self) -> bl_states.BeamlineStateConfig | None:
|
||||
"""Validated configuration after the dialog was accepted."""
|
||||
return self._config
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Release the form's device subscriptions."""
|
||||
self.form.cleanup()
|
||||
|
||||
|
||||
class BeamlineStatesPortBase(PortedPropertiesMixin, BECWidget, QWidget):
|
||||
"""Shared implementation of the beamline state manager ports.
|
||||
|
||||
Subclasses build the renderer in :meth:`_init_view` and redraw it from :meth:`view_state` in
|
||||
:meth:`_sync_view`.
|
||||
"""
|
||||
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
rpc_widget_class = "BeamlineStateManager"
|
||||
ICON_NAME = "format_list_bulleted"
|
||||
USER_ACCESS = [
|
||||
"clear_filters",
|
||||
"collapse_all",
|
||||
"state_summary",
|
||||
"remove",
|
||||
"attach",
|
||||
"detach",
|
||||
"screenshot",
|
||||
]
|
||||
|
||||
scan_interlock_enabled_changed = Signal(bool)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent: QWidget | None = None,
|
||||
client=None,
|
||||
config: ConnectionConfig | None = None,
|
||||
gui_id: str | None = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
|
||||
self._configs: dict[str, messages.BeamlineStateConfig] = {}
|
||||
self._order: list[str] = []
|
||||
self._states: dict[str, tuple[str, str]] = {}
|
||||
self._subscribed: set[str] = set()
|
||||
self._expanded: set[str] = set()
|
||||
self._selected_statuses: set[str] | None = None
|
||||
self._selected_devices: set[str] | None = None
|
||||
self._filter_text = ""
|
||||
self._show_hidden = False
|
||||
self._scan_interlock = self.client.builtin_actors.scan_interlock
|
||||
self._interlock_enabled = False
|
||||
self._interlock_states: dict[str, list[str]] = {}
|
||||
self._preferred_statuses: dict[str, list[str]] = {}
|
||||
self._enabled_config = None
|
||||
|
||||
self._init_view()
|
||||
|
||||
self.bec_dispatcher.connect_slot(
|
||||
self.update_available_states, MessageEndpoints.available_beamline_states()
|
||||
)
|
||||
self.bec_dispatcher.connect_slot(
|
||||
self._refresh_scan_interlock,
|
||||
MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"),
|
||||
)
|
||||
self.scan_interlock_enabled_changed.connect(self._on_interlock_enabled_changed)
|
||||
self._enabled_config = getattr(self._scan_interlock, "_enabled", None)
|
||||
if self._enabled_config is not None and hasattr(self._enabled_config, "subscribe"):
|
||||
self._enabled_config.subscribe(self._on_interlock_enabled_stream)
|
||||
self._refresh_scan_interlock()
|
||||
self.refresh_states()
|
||||
self._refresh()
|
||||
|
||||
# ---- renderer hooks ------------------------------------------------------------------
|
||||
|
||||
def _init_view(self) -> None:
|
||||
"""Build the renderer."""
|
||||
|
||||
def _sync_view(self) -> None:
|
||||
"""Redraw the renderer from :meth:`view_state`."""
|
||||
|
||||
def _refresh(self) -> None:
|
||||
self._sync_view()
|
||||
|
||||
# ---- original public API -------------------------------------------------------------
|
||||
|
||||
@SafeSlot()
|
||||
def clear_filters(self) -> None:
|
||||
"""Remove every status, device and text filter."""
|
||||
self._selected_statuses = None
|
||||
self._selected_devices = None
|
||||
self._filter_text = ""
|
||||
self._show_hidden = False
|
||||
self._refresh()
|
||||
|
||||
@SafeSlot()
|
||||
def collapse_all(self) -> None:
|
||||
"""Collapse the details of all states."""
|
||||
self._expanded.clear()
|
||||
self._refresh()
|
||||
|
||||
def state_summary(self) -> dict[str, dict[str, str]]:
|
||||
"""
|
||||
Return all beamline states (including filtered ones) with their current status and label.
|
||||
|
||||
Returns:
|
||||
dict: Mapping of state name to a dictionary with ``status`` and ``label`` keys.
|
||||
"""
|
||||
return {
|
||||
name: {"status": self._status(name), "label": self._label(name)} for name in self._order
|
||||
}
|
||||
|
||||
@SafeSlot()
|
||||
def refresh_states(self) -> None:
|
||||
"""Fetch the latest cached available beamline states and update the list immediately."""
|
||||
msg = self.client.connector.get_last(
|
||||
MessageEndpoints.available_beamline_states(), key="data"
|
||||
)
|
||||
if msg is not None:
|
||||
self.update_available_states(msg.content, msg.metadata)
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def update_available_states(
|
||||
self, content: dict[str, Any], _metadata: dict[str, Any] | None = None
|
||||
) -> None:
|
||||
"""Update the displayed states from ``AvailableBeamlineStatesMessage`` content."""
|
||||
state_configs: list[messages.BeamlineStateConfig] = content.get("states", [])
|
||||
self._configs = {state.name: state for state in state_configs}
|
||||
self._order = [state.name for state in state_configs]
|
||||
names = set(self._order)
|
||||
for name in sorted(self._subscribed - names):
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.update_state, MessageEndpoints.beamline_state(name)
|
||||
)
|
||||
self._subscribed.discard(name)
|
||||
self._states.pop(name, None)
|
||||
self._expanded.discard(name)
|
||||
for name in self._order:
|
||||
if name in self._subscribed:
|
||||
continue
|
||||
self._subscribed.add(name)
|
||||
msg = self.client.connector.get_last(MessageEndpoints.beamline_state(name), key="data")
|
||||
if msg is not None:
|
||||
self._store_state(msg.content)
|
||||
self.bec_dispatcher.connect_slot(
|
||||
self.update_state, MessageEndpoints.beamline_state(name)
|
||||
)
|
||||
self._refresh()
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def update_state(
|
||||
self, content: dict[str, Any], _metadata: dict[str, Any] | None = None
|
||||
) -> None:
|
||||
"""Update one state from a ``BeamlineStateMessage`` content dictionary."""
|
||||
if self._store_state(content):
|
||||
self._refresh()
|
||||
|
||||
@SafeSlot()
|
||||
def open_add_state_dialog(self) -> None:
|
||||
"""Ask for a new state and add it, optionally to the scan interlock."""
|
||||
dialog = AddBeamlineStateDialog(self, client=self.client)
|
||||
config = None
|
||||
add_to_interlock = False
|
||||
statuses = list(DEFAULT_INTERLOCK_STATUSES)
|
||||
try:
|
||||
if dialog.exec() == QDialog.DialogCode.Accepted:
|
||||
config = dialog.config_result
|
||||
add_to_interlock = dialog.add_to_interlock()
|
||||
statuses = dialog.interlock_statuses()
|
||||
finally:
|
||||
dialog.cleanup()
|
||||
dialog.deleteLater()
|
||||
if config is None:
|
||||
return
|
||||
try:
|
||||
self.client.beamline_states.add(config)
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Cannot Add State", str(exc))
|
||||
return
|
||||
self._preferred_statuses[config.name] = statuses
|
||||
if add_to_interlock:
|
||||
self._call_interlock(self._scan_interlock.add_state_to_interlock, config.name, statuses)
|
||||
|
||||
@SafeSlot()
|
||||
def open_status_filter_dialog(self) -> None:
|
||||
"""Choose the shown statuses in a dialog (the chips do the same inline)."""
|
||||
dialog = StatusFilterDialog(self._selected_statuses, self)
|
||||
if dialog.exec() != QDialog.DialogCode.Accepted:
|
||||
return
|
||||
self._selected_statuses = dialog.selected_statuses()
|
||||
self._refresh()
|
||||
|
||||
@SafeSlot()
|
||||
def open_device_filter_dialog(self) -> None:
|
||||
"""Choose the shown devices in a dialog (the search field does the same inline)."""
|
||||
devices = sorted(
|
||||
{device for config in self._configs.values() if (device := state_device(config))}
|
||||
)
|
||||
dialog = DeviceFilterDialog(devices, self._selected_devices, self._filter_text, self)
|
||||
if dialog.exec() != QDialog.DialogCode.Accepted:
|
||||
return
|
||||
self._selected_devices = dialog.selected_devices()
|
||||
self._filter_text = dialog.filter_text()
|
||||
self._refresh()
|
||||
|
||||
# ---- UX actions ----------------------------------------------------------------------
|
||||
|
||||
def toggle_status_filter(self, status: str) -> None:
|
||||
"""Show only states of the clicked status; click more chips to add statuses.
|
||||
|
||||
Clicking the last selected chip again shows all statuses.
|
||||
"""
|
||||
selected = set(self._selected_statuses or ())
|
||||
if status in selected:
|
||||
selected.discard(status)
|
||||
else:
|
||||
selected.add(status)
|
||||
self._selected_statuses = selected or None
|
||||
self._refresh()
|
||||
|
||||
def set_filter_text(self, text: str) -> None:
|
||||
"""Filter by state name or device; commas separate alternatives."""
|
||||
if text == self._filter_text:
|
||||
return
|
||||
self._filter_text = text
|
||||
self._refresh()
|
||||
|
||||
def set_expanded(self, name: str, expanded: bool) -> None:
|
||||
"""Expand or collapse the details of one state."""
|
||||
if expanded:
|
||||
self._expanded.add(name)
|
||||
else:
|
||||
self._expanded.discard(name)
|
||||
self._refresh()
|
||||
|
||||
def toggle_expanded(self, name: str) -> None:
|
||||
"""Toggle the details of one state."""
|
||||
self.set_expanded(name, name not in self._expanded)
|
||||
|
||||
def set_show_hidden(self, show: bool) -> None:
|
||||
"""Show or hide the states removed by the filters."""
|
||||
self._show_hidden = bool(show)
|
||||
self._refresh()
|
||||
|
||||
def set_interlock_enabled(self, enabled: bool) -> None:
|
||||
"""Arm or disarm the scan interlock."""
|
||||
enabled = bool(enabled)
|
||||
if enabled == self._interlock_enabled:
|
||||
return
|
||||
try:
|
||||
self._scan_interlock.enabled = enabled
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Cannot Toggle Scan Interlock", str(exc))
|
||||
self._refresh_scan_interlock()
|
||||
return
|
||||
self._interlock_enabled = enabled
|
||||
self._refresh()
|
||||
|
||||
def toggle_interlock_for(self, name: str) -> None:
|
||||
"""Add a state to the scan interlock or remove it."""
|
||||
if name in self._interlock_states:
|
||||
self._call_interlock(self._scan_interlock.remove_state_from_interlock, name)
|
||||
else:
|
||||
self._call_interlock(
|
||||
self._scan_interlock.add_state_to_interlock, name, self._accepted_statuses(name)
|
||||
)
|
||||
|
||||
def set_trip_on_warning(self, name: str, trip: bool) -> None:
|
||||
"""Whether a WARNING status of the state trips the scan interlock."""
|
||||
statuses = ["valid"] if trip else list(DEFAULT_INTERLOCK_STATUSES)
|
||||
if statuses == self._accepted_statuses(name):
|
||||
return
|
||||
self._preferred_statuses[name] = statuses
|
||||
if name in self._interlock_states:
|
||||
self._call_interlock(self._scan_interlock.add_state_to_interlock, name, statuses)
|
||||
else:
|
||||
self._refresh()
|
||||
|
||||
def edit_state(self, name: str) -> None:
|
||||
"""Edit the parameters of a state in a form dialog."""
|
||||
config = self._configs.get(name)
|
||||
if config is None:
|
||||
return
|
||||
try:
|
||||
dialog = EditBeamlineStateDialog(config, self, client=self.client)
|
||||
except ValueError as exc:
|
||||
QMessageBox.warning(self, "Cannot Edit State", str(exc))
|
||||
return
|
||||
new_config = None
|
||||
try:
|
||||
if dialog.exec() == QDialog.DialogCode.Accepted:
|
||||
new_config = dialog.config_result
|
||||
finally:
|
||||
dialog.cleanup()
|
||||
dialog.deleteLater()
|
||||
if new_config is not None:
|
||||
self.update_state_parameters(name, new_config)
|
||||
|
||||
def update_state_parameters(self, name: str, config: bl_states.BeamlineStateConfig) -> bool:
|
||||
"""Send new parameters of a state to BEC.
|
||||
|
||||
Returns:
|
||||
bool: Whether BEC accepted the update.
|
||||
"""
|
||||
state_client = getattr(self.client.beamline_states, name, None)
|
||||
if state_client is None:
|
||||
QMessageBox.warning(
|
||||
self, "Cannot Update State", f"Beamline state '{name}' is not available."
|
||||
)
|
||||
return False
|
||||
try:
|
||||
state_client.update_parameters(**config.model_dump(exclude={"name"}))
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Cannot Update State", str(exc))
|
||||
return False
|
||||
return True
|
||||
|
||||
def remove_state(self, name: str) -> None:
|
||||
"""Delete a state after confirmation."""
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Remove Beamline State",
|
||||
f"Remove beamline state '{name}'?",
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||||
QMessageBox.StandardButton.No,
|
||||
)
|
||||
if reply != QMessageBox.StandardButton.Yes:
|
||||
return
|
||||
try:
|
||||
self.client.beamline_states.delete(name)
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Cannot Remove State", str(exc))
|
||||
|
||||
# ---- scan interlock ------------------------------------------------------------------
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def _refresh_scan_interlock(
|
||||
self, _content: dict[str, Any] | None = None, _metadata: dict[str, Any] | None = None
|
||||
) -> None:
|
||||
"""Re-read the scan-interlock state from BEC."""
|
||||
try:
|
||||
self._interlock_enabled = bool(self._scan_interlock.enabled)
|
||||
self._interlock_states = {
|
||||
name: list(statuses)
|
||||
for name, statuses in dict(self._scan_interlock.states_watched).items()
|
||||
}
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Scan Interlock Unavailable", str(exc))
|
||||
return
|
||||
self._refresh()
|
||||
|
||||
def _on_interlock_enabled_stream(self, enabled: bool) -> None:
|
||||
"""Connector-thread callback; hands the value to the Qt thread through a signal."""
|
||||
self.scan_interlock_enabled_changed.emit(bool(enabled))
|
||||
|
||||
@SafeSlot(bool)
|
||||
def _on_interlock_enabled_changed(self, enabled: bool) -> None:
|
||||
self._interlock_enabled = bool(enabled)
|
||||
self._refresh()
|
||||
|
||||
def _call_interlock(self, method, *args) -> None:
|
||||
try:
|
||||
method(*args)
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
QMessageBox.warning(self, "Cannot Update Scan Interlock", str(exc))
|
||||
return
|
||||
self._refresh_scan_interlock()
|
||||
|
||||
def _accepted_statuses(self, name: str) -> list[str]:
|
||||
if name in self._interlock_states:
|
||||
return list(self._interlock_states[name])
|
||||
return list(self._preferred_statuses.get(name, DEFAULT_INTERLOCK_STATUSES))
|
||||
|
||||
def _is_tripped(self, name: str) -> bool:
|
||||
accepted = self._interlock_states.get(name)
|
||||
if not accepted or not self._interlock_enabled:
|
||||
return False
|
||||
return self._status(name) not in accepted
|
||||
|
||||
# ---- state bookkeeping ---------------------------------------------------------------
|
||||
|
||||
def _store_state(self, content: dict[str, Any]) -> bool:
|
||||
name = content.get("name")
|
||||
if not name or name not in self._configs:
|
||||
return False
|
||||
status = str(content.get("status", "unknown")).lower()
|
||||
if status not in BEAMLINE_STATE_STATUS_LABELS:
|
||||
status = "unknown"
|
||||
label = str(content.get("label", "No state information available."))
|
||||
if self._states.get(name) == (status, label):
|
||||
return False
|
||||
self._states[name] = (status, label)
|
||||
return True
|
||||
|
||||
def _status(self, name: str) -> str:
|
||||
return self._states.get(name, ("unknown", ""))[0]
|
||||
|
||||
def _label(self, name: str) -> str:
|
||||
return self._states.get(name, ("unknown", "No state information available."))[1]
|
||||
|
||||
def _matches_filters(self, name: str) -> bool:
|
||||
if (
|
||||
self._selected_statuses is not None
|
||||
and self._status(name) not in self._selected_statuses
|
||||
):
|
||||
return False
|
||||
device = state_device(self._configs.get(name))
|
||||
if self._selected_devices is not None and device not in self._selected_devices:
|
||||
return False
|
||||
tokens = [token.strip().casefold() for token in self._filter_text.split(",")]
|
||||
tokens = [token for token in tokens if token]
|
||||
if tokens:
|
||||
haystack = f"{name} {device or ''}".casefold()
|
||||
if not any(token in haystack for token in tokens):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _filters_active(self) -> bool:
|
||||
return (
|
||||
self._selected_statuses is not None
|
||||
or self._selected_devices is not None
|
||||
or bool(self._filter_text.strip())
|
||||
)
|
||||
|
||||
# ---- view state ----------------------------------------------------------------------
|
||||
|
||||
def _row(self, name: str) -> dict:
|
||||
config = self._configs[name]
|
||||
status = self._status(name)
|
||||
watched = name in self._interlock_states
|
||||
accepted = self._accepted_statuses(name)
|
||||
params = [
|
||||
{"key": key, "value": format_parameter(value)}
|
||||
for key, value in config.parameters.items()
|
||||
]
|
||||
return {
|
||||
"kind": "state",
|
||||
"key": f"state:{name}",
|
||||
"name": name,
|
||||
"label": self._label(name),
|
||||
"status": status,
|
||||
"statusText": BEAMLINE_STATE_STATUS_LABELS[status].capitalize(),
|
||||
"tone": STATUS_TONES[status],
|
||||
"icon": STATUS_ICONS[status],
|
||||
"device": state_device(config) or "",
|
||||
"stateType": config.state_type,
|
||||
"params": params,
|
||||
"watched": watched,
|
||||
"tripped": self._is_tripped(name),
|
||||
"armed": watched and self._interlock_enabled,
|
||||
"acceptedText": " or ".join(s.upper() for s in accepted),
|
||||
"tripOnWarning": "warning" not in accepted,
|
||||
"expanded": name in self._expanded,
|
||||
"title": "",
|
||||
"count": 0,
|
||||
}
|
||||
|
||||
def view_state(self) -> dict:
|
||||
"""Display-ready state shared by both renderers."""
|
||||
rank = {status: index for index, status in enumerate(STATUS_ORDER)}
|
||||
ordered = sorted(self._order, key=lambda name: rank[self._status(name)])
|
||||
visible, hidden = [], []
|
||||
for name in ordered:
|
||||
# States watched by the scan interlock are never filtered away.
|
||||
if name in self._interlock_states or self._matches_filters(name):
|
||||
visible.append(name)
|
||||
else:
|
||||
hidden.append(name)
|
||||
shown = set(visible) | (set(hidden) if self._show_hidden else set())
|
||||
|
||||
rows: list[dict] = []
|
||||
sections = (
|
||||
("interlock", "Scan interlock", [n for n in ordered if n in self._interlock_states]),
|
||||
(
|
||||
"others",
|
||||
"Not in scan interlock",
|
||||
[n for n in ordered if n not in self._interlock_states],
|
||||
),
|
||||
)
|
||||
# Like the original manager, headers only appear once some state is in the interlock.
|
||||
with_headers = any(name in shown for name in sections[0][2])
|
||||
for key, title, names in sections:
|
||||
names = [name for name in names if name in shown]
|
||||
if not names:
|
||||
continue
|
||||
if not with_headers:
|
||||
rows.extend(self._row(name) for name in names)
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"kind": "header",
|
||||
"key": f"header:{key}",
|
||||
"name": key,
|
||||
"title": title,
|
||||
"count": len(names),
|
||||
"armed": key == "interlock" and self._interlock_enabled,
|
||||
"expanded": False,
|
||||
"params": [],
|
||||
}
|
||||
)
|
||||
rows.extend(self._row(name) for name in names)
|
||||
|
||||
counts = {status: 0 for status in STATUS_ORDER}
|
||||
for name in self._order:
|
||||
counts[self._status(name)] += 1
|
||||
chips = [
|
||||
{
|
||||
"status": status,
|
||||
"text": f"{counts[status]} {STATUS_WORDS[status]}",
|
||||
"tone": STATUS_TONES[status],
|
||||
"active": self._selected_statuses is not None and status in self._selected_statuses,
|
||||
}
|
||||
for status in STATUS_ORDER
|
||||
if counts[status] or (self._selected_statuses and status in self._selected_statuses)
|
||||
]
|
||||
tripped = [name for name in ordered if self._is_tripped(name)]
|
||||
if tripped:
|
||||
banner = (
|
||||
"Scans are blocked: "
|
||||
+ ", ".join(tripped)
|
||||
+ (" is" if len(tripped) == 1 else " are")
|
||||
+ " outside the accepted status."
|
||||
)
|
||||
else:
|
||||
banner = ""
|
||||
if self._interlock_enabled:
|
||||
interlock_text = "Armed" if not tripped else "Tripped"
|
||||
else:
|
||||
interlock_text = "Off"
|
||||
hidden_count = len(hidden)
|
||||
noun = "state" if hidden_count == 1 else "states"
|
||||
return {
|
||||
"empty": not self._order,
|
||||
"rows": rows,
|
||||
"chips": chips,
|
||||
"total": len(self._order),
|
||||
"filterText": self._filter_text,
|
||||
"filtersActive": self._filters_active(),
|
||||
"noMatches": bool(self._order) and not shown,
|
||||
"hiddenCount": hidden_count,
|
||||
"hiddenText": (f"{hidden_count} {noun} hidden by filters" if hidden_count else ""),
|
||||
"showHidden": self._show_hidden,
|
||||
"interlockEnabled": self._interlock_enabled,
|
||||
"interlockText": interlock_text,
|
||||
"interlockTone": (
|
||||
("danger" if tripped else "success") if self._interlock_enabled else "muted"
|
||||
),
|
||||
"interlockCount": len(self._interlock_states),
|
||||
"banner": banner,
|
||||
}
|
||||
|
||||
# ---- teardown ------------------------------------------------------------------------
|
||||
|
||||
def cleanup(self) -> None:
|
||||
if self._enabled_config is not None:
|
||||
self._enabled_config.unsubscribe(self._on_interlock_enabled_stream)
|
||||
self._enabled_config = None
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.update_available_states, MessageEndpoints.available_beamline_states()
|
||||
)
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self._refresh_scan_interlock,
|
||||
MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"),
|
||||
)
|
||||
for name in sorted(self._subscribed):
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.update_state, MessageEndpoints.beamline_state(name)
|
||||
)
|
||||
self._subscribed.clear()
|
||||
super().cleanup()
|
||||
|
||||
|
||||
BeamlineStatesPortBase.PORTED_FROM = BeamlineStatesPortBase
|
||||
@@ -0,0 +1,486 @@
|
||||
import QtQuick
|
||||
import QtQuick.Controls.Basic
|
||||
import QtQuick.Layouts
|
||||
import BecUi
|
||||
|
||||
// Beamline state manager: status counts as filters, search, scan-interlock switch and a
|
||||
// list of states grouped by scan-interlock membership.
|
||||
// backend: BeamlineStatesBackend (state map, rows model, 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
|
||||
}
|
||||
function soft(c, a) { return Qt.rgba(c.r, c.g, c.b, a) }
|
||||
|
||||
color: theme.bg
|
||||
implicitWidth: 420
|
||||
implicitHeight: 480
|
||||
|
||||
Card {
|
||||
id: card
|
||||
anchors.fill: parent
|
||||
anchors.margins: 8
|
||||
spacing: 8
|
||||
fillContent: true
|
||||
title: "Beamline states"
|
||||
subtitle: root.s.total > 0 ? root.s.total + (root.s.total === 1 ? " state" : " states") : ""
|
||||
|
||||
// ---- scan interlock --------------------------------------------------------------
|
||||
Rectangle {
|
||||
Layout.fillWidth: true
|
||||
implicitHeight: interlockRow.implicitHeight + 12
|
||||
radius: 8
|
||||
color: root.s.interlockEnabled ? root.soft(root.toneColor(root.s.interlockTone), 0.10) : theme.field
|
||||
border.color: root.s.interlockEnabled ? root.soft(root.toneColor(root.s.interlockTone), 0.45) : theme.border
|
||||
Behavior on color { ColorAnimation { duration: 160 } }
|
||||
RowLayout {
|
||||
id: interlockRow
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 10
|
||||
anchors.rightMargin: 6
|
||||
spacing: 8
|
||||
Icon {
|
||||
name: root.s.interlockEnabled ? "lock" : "no_encryption"
|
||||
filled: true
|
||||
color: root.toneColor(root.s.interlockTone)
|
||||
}
|
||||
ColumnLayout {
|
||||
spacing: 0
|
||||
Layout.fillWidth: true
|
||||
Text {
|
||||
text: "Scan interlock"
|
||||
color: theme.fg
|
||||
font.pixelSize: 13
|
||||
font.weight: Font.DemiBold
|
||||
}
|
||||
Text {
|
||||
text: root.s.interlockCount === 0 ? "No states watched"
|
||||
: root.s.interlockCount + (root.s.interlockCount === 1 ? " state watched" : " states watched")
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 11
|
||||
}
|
||||
}
|
||||
StatusPill {
|
||||
text: root.s.interlockText || ""
|
||||
tone: root.toneColor(root.s.interlockTone)
|
||||
pulse: root.s.banner !== ""
|
||||
}
|
||||
SwitchField {
|
||||
checked: root.s.interlockEnabled === true
|
||||
onToggled: root.backend.setInterlockEnabled(checked)
|
||||
Accessible.name: "Arm scan interlock"
|
||||
ToolTip.visible: hovered
|
||||
ToolTip.delay: 500
|
||||
ToolTip.text: checked ? "Disarm the scan interlock" : "Arm the scan interlock"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- tripped banner ----------------------------------------------------------------
|
||||
Rectangle {
|
||||
visible: root.s.banner !== ""
|
||||
Layout.fillWidth: true
|
||||
implicitHeight: bannerText.implicitHeight + 14
|
||||
radius: 8
|
||||
color: root.soft(theme.danger, 0.14)
|
||||
border.color: root.soft(theme.danger, 0.5)
|
||||
RowLayout {
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 10
|
||||
anchors.rightMargin: 10
|
||||
spacing: 8
|
||||
Icon { name: "block"; color: theme.danger; size: 16 }
|
||||
Text {
|
||||
id: bannerText
|
||||
Layout.fillWidth: true
|
||||
text: root.s.banner || ""
|
||||
color: theme.fg
|
||||
font.pixelSize: 12
|
||||
wrapMode: Text.WordWrap
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- search, filters and add -------------------------------------------------------
|
||||
RowLayout {
|
||||
visible: root.s.empty !== true
|
||||
Layout.fillWidth: true
|
||||
spacing: 6
|
||||
InputField {
|
||||
id: search
|
||||
Layout.fillWidth: true
|
||||
leftPadding: 30
|
||||
placeholderText: "Filter by name or device"
|
||||
onTextEdited: root.backend.setFilterText(text)
|
||||
Accessible.name: "Filter states"
|
||||
Icon {
|
||||
name: "search"
|
||||
size: 16
|
||||
color: theme.fgSubtle
|
||||
anchors.left: parent.left
|
||||
anchors.leftMargin: 9
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
}
|
||||
Connections {
|
||||
target: root.backend
|
||||
function onChanged() {
|
||||
if (!search.activeFocus && search.text !== root.s.filterText)
|
||||
search.text = root.s.filterText || ""
|
||||
}
|
||||
}
|
||||
}
|
||||
IconButton {
|
||||
visible: root.s.filtersActive === true
|
||||
iconName: "filter_alt_off"
|
||||
tip: "Clear filters"
|
||||
onClicked: root.backend.clearFilters()
|
||||
}
|
||||
TextButton {
|
||||
text: "Add"
|
||||
iconName: "add"
|
||||
variant: "primary"
|
||||
onClicked: root.backend.addState()
|
||||
}
|
||||
}
|
||||
|
||||
// ---- status chips (quick filters) -------------------------------------------------
|
||||
Flow {
|
||||
visible: root.s.empty !== true
|
||||
Layout.fillWidth: true
|
||||
spacing: 6
|
||||
Repeater {
|
||||
model: root.s.chips || []
|
||||
delegate: AbstractButton {
|
||||
id: chip
|
||||
required property var modelData
|
||||
readonly property color tone: root.toneColor(modelData.tone)
|
||||
hoverEnabled: true
|
||||
implicitHeight: 24
|
||||
implicitWidth: chipLabel.implicitWidth + 28
|
||||
Accessible.name: "Show only " + modelData.status + " states"
|
||||
background: Rectangle {
|
||||
radius: height / 2
|
||||
color: chip.modelData.active ? root.soft(chip.tone, 0.28)
|
||||
: chip.hovered ? root.soft(chip.tone, 0.16) : "transparent"
|
||||
border.color: chip.modelData.active ? chip.tone : root.soft(chip.tone, 0.45)
|
||||
Behavior on color { ColorAnimation { duration: 100 } }
|
||||
}
|
||||
contentItem: Item {
|
||||
Rectangle {
|
||||
width: 7; height: 7; radius: 3.5
|
||||
color: chip.tone
|
||||
anchors.left: parent.left
|
||||
anchors.leftMargin: 10
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
}
|
||||
Text {
|
||||
id: chipLabel
|
||||
x: 22
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
text: chip.modelData.text
|
||||
color: chip.modelData.active ? theme.fg : theme.fgMuted
|
||||
font.pixelSize: 12
|
||||
font.weight: chip.modelData.active ? Font.DemiBold : Font.Normal
|
||||
}
|
||||
}
|
||||
onClicked: root.backend.toggleStatus(modelData.status)
|
||||
ToolTip.visible: hovered
|
||||
ToolTip.delay: 600
|
||||
ToolTip.text: modelData.active ? "Click to stop filtering by this status"
|
||||
: "Click to show only this status"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- list of states ----------------------------------------------------------------
|
||||
ListView {
|
||||
id: list
|
||||
visible: root.s.empty !== true && root.s.noMatches !== true
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: true
|
||||
clip: true
|
||||
spacing: 4
|
||||
boundsBehavior: Flickable.StopAtBounds
|
||||
model: root.backend.rows
|
||||
ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded }
|
||||
|
||||
delegate: Item {
|
||||
id: row
|
||||
required property var model
|
||||
readonly property bool isHeader: model.kind === "header"
|
||||
readonly property color tone: root.toneColor(model.tone)
|
||||
width: ListView.view.width - 2
|
||||
height: isHeader ? 26 : stateCard.height
|
||||
|
||||
// section header
|
||||
RowLayout {
|
||||
visible: row.isHeader
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 4
|
||||
spacing: 6
|
||||
Icon {
|
||||
name: row.model.name === "interlock" ? "lock" : "lock_open_right"
|
||||
filled: row.model.armed === true
|
||||
size: 14
|
||||
color: row.model.armed === true ? theme.fg : theme.fgMuted
|
||||
}
|
||||
Text {
|
||||
text: (row.model.title || "") + " · " + row.model.count
|
||||
color: row.model.armed === true ? theme.fg : theme.fgMuted
|
||||
font.pixelSize: 11
|
||||
font.weight: Font.Bold
|
||||
}
|
||||
Rectangle { Layout.fillWidth: true; height: 1; color: theme.border }
|
||||
}
|
||||
|
||||
// state card
|
||||
Rectangle {
|
||||
id: stateCard
|
||||
visible: !row.isHeader
|
||||
width: parent.width
|
||||
height: column.implicitHeight
|
||||
radius: 8
|
||||
property real flash: 0
|
||||
color: row.model.tripped === true ? root.soft(theme.danger, 0.08 + 0.10 * flash)
|
||||
: row.model.expanded === true ? theme.field
|
||||
: headerArea.containsMouse ? theme.hover : "transparent"
|
||||
border.width: row.model.tripped === true ? 2 : 1
|
||||
border.color: row.model.tripped === true ? root.soft(theme.danger, 0.5 + 0.5 * flash)
|
||||
: row.model.expanded === true ? theme.border : "transparent"
|
||||
Behavior on color { enabled: row.model.tripped !== true; ColorAnimation { duration: 120 } }
|
||||
SequentialAnimation on flash {
|
||||
running: row.model.tripped === true
|
||||
loops: Animation.Infinite
|
||||
NumberAnimation { from: 0; to: 1; duration: 700; easing.type: Easing.InOutSine }
|
||||
NumberAnimation { from: 1; to: 0; duration: 700; easing.type: Easing.InOutSine }
|
||||
}
|
||||
|
||||
ColumnLayout {
|
||||
id: column
|
||||
width: parent.width
|
||||
spacing: 0
|
||||
|
||||
Item {
|
||||
Layout.fillWidth: true
|
||||
implicitHeight: 52
|
||||
MouseArea {
|
||||
id: headerArea
|
||||
anchors.fill: parent
|
||||
hoverEnabled: true
|
||||
cursorShape: Qt.PointingHandCursor
|
||||
onClicked: root.backend.toggleExpanded(row.model.name)
|
||||
}
|
||||
RowLayout {
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 10
|
||||
anchors.rightMargin: 4
|
||||
spacing: 10
|
||||
Rectangle {
|
||||
width: 30; height: 30; radius: 15
|
||||
color: root.soft(row.tone, 0.18)
|
||||
Icon {
|
||||
anchors.centerIn: parent
|
||||
name: row.model.icon || "help"
|
||||
filled: true
|
||||
color: row.tone
|
||||
}
|
||||
}
|
||||
ColumnLayout {
|
||||
Layout.fillWidth: true
|
||||
spacing: 1
|
||||
Text {
|
||||
Layout.fillWidth: true
|
||||
text: row.model.name || ""
|
||||
color: theme.fg
|
||||
font.pixelSize: 13
|
||||
font.weight: Font.DemiBold
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
Text {
|
||||
Layout.fillWidth: true
|
||||
text: row.model.label || ""
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 11
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
}
|
||||
StatusPill {
|
||||
text: row.model.statusText || ""
|
||||
tone: row.tone
|
||||
}
|
||||
IconButton {
|
||||
iconName: row.model.watched === true ? "lock" : "lock_open_right"
|
||||
iconColor: row.model.tripped === true ? theme.danger
|
||||
: row.model.watched === true ? theme.fg : theme.fgSubtle
|
||||
tip: row.model.watched === true
|
||||
? "Watched by the scan interlock (accepts " + row.model.acceptedText + "). Click to stop watching."
|
||||
: "Not watched by the scan interlock. Click to watch this state."
|
||||
onClicked: root.backend.toggleInterlockFor(row.model.name)
|
||||
}
|
||||
Icon {
|
||||
name: row.model.expanded === true ? "expand_less" : "expand_more"
|
||||
color: theme.fgMuted
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// details
|
||||
ColumnLayout {
|
||||
visible: row.model.expanded === true
|
||||
Layout.fillWidth: true
|
||||
Layout.leftMargin: 50
|
||||
Layout.rightMargin: 12
|
||||
Layout.bottomMargin: 12
|
||||
spacing: 8
|
||||
GridLayout {
|
||||
columns: 2
|
||||
columnSpacing: 14
|
||||
rowSpacing: 3
|
||||
Text { text: "type"; color: theme.fgSubtle; font.pixelSize: 12 }
|
||||
Text { text: row.model.stateType || ""; color: theme.fg; font.pixelSize: 12 }
|
||||
Repeater {
|
||||
model: row.model.params || []
|
||||
delegate: Text {
|
||||
required property var modelData
|
||||
required property int index
|
||||
text: modelData.key
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 12
|
||||
Layout.row: index + 1
|
||||
Layout.column: 0
|
||||
}
|
||||
}
|
||||
Repeater {
|
||||
model: row.model.params || []
|
||||
delegate: Text {
|
||||
required property var modelData
|
||||
required property int index
|
||||
text: modelData.value
|
||||
color: theme.fg
|
||||
font.pixelSize: 12
|
||||
font.family: "monospace"
|
||||
Layout.row: index + 1
|
||||
Layout.column: 1
|
||||
}
|
||||
}
|
||||
}
|
||||
SwitchField {
|
||||
text: "WARNING trips the scan interlock"
|
||||
checked: row.model.tripOnWarning === true
|
||||
onToggled: root.backend.setTripOnWarning(row.model.name, checked)
|
||||
}
|
||||
RowLayout {
|
||||
spacing: 6
|
||||
TextButton {
|
||||
text: "Edit…"
|
||||
iconName: "edit"
|
||||
onClicked: root.backend.editState(row.model.name)
|
||||
}
|
||||
TextButton {
|
||||
text: "Remove"
|
||||
iconName: "delete"
|
||||
variant: "ghost"
|
||||
onClicked: root.backend.removeState(row.model.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- no matches -------------------------------------------------------------------
|
||||
ColumnLayout {
|
||||
visible: root.s.noMatches === true
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: true
|
||||
spacing: 8
|
||||
Item { Layout.fillHeight: true }
|
||||
Text {
|
||||
Layout.alignment: Qt.AlignHCenter
|
||||
text: "No states match the filters."
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 13
|
||||
}
|
||||
TextButton {
|
||||
Layout.alignment: Qt.AlignHCenter
|
||||
text: "Clear filters"
|
||||
iconName: "filter_alt_off"
|
||||
onClicked: root.backend.clearFilters()
|
||||
}
|
||||
Item { Layout.fillHeight: true }
|
||||
}
|
||||
|
||||
// ---- hidden by filters ----------------------------------------------------------------
|
||||
AbstractButton {
|
||||
id: hiddenToggle
|
||||
visible: root.s.hiddenCount > 0 && root.s.noMatches !== true
|
||||
Layout.fillWidth: true
|
||||
implicitHeight: 28
|
||||
hoverEnabled: true
|
||||
background: Rectangle {
|
||||
radius: 6
|
||||
color: hiddenToggle.hovered ? theme.hover : "transparent"
|
||||
}
|
||||
contentItem: RowLayout {
|
||||
spacing: 6
|
||||
Icon {
|
||||
name: root.s.showHidden ? "visibility_off" : "visibility"
|
||||
size: 16
|
||||
color: theme.fgMuted
|
||||
Layout.leftMargin: 6
|
||||
}
|
||||
Text {
|
||||
Layout.fillWidth: true
|
||||
text: (root.s.hiddenText || "") + (root.s.showHidden ? " · Hide them" : " · Show them")
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 12
|
||||
}
|
||||
}
|
||||
onClicked: root.backend.setShowHidden(!root.s.showHidden)
|
||||
}
|
||||
|
||||
// ---- empty state ----------------------------------------------------------------------
|
||||
ColumnLayout {
|
||||
visible: root.s.empty === true
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: true
|
||||
spacing: 8
|
||||
Item { Layout.fillHeight: true }
|
||||
Icon {
|
||||
Layout.alignment: Qt.AlignHCenter
|
||||
name: "rule"
|
||||
size: 36
|
||||
color: theme.fgSubtle
|
||||
}
|
||||
Text {
|
||||
Layout.alignment: Qt.AlignHCenter
|
||||
text: "No beamline states yet"
|
||||
color: theme.fg
|
||||
font.pixelSize: 14
|
||||
font.weight: Font.DemiBold
|
||||
}
|
||||
Text {
|
||||
Layout.fillWidth: true
|
||||
horizontalAlignment: Text.AlignHCenter
|
||||
wrapMode: Text.WordWrap
|
||||
text: "A state watches a device, for example whether a motor is within limits, and can block scans through the scan interlock."
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 12
|
||||
}
|
||||
TextButton {
|
||||
Layout.alignment: Qt.AlignHCenter
|
||||
text: "Add state"
|
||||
iconName: "add"
|
||||
variant: "primary"
|
||||
onClicked: root.backend.addState()
|
||||
}
|
||||
Item { Layout.fillHeight: true }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,363 @@
|
||||
"""Tests for the QML and QWidget ports of the beamline state manager."""
|
||||
|
||||
# pylint: disable=redefined-outer-name, protected-access, missing-function-docstring
|
||||
import pytest
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from qtpy.QtQuickWidgets import QQuickWidget
|
||||
from qtpy.QtWidgets import QDialog, QMessageBox
|
||||
|
||||
from bec_widgets.widgets.services.beamline_states import beamline_states_ux_common as common
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_states_qml import BeamlineStatesQML
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_states_qwidget import (
|
||||
BeamlineStatesQWidget,
|
||||
)
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_states_ux_common import (
|
||||
EditBeamlineStateDialog,
|
||||
format_parameter,
|
||||
state_device,
|
||||
)
|
||||
|
||||
from .conftest import create_widget
|
||||
from .test_beamline_state_pill import _FakeScanInterlock, _limits_state
|
||||
|
||||
|
||||
def _message(name: str, status: str, label: str = "") -> dict:
|
||||
return {"name": name, "status": status, "label": label or f"{name} is {status}"}
|
||||
|
||||
|
||||
@pytest.fixture(params=[BeamlineStatesQML, BeamlineStatesQWidget])
|
||||
def manager(request, qtbot, mocked_client):
|
||||
widget = create_widget(qtbot, request.param, client=mocked_client)
|
||||
widget._scan_interlock = _FakeScanInterlock()
|
||||
widget._refresh_scan_interlock()
|
||||
return widget
|
||||
|
||||
|
||||
def _load(manager, *names_devices):
|
||||
states = [_limits_state(name=name, device=device) for name, device in names_devices]
|
||||
manager.update_available_states({"states": states}, {})
|
||||
|
||||
|
||||
def _state_rows(manager) -> list[dict]:
|
||||
return [row for row in manager.view_state()["rows"] if row["kind"] == "state"]
|
||||
|
||||
|
||||
def test_helpers():
|
||||
assert format_parameter(0.1) == "0.1"
|
||||
assert format_parameter(10.0) == "10"
|
||||
assert format_parameter(None) == "—"
|
||||
assert state_device(_limits_state(device="samy")) == "samy"
|
||||
assert state_device(None) is None
|
||||
|
||||
|
||||
def test_empty_state(manager):
|
||||
state = manager.view_state()
|
||||
assert state["empty"]
|
||||
assert state["rows"] == []
|
||||
assert state["interlockText"] == "Off"
|
||||
|
||||
|
||||
def test_adds_and_removes_states_and_subscriptions(manager, mocked_client):
|
||||
_load(manager, ("limits", "samx"), ("samy_limits", "samy"))
|
||||
assert manager._subscribed == {"limits", "samy_limits"}
|
||||
assert [row["name"] for row in _state_rows(manager)] == ["limits", "samy_limits"]
|
||||
|
||||
manager.update_state(_message("limits", "valid", "Within limits."))
|
||||
summary = manager.state_summary()
|
||||
assert summary["limits"] == {"status": "valid", "label": "Within limits."}
|
||||
assert summary["samy_limits"]["status"] == "unknown"
|
||||
|
||||
_load(manager, ("limits", "samx"))
|
||||
assert manager._subscribed == {"limits"}
|
||||
assert list(manager.state_summary()) == ["limits"]
|
||||
|
||||
|
||||
def test_reads_cached_state_on_add(qtbot, mocked_client):
|
||||
from bec_lib import messages
|
||||
|
||||
mocked_client.connector.xadd(
|
||||
MessageEndpoints.beamline_state("cached_limits"),
|
||||
{
|
||||
"data": messages.BeamlineStateMessage(
|
||||
name="cached_limits", status="invalid", label="Out of range"
|
||||
)
|
||||
},
|
||||
)
|
||||
widget = create_widget(qtbot, BeamlineStatesQWidget, client=mocked_client)
|
||||
widget._scan_interlock = _FakeScanInterlock()
|
||||
_load(widget, ("cached_limits", "samx"))
|
||||
assert widget.state_summary()["cached_limits"] == {"status": "invalid", "label": "Out of range"}
|
||||
|
||||
|
||||
def test_ignores_messages_for_unknown_states(manager):
|
||||
_load(manager, ("limits", "samx"))
|
||||
manager.update_state(_message("other", "invalid"))
|
||||
assert "other" not in manager.state_summary()
|
||||
assert manager.state_summary()["limits"]["status"] == "unknown"
|
||||
|
||||
|
||||
def test_orders_by_severity_and_groups_by_interlock(manager):
|
||||
_load(manager, ("a", "samx"), ("b", "samy"), ("c", "samz"), ("d", "bpm4i"))
|
||||
for name, status in (("a", "valid"), ("b", "warning"), ("c", "invalid"), ("d", "valid")):
|
||||
manager.update_state(_message(name, status))
|
||||
manager._scan_interlock = _FakeScanInterlock(states_watched={"d": "valid", "b": "valid"})
|
||||
manager._refresh_scan_interlock()
|
||||
|
||||
rows = manager.view_state()["rows"]
|
||||
assert [(row["kind"], row["name"]) for row in rows] == [
|
||||
("header", "interlock"),
|
||||
("state", "b"),
|
||||
("state", "d"),
|
||||
("header", "others"),
|
||||
("state", "c"),
|
||||
("state", "a"),
|
||||
]
|
||||
assert rows[0]["count"] == 2 and rows[3]["count"] == 2
|
||||
|
||||
|
||||
def test_status_chips_filter_and_watched_states_bypass(manager):
|
||||
_load(manager, ("a", "samx"), ("b", "samy"), ("c", "samz"))
|
||||
manager.update_state(_message("a", "valid"))
|
||||
manager.update_state(_message("b", "invalid"))
|
||||
manager.update_state(_message("c", "valid"))
|
||||
manager._scan_interlock = _FakeScanInterlock(states_watched={"c": "valid"})
|
||||
manager._refresh_scan_interlock()
|
||||
|
||||
chips = {chip["status"]: chip for chip in manager.view_state()["chips"]}
|
||||
assert chips["valid"]["text"] == "2 valid"
|
||||
assert chips["invalid"]["text"] == "1 invalid"
|
||||
|
||||
manager.toggle_status_filter("invalid")
|
||||
state = manager.view_state()
|
||||
assert [row["name"] for row in _state_rows(manager)] == ["c", "b"]
|
||||
assert state["hiddenCount"] == 1
|
||||
assert state["filtersActive"]
|
||||
assert {chip["status"]: chip["active"] for chip in state["chips"]}["invalid"]
|
||||
|
||||
manager.set_show_hidden(True)
|
||||
assert {row["name"] for row in _state_rows(manager)} == {"a", "b", "c"}
|
||||
|
||||
manager.toggle_status_filter("invalid")
|
||||
assert manager._selected_statuses is None
|
||||
|
||||
|
||||
def test_text_filter_matches_name_and_device(manager):
|
||||
_load(manager, ("limits", "samx"), ("temp", "bpm4i"))
|
||||
manager.set_filter_text("bpm")
|
||||
assert [row["name"] for row in _state_rows(manager)] == ["temp"]
|
||||
manager.set_filter_text("lim, temp")
|
||||
assert len(_state_rows(manager)) == 2
|
||||
manager.set_filter_text("nothing")
|
||||
assert manager.view_state()["noMatches"]
|
||||
manager.clear_filters()
|
||||
assert manager.view_state()["filterText"] == ""
|
||||
assert len(_state_rows(manager)) == 2
|
||||
|
||||
|
||||
def test_toggle_interlock_for_state(manager):
|
||||
fake = _FakeScanInterlock()
|
||||
manager._scan_interlock = fake
|
||||
_load(manager, ("limits", "samx"))
|
||||
|
||||
manager.toggle_interlock_for("limits")
|
||||
assert fake.added == [("limits", ["valid", "warning"])]
|
||||
assert _state_rows(manager)[0]["watched"]
|
||||
|
||||
manager.toggle_interlock_for("limits")
|
||||
assert fake.removed == ["limits"]
|
||||
assert not _state_rows(manager)[0]["watched"]
|
||||
|
||||
|
||||
def test_trip_on_warning(manager):
|
||||
fake = _FakeScanInterlock()
|
||||
manager._scan_interlock = fake
|
||||
_load(manager, ("limits", "samx"))
|
||||
|
||||
# not watched yet: kept as preference for when it is added
|
||||
manager.set_trip_on_warning("limits", True)
|
||||
assert fake.added == []
|
||||
assert _state_rows(manager)[0]["tripOnWarning"]
|
||||
manager.toggle_interlock_for("limits")
|
||||
assert fake.added == [("limits", ["valid"])]
|
||||
|
||||
# watched: written to the interlock right away
|
||||
manager.set_trip_on_warning("limits", False)
|
||||
assert fake.added[-1] == ("limits", ["valid", "warning"])
|
||||
assert _state_rows(manager)[0]["acceptedText"] == "VALID or WARNING"
|
||||
|
||||
|
||||
def test_tripped_state_and_banner(manager):
|
||||
manager._scan_interlock = _FakeScanInterlock(enabled=True, states_watched={"limits": "valid"})
|
||||
_load(manager, ("limits", "samx"))
|
||||
manager._refresh_scan_interlock()
|
||||
manager.update_state(_message("limits", "valid"))
|
||||
state = manager.view_state()
|
||||
assert state["interlockText"] == "Armed" and state["banner"] == ""
|
||||
|
||||
manager.update_state(_message("limits", "warning"))
|
||||
state = manager.view_state()
|
||||
assert state["interlockText"] == "Tripped"
|
||||
assert state["interlockTone"] == "danger"
|
||||
assert "limits" in state["banner"]
|
||||
assert _state_rows(manager)[0]["tripped"]
|
||||
|
||||
manager._on_interlock_enabled_changed(False)
|
||||
assert not _state_rows(manager)[0]["tripped"]
|
||||
assert manager.view_state()["banner"] == ""
|
||||
|
||||
|
||||
def test_set_interlock_enabled_writes_backend(manager):
|
||||
fake = _FakeScanInterlock()
|
||||
manager._scan_interlock = fake
|
||||
manager.set_interlock_enabled(True)
|
||||
assert fake.enabled is True
|
||||
assert manager.view_state()["interlockEnabled"]
|
||||
|
||||
|
||||
def test_interlock_failure_is_reported(manager, monkeypatch):
|
||||
class Broken(_FakeScanInterlock):
|
||||
def add_state_to_interlock(self, state_name, required_value="valid"):
|
||||
raise RuntimeError("no actor")
|
||||
|
||||
warnings = []
|
||||
monkeypatch.setattr(QMessageBox, "warning", lambda *args: warnings.append(args[-1]))
|
||||
manager._scan_interlock = Broken()
|
||||
_load(manager, ("limits", "samx"))
|
||||
manager.toggle_interlock_for("limits")
|
||||
assert warnings == ["no actor"]
|
||||
|
||||
|
||||
def test_expand_and_collapse_all(manager):
|
||||
_load(manager, ("a", "samx"), ("b", "samy"))
|
||||
manager.toggle_expanded("a")
|
||||
manager.set_expanded("b", True)
|
||||
assert all(row["expanded"] for row in _state_rows(manager))
|
||||
params = {p["key"]: p["value"] for p in _state_rows(manager)[0]["params"]}
|
||||
assert params["device"] == "samx" and params["high_limit"] == "10"
|
||||
manager.collapse_all()
|
||||
assert not any(row["expanded"] for row in _state_rows(manager))
|
||||
|
||||
|
||||
def test_remove_state_confirms(manager, mocked_client, monkeypatch):
|
||||
class StateManager:
|
||||
deleted = None
|
||||
|
||||
def delete(self, name):
|
||||
self.deleted = name
|
||||
|
||||
mocked_client.beamline_states = StateManager()
|
||||
monkeypatch.setattr(QMessageBox, "question", lambda *a, **k: QMessageBox.StandardButton.No)
|
||||
manager.remove_state("limits")
|
||||
assert mocked_client.beamline_states.deleted is None
|
||||
monkeypatch.setattr(QMessageBox, "question", lambda *a, **k: QMessageBox.StandardButton.Yes)
|
||||
manager.remove_state("limits")
|
||||
assert mocked_client.beamline_states.deleted == "limits"
|
||||
|
||||
|
||||
def test_edit_state_sends_new_parameters(manager, mocked_client, monkeypatch):
|
||||
class StateClient:
|
||||
parameters = None
|
||||
|
||||
def update_parameters(self, **kwargs):
|
||||
self.parameters = kwargs
|
||||
|
||||
class StateManager:
|
||||
limits = StateClient()
|
||||
|
||||
mocked_client.beamline_states = StateManager()
|
||||
_load(manager, ("limits", "samx"))
|
||||
|
||||
def fake_exec(dialog):
|
||||
dialog.form.input_widget("high_limit").setValue(20.0)
|
||||
dialog.accept()
|
||||
return QDialog.DialogCode.Accepted
|
||||
|
||||
monkeypatch.setattr(EditBeamlineStateDialog, "exec", fake_exec)
|
||||
manager.edit_state("limits")
|
||||
assert mocked_client.beamline_states.limits.parameters["high_limit"] == 20.0
|
||||
assert mocked_client.beamline_states.limits.parameters["device"] == "samx"
|
||||
|
||||
|
||||
def test_update_parameters_unknown_state(manager, mocked_client, monkeypatch):
|
||||
warnings = []
|
||||
monkeypatch.setattr(QMessageBox, "warning", lambda *args: warnings.append(args[-1]))
|
||||
mocked_client.beamline_states = object()
|
||||
assert not manager.update_state_parameters("limits", _limits_state())
|
||||
assert "not available" in warnings[0]
|
||||
|
||||
|
||||
def test_add_state_dialog_adds_to_interlock(manager, mocked_client, monkeypatch):
|
||||
added = []
|
||||
|
||||
class StateManager:
|
||||
def add(self, config):
|
||||
added.append(config)
|
||||
|
||||
class FakeDialog:
|
||||
config_result = _limits_state(name="new_state")
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def exec(self):
|
||||
return QDialog.DialogCode.Accepted
|
||||
|
||||
def add_to_interlock(self):
|
||||
return True
|
||||
|
||||
def interlock_statuses(self):
|
||||
return ["valid"]
|
||||
|
||||
def cleanup(self):
|
||||
pass
|
||||
|
||||
def deleteLater(self): # pylint: disable=invalid-name
|
||||
pass
|
||||
|
||||
fake = _FakeScanInterlock()
|
||||
manager._scan_interlock = fake
|
||||
mocked_client.beamline_states = StateManager()
|
||||
monkeypatch.setattr(common, "AddBeamlineStateDialog", FakeDialog)
|
||||
manager.open_add_state_dialog()
|
||||
assert added[0].name == "new_state"
|
||||
assert fake.added == [("new_state", ["valid"])]
|
||||
|
||||
|
||||
def test_renderer_follows_state(manager, qtbot):
|
||||
_load(manager, ("a", "samx"), ("b", "samy"))
|
||||
manager.update_state(_message("b", "invalid"))
|
||||
manager.set_expanded("a", True)
|
||||
if isinstance(manager, BeamlineStatesQML):
|
||||
assert isinstance(manager.view, QQuickWidget)
|
||||
assert manager.view.status() == QQuickWidget.Status.Ready
|
||||
rows = manager.backend._rows.items
|
||||
assert [row["name"] for row in rows] == ["b", "a"]
|
||||
assert manager.backend.state["total"] == 2
|
||||
else:
|
||||
rows = manager.state_rows
|
||||
assert set(rows) == {"a", "b"}
|
||||
assert rows["a"].details.isVisibleTo(manager)
|
||||
assert not rows["b"].details.isVisibleTo(manager)
|
||||
assert rows["b"].status_pill.text == "Invalid"
|
||||
first = rows["a"]
|
||||
manager.update_state(_message("a", "warning"))
|
||||
assert manager.state_rows["a"] is first
|
||||
|
||||
|
||||
def test_qwidget_tripped_row_animates(qtbot, mocked_client):
|
||||
widget = create_widget(qtbot, BeamlineStatesQWidget, client=mocked_client)
|
||||
widget._scan_interlock = _FakeScanInterlock(enabled=True, states_watched={"a": "valid"})
|
||||
_load(widget, ("a", "samx"))
|
||||
widget._refresh_scan_interlock()
|
||||
widget.update_state(_message("a", "invalid"))
|
||||
row = widget.state_rows["a"]
|
||||
assert row._flash_animation.state() == row._flash_animation.State.Running
|
||||
assert widget.banner.isVisibleTo(widget)
|
||||
widget.update_state(_message("a", "valid"))
|
||||
assert row._flash_animation.state() == row._flash_animation.State.Stopped
|
||||
|
||||
|
||||
def test_ports_are_not_rpc_or_designer_plugins():
|
||||
for cls in (BeamlineStatesQML, BeamlineStatesQWidget):
|
||||
assert cls.PLUGIN is False and cls.RPC is False
|
||||
assert cls.rpc_widget_class == "BeamlineStateManager"
|
||||
Reference in new issue
Block a user