From cba33152e84f5e0639a18212c8efe88fe02e3c76 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 20:30:53 +0000 Subject: [PATCH] feat(beamline_states): add QML and QWidget prototypes with modernized UX for comparison --- bec_widgets/utils/qml_host.py | 152 ++++ .../beamline_states/modern/__init__.py | 0 .../beamline_states/modern/actions.py | 140 +++ .../modern/beamline_states_qml.py | 92 ++ .../modern/beamline_states_widget.py | 860 ++++++++++++++++++ .../beamline_states/modern/compare_demo.py | 46 + .../modern/qml/BeamlineStates.qml | 616 +++++++++++++ .../modern/states_controller.py | 497 ++++++++++ .../unit_tests/test_beamline_states_modern.py | 212 +++++ 9 files changed, 2615 insertions(+) create mode 100644 bec_widgets/utils/qml_host.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/__init__.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/actions.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/beamline_states_qml.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/beamline_states_widget.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/compare_demo.py create mode 100644 bec_widgets/widgets/services/beamline_states/modern/qml/BeamlineStates.qml create mode 100644 bec_widgets/widgets/services/beamline_states/modern/states_controller.py create mode 100644 tests/unit_tests/test_beamline_states_modern.py diff --git a/bec_widgets/utils/qml_host.py b/bec_widgets/utils/qml_host.py new file mode 100644 index 00000000..893e55f1 --- /dev/null +++ b/bec_widgets/utils/qml_host.py @@ -0,0 +1,152 @@ +""" +Helpers for hosting QML views inside BEC widgets. + +A QML view is embedded with ``QQuickWidget`` so it lives inside the usual ``BECWidget`` shell +(dock areas, RPC, cleanup). The view gets two shared helpers: + +* ``theme``: a :class:`QmlTheme` context property exposing the active bec_qthemes colors, and +* ``image://material/?color=&filled=<0|1>``: Material icons rendered by bec_qthemes. +""" + +from __future__ import annotations + +from pathlib import Path +from urllib.parse import parse_qs, urlsplit + +from bec_qthemes import material_icon +from qtpy.QtCore import Property, QObject, QSize, QUrl, Signal +from qtpy.QtGui import QColor, QImage +from qtpy.QtQuick import QQuickImageProvider +from qtpy.QtQuickWidgets import QQuickWidget +from qtpy.QtWidgets import QApplication, QWidget + +from bec_widgets.utils.colors import Colors + +_DARK_FALLBACK = { + "BG": "#1e1f22", + "CARD_BG": "#2b2c2f", + "FIELD_BG": "#26272a", + "FG": "#dddddd", + "BORDER": "#464749", + "PRIMARY": "#6089ef", + "ON_PRIMARY": "#ffffff", + "ACCENT_DEFAULT": "#40B6E0", + "ACCENT_WARNING": "#EDA200", + "ACCENT_EMERGENCY": "#E0534B", + "ACCENT_SUCCESS": "#59A869", +} + + +def _app_theme(): + app = QApplication.instance() + return getattr(app, "theme", None) if app is not None else None + + +class QmlTheme(QObject): + """Theme colors of the running BEC application, exposed to QML as ``theme``.""" + + changed = Signal() + + def __init__(self, parent: QObject | None = None) -> None: + super().__init__(parent) + self._colors: dict[str, QColor] = {} + self._dark = True + theme = _app_theme() + if theme is not None: + theme.theme_changed.connect(self.refresh) + self.refresh() + + def refresh(self, *_args) -> None: + """Re-read the colors from the application theme.""" + theme = _app_theme() + self._dark = theme is None or theme.theme != "light" + self._colors = { + key: theme.color(key, fallback) if theme is not None else QColor(fallback) + for key, fallback in _DARK_FALLBACK.items() + } + self.changed.emit() + + def get(self, key: str) -> QColor: + """Return a theme color by its bec_qthemes key.""" + return QColor(self._colors.get(key, QColor(_DARK_FALLBACK.get(key, "#000000")))) + + def _color(key: str, notify=changed): # pylint: disable=no-self-argument + return Property(QColor, lambda self: self.get(key), notify=notify) + + dark = Property(bool, lambda self: self._dark, notify=changed) + background = _color("BG") + card = _color("CARD_BG") + field = _color("FIELD_BG") + foreground = _color("FG") + border = _color("BORDER") + primary = _color("PRIMARY") + onPrimary = _color("ON_PRIMARY") # noqa: N815 + info = _color("ACCENT_DEFAULT") + success = _color("ACCENT_SUCCESS") + warning = _color("ACCENT_WARNING") + danger = _color("ACCENT_EMERGENCY") + + @Property(QColor, notify=changed) + def muted(self) -> QColor: + """Secondary text color, between card background and foreground.""" + return Colors._blend(self.get("CARD_BG"), self.get("FG"), 0.62) + + del _color + + +class MaterialIconProvider(QQuickImageProvider): + """Serves ``image://material/?color=&filled=<0|1>`` from bec_qthemes.""" + + def __init__(self) -> None: + super().__init__(QQuickImageProvider.ImageType.Image) + + def requestImage(self, icon_id: str, size: QSize, requested: QSize) -> QImage: # noqa: N802 + parts = urlsplit(icon_id) + query = parse_qs(parts.query) + color = query.get("color", [None])[0] + filled = query.get("filled", ["0"])[0] in ("1", "true") + width = requested.width() if requested.width() > 0 else 24 + height = requested.height() if requested.height() > 0 else width + try: + image = material_icon( + parts.path, size=(width, height), color=color, filled=filled + ).toImage() + except KeyError: + image = QImage(width, height, QImage.Format.Format_ARGB32) + image.fill(0) + if size is not None: + size.setWidth(image.width()) + size.setHeight(image.height()) + return image + + +def create_quick_widget( + parent: QWidget, qml_file: str | Path, context: dict[str, QObject] | None = None +) -> QQuickWidget: + """ + Create a ``QQuickWidget`` showing ``qml_file`` with the shared theme and icon helpers. + + Args: + parent: Widget that owns the view. + qml_file: Path to the root QML file. + context: Extra objects exposed to QML as context properties. + + Returns: + QQuickWidget: The loaded view. Its ``theme`` context property is a :class:`QmlTheme`. + """ + view = QQuickWidget(parent) + view.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView) + view.setClearColor(QColor(0, 0, 0, 0)) + engine = view.engine() + engine.addImageProvider("material", MaterialIconProvider()) + theme = QmlTheme(view) + view.setProperty("qml_theme", theme) + root_context = view.rootContext() + root_context.setContextProperty("theme", theme) + for name, obj in (context or {}).items(): + root_context.setContextProperty(name, obj) + view.setSource(QUrl.fromLocalFile(str(Path(qml_file).resolve()))) + if view.status() == QQuickWidget.Status.Error: + errors = "\n".join(error.toString() for error in view.errors()) + raise RuntimeError(f"Failed to load {qml_file}:\n{errors}") + return view diff --git a/bec_widgets/widgets/services/beamline_states/modern/__init__.py b/bec_widgets/widgets/services/beamline_states/modern/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/services/beamline_states/modern/actions.py b/bec_widgets/widgets/services/beamline_states/modern/actions.py new file mode 100644 index 00000000..530a9f6c --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/actions.py @@ -0,0 +1,140 @@ +"""Dialog-based actions shared by the QML and QWidget beamline-state views.""" + +from __future__ import annotations + +from bec_lib import bl_states +from qtpy.QtWidgets import ( + QDialog, + QDialogButtonBox, + QFormLayout, + QLabel, + QMessageBox, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.utils.forms_from_types.pydantic_widget_form import PydanticWidgetForm +from bec_widgets.widgets.services.beamline_states.dialogs import ( + SUPPORTED_BEAMLINE_STATES, + AddBeamlineStateDialog, +) +from bec_widgets.widgets.services.beamline_states.modern.states_controller import ( + BeamlineStatesController, +) + + +def config_class_for(state_type: str) -> type[bl_states.BeamlineStateConfig] | None: + """Return the 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): + """Edit the parameters of one beamline state with the generated pydantic form.""" + + def __init__( + self, + name: str, + config_class: type[bl_states.BeamlineStateConfig], + parameters: dict, + parent: QWidget | None = None, + client=None, + ) -> None: + super().__init__(parent) + self.setWindowTitle(f"Edit {name}") + self.setMinimumWidth(420) + self.result_config: bl_states.BeamlineStateConfig | None = None + self.form = PydanticWidgetForm( + config_class, parent=self, client=client, read_only_fields={"name"} + ) + data = {"name": name} + data.update({k: v for k, v in parameters.items() if k in config_class.model_fields}) + self.form.set_partial_data(data) + self.form.mark_clean() + + header = QFormLayout() + header.addRow("Type", QLabel(config_class.__name__, self)) + + self._buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Save | QDialogButtonBox.StandardButton.Cancel, self + ) + self._save = self._buttons.button(QDialogButtonBox.StandardButton.Save) + self._save.setEnabled(False) + self._buttons.accepted.connect(self._accept) + self._buttons.rejected.connect(self.reject) + self.form.changed.connect(self._on_changed) + + layout = QVBoxLayout(self) + layout.addLayout(header) + layout.addWidget(self.form) + layout.addWidget(self._buttons) + + def _on_changed(self, *_args) -> None: + self._save.setEnabled(bool(self.form.dirty_fields() - {"name"})) + + @SafeSlot() + def _accept(self) -> None: + try: + self.result_config = self.form.model_instance() + except ValueError as exc: + QMessageBox.warning(self, "Invalid Beamline State", str(exc)) + return + self.accept() + + def cleanup(self) -> None: + """Release the form's BEC widgets.""" + self.form.cleanup() + + +def open_add_dialog(parent: QWidget, controller: BeamlineStatesController) -> None: + """Show the add-state dialog and create the state (and interlock entry) in BEC.""" + dialog = AddBeamlineStateDialog(parent, client=controller.client) + try: + if dialog.exec() != QDialog.DialogCode.Accepted: + return + config = dialog.config_result + add_to_interlock = dialog.add_to_interlock() + accepted = dialog.interlock_statuses() + finally: + dialog.cleanup() + dialog.deleteLater() + if config is None: + return + try: + controller.client.beamline_states.add(config) + except Exception as exc: # pylint: disable=broad-except + QMessageBox.warning(parent, "Cannot Add State", str(exc)) + return + controller.set_preferred_accepted(config.name, accepted) + if add_to_interlock: + controller.toggleWatched(config.name) + + +def open_edit_dialog(parent: QWidget, controller: BeamlineStatesController, name: str) -> None: + """Show the edit dialog for ``name`` and push changed parameters to BEC.""" + config = controller.config(name) + config_class = config_class_for(config.state_type) if config is not None else None + if config_class is None: + QMessageBox.warning(parent, "Cannot Edit State", f"'{name}' has an unsupported type.") + return + dialog = EditBeamlineStateDialog( + name, config_class, dict(config.parameters), parent, client=controller.client + ) + try: + if dialog.exec() != QDialog.DialogCode.Accepted or dialog.result_config is None: + return + new_config = dialog.result_config + finally: + dialog.cleanup() + dialog.deleteLater() + state_client = getattr(controller.client.beamline_states, name, None) + if state_client is None: + QMessageBox.warning(parent, "Cannot Update State", f"'{name}' is not available.") + return + try: + state_client.update_parameters(**new_config.model_dump(exclude={"name"})) + except Exception as exc: # pylint: disable=broad-except + QMessageBox.warning(parent, "Cannot Update State", str(exc)) diff --git a/bec_widgets/widgets/services/beamline_states/modern/beamline_states_qml.py b/bec_widgets/widgets/services/beamline_states/modern/beamline_states_qml.py new file mode 100644 index 00000000..3ebf1e01 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/beamline_states_qml.py @@ -0,0 +1,92 @@ +"""Beamline states view rendered with QML, hosted in a ``QQuickWidget``.""" + +from __future__ import annotations + +from pathlib import Path + +from qtpy.QtCore import QMetaObject, QUrl +from qtpy.QtWidgets import 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.qml_host import create_quick_widget +from bec_widgets.widgets.services.beamline_states.modern.actions import ( + open_add_dialog, + open_edit_dialog, +) +from bec_widgets.widgets.services.beamline_states.modern.states_controller import ( + BeamlineStatesController, +) + +QML_FILE = Path(__file__).parent / "qml" / "BeamlineStates.qml" + + +class BeamlineStatesQML(BECWidget, QWidget): + """ + Beamline states with the scan interlock, rendered in QML. + + Prototype for comparison with :class:`BeamlineStatesWidget` (same UX in QWidgets) and the + original ``BeamlineStateManager``. + """ + + PLUGIN = False + RPC = False + ICON_NAME = "format_list_bulleted" + USER_ACCESS = ["clear_filters", "collapse_all", "state_summary", "remove", "attach", "detach"] + + 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.controller = BeamlineStatesController(self.client, self.bec_dispatcher, self) + self.controller.error_occurred.connect(self._show_error) + self.controller.add_requested.connect(self._open_add_dialog) + self.controller.edit_requested.connect(self._open_edit_dialog) + self.view = create_quick_widget( + self, QML_FILE, {"controller": self.controller, "statesModel": self.controller.model} + ) + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.addWidget(self.view) + + @SafeSlot() + def clear_filters(self) -> None: + """Reset the status chips and the search text.""" + self.controller.clearFilters() + + @SafeSlot() + def collapse_all(self) -> None: + """Collapse every expanded state.""" + QMetaObject.invokeMethod(self.view.rootObject(), "collapseAll") + + 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 self.controller.state_summary() + + @SafeSlot(str, str) + def _show_error(self, title: str, text: str) -> None: + QMessageBox.warning(self, title, text) + + @SafeSlot() + def _open_add_dialog(self) -> None: + open_add_dialog(self, self.controller) + + @SafeSlot(str) + def _open_edit_dialog(self, name: str) -> None: + open_edit_dialog(self, self.controller, name) + + def cleanup(self) -> None: + self.controller.cleanup() + self.view.setSource(QUrl()) + super().cleanup() diff --git a/bec_widgets/widgets/services/beamline_states/modern/beamline_states_widget.py b/bec_widgets/widgets/services/beamline_states/modern/beamline_states_widget.py new file mode 100644 index 00000000..99d3a371 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/beamline_states_widget.py @@ -0,0 +1,860 @@ +"""Beamline states view built from QWidgets, with the same UX as the QML version.""" + +from __future__ import annotations + +from functools import lru_cache +from typing import Any + +from bec_qthemes import material_icon +from qtpy.QtCore import Property, QEasingCurve, QPropertyAnimation, QSize, Qt, QTimer, Signal +from qtpy.QtGui import QColor, QIcon, QPainter, QPen +from qtpy.QtWidgets import ( + QAbstractButton, + QFrame, + QGraphicsOpacityEffect, + QGridLayout, + QHBoxLayout, + QLabel, + QLineEdit, + QMessageBox, + QPushButton, + QScrollArea, + QSizePolicy, + QToolButton, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.bec_connector import ConnectionConfig +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.eliding_label import ElidingLabel +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.utils.qml_host import QmlTheme +from bec_widgets.widgets.services.beamline_states.modern.actions import ( + open_add_dialog, + open_edit_dialog, +) +from bec_widgets.widgets.services.beamline_states.modern.states_controller import ( + INTERLOCK_SECTION, + STATUSES, + BeamlineStatesController, +) +from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch + +STATUS_TITLES = {"invalid": "Invalid", "warning": "Warning", "unknown": "Unknown", "valid": "Valid"} +STATUS_ICONS = { + "invalid": "cancel", + "warning": "warning", + "unknown": "help", + "valid": "check_circle", +} + + +def _status_color(theme: QmlTheme, status: str) -> QColor: + return {"invalid": theme.danger, "warning": theme.warning, "valid": theme.success}.get( + status, theme.muted + ) + + +def _rgba(color: QColor, alpha: float) -> str: + return f"rgba({color.red()}, {color.green()}, {color.blue()}, {int(alpha * 255)})" + + +@lru_cache(maxsize=256) +def _cached_icon(name: str, color: str, size: int, filled: bool): + return material_icon(name, size=(size, size), color=color, filled=filled) + + +def _icon(name: str, color: QColor, size: int = 18, filled: bool = True): + return _cached_icon(name, color.name(), size, filled) + + +def _set_prop(widget: QWidget, name: str, value) -> None: + """Set a dynamic property used by the stylesheet and re-polish only if it changed.""" + if widget.property(name) == value: + return + widget.setProperty(name, value) + widget.style().unpolish(widget) + widget.style().polish(widget) + + +def _card_stylesheet(theme: QmlTheme) -> str: + """Stylesheet for the cards, set once on the top-level widget per theme.""" + hover = theme.card.lighter(108) if theme.dark else theme.card.darker(102) + rules = [ + "QFrame#state_card {" + f"background: {theme.card.name()}; border: 1px solid {theme.border.name()};" + "border-left-width: 4px; border-radius: 8px; }", + f"QFrame#state_card:hover {{ background: {hover.name()}; }}", + f'QFrame#state_card[expanded="true"] {{ background: {hover.name()}; }}', + "QLabel#card_name { font-size: 13px; font-weight: 600; }", + "QLabel#device_chip {" + f"background: {_rgba(theme.foreground, 0.08)}; color: {theme.muted.name()};" + "border-radius: 4px; padding: 1px 6px; font-size: 10px; font-family: monospace; }", + f"QLabel#muted {{ color: {theme.muted.name()}; font-size: 11px; }}", + "QLabel#detail_value { font-size: 11px; font-family: monospace; }", + "QLabel#status_badge { border-radius: 15px; }", + "QLabel#status_chip {" + "border-radius: 11px; padding: 0px 8px; font-size: 10px; font-weight: 700; }", + f"QFrame#rule {{ background: {theme.border.name()}; }}", + "QPushButton#danger_button {" + f"background: transparent; color: {theme.danger.name()};" + f"border: 1px solid {_rgba(theme.danger, 0.6)}; border-radius: 6px;" + "padding: 5px 12px; font-weight: 600; }", + f"QPushButton#danger_button:hover {{ background: {_rgba(theme.danger, 0.14)}; }}", + 'QPushButton#danger_button[solid="true"] {' + f"background: {theme.danger.name()}; color: white; }}", + ] + for status in STATUSES: + accent = _status_color(theme, status) + rules += [ + f'QFrame#state_card[status="{status}"] {{ border-left-color: {accent.name()}; }}', + f'QFrame#state_card[status="{status}"][expanded="true"] {{' + f"border-color: {_rgba(accent, 0.6)}; border-left-color: {accent.name()}; }}", + f'QLabel#status_badge[status="{status}"] {{ background: {_rgba(accent, 0.18)}; }}', + f'QLabel#status_chip[status="{status}"] {{' + f"background: {_rgba(accent, 0.18)}; color: {accent.name()}; }}", + ] + rules.append( + 'QFrame#state_card[triggered="true"] {' + f"border-color: {theme.danger.name()}; border-left-color: {theme.danger.name()}; }}" + ) + return "".join(rules) + + +class _Pulse(QFrame): + """Frame that can draw a pulsing red outline (scan-blocking marker).""" + + def __init__(self, theme: QmlTheme, parent: QWidget | None = None) -> None: + super().__init__(parent) + self._theme = theme + self._phase = 0.0 + self._pulsing = False + self._animation = QPropertyAnimation(self, b"pulse_phase", self) + self._animation.setDuration(1400) + self._animation.setStartValue(0.0) + self._animation.setEndValue(1.0) + self._animation.setEasingCurve(QEasingCurve.Type.InOutSine) + self._animation.setLoopCount(-1) + + @Property(float) + def pulse_phase(self) -> float: + """Animation phase of the outline.""" + return self._phase + + @pulse_phase.setter + def pulse_phase(self, phase: float) -> None: + self._phase = phase + self.update() + + def set_pulsing(self, pulsing: bool) -> None: + """Start or stop the pulsing outline.""" + if pulsing == self._pulsing: + return + self._pulsing = pulsing + if pulsing: + self._animation.start() + else: + self._animation.stop() + self.update() + + def paintEvent(self, event) -> None: # noqa: N802 + super().paintEvent(event) + if not self._pulsing: + return + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + color = QColor(self._theme.danger) + color.setAlphaF(0.25 + 0.75 * abs(1.0 - 2.0 * self._phase)) + painter.setPen(QPen(color, 2)) + painter.drawRoundedRect(self.rect().adjusted(1, 1, -1, -1), 8, 8) + + def cleanup(self) -> None: + """Stop the animation.""" + self._animation.stop() + + +class _Chip(QAbstractButton): + """Checkable status chip with a colored dot and a count.""" + + def __init__(self, status: str, theme: QmlTheme, parent: QWidget | None = None) -> None: + super().__init__(parent) + self.status = status + self._theme = theme + self._count = 0 + self.setCheckable(True) + self.setCursor(Qt.CursorShape.PointingHandCursor) + self.setAttribute(Qt.WidgetAttribute.WA_Hover, True) + self.set_count(0) + + def set_count(self, count: int) -> None: + """Set the number shown on the chip.""" + self._count = count + self.setText(f"{count} {STATUS_TITLES[self.status]}") + self.setToolTip( + f"Stop filtering by {STATUS_TITLES[self.status]}" + if self.isChecked() + else f"Show only {STATUS_TITLES[self.status]} states" + ) + self.updateGeometry() + self.update() + + def sizeHint(self) -> QSize: # noqa: N802 + return QSize(self.fontMetrics().horizontalAdvance(self.text()) + 34, 28) + + def paintEvent(self, _event) -> None: # noqa: N802 + theme = self._theme + accent = _status_color(theme, self.status) + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + if self._count == 0 and not self.isChecked(): + painter.setOpacity(0.45) + rect = self.rect().adjusted(0, 0, -1, -1) + fill = QColor(Qt.GlobalColor.transparent) + if self.isChecked(): + fill = QColor(accent) + fill.setAlphaF(0.22) + elif self.underMouse(): + fill = QColor(theme.foreground) + fill.setAlphaF(0.08) + painter.setBrush(fill) + painter.setPen(QPen(accent if self.isChecked() else theme.border, 1)) + painter.drawRoundedRect(rect, rect.height() / 2, rect.height() / 2) + painter.setPen(Qt.PenStyle.NoPen) + painter.setBrush(accent) + painter.drawEllipse(10, rect.height() // 2 - 3, 8, 8) + font = painter.font() + font.setPixelSize(12) + font.setBold(self.isChecked()) + painter.setFont(font) + painter.setPen(theme.foreground) + painter.drawText(rect.adjusted(24, 0, 0, 0), Qt.AlignmentFlag.AlignVCenter, self.text()) + + +class _InterlockBanner(_Pulse): + """Banner showing whether the scan interlock is off, armed, or blocking scans.""" + + arm_requested = Signal(bool) + + def __init__(self, theme: QmlTheme, parent: QWidget | None = None) -> None: + super().__init__(theme, parent) + self.setObjectName("interlock_banner") + self.setMinimumHeight(52) + self._style_key = None + self._badge = QLabel(self) + self._badge.setObjectName("banner_badge") + self._badge.setFixedSize(30, 30) + self._badge.setAlignment(Qt.AlignmentFlag.AlignCenter) + self._title = ElidingLabel(self) + self._title.setObjectName("banner_title") + self._subtitle = ElidingLabel(self) + self._subtitle.setObjectName("banner_subtitle") + self._state = QLabel(self) + self._state.setObjectName("muted") + self.toggle = ToggleSwitch(self, checked=False) + self.toggle.setCursor(Qt.CursorShape.PointingHandCursor) + self.toggle.stateChanged.connect(self.arm_requested) + + text = QVBoxLayout() + text.setSpacing(1) + text.addWidget(self._title) + text.addWidget(self._subtitle) + layout = QHBoxLayout(self) + layout.setContentsMargins(12, 8, 12, 8) + layout.setSpacing(10) + layout.addWidget(self._badge) + layout.addLayout(text, 1) + layout.addWidget(self._state) + layout.addWidget(self.toggle) + + def update_from(self, controller: BeamlineStatesController) -> None: + """Refresh the banner from the controller state.""" + theme = self._theme + armed = controller.interlockArmed + blocking = controller.blockingStates + watched = controller.watchedCount + tripped = armed and bool(blocking) + accent = theme.danger if tripped else (theme.success if armed else theme.muted) + if tripped: + title = "Scans are blocked by the scan interlock" + subtitle = "Blocking: " + ", ".join(blocking) + elif armed: + title = "Scan interlock armed, scans can run" + subtitle = f"{watched} {'state is' if watched == 1 else 'states are'} watched." + else: + title = "Scan interlock is off" + subtitle = ( + "No states are watched. Use the lock on a state to watch it." + if watched == 0 + else f"{watched} {'state is' if watched == 1 else 'states are'} watched. " + "Arm the interlock to block scans when they fail." + ) + self._title.setText(title) + self._subtitle.setText(subtitle) + self._state.setText("Armed" if armed else "Off") + self.toggle.blockSignals(True) + self.toggle.active_track_color = theme.danger if tripped else theme.success + self.toggle.setChecked(armed) + self.toggle.blockSignals(False) + style_key = (armed, tripped, accent.name(), theme.muted.name()) + if style_key != self._style_key: + self._style_key = style_key + icon_name = "block" if tripped else ("lock" if armed else "lock_open_right") + self._badge.setPixmap(_icon(icon_name, accent)) + self.setStyleSheet( + "QFrame#interlock_banner {" + f"background: {_rgba(accent, 0.16 if armed else 0.08)};" + f"border: 1px solid {_rgba(accent, 0.9 if tripped else 0.45)};" + "border-radius: 8px; }" + f"QLabel#banner_badge {{ background: {_rgba(accent, 0.22)}; border-radius: 15px; }}" + "QLabel#banner_title { font-size: 13px; font-weight: 600; }" + f"QLabel#banner_subtitle {{ font-size: 11px; color: " + f"{(theme.danger if tripped else theme.muted).name()}; }}" + f"QLabel#muted {{ color: {theme.muted.name()}; }}" + ) + self.set_pulsing(tripped) + + +class _StateCard(_Pulse): + """One beamline state: header row plus an expandable details panel.""" + + expand_toggled = Signal(str) + + def __init__( + self, name: str, controller: BeamlineStatesController, theme: QmlTheme, parent=None + ) -> None: + super().__init__(theme, parent) + self.name = name + self._controller = controller + self._row: dict[str, Any] = {} + self._expanded = False + self.setObjectName("state_card") + self.setAttribute(Qt.WidgetAttribute.WA_Hover, True) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + + self._header = QWidget(self) + self._header.setCursor(Qt.CursorShape.PointingHandCursor) + self._header.mousePressEvent = self._header_clicked + self._badge = QLabel(self._header) + self._badge.setObjectName("status_badge") + self._badge.setFixedSize(30, 30) + self._badge.setAlignment(Qt.AlignmentFlag.AlignCenter) + self._name = QLabel(name, self._header) + self._name.setObjectName("card_name") + self._device = QLabel(self._header) + self._device.setObjectName("device_chip") + self._label = ElidingLabel(self._header) + self._label.setObjectName("muted") + self._status = QLabel(self._header) + self._status.setObjectName("status_chip") + self._status.setFixedHeight(22) + self._status.setAlignment(Qt.AlignmentFlag.AlignCenter) + self._lock = QToolButton(self._header) + self._lock.setAutoRaise(True) + self._lock.setIconSize(QSize(18, 18)) + self._lock.setCursor(Qt.CursorShape.PointingHandCursor) + self._lock.clicked.connect(lambda: controller.toggleWatched(self.name)) + self._chevron = QLabel(self._header) + for child in (self._badge, self._name, self._device, self._label, self._status): + child.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + + title = QHBoxLayout() + title.setSpacing(6) + title.addWidget(self._name) + title.addWidget(self._device) + title.addStretch(1) + text = QVBoxLayout() + text.setSpacing(2) + text.addLayout(title) + text.addWidget(self._label) + header = QHBoxLayout(self._header) + header.setContentsMargins(14, 8, 8, 8) + header.setSpacing(10) + header.addWidget(self._badge) + header.addLayout(text, 1) + header.addWidget(self._status) + header.addWidget(self._lock) + header.addWidget(self._chevron) + + self._body = QWidget(self) + self._body.setObjectName("card_body") + self._body.setVisible(False) + self._details = QGridLayout() + self._details.setHorizontalSpacing(14) + self._details.setVerticalSpacing(4) + self._details.setColumnStretch(1, 1) + self._warning_toggle = ToggleSwitch(self._body, checked=False) + self._warning_toggle.stateChanged.connect( + lambda value: controller.setTriggerOnWarning(self.name, bool(value)) + ) + self._warning_label = QLabel(self._body) + warning_row = QHBoxLayout() + warning_row.setSpacing(8) + warning_row.addWidget(self._warning_toggle) + warning_row.addWidget(self._warning_label, 1) + self._edit = QPushButton("Edit parameters", self._body) + self._edit.setObjectName("outline_button") + self._edit.clicked.connect(lambda: controller.requestEdit(self.name)) + self._remove = QPushButton("Remove", self._body) + self._remove.setObjectName("danger_button") + self._remove.clicked.connect(self._remove_clicked) + self._confirm_timer = QTimer(self) + self._confirm_timer.setSingleShot(True) + self._confirm_timer.setInterval(3000) + self._confirm_timer.timeout.connect(lambda: self._set_confirming(False)) + buttons = QHBoxLayout() + buttons.addWidget(self._edit) + buttons.addStretch(1) + buttons.addWidget(self._remove) + rule = QFrame(self._body) + rule.setObjectName("rule") + rule.setFixedHeight(1) + body = QVBoxLayout(self._body) + body.setContentsMargins(14, 0, 14, 10) + body.setSpacing(8) + body.addWidget(rule) + body.addLayout(self._details) + body.addLayout(warning_row) + body.addLayout(buttons) + + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + layout.addWidget(self._header) + layout.addWidget(self._body) + self._confirming = False + + def _header_clicked(self, event) -> None: + if event.button() == Qt.MouseButton.LeftButton: + self.expand_toggled.emit(self.name) + + def is_expanded(self) -> bool: + """Whether the details panel is open.""" + return self._expanded + + def set_expanded(self, expanded: bool) -> None: + """Open or close the details panel.""" + if expanded == self._expanded: + return + self._expanded = expanded + self._body.setVisible(expanded) + self._set_confirming(False) + self._apply_style() + + def _remove_clicked(self) -> None: + if self._confirming: + self._set_confirming(False) + self._controller.removeState(self.name) + else: + self._set_confirming(True) + self._confirm_timer.start() + + def _set_confirming(self, confirming: bool) -> None: + self._confirming = confirming + self._remove.setText("Click again to remove" if confirming else "Remove") + _set_prop(self._remove, "solid", confirming) + self._apply_style() + + def update_row(self, row: dict[str, Any]) -> None: + """Show the data of a controller row.""" + if row == self._row: + return + details_changed = row.get("details") != self._row.get("details") or row.get( + "stateType" + ) != self._row.get("stateType") + self._row = row + self._device.setText(row["device"]) + self._device.setVisible(bool(row["device"])) + self._label.setText(row["label"]) + self.setToolTip(row["label"]) + self._status.setText(row["statusText"]) + self._lock.setToolTip( + f"Watched by the scan interlock (accepts {row['accepted']}). Click to stop watching." + if row["watched"] + else "Not watched by the scan interlock. Click to watch this state." + ) + self._warning_toggle.blockSignals(True) + self._warning_toggle.setChecked(row["triggerOnWarning"]) + self._warning_toggle.blockSignals(False) + self._warning_label.setText( + "Block scans on WARNING too" + ("" if row["watched"] else " (applies once watched)") + ) + if details_changed: + self._fill_details() + self.set_pulsing(row["triggered"]) + self._apply_style() + + def _fill_details(self) -> None: + while self._details.count(): + item = self._details.takeAt(0) + if item.widget() is not None: + item.widget().deleteLater() + pairs = [("Type", self._row["stateType"])] + [ + (d["key"], d["value"]) for d in self._row["details"] + ] + for i, (key, value) in enumerate(pairs): + key_label = QLabel(key, self._body) + key_label.setObjectName("muted") + value_label = QLabel(value, self._body) + value_label.setObjectName("detail_value" if i else "") + self._details.addWidget(key_label, i, 0) + self._details.addWidget(value_label, i, 1) + + def apply_theme(self) -> None: + """Re-apply theme colors.""" + self._apply_style() + + def _apply_style(self) -> None: + if not self._row: + return + theme = self._theme + row = self._row + accent = _status_color(theme, row["status"]) + for widget in (self, self._badge, self._status): + _set_prop(widget, "status", row["status"]) + _set_prop(self, "expanded", self._expanded) + _set_prop(self, "triggered", row["triggered"]) + lock_color = ( + theme.danger + if row["triggered"] + else (theme.foreground if row["watched"] else theme.muted) + ) + self._badge.setPixmap(_icon(STATUS_ICONS[row["status"]], accent)) + self._lock.setIcon( + QIcon( + _icon( + "lock" if row["watched"] else "lock_open_right", lock_color, 18, row["watched"] + ) + ) + ) + self._chevron.setPixmap( + _icon("expand_less" if self._expanded else "expand_more", theme.muted) + ) + self._edit.setIcon(QIcon(_icon("edit", theme.primary, 16))) + self._remove.setIcon( + QIcon(_icon("delete", QColor("#ffffff") if self._confirming else theme.danger, 16)) + ) + + +class _SectionHeader(QWidget): + """Small caps section title above the watched / not watched groups.""" + + def __init__(self, section: str, theme: QmlTheme, parent: QWidget | None = None) -> None: + super().__init__(parent) + self._section = section + self._theme = theme + self._icon = QLabel(self) + self._text = QLabel( + "WATCHED BY SCAN INTERLOCK" if section == INTERLOCK_SECTION else "NOT WATCHED", self + ) + layout = QHBoxLayout(self) + layout.setContentsMargins(2, 8, 0, 2) + layout.setSpacing(6) + layout.addWidget(self._icon) + layout.addWidget(self._text, 1) + self.apply_theme() + + def apply_theme(self) -> None: + """Re-apply theme colors.""" + name = "lock" if self._section == INTERLOCK_SECTION else "lock_open_right" + self._icon.setPixmap(_icon(name, self._theme.muted, 14)) + self._text.setStyleSheet( + f"color: {self._theme.muted.name()}; font-size: 10px; font-weight: 700;" + "letter-spacing: 0.6px;" + ) + + +class BeamlineStatesWidget(BECWidget, QWidget): + """ + Beamline states with the scan interlock, built from QWidgets. + + Same UX as :class:`BeamlineStatesQML`, for comparison with the QML version and the original + ``BeamlineStateManager``. + """ + + PLUGIN = False + RPC = False + ICON_NAME = "format_list_bulleted" + USER_ACCESS = ["clear_filters", "collapse_all", "state_summary", "remove", "attach", "detach"] + + 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.setObjectName("beamline_states_widget") + self.theme = QmlTheme(self) + self._cards: dict[str, _StateCard] = {} + self._expanded: set[str] = set() + + # Summary chips, search and add. + self._chips = {status: _Chip(status, self.theme, self) for status in STATUSES} + self._search = QLineEdit(self) + self._search.setPlaceholderText("Search name or device") + self._search.setClearButtonEnabled(True) + self._search.setMinimumWidth(150) + self._search_action = self._search.addAction( + material_icon("search", size=(16, 16)), QLineEdit.ActionPosition.LeadingPosition + ) + self._add = QPushButton("Add state", self) + self._add.setObjectName("primary_button") + self._add.setCursor(Qt.CursorShape.PointingHandCursor) + top = QHBoxLayout() + top.setSpacing(6) + for chip in self._chips.values(): + top.addWidget(chip) + top.addWidget(self._search, 1) + top.addWidget(self._add) + + self._banner = _InterlockBanner(self.theme, self) + + # Scrollable list of section headers and cards. + self._list = QWidget() + self._list.setObjectName("states_list") + self._list_layout = QVBoxLayout(self._list) + self._list_layout.setContentsMargins(0, 0, 0, 0) + self._list_layout.setSpacing(6) + self._list_layout.addStretch(1) + self._sections = { + section: _SectionHeader(section, self.theme, self._list) + for section in (INTERLOCK_SECTION, "others") + } + self._scroll = QScrollArea(self) + self._scroll.setWidgetResizable(True) + self._scroll.setFrameShape(QFrame.Shape.NoFrame) + self._scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) + self._scroll.setWidget(self._list) + + self._footer = QWidget(self) + self._hidden_label = QLabel(self._footer) + self._show_hidden = QPushButton(self._footer) + self._show_hidden.setObjectName("outline_button") + self._clear = QPushButton("Clear filters", self._footer) + self._clear.setObjectName("outline_button") + footer = QHBoxLayout(self._footer) + footer.setContentsMargins(0, 2, 0, 2) + footer.addStretch(1) + footer.addWidget(self._hidden_label) + footer.addWidget(self._show_hidden) + footer.addWidget(self._clear) + footer.addStretch(1) + + self._empty = QWidget(self) + self._empty_icon = QLabel(self._empty) + self._empty_title = QLabel(self._empty) + self._empty_title.setObjectName("empty_title") + self._empty_text = QLabel(self._empty) + self._empty_text.setObjectName("muted") + self._empty_button = QPushButton(self._empty) + empty = QVBoxLayout(self._empty) + empty.addStretch(1) + for child in (self._empty_icon, self._empty_title, self._empty_text, self._empty_button): + empty.addWidget(child, 0, Qt.AlignmentFlag.AlignHCenter) + empty.addStretch(1) + + layout = QVBoxLayout(self) + layout.setContentsMargins(10, 10, 10, 10) + layout.setSpacing(8) + layout.addLayout(top) + layout.addWidget(self._banner) + layout.addWidget(self._scroll, 1) + layout.addWidget(self._empty, 1) + layout.addWidget(self._footer) + + self.controller = BeamlineStatesController(self.client, self.bec_dispatcher, self) + self.controller.changed.connect(self._sync) + self.controller.error_occurred.connect(self._show_error) + self.controller.add_requested.connect(self._open_add_dialog) + self.controller.edit_requested.connect(self._open_edit_dialog) + for chip in self._chips.values(): + chip.clicked.connect( + lambda _checked=False, s=chip.status: self.controller.toggleStatusFilter(s) + ) + self._search.textEdited.connect(self.controller.setSearchText) + self._add.clicked.connect(self.controller.requestAdd) + self._banner.arm_requested.connect(self.controller.setInterlockArmed) + self._show_hidden.clicked.connect( + lambda: self.controller.setShowFiltered(not self.controller.showFiltered) + ) + self._clear.clicked.connect(self.controller.clearFilters) + self._empty_button.clicked.connect(self._empty_action) + self.theme.changed.connect(self._restyle) + self._restyle() + + # ------------------------------------------------------------------ rendering + + @SafeSlot() + def _sync(self) -> None: + controller = self.controller + rows = controller.model.rows() + names = {row["name"] for row in rows} + for name in list(self._cards): + if name not in names: + card = self._cards.pop(name) + card.cleanup() + card.deleteLater() + self._expanded.discard(name) + + # Re-insert headers and cards in display order; widgets that keep their slot do not move. + wanted: list[QWidget] = [] + section = None + for row in rows: + if row["section"] != section: + section = row["section"] + wanted.append(self._sections[section]) + card = self._cards.get(row["name"]) + if card is None: + card = _StateCard(row["name"], controller, self.theme, self._list) + card.expand_toggled.connect(self._toggle_expanded) + self._cards[row["name"]] = card + card.update_row(row) + card.set_expanded(row["name"] in self._expanded) + wanted.append(card) + for i, widget in enumerate(wanted): + if self._list_layout.indexOf(widget) != i: + self._list_layout.removeWidget(widget) + self._list_layout.insertWidget(i, widget) + widget.show() + for header in self._sections.values(): + if header not in wanted: + self._list_layout.removeWidget(header) + header.hide() + for row in rows: + self._set_dimmed(self._cards[row["name"]], row["filteredOut"]) + + counts = controller.counts + filters = set(controller.statusFilter) + for status, chip in self._chips.items(): + chip.setChecked(status in filters) + chip.set_count(counts[status]) + if self._search.text() != controller.searchText: + self._search.setText(controller.searchText) + self._banner.update_from(controller) + + hidden = controller.hiddenCount + has_rows = bool(rows) + self._scroll.setVisible(has_rows) + self._empty.setVisible(not has_rows) + self._footer.setVisible(has_rows and hidden > 0) + self._hidden_label.setText( + f"{hidden} {'state' if hidden == 1 else 'states'} hidden by filters" + ) + self._show_hidden.setText("Hide them" if controller.showFiltered else "Show them") + no_states = controller.totalCount == 0 + self._empty_icon.setPixmap( + _icon("playlist_add" if no_states else "filter_alt_off", self.theme.muted, 36) + ) + self._empty_title.setText( + "No beamline states yet" if no_states else "No states match the filters" + ) + self._empty_text.setText( + "States check conditions such as a motor within limits." + if no_states + else f"{hidden} states are hidden." + ) + self._empty_button.setText("Add state" if no_states else "Clear filters") + self._empty_button.setObjectName("primary_button" if no_states else "outline_button") + self._empty_button.style().unpolish(self._empty_button) + self._empty_button.style().polish(self._empty_button) + + @staticmethod + def _set_dimmed(card: _StateCard, dimmed: bool) -> None: + # Opacity effects are costly, so only filtered-out cards get one. + if dimmed == (card.graphicsEffect() is not None): + return + if dimmed: + effect = QGraphicsOpacityEffect(card) + effect.setOpacity(0.5) + card.setGraphicsEffect(effect) + else: + card.setGraphicsEffect(None) + + @SafeSlot(str) + def _toggle_expanded(self, name: str) -> None: + self._expanded ^= {name} + card = self._cards.get(name) + if card is not None: + card.set_expanded(name in self._expanded) + + @SafeSlot() + def _restyle(self) -> None: + theme = self.theme + self._search_action.setIcon( + material_icon("search", size=(16, 16), color=theme.muted.name()) + ) + self._add.setIcon(material_icon("add", size=(16, 16), color=theme.onPrimary.name())) + self.setStyleSheet( + f"QWidget#beamline_states_widget {{ background: {theme.background.name()}; }}" + "QWidget#states_list { background: transparent; }" + "QLineEdit { border-radius: 6px; padding: 4px 6px; min-height: 18px; }" + "QPushButton#primary_button {" + f"background: {theme.primary.name()}; color: {theme.onPrimary.name()};" + "border: none; border-radius: 6px; padding: 5px 12px; font-weight: 600; }" + "QPushButton#primary_button:hover {" + f"background: {theme.primary.darker(112).name()}; }}" + "QPushButton#outline_button {" + f"background: transparent; color: {theme.primary.name()};" + f"border: 1px solid {_rgba(theme.primary, 0.6)}; border-radius: 6px;" + "padding: 5px 12px; font-weight: 600; }" + "QLabel#empty_title { font-size: 13px; font-weight: 600; }" + f"QLabel#muted {{ color: {theme.muted.name()}; font-size: 11px; }}" + + _card_stylesheet(theme) + ) + for header in self._sections.values(): + header.apply_theme() + for card in self._cards.values(): + card.apply_theme() + self._sync() + + @SafeSlot() + def _empty_action(self) -> None: + if self.controller.totalCount == 0: + self.controller.requestAdd() + else: + self.controller.clearFilters() + + # ------------------------------------------------------------------ public API + + @SafeSlot() + def clear_filters(self) -> None: + """Reset the status chips and the search text.""" + self.controller.clearFilters() + + @SafeSlot() + def collapse_all(self) -> None: + """Collapse every expanded state.""" + self._expanded.clear() + for card in self._cards.values(): + card.set_expanded(False) + + 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 self.controller.state_summary() + + @SafeSlot(str, str) + def _show_error(self, title: str, text: str) -> None: + QMessageBox.warning(self, title, text) + + @SafeSlot() + def _open_add_dialog(self) -> None: + open_add_dialog(self, self.controller) + + @SafeSlot(str) + def _open_edit_dialog(self, name: str) -> None: + open_edit_dialog(self, self.controller, name) + + def cleanup(self) -> None: + self.controller.cleanup() + self._banner.cleanup() + for card in self._cards.values(): + card.cleanup() + super().cleanup() diff --git a/bec_widgets/widgets/services/beamline_states/modern/compare_demo.py b/bec_widgets/widgets/services/beamline_states/modern/compare_demo.py new file mode 100644 index 00000000..cfc0ad4e --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/compare_demo.py @@ -0,0 +1,46 @@ +""" +Show the original, QML and QWidget beamline-state views side by side. + +Run against a live BEC session: ``python -m bec_widgets.widgets.services.beamline_states.modern.compare_demo`` +""" + +from __future__ import annotations + +import sys + +from qtpy.QtWidgets import QApplication, QHBoxLayout, QLabel, QVBoxLayout, QWidget + +from bec_widgets.utils.colors import apply_theme +from bec_widgets.widgets.services.beamline_states.beamline_state_manager import BeamlineStateManager +from bec_widgets.widgets.services.beamline_states.modern.beamline_states_qml import ( + BeamlineStatesQML, +) +from bec_widgets.widgets.services.beamline_states.modern.beamline_states_widget import ( + BeamlineStatesWidget, +) +from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + +if __name__ == "__main__": # pragma: no cover + app = QApplication(sys.argv) + apply_theme("dark") + window = QWidget() + window.setWindowTitle("Beamline States: original vs QML vs QWidget") + columns = QHBoxLayout() + for title, cls in ( + ("Original", BeamlineStateManager), + ("QML", BeamlineStatesQML), + ("QWidget", BeamlineStatesWidget), + ): + column = QVBoxLayout() + column.addWidget(QLabel(f"{title}")) + column.addWidget(cls(parent=window), 1) + columns.addLayout(column, 1) + top = QHBoxLayout() + top.addStretch(1) + top.addWidget(DarkModeButton(parent=window)) + layout = QVBoxLayout(window) + layout.addLayout(top) + layout.addLayout(columns, 1) + window.resize(1800, 700) + window.show() + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/services/beamline_states/modern/qml/BeamlineStates.qml b/bec_widgets/widgets/services/beamline_states/modern/qml/BeamlineStates.qml new file mode 100644 index 00000000..21da17a0 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/qml/BeamlineStates.qml @@ -0,0 +1,616 @@ +import QtQuick +import QtQuick.Controls.Basic +import QtQuick.Layouts + +// Beamline states view. Context properties: theme (QmlTheme), controller +// (BeamlineStatesController) and statesModel (BeamlineStatesModel). +Rectangle { + id: root + color: theme.background + focus: true + + property var expanded: ({}) + readonly property var statusOrder: ["invalid", "warning", "unknown", "valid"] + readonly property var statusTitles: ({ invalid: "Invalid", warning: "Warning", unknown: "Unknown", valid: "Valid" }) + readonly property var statusIcons: ({ invalid: "cancel", warning: "warning", unknown: "help", valid: "check_circle" }) + + function statusColor(status) { + if (status === "invalid") return theme.danger + if (status === "warning") return theme.warning + if (status === "valid") return theme.success + return theme.muted + } + function tint(c, alpha) { return Qt.rgba(c.r, c.g, c.b, alpha) } + function icon(name, c, filled) { + return "image://material/" + name + "?color=" + encodeURIComponent(c.toString()) + "&filled=" + (filled ? 1 : 0) + } + function isExpanded(name) { return expanded[name] === true } + function toggleExpanded(name) { + var copy = Object.assign({}, expanded) + copy[name] = !copy[name] + expanded = copy + } + function collapseAll() { expanded = ({}) } + + // ---------------------------------------------------------------- shared controls + + component Icon: Image { + property string name + property color tintColor: theme.foreground + property bool filled: true + property int size: 18 + source: name ? root.icon(name, tintColor, filled) : "" + sourceSize: Qt.size(size, size) + width: size + height: size + smooth: true + } + + component IconButton: Rectangle { + id: iconButton + property string iconName + property color iconColor: theme.foreground + property bool filled: true + property string tip + signal clicked() + implicitWidth: 28 + implicitHeight: 28 + radius: 6 + color: mouse.containsMouse ? root.tint(theme.foreground, 0.10) : "transparent" + Icon { anchors.centerIn: parent; name: iconButton.iconName; tintColor: iconButton.iconColor; filled: iconButton.filled } + MouseArea { + id: mouse + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: iconButton.clicked() + } + ToolTip.visible: tip.length > 0 && mouse.containsMouse + ToolTip.delay: 500 + ToolTip.text: tip + } + + component TextButton: Rectangle { + id: textButton + property string text + property string iconName + property color accent: theme.primary + property bool solid: false + signal clicked() + implicitHeight: 28 + implicitWidth: label.implicitWidth + (iconName ? 40 : 24) + radius: 6 + color: solid ? (mouse.containsMouse ? Qt.darker(accent, 1.12) : accent) + : (mouse.containsMouse ? root.tint(accent, 0.14) : "transparent") + border.width: solid ? 0 : 1 + border.color: root.tint(accent, 0.6) + Row { + anchors.centerIn: parent + spacing: 6 + Icon { + visible: textButton.iconName.length > 0 + name: textButton.iconName + size: 16 + anchors.verticalCenter: parent.verticalCenter + tintColor: textButton.solid ? theme.onPrimary : textButton.accent + } + Text { + id: label + text: textButton.text + color: textButton.solid ? theme.onPrimary : textButton.accent + font.pixelSize: 12 + font.weight: Font.DemiBold + anchors.verticalCenter: parent.verticalCenter + } + } + MouseArea { + id: mouse + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: textButton.clicked() + } + } + + // Pulsing red outline marking something that blocks scans. + component Pulse: Rectangle { + id: pulse + property bool running + anchors.fill: parent + radius: parent.radius + color: "transparent" + border.width: 2 + border.color: theme.danger + SequentialAnimation on opacity { + running: pulse.running + loops: Animation.Infinite + NumberAnimation { from: 1.0; to: 0.25; duration: 700; easing.type: Easing.InOutSine } + NumberAnimation { from: 0.25; to: 1.0; duration: 700; easing.type: Easing.InOutSine } + } + } + + component Toggle: Rectangle { + id: toggle + property bool checked + property color accent: theme.success + signal toggled(bool value) + implicitWidth: 38 + implicitHeight: 22 + radius: height / 2 + color: checked ? accent : root.tint(theme.foreground, 0.22) + Behavior on color { ColorAnimation { duration: 140 } } + Rectangle { + width: 16; height: 16; radius: 8 + y: 3 + x: toggle.checked ? toggle.width - width - 3 : 3 + color: "white" + Behavior on x { NumberAnimation { duration: 140; easing.type: Easing.OutCubic } } + } + MouseArea { + anchors.fill: parent + cursorShape: Qt.PointingHandCursor + onClicked: toggle.toggled(!toggle.checked) + } + } + + // ---------------------------------------------------------------- layout + + ColumnLayout { + anchors.fill: parent + anchors.margins: 10 + spacing: 8 + + // Summary chips (status filters), search and add. + RowLayout { + Layout.fillWidth: true + spacing: 6 + + Repeater { + model: root.statusOrder + delegate: Rectangle { + id: chip + required property string modelData + readonly property int count: controller.counts[modelData] || 0 + readonly property bool active: controller.statusFilter.indexOf(modelData) >= 0 + readonly property color accent: root.statusColor(modelData) + implicitHeight: 28 + implicitWidth: chipRow.implicitWidth + 20 + radius: 14 + opacity: count === 0 && !active ? 0.45 : 1.0 + color: active ? root.tint(accent, 0.22) : (chipMouse.containsMouse ? root.tint(theme.foreground, 0.08) : "transparent") + border.width: 1 + border.color: active ? accent : theme.border + Row { + id: chipRow + anchors.centerIn: parent + spacing: 6 + Rectangle { width: 8; height: 8; radius: 4; color: chip.accent; anchors.verticalCenter: parent.verticalCenter } + Text { + text: chip.count + " " + root.statusTitles[chip.modelData] + color: theme.foreground + font.pixelSize: 12 + font.weight: chip.active ? Font.DemiBold : Font.Normal + } + } + MouseArea { + id: chipMouse + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: controller.toggleStatusFilter(chip.modelData) + } + ToolTip.visible: chipMouse.containsMouse + ToolTip.delay: 600 + ToolTip.text: chip.active ? "Stop filtering by " + root.statusTitles[chip.modelData] : "Show only " + root.statusTitles[chip.modelData] + " states" + } + } + + Rectangle { + Layout.fillWidth: true + Layout.minimumWidth: 150 + implicitHeight: 28 + radius: 6 + color: theme.field + border.width: 1 + border.color: search.activeFocus ? theme.primary : theme.border + Icon { id: searchIcon; name: "search"; size: 16; tintColor: theme.muted; anchors.left: parent.left; anchors.leftMargin: 8; anchors.verticalCenter: parent.verticalCenter } + TextInput { + id: search + anchors.left: searchIcon.right + anchors.right: clearSearch.left + anchors.margins: 6 + anchors.verticalCenter: parent.verticalCenter + color: theme.foreground + font.pixelSize: 12 + clip: true + selectByMouse: true + text: controller.searchText + onTextEdited: controller.setSearchText(text) + Keys.onEscapePressed: { text = ""; controller.setSearchText("") } + Text { + visible: !search.text && !search.activeFocus + text: "Search name or device" + color: theme.muted + font: search.font + } + } + IconButton { + id: clearSearch + visible: search.text.length > 0 + width: visible ? 24 : 0; height: 24 + anchors.right: parent.right; anchors.rightMargin: 2 + anchors.verticalCenter: parent.verticalCenter + iconName: "close"; iconColor: theme.muted + onClicked: controller.setSearchText("") + } + } + + TextButton { text: "Add state"; iconName: "add"; solid: true; onClicked: controller.requestAdd() } + } + + // Scan interlock banner. + Rectangle { + id: banner + Layout.fillWidth: true + readonly property bool armed: controller.interlockArmed + readonly property var blocking: controller.blockingStates + readonly property bool tripped: armed && blocking.length > 0 + readonly property color accent: tripped ? theme.danger : (armed ? theme.success : theme.muted) + implicitHeight: 52 + radius: 8 + color: root.tint(accent, armed ? 0.16 : 0.08) + border.width: 1 + border.color: root.tint(accent, tripped ? 0.9 : 0.45) + + Pulse { visible: banner.tripped; running: banner.tripped } + + RowLayout { + anchors.fill: parent + anchors.leftMargin: 12 + anchors.rightMargin: 12 + spacing: 10 + Rectangle { + width: 30; height: 30; radius: 15 + color: root.tint(banner.accent, 0.22) + Icon { + anchors.centerIn: parent + name: banner.tripped ? "block" : (banner.armed ? "lock" : "lock_open_right") + tintColor: banner.accent + } + } + ColumnLayout { + Layout.fillWidth: true + spacing: 1 + Text { + Layout.fillWidth: true + text: banner.tripped ? "Scans are blocked by the scan interlock" + : banner.armed ? "Scan interlock armed, scans can run" + : "Scan interlock is off" + color: theme.foreground + font.pixelSize: 13 + font.weight: Font.DemiBold + elide: Text.ElideRight + } + Text { + Layout.fillWidth: true + text: banner.tripped ? "Blocking: " + banner.blocking.join(", ") + : controller.watchedCount === 0 ? "No states are watched. Use the lock on a state to watch it." + : controller.watchedCount + (controller.watchedCount === 1 ? " state is" : " states are") + " watched" + + (banner.armed ? "." : ". Arm the interlock to block scans when they fail.") + color: banner.tripped ? theme.danger : theme.muted + font.pixelSize: 11 + elide: Text.ElideRight + } + } + Text { text: banner.armed ? "Armed" : "Off"; color: theme.muted; font.pixelSize: 12 } + Toggle { + checked: banner.armed + accent: banner.tripped ? theme.danger : theme.success + onToggled: function(value) { controller.setInterlockArmed(value) } + } + } + } + + // State list. + Item { + Layout.fillWidth: true + Layout.fillHeight: true + + ListView { + id: list + anchors.fill: parent + clip: true + spacing: 6 + model: statesModel + boundsBehavior: Flickable.StopAtBounds + reuseItems: false + ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded } + + section.property: "section" + section.delegate: Item { + required property string section + width: list.width + height: 28 + Row { + anchors.left: parent.left + anchors.leftMargin: 2 + anchors.bottom: parent.bottom + anchors.bottomMargin: 4 + spacing: 6 + Icon { + name: section === "interlock" ? "lock" : "lock_open_right" + size: 14 + tintColor: theme.muted + } + Text { + text: section === "interlock" ? "WATCHED BY SCAN INTERLOCK" : "NOT WATCHED" + color: theme.muted + font.pixelSize: 10 + font.weight: Font.Bold + font.letterSpacing: 0.6 + } + } + } + + move: Transition { NumberAnimation { properties: "y"; duration: 180; easing.type: Easing.OutCubic } } + displaced: Transition { NumberAnimation { properties: "y"; duration: 180; easing.type: Easing.OutCubic } } + add: Transition { NumberAnimation { property: "opacity"; from: 0; to: 1; duration: 160 } } + + delegate: Rectangle { + id: card + required property string name + required property string status + required property string statusText + required property string label + required property string device + required property string stateType + required property var details + required property bool watched + required property string accepted + required property bool triggerOnWarning + required property bool triggered + required property bool filteredOut + + readonly property bool open: root.isExpanded(name) + readonly property color accent: root.statusColor(status) + property bool confirmRemove: false + + width: list.width - (list.ScrollBar.vertical.visible ? 10 : 0) + height: header.height + (open ? body.implicitHeight : 0) + Behavior on height { NumberAnimation { duration: 140; easing.type: Easing.OutCubic } } + clip: true + radius: 8 + opacity: filteredOut ? 0.5 : 1.0 + color: headerMouse.containsMouse || open ? Qt.lighter(theme.card, theme.dark ? 1.08 : 0.98) : theme.card + border.width: 1 + border.color: triggered ? theme.danger : (open ? root.tint(accent, 0.6) : theme.border) + + Pulse { visible: card.triggered; running: card.triggered; z: 2 } + + // Status stripe. + Rectangle { + x: 0; y: 0 + width: 4; height: header.height + radius: 2 + color: card.accent + } + + Item { + id: header + width: parent.width + height: 52 + MouseArea { + id: headerMouse + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: root.toggleExpanded(card.name) + } + RowLayout { + anchors.fill: parent + anchors.leftMargin: 14 + anchors.rightMargin: 8 + spacing: 10 + Rectangle { + width: 30; height: 30; radius: 15 + color: root.tint(card.accent, 0.18) + Icon { anchors.centerIn: parent; name: root.statusIcons[card.status]; tintColor: card.accent } + } + ColumnLayout { + Layout.fillWidth: true + spacing: 2 + Row { + id: titleRow + Layout.fillWidth: true + spacing: 6 + Text { + width: Math.min(implicitWidth, titleRow.width - (deviceChip.visible ? deviceChip.width + 6 : 0)) + text: card.name + color: theme.foreground + font.pixelSize: 13 + font.weight: Font.DemiBold + elide: Text.ElideRight + } + Rectangle { + id: deviceChip + visible: card.device.length > 0 + anchors.verticalCenter: parent.verticalCenter + implicitHeight: 18 + implicitWidth: deviceText.implicitWidth + 12 + radius: 4 + color: root.tint(theme.foreground, 0.08) + Text { id: deviceText; anchors.centerIn: parent; text: card.device; color: theme.muted; font.pixelSize: 10; font.family: "monospace" } + } + } + Text { + Layout.fillWidth: true + text: card.label + color: theme.muted + font.pixelSize: 11 + elide: Text.ElideRight + } + } + Rectangle { + implicitHeight: 22 + implicitWidth: statusLabel.implicitWidth + 14 + radius: 11 + color: root.tint(card.accent, 0.18) + Text { id: statusLabel; anchors.centerIn: parent; text: card.statusText; color: card.accent; font.pixelSize: 10; font.weight: Font.Bold; font.letterSpacing: 0.5 } + } + IconButton { + iconName: card.watched ? "lock" : "lock_open_right" + filled: card.watched + iconColor: card.triggered ? theme.danger : (card.watched ? theme.foreground : theme.muted) + tip: card.watched ? "Watched by the scan interlock (accepts " + card.accepted + "). Click to stop watching." + : "Not watched by the scan interlock. Click to watch this state." + onClicked: controller.toggleWatched(card.name) + } + Icon { name: card.open ? "expand_less" : "expand_more"; tintColor: theme.muted } + } + } + + ColumnLayout { + id: body + y: header.height + x: 14 + width: parent.width - 28 + spacing: 8 + visible: card.open || card.height > header.height + + Rectangle { Layout.fillWidth: true; height: 1; color: theme.border } + + GridLayout { + columns: 2 + columnSpacing: 14 + rowSpacing: 4 + Layout.fillWidth: true + Text { text: "Type"; color: theme.muted; font.pixelSize: 11 } + Text { text: card.stateType; color: theme.foreground; font.pixelSize: 11; Layout.fillWidth: true; elide: Text.ElideRight } + Repeater { + model: card.details + delegate: Text { + required property var modelData + required property int index + text: modelData.key + color: theme.muted + font.pixelSize: 11 + Layout.row: index + 1 + Layout.column: 0 + } + } + Repeater { + model: card.details + delegate: Text { + required property var modelData + required property int index + text: modelData.value + color: theme.foreground + font.pixelSize: 11 + font.family: "monospace" + elide: Text.ElideRight + Layout.fillWidth: true + Layout.row: index + 1 + Layout.column: 1 + } + } + } + + RowLayout { + spacing: 8 + Toggle { + checked: card.triggerOnWarning + accent: theme.warning + onToggled: function(value) { controller.setTriggerOnWarning(card.name, value) } + } + Text { + Layout.fillWidth: true + text: "Block scans on WARNING too" + (card.watched ? "" : " (applies once watched)") + color: theme.foreground + font.pixelSize: 11 + elide: Text.ElideRight + } + } + + RowLayout { + spacing: 8 + Layout.bottomMargin: 10 + TextButton { text: "Edit parameters"; iconName: "edit"; onClicked: controller.requestEdit(card.name) } + Item { Layout.fillWidth: true } + TextButton { + text: card.confirmRemove ? "Click again to remove" : "Remove" + iconName: "delete" + accent: theme.danger + solid: card.confirmRemove + onClicked: { + if (card.confirmRemove) { + card.confirmRemove = false + controller.removeState(card.name) + } else { + card.confirmRemove = true + confirmTimer.restart() + } + } + } + Timer { id: confirmTimer; interval: 3000; onTriggered: card.confirmRemove = false } + } + } + } + + footer: Item { + width: list.width + height: visible ? 40 : 0 + visible: controller.hiddenCount > 0 && list.count > 0 + Row { + anchors.centerIn: parent + spacing: 10 + Text { + anchors.verticalCenter: parent.verticalCenter + text: controller.hiddenCount + (controller.hiddenCount === 1 ? " state hidden" : " states hidden") + " by filters" + color: theme.muted + font.pixelSize: 11 + } + TextButton { + text: controller.showFiltered ? "Hide them" : "Show them" + accent: theme.foreground + onClicked: controller.setShowFiltered(!controller.showFiltered) + } + TextButton { text: "Clear filters"; accent: theme.primary; onClicked: controller.clearFilters() } + } + } + } + + // Empty states. + Column { + anchors.centerIn: parent + spacing: 10 + visible: list.count === 0 + Icon { + anchors.horizontalCenter: parent.horizontalCenter + name: controller.totalCount === 0 ? "playlist_add" : "filter_alt_off" + size: 36 + tintColor: theme.muted + } + Text { + anchors.horizontalCenter: parent.horizontalCenter + text: controller.totalCount === 0 ? "No beamline states yet" : "No states match the filters" + color: theme.foreground + font.pixelSize: 13 + font.weight: Font.DemiBold + } + Text { + anchors.horizontalCenter: parent.horizontalCenter + text: controller.totalCount === 0 ? "States check conditions such as a motor within limits." : controller.hiddenCount + " states are hidden." + color: theme.muted + font.pixelSize: 11 + } + TextButton { + anchors.horizontalCenter: parent.horizontalCenter + text: controller.totalCount === 0 ? "Add state" : "Clear filters" + iconName: controller.totalCount === 0 ? "add" : "" + solid: controller.totalCount === 0 + onClicked: controller.totalCount === 0 ? controller.requestAdd() : controller.clearFilters() + } + } + } + } +} diff --git a/bec_widgets/widgets/services/beamline_states/modern/states_controller.py b/bec_widgets/widgets/services/beamline_states/modern/states_controller.py new file mode 100644 index 00000000..9f4e60b9 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/modern/states_controller.py @@ -0,0 +1,497 @@ +""" +View-independent logic shared by the QML and QWidget beamline-state views. + +:class:`BeamlineStatesController` owns every BEC subscription and backend call (available +states, per-state status, scan interlock) plus the filter state. It publishes the rows to show +through :class:`BeamlineStatesModel`, a flat list model with roles that both views read: QML +through role names, the QWidget view through :meth:`BeamlineStatesModel.row`. +""" + +from __future__ import annotations + +from typing import Any + +from bec_lib import messages +from bec_lib.endpoints import MessageEndpoints +from qtpy.QtCore import ( + Property, + QAbstractListModel, + QByteArray, + QModelIndex, + QObject, + Qt, + Signal, + Slot, +) + +from bec_widgets.widgets.services.beamline_states.dialogs import BEAMLINE_STATE_STATUS_LABELS + +STATUSES = ("invalid", "warning", "unknown", "valid") +DEFAULT_ACCEPTED = ["valid", "warning"] +NO_INFO_LABEL = "No state information available." + +INTERLOCK_SECTION = "interlock" +OTHERS_SECTION = "others" +SECTION_TITLES = { + INTERLOCK_SECTION: "Watched by scan interlock", + OTHERS_SECTION: "Not watched by scan interlock", +} + +ROLES = ( + "name", + "status", + "statusText", + "label", + "device", + "stateType", + "details", + "watched", + "accepted", + "triggerOnWarning", + "triggered", + "section", + "filteredOut", +) + + +class BeamlineStatesModel(QAbstractListModel): + """Flat list of displayed beamline states, updated by key-based diffs.""" + + _ROLE_IDS = {name: Qt.ItemDataRole.UserRole + 1 + i for i, name in enumerate(ROLES)} + + def __init__(self, parent: QObject | None = None) -> None: + super().__init__(parent) + self._rows: list[dict[str, Any]] = [] + + def roleNames(self) -> dict[int, QByteArray]: # noqa: N802 + return {role: QByteArray(name.encode()) for name, role in self._ROLE_IDS.items()} + + def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: # noqa: N802 + return 0 if parent.isValid() else len(self._rows) + + def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole) -> Any: + if not index.isValid() or not 0 <= index.row() < len(self._rows): + return None + row = self._rows[index.row()] + if role == Qt.ItemDataRole.DisplayRole: + return row["name"] + for name, role_id in self._ROLE_IDS.items(): + if role_id == role: + return row[name] + return None + + def row(self, i: int) -> dict[str, Any]: + """Return the row dictionary at position ``i``.""" + return self._rows[i] + + def rows(self) -> list[dict[str, Any]]: + """Return all rows in display order.""" + return list(self._rows) + + def set_rows(self, rows: list[dict[str, Any]]) -> None: + """Replace the rows, emitting minimal remove/insert/move/change notifications.""" + new_names = [row["name"] for row in rows] + new_set = set(new_names) + for i in reversed(range(len(self._rows))): + if self._rows[i]["name"] not in new_set: + self.beginRemoveRows(QModelIndex(), i, i) + self._rows.pop(i) + self.endRemoveRows() + for target, new_row in enumerate(rows): + current = next( + (i for i, row in enumerate(self._rows) if row["name"] == new_row["name"]), None + ) + if current is None: + self.beginInsertRows(QModelIndex(), target, target) + self._rows.insert(target, new_row) + self.endInsertRows() + continue + if current != target: + self.beginMoveRows(QModelIndex(), current, current, QModelIndex(), target) + self._rows.insert(target, self._rows.pop(current)) + self.endMoveRows() + if self._rows[target] != new_row: + changed = [ + self._ROLE_IDS[key] for key in ROLES if self._rows[target][key] != new_row[key] + ] + self._rows[target] = new_row + index = self.index(target, 0) + self.dataChanged.emit(index, index, changed) + + +class BeamlineStatesController(QObject): + """ + BEC-facing logic of the beamline-state views. + + Args: + client: BEC client, used for ``beamline_states`` and the scan-interlock actor. + dispatcher: ``BECDispatcher`` of the hosting widget. + parent: Owning QObject. + """ + + changed = Signal() + error_occurred = Signal(str, str) + add_requested = Signal() + edit_requested = Signal(str) + _interlock_enabled_streamed = Signal(bool) + + def __init__(self, client, dispatcher, parent: QObject | None = None) -> None: + super().__init__(parent) + self.client = client + self._dispatcher = dispatcher + self.model = BeamlineStatesModel(self) + self._configs: dict[str, messages.BeamlineStateConfig] = {} + self._order: list[str] = [] + self._status: dict[str, tuple[str, str]] = {} + self._subscribed: set[str] = set() + self._interlock = client.builtin_actors.scan_interlock + self._interlock_enabled = False + self._watched: dict[str, list[str]] = {} + self._preferred_accepted: dict[str, list[str]] = {} + self._status_filter: set[str] = set() + self._search = "" + self._show_filtered = False + self._hidden_count = 0 + + dispatcher.connect_slot( + self.update_available_states, MessageEndpoints.available_beamline_states() + ) + dispatcher.connect_slot( + self.refresh_interlock, + MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"), + ) + self._interlock_enabled_streamed.connect(self._set_interlock_enabled) + self._enabled_config = getattr(self._interlock, "_enabled", None) + if self._enabled_config is not None and hasattr(self._enabled_config, "subscribe"): + self._enabled_config.subscribe(self._on_enabled_stream) + self.refresh_interlock() + self.refresh_states() + + # ------------------------------------------------------------------ BEC input + + def refresh_states(self) -> None: + """Load the cached list of available states.""" + 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) + + def update_available_states(self, content: dict, _metadata: dict | None = None) -> None: + """Apply ``AvailableBeamlineStatesMessage`` content.""" + configs: list[messages.BeamlineStateConfig] = content.get("states", []) + self._configs = {state.name: state for state in configs} + self._order = [state.name for state in configs] + removed = self._subscribed - set(self._order) + added = [name for name in self._order if name not in self._subscribed] + if removed: + self._dispatcher.disconnect_slot( + self.update_state, [MessageEndpoints.beamline_state(name) for name in removed] + ) + for name in removed: + self._status.pop(name, None) + for name in added: + msg = self.client.connector.get_last(MessageEndpoints.beamline_state(name), key="data") + if msg is not None: + self._store_status(name, msg.content) + if added: + self._dispatcher.connect_slot( + self.update_state, [MessageEndpoints.beamline_state(name) for name in added] + ) + self._subscribed = set(self._order) + self._rebuild() + + def update_state(self, content: dict, _metadata: dict | None = None) -> None: + """Apply a ``BeamlineStateMessage`` for one of the displayed states.""" + name = content.get("name") + if name not in self._configs: + return + if self._store_status(name, content): + self._rebuild() + + def _store_status(self, name: str, content: dict) -> bool: + status = str(content.get("status", "unknown")).lower() + status = status if status in BEAMLINE_STATE_STATUS_LABELS else "unknown" + value = (status, str(content.get("label", NO_INFO_LABEL))) + if self._status.get(name) == value: + return False + self._status[name] = value + return True + + def refresh_interlock(self, *_args) -> None: + """Re-read the scan-interlock configuration from BEC.""" + try: + self._interlock_enabled = bool(self._interlock.enabled) + self._watched = { + name: list(statuses) for name, statuses in self._interlock.states_watched.items() + } + except Exception as exc: # pylint: disable=broad-except + self.error_occurred.emit("Scan Interlock Unavailable", str(exc)) + return + self._rebuild() + + def _on_enabled_stream(self, enabled: bool) -> None: + # Called from the connector thread; hop to the Qt thread through a queued signal. + self._interlock_enabled_streamed.emit(bool(enabled)) + + def _set_interlock_enabled(self, enabled: bool) -> None: + if enabled != self._interlock_enabled: + self._interlock_enabled = enabled + self._rebuild() + + # ------------------------------------------------------------------ derived state + + def status_of(self, name: str) -> tuple[str, str]: + """Return ``(status, label)`` of a state.""" + return self._status.get(name, ("unknown", NO_INFO_LABEL)) + + def _accepted(self, name: str) -> list[str]: + return self._watched.get(name) or self._preferred_accepted.get(name, DEFAULT_ACCEPTED) + + def _is_triggered(self, name: str) -> bool: + accepted = self._watched.get(name) + if not self._interlock_enabled or not accepted: + return False + return self.status_of(name)[0] not in accepted + + def _passes_filters(self, name: str) -> bool: + if self._status_filter and self.status_of(name)[0] not in self._status_filter: + return False + tokens = [token.strip().casefold() for token in self._search.split(",") if token.strip()] + if not tokens: + return True + device = self._device(name) or "" + haystack = f"{name} {device}".casefold() + return any(token in haystack for token in tokens) + + def _device(self, name: str) -> str | None: + config = self._configs.get(name) + device = config.parameters.get("device") if config is not None else None + return str(device) if device else None + + def _make_row(self, name: str, filtered_out: bool) -> dict[str, Any]: + config = self._configs[name] + status, label = self.status_of(name) + accepted = self._accepted(name) + details = [ + {"key": key, "value": str(value)} + for key, value in config.parameters.items() + if value is not None and value != "" + ] + return { + "name": name, + "status": status, + "statusText": BEAMLINE_STATE_STATUS_LABELS[status], + "label": label, + "device": self._device(name) or "", + "stateType": config.state_type, + "details": details, + "watched": name in self._watched, + "accepted": ", ".join(s.upper() for s in accepted), + "triggerOnWarning": "warning" not in accepted, + "triggered": self._is_triggered(name), + "section": INTERLOCK_SECTION if name in self._watched else OTHERS_SECTION, + "filteredOut": filtered_out, + } + + def _rebuild(self) -> None: + rank = {status: i for i, status in enumerate(STATUSES)} + position = {name: i for i, name in enumerate(self._order)} + ordered = sorted( + self._order, + key=lambda name: ( + name not in self._watched, + rank[self.status_of(name)[0]], + position[name], + ), + ) + rows = [] + hidden = 0 + for name in ordered: + # States watched by the interlock are always shown: they can block scans. + passes = name in self._watched or self._passes_filters(name) + if not passes: + hidden += 1 + if not self._show_filtered: + continue + rows.append(self._make_row(name, filtered_out=not passes)) + self._hidden_count = hidden + self.model.set_rows(rows) + self.changed.emit() + + # ------------------------------------------------------------------ properties for views + + @Property(bool, notify=changed) + def interlockArmed(self) -> bool: # noqa: N802 + """Whether the scan interlock is armed.""" + return self._interlock_enabled + + @Property(int, notify=changed) + def watchedCount(self) -> int: # noqa: N802 + """Number of states watched by the scan interlock.""" + return sum(1 for name in self._order if name in self._watched) + + @Property("QStringList", notify=changed) + def blockingStates(self) -> list[str]: # noqa: N802 + """Watched states currently tripping the armed interlock.""" + return [name for name in self._order if self._is_triggered(name)] + + @Property("QVariantMap", notify=changed) + def counts(self) -> dict[str, int]: + """Number of states per status.""" + counts = {status: 0 for status in STATUSES} + for name in self._order: + counts[self.status_of(name)[0]] += 1 + return counts + + @Property(int, notify=changed) + def totalCount(self) -> int: # noqa: N802 + """Number of configured states.""" + return len(self._order) + + @Property("QStringList", notify=changed) + def statusFilter(self) -> list[str]: # noqa: N802 + """Statuses selected in the summary chips; empty means all.""" + return sorted(self._status_filter, key=STATUSES.index) + + @Property(str, notify=changed) + def searchText(self) -> str: # noqa: N802 + """Name/device search text.""" + return self._search + + @Property(int, notify=changed) + def hiddenCount(self) -> int: # noqa: N802 + """Number of states hidden by the filters.""" + return self._hidden_count + + @Property(bool, notify=changed) + def showFiltered(self) -> bool: # noqa: N802 + """Whether states hidden by the filters are shown dimmed.""" + return self._show_filtered + + @Property(bool, notify=changed) + def filtersActive(self) -> bool: # noqa: N802 + """Whether any filter is set.""" + return bool(self._status_filter or self._search.strip()) + + # ------------------------------------------------------------------ actions + + @Slot(str) + def toggleStatusFilter(self, status: str) -> None: # noqa: N802 + """Add or remove a status from the status filter.""" + self._status_filter ^= {status} + self._rebuild() + + @Slot(str) + def setSearchText(self, text: str) -> None: # noqa: N802 + """Filter by state name or device (comma-separated tokens).""" + if text != self._search: + self._search = text + self._rebuild() + + @Slot(bool) + def setShowFiltered(self, show: bool) -> None: # noqa: N802 + """Show or hide the states hidden by the filters.""" + self._show_filtered = bool(show) + self._rebuild() + + @Slot() + def clearFilters(self) -> None: # noqa: N802 + """Reset all filters.""" + self._status_filter = set() + self._search = "" + self._show_filtered = False + self._rebuild() + + @Slot(bool) + def setInterlockArmed(self, armed: bool) -> None: # noqa: N802 + """Arm or disarm the scan interlock.""" + try: + self._interlock.enabled = bool(armed) + except Exception as exc: # pylint: disable=broad-except + self.error_occurred.emit("Cannot Toggle Scan Interlock", str(exc)) + self.refresh_interlock() + return + self._interlock_enabled = bool(armed) + self._rebuild() + + @Slot(str) + def toggleWatched(self, name: str) -> None: # noqa: N802 + """Add a state to, or remove it from, the scan interlock.""" + try: + if name in self._watched: + self._interlock.remove_state_from_interlock(name) + else: + self._interlock.add_state_to_interlock(name, self._accepted(name)) + except Exception as exc: # pylint: disable=broad-except + self.error_occurred.emit("Cannot Update Scan Interlock", str(exc)) + return + self.refresh_interlock() + + @Slot(str, bool) + def setTriggerOnWarning(self, name: str, trigger: bool) -> None: # noqa: N802 + """Choose whether WARNING also trips the interlock for ``name``.""" + accepted = ["valid"] if trigger else list(DEFAULT_ACCEPTED) + self._preferred_accepted[name] = accepted + if name not in self._watched: + self._rebuild() + return + try: + self._interlock.add_state_to_interlock(name, accepted) + except Exception as exc: # pylint: disable=broad-except + self.error_occurred.emit("Cannot Update Scan Interlock", str(exc)) + return + self.refresh_interlock() + + def set_preferred_accepted(self, name: str, accepted: list[str]) -> None: + """Remember the accepted statuses for a state not yet watched by the interlock.""" + self._preferred_accepted[name] = list(accepted) + + @Slot(str) + def removeState(self, name: str) -> None: # noqa: N802 + """Delete a beamline state from BEC (the view asks for confirmation first).""" + try: + self.client.beamline_states.delete(name) + except Exception as exc: # pylint: disable=broad-except + self.error_occurred.emit("Cannot Remove State", str(exc)) + + @Slot() + def requestAdd(self) -> None: # noqa: N802 + """Ask the hosting widget to open the add-state dialog.""" + self.add_requested.emit() + + @Slot(str) + def requestEdit(self, name: str) -> None: # noqa: N802 + """Ask the hosting widget to open the edit dialog for ``name``.""" + self.edit_requested.emit(name) + + def config(self, name: str) -> messages.BeamlineStateConfig | None: + """Return the configuration of a state.""" + return self._configs.get(name) + + def state_summary(self) -> dict[str, dict[str, str]]: + """Return every configured state with its ``status`` and ``label``.""" + return { + name: {"status": status, "label": label} + for name in self._order + for status, label in [self.status_of(name)] + } + + def cleanup(self) -> None: + """Release all BEC subscriptions.""" + if self._enabled_config is not None: + self._enabled_config.unsubscribe(self._on_enabled_stream) + self._enabled_config = None + self._dispatcher.disconnect_slot( + self.update_available_states, MessageEndpoints.available_beamline_states() + ) + self._dispatcher.disconnect_slot( + self.refresh_interlock, + MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"), + ) + if self._subscribed: + self._dispatcher.disconnect_slot( + self.update_state, + [MessageEndpoints.beamline_state(name) for name in self._subscribed], + ) + self._subscribed = set() diff --git a/tests/unit_tests/test_beamline_states_modern.py b/tests/unit_tests/test_beamline_states_modern.py new file mode 100644 index 00000000..08f41453 --- /dev/null +++ b/tests/unit_tests/test_beamline_states_modern.py @@ -0,0 +1,212 @@ +import pytest +from bec_lib import bl_states, messages + +from bec_widgets.widgets.services.beamline_states.modern import actions +from bec_widgets.widgets.services.beamline_states.modern.beamline_states_qml import ( + BeamlineStatesQML, +) +from bec_widgets.widgets.services.beamline_states.modern.beamline_states_widget import ( + BeamlineStatesWidget, +) + +from .conftest import create_widget + + +class _FakeScanInterlock: + def __init__(self, enabled=False, states_watched=None): + self.enabled = enabled + self._states = {name: list(v) for name, v in (states_watched or {}).items()} + self.added = [] + self.removed = [] + + @property + def states_watched(self): + return {name: list(v) for name, v in self._states.items()} + + def add_state_to_interlock(self, state_name, required_value="valid"): + self.added.append((state_name, list(required_value))) + self._states[state_name] = list(required_value) + + def remove_state_from_interlock(self, state_name): + self.removed.append(state_name) + self._states.pop(state_name, None) + + +def _limits(name, device="samx"): + config = bl_states.DeviceWithinLimitsState.CONFIG_CLASS( + name=name, device=device, signal=device, low_limit=0.0, high_limit=10.0, tolerance=0.1 + ) + return messages.BeamlineStateConfig( + name=name, + state_type="DeviceWithinLimitsState", + parameters=config.model_dump(exclude={"name"}), + ) + + +@pytest.fixture(params=[BeamlineStatesQML, BeamlineStatesWidget], ids=["qml", "qwidget"]) +def view(request, qtbot, mocked_client): + widget = create_widget(qtbot, request.param, client=mocked_client) + widget.controller._interlock = _FakeScanInterlock( + enabled=True, states_watched={"samx_ok": ["valid", "warning"]} + ) + widget.controller.refresh_interlock() + widget.controller.update_available_states( + {"states": [_limits("samx_ok"), _limits("samy_ok", "samy"), _limits("temp_ok", "temp")]}, {} + ) + return widget + + +def _names(widget): + return [row["name"] for row in widget.controller.model.rows()] + + +def _status(widget, name, status, label="label"): + widget.controller.update_state({"name": name, "status": status, "label": label}, {}) + + +def test_rows_grouped_watched_first_then_by_severity(view): + _status(view, "temp_ok", "invalid") + _status(view, "samy_ok", "valid") + assert _names(view) == ["samx_ok", "temp_ok", "samy_ok"] + sections = [row["section"] for row in view.controller.model.rows()] + assert sections == ["interlock", "others", "others"] + + +def test_state_summary_includes_all_states(view): + _status(view, "samy_ok", "warning", "close to limit") + view.controller.toggleStatusFilter("warning") + summary = view.state_summary() + assert set(summary) == {"samx_ok", "samy_ok", "temp_ok"} + assert summary["samy_ok"] == {"status": "warning", "label": "close to limit"} + assert summary["temp_ok"]["status"] == "unknown" + + +def test_status_chip_filter_keeps_watched_states(view): + _status(view, "samy_ok", "valid") + _status(view, "temp_ok", "invalid") + view.controller.toggleStatusFilter("invalid") + assert _names(view) == ["samx_ok", "temp_ok"] + assert view.controller.hiddenCount == 1 + view.controller.setShowFiltered(True) + rows = {row["name"]: row for row in view.controller.model.rows()} + assert rows["samy_ok"]["filteredOut"] is True + view.clear_filters() + assert view.controller.hiddenCount == 0 + assert len(_names(view)) == 3 + + +def test_search_matches_name_or_device(view): + view.controller.setSearchText("temp") + assert _names(view) == ["samx_ok", "temp_ok"] + view.controller.setSearchText("samy, temp") + assert set(_names(view)) == {"samx_ok", "samy_ok", "temp_ok"} + + +def test_triggered_state_blocks_banner(view): + # A watched state without a status yet (unknown) is not accepted, so it blocks scans. + assert view.controller.blockingStates == ["samx_ok"] + _status(view, "samx_ok", "valid") + assert view.controller.blockingStates == [] + _status(view, "samx_ok", "invalid") + assert view.controller.blockingStates == ["samx_ok"] + assert view.controller.model.rows()[0]["triggered"] is True + view.controller.setInterlockArmed(False) + assert view.controller._interlock.enabled is False + assert view.controller.blockingStates == [] + + +def test_lock_toggles_interlock_membership(view): + view.controller.toggleWatched("samy_ok") + assert view.controller._interlock.added == [("samy_ok", ["valid", "warning"])] + assert "samy_ok" in [r["name"] for r in view.controller.model.rows() if r["watched"]] + view.controller.toggleWatched("samx_ok") + assert view.controller._interlock.removed == ["samx_ok"] + + +def test_trigger_on_warning_reenrolls_watched_state(view): + view.controller.setTriggerOnWarning("samx_ok", True) + assert view.controller._interlock.added == [("samx_ok", ["valid"])] + view.controller.setTriggerOnWarning("samy_ok", True) + assert len(view.controller._interlock.added) == 1 + view.controller.toggleWatched("samy_ok") + assert view.controller._interlock.added[-1] == ("samy_ok", ["valid"]) + + +class _StateClient: + def __init__(self): + self.parameters = None + + def update_parameters(self, **kwargs): + self.parameters = kwargs + + +class _StateManager: + def __init__(self): + self.deleted = None + self.samx_ok = _StateClient() + + def delete(self, state_name): + self.deleted = state_name + + +def test_remove_state_calls_backend(view): + view.controller.client.beamline_states = _StateManager() + view.controller.removeState("samy_ok") + assert view.controller.client.beamline_states.deleted == "samy_ok" + + +def test_removed_state_disappears(view): + view.controller.update_available_states({"states": [_limits("samx_ok")]}, {}) + assert _names(view) == ["samx_ok"] + assert view.controller._subscribed == {"samx_ok"} + + +def test_edit_pushes_parameters(view, monkeypatch): + + class _Dialog: + def __init__(self, name, config_class, parameters, parent, client=None): + data = dict(parameters, high_limit=20.0) + self.result_config = config_class(name=name, **data) + + def exec(self): + return actions.QDialog.DialogCode.Accepted + + def cleanup(self): + pass + + def deleteLater(self): + pass + + monkeypatch.setattr(actions, "EditBeamlineStateDialog", _Dialog) + view.controller.client.beamline_states = _StateManager() + view.controller.requestEdit("samx_ok") + assert view.controller.client.beamline_states.samx_ok.parameters["high_limit"] == 20.0 + + +def test_qwidget_cards_follow_rows(qtbot, mocked_client): + widget = create_widget(qtbot, BeamlineStatesWidget, client=mocked_client) + widget.controller._interlock = _FakeScanInterlock() + widget.controller.update_available_states({"states": [_limits("a"), _limits("b")]}, {}) + card_a = widget._cards["a"] + widget.controller.update_state({"name": "b", "status": "invalid", "label": "x"}, {}) + assert widget._cards["a"] is card_a + assert widget._list_layout.indexOf(widget._cards["b"]) < widget._list_layout.indexOf(card_a) + widget._toggle_expanded("a") + assert card_a.is_expanded() + widget.collapse_all() + assert not card_a.is_expanded() + + +def test_qml_view_loads_and_collapses(qtbot, mocked_client): + widget = create_widget(qtbot, BeamlineStatesQML, client=mocked_client) + root = widget.view.rootObject() + assert root is not None + root.setProperty("expanded", {"a": True}) + + def expanded(): + value = root.property("expanded") + return value.toVariant() if hasattr(value, "toVariant") else value + + assert expanded() == {"a": True} + widget.collapse_all() + assert expanded() == {}