From 78e20e699eb75013eac0934363fc64425470f021 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 20:57:51 +0000 Subject: [PATCH] feat(beamline_states): add QML and QWidget ports with status chips, search, interlock switch and tripped banner --- bec_widgets/utils/quick/qml/BecUi/Card.qml | 3 + .../beamline_states/beamline_states_qml.py | 142 ++++ .../beamline_states_qwidget.py | 713 ++++++++++++++++++ .../beamline_states_ux_common.py | 705 +++++++++++++++++ .../qml/BeamlineStatesView.qml | 486 ++++++++++++ .../unit_tests/test_beamline_states_ports.py | 363 +++++++++ 6 files changed, 2412 insertions(+) create mode 100644 bec_widgets/widgets/services/beamline_states/beamline_states_qml.py create mode 100644 bec_widgets/widgets/services/beamline_states/beamline_states_qwidget.py create mode 100644 bec_widgets/widgets/services/beamline_states/beamline_states_ux_common.py create mode 100644 bec_widgets/widgets/services/beamline_states/qml/BeamlineStatesView.qml create mode 100644 tests/unit_tests/test_beamline_states_ports.py diff --git a/bec_widgets/utils/quick/qml/BecUi/Card.qml b/bec_widgets/utils/quick/qml/BecUi/Card.qml index eadf5d56..2451958f 100644 --- a/bec_widgets/utils/quick/qml/BecUi/Card.qml +++ b/bec_widgets/utils/quick/qml/BecUi/Card.qml @@ -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 } } diff --git a/bec_widgets/widgets/services/beamline_states/beamline_states_qml.py b/bec_widgets/widgets/services/beamline_states/beamline_states_qml.py new file mode 100644 index 00000000..28b2dc62 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_states_qml.py @@ -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()) diff --git a/bec_widgets/widgets/services/beamline_states/beamline_states_qwidget.py b/bec_widgets/widgets/services/beamline_states/beamline_states_qwidget.py new file mode 100644 index 00000000..32afd3bf --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_states_qwidget.py @@ -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()) diff --git a/bec_widgets/widgets/services/beamline_states/beamline_states_ux_common.py b/bec_widgets/widgets/services/beamline_states/beamline_states_ux_common.py new file mode 100644 index 00000000..c0a437de --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_states_ux_common.py @@ -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 diff --git a/bec_widgets/widgets/services/beamline_states/qml/BeamlineStatesView.qml b/bec_widgets/widgets/services/beamline_states/qml/BeamlineStatesView.qml new file mode 100644 index 00000000..d7d0c783 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/qml/BeamlineStatesView.qml @@ -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 } + } + } +} diff --git a/tests/unit_tests/test_beamline_states_ports.py b/tests/unit_tests/test_beamline_states_ports.py new file mode 100644 index 00000000..3c9f43ce --- /dev/null +++ b/tests/unit_tests/test_beamline_states_ports.py @@ -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"