diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 1f84fa5a..5bf8e6dd 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -32,6 +32,8 @@ _Widgets = { "BECQueue": "BECQueue", "BECShell": "BECShell", "BECStatusBox": "BECStatusBox", + "BeamlineStateManager": "BeamlineStateManager", + "BeamlineStatePill": "BeamlineStatePill", "BecConsole": "BecConsole", "DapComboBox": "DapComboBox", "DeviceBrowser": "DeviceBrowser", @@ -717,6 +719,110 @@ class BaseROI(RPCBase): """ +class BeamlineStateManager(RPCBase): + """Widget displaying and managing all BEC beamline states.""" + + _IMPORT_MODULE = "bec_widgets.widgets.services.beamline_states.beamline_state_pill" + + @property + @rpc_call + def idle_card_background(self) -> "bool": + """ + Whether idle collapsed pills keep the status-tinted card background. + """ + + @rpc_call + def set_idle_card_background(self, enabled: "bool") -> "None": + """ + Set whether idle collapsed pills keep the status-tinted card background. + """ + + @rpc_call + def refresh_states(self) -> "None": + """ + Fetch the latest cached available beamline states and update the list immediately. + """ + + @rpc_call + def clear_filters(self) -> "None": + """ + None + """ + + @rpc_call + def remove(self): + """ + Cleanup the BECConnector + """ + + @rpc_call + def attach(self): + """ + None + """ + + @rpc_call + def detach(self): + """ + Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget. + """ + + @rpc_timeout(None) + @rpc_call + def screenshot(self, file_name: "str | None" = None): + """ + Take a screenshot of the dock area and save it to a file. + """ + + +class BeamlineStatePill(RPCBase): + """Compact widget showing one BEC beamline state.""" + + _IMPORT_MODULE = "bec_widgets.widgets.services.beamline_states.beamline_state_pill" + + @property + @rpc_call + def state_name(self) -> "str | None": + """ + Name of the BEC beamline state displayed by this pill. + """ + + @rpc_call + def set_state_name(self, state_name: "str | None", title: "str | None" = None) -> "None": + """ + Set the BEC beamline state this pill displays. + + Args: + state_name: State name as published by ``AvailableBeamlineStatesMessage``. + title: Optional human-readable title for the state. + """ + + @rpc_call + def remove(self): + """ + Cleanup the BECConnector + """ + + @rpc_call + def attach(self): + """ + None + """ + + @rpc_call + def detach(self): + """ + Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget. + """ + + @rpc_timeout(None) + @rpc_call + def screenshot(self, file_name: "str | None" = None): + """ + Take a screenshot of the dock area and save it to a file. + """ + + class BecConsole(RPCBase): """A console widget with access to a shared registry of terminals, such that instances can be moved around.""" diff --git a/bec_widgets/cli/designer_plugins.py b/bec_widgets/cli/designer_plugins.py index bdbdcb6f..6315596e 100644 --- a/bec_widgets/cli/designer_plugins.py +++ b/bec_widgets/cli/designer_plugins.py @@ -19,6 +19,14 @@ designer_plugins = { "BECShell": ("bec_widgets.widgets.editors.bec_console.bec_console", "BECShell"), "BECSpinBox": ("bec_widgets.widgets.utility.spinbox.decimal_spinbox", "BECSpinBox"), "BECStatusBox": ("bec_widgets.widgets.services.bec_status_box.bec_status_box", "BECStatusBox"), + "BeamlineStateManager": ( + "bec_widgets.widgets.services.beamline_states.beamline_state_pill", + "BeamlineStateManager", + ), + "BeamlineStatePill": ( + "bec_widgets.widgets.services.beamline_states.beamline_state_pill", + "BeamlineStatePill", + ), "BecConsole": ("bec_widgets.widgets.editors.bec_console.bec_console", "BecConsole"), "ColorButton": ("bec_widgets.widgets.utility.visual.color_button.color_button", "ColorButton"), "ColorButtonNative": ( @@ -118,6 +126,8 @@ widget_icons = { "BECShell": "hub", "BECSpinBox": "123", "BECStatusBox": "widgets", + "BeamlineStateManager": "format_list_bulleted", + "BeamlineStatePill": "info", "BecConsole": "terminal", "ColorButton": "colors", "ColorButtonNative": "colors", diff --git a/bec_widgets/widgets/containers/dock_area/dock_area.py b/bec_widgets/widgets/containers/dock_area/dock_area.py index 01e0656e..acd03392 100644 --- a/bec_widgets/widgets/containers/dock_area/dock_area.py +++ b/bec_widgets/widgets/containers/dock_area/dock_area.py @@ -385,6 +385,11 @@ class BECDockArea(DockAreaWidget): "bec_shell": (widget_icons["BECShell"], "Add BEC Shell", "BECShell"), "sbb_monitor": (widget_icons["SBBMonitor"], "Add SBB Monitor", "SBBMonitor"), "log_panel": (widget_icons["LogPanel"], "Add LogPanel", "LogPanel"), + "beamline_state_manager": ( + widget_icons["BeamlineStateManager"], + "Add Beamline State Manager", + "BeamlineStateManager", + ), } # Create expandable menu actions (original behavior) diff --git a/bec_widgets/widgets/containers/main_window/main_window.py b/bec_widgets/widgets/containers/main_window/main_window.py index 84b92de7..83a18b04 100644 --- a/bec_widgets/widgets/containers/main_window/main_window.py +++ b/bec_widgets/widgets/containers/main_window/main_window.py @@ -413,6 +413,8 @@ class BECMainWindow(BECWidget, QMainWindow): apply_theme(theme) # emits theme_updated and applies palette globally def event(self, event): + if not isinstance(event, QEvent): + return False if event.type() == QEvent.Type.StatusTip: return True return super().event(event) diff --git a/bec_widgets/widgets/services/beamline_states/__init__.py b/bec_widgets/widgets/services/beamline_states/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.pyproject b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.pyproject new file mode 100644 index 00000000..340bae9a --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.pyproject @@ -0,0 +1 @@ +{'files': ['beamline_state_pill.py']} \ No newline at end of file diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_manager_plugin.py b/bec_widgets/widgets/services/beamline_states/beamline_state_manager_plugin.py new file mode 100644 index 00000000..ef7b28e0 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager_plugin.py @@ -0,0 +1,57 @@ +# Copyright (C) 2022 The Qt Company Ltd. +# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause + +from qtpy.QtDesigner import QDesignerCustomWidgetInterface +from qtpy.QtWidgets import QWidget + +from bec_widgets.utils.bec_designer import designer_material_icon +from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStateManager + +DOM_XML = """ + + + + +""" + + +class BeamlineStateManagerPlugin(QDesignerCustomWidgetInterface): # pragma: no cover + def __init__(self): + super().__init__() + self._form_editor = None + + def createWidget(self, parent): + if parent is None: + return QWidget() + t = BeamlineStateManager(parent) + return t + + def domXml(self): + return DOM_XML + + def group(self): + return "" + + def icon(self): + return designer_material_icon(BeamlineStateManager.ICON_NAME) + + def includeFile(self): + return "beamline_state_manager" + + def initialize(self, form_editor): + self._form_editor = form_editor + + def isContainer(self): + return False + + def isInitialized(self): + return self._form_editor is not None + + def name(self): + return "BeamlineStateManager" + + def toolTip(self): + return "" + + def whatsThis(self): + return self.toolTip() diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py new file mode 100644 index 00000000..01dba742 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py @@ -0,0 +1,1491 @@ +from __future__ import annotations + +import sys +from typing import Any + +from bec_lib.endpoints import MessageEndpoints +from bec_qthemes import material_icon +from qtpy.QtCore import QAbstractListModel, QModelIndex, QSize, Qt, Signal +from qtpy.QtGui import QColor, QMouseEvent, QPalette +from qtpy.QtWidgets import ( + QAbstractItemView, + QApplication, + QCheckBox, + QDialog, + QFormLayout, + QGraphicsDropShadowEffect, + QHBoxLayout, + QLabel, + QLineEdit, + QListView, + QMessageBox, + QPushButton, + QSizePolicy, + QStyledItemDelegate, + QStyleOptionViewItem, + QToolButton, + 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 SafeProperty, SafeSlot +from bec_widgets.utils.toolbars.actions import MaterialIconAction +from bec_widgets.utils.toolbars.bundles import ToolbarBundle +from bec_widgets.utils.toolbars.toolbar import ModularToolBar +from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox +from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox +from bec_widgets.widgets.services.beamline_states.dialogs import ( + BEAMLINE_STATE_STATUS_LABELS, + AddBeamlineStateDialog, + DeviceFilterDialog, + StatusFilterDialog, +) +from bec_widgets.widgets.utility.spinbox.decimal_spinbox import BECSpinBox + +__all__ = [ + "AddBeamlineStateDialog", + "BeamlineStateManager", + "BeamlineStatePill", + "DeviceFilterDialog", + "StatusFilterDialog", +] + + +def _coerce_bool(value: Any) -> bool: + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on"} + return bool(value) + + +class _BeamlineStatePillHeader(QWidget): + """Header surface responsible for pill click gestures.""" + + clicked = Signal() + + def __init__(self, parent: QWidget | None = None) -> None: + super().__init__(parent) + self._press_active = False + + def mousePressEvent(self, event: QMouseEvent) -> None: # noqa: N802 + if event.button() == Qt.MouseButton.LeftButton: + self._press_active = True + event.ignore() + return + super().mousePressEvent(event) + + def mouseReleaseEvent(self, event: QMouseEvent) -> None: # noqa: N802 + clicked = ( + event.button() == Qt.MouseButton.LeftButton + and self._press_active + and self.rect().contains(event.pos()) + ) + self._press_active = False + if clicked: + self.clicked.emit() + event.accept() + return + super().mouseReleaseEvent(event) + + +class BeamlineStatePill(BECWidget, QWidget): + """ + Compact widget showing one BEC beamline state. + + The pill subscribes to ``MessageEndpoints.beamline_state(state_name)`` and updates whenever + a ``BeamlineStateMessage`` is published for that state. + """ + + PLUGIN = True + ICON_NAME = "info" + USER_ACCESS = ["state_name", "set_state_name", "remove", "attach", "detach", "screenshot"] + + state_changed = Signal(str, str, str) + update_requested = Signal(str, dict) + remove_requested = Signal(str) + size_hint_changed = Signal() + + _STATUS_LABELS = BEAMLINE_STATE_STATUS_LABELS + _STATUS_ICONS = { + "valid": "check_circle", + "invalid": "cancel", + "warning": "warning", + "unknown": "help", + } + _SETTINGS_FIELD_WIDTH = 180 + + def __init__( + self, + parent: QWidget | None = None, + state_name: str | None = None, + title: str | 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, theme_update=True, **kwargs + ) + self.setObjectName("BeamlineStatePill") + self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True) + self._state_name: str | None = None + self._title: str | None = None + self._state_config: dict[str, Any] = {} + self._status = "unknown" + self._label = "No state information available." + self._expanded = False + self._idle_card_background = False + self._populating_settings = False + self._settings_baseline: dict[str, Any] = {} + self._settings_dirty_fields: set[str] = set() + + self._shadow = QGraphicsDropShadowEffect(self) + self._shadow.setBlurRadius(18) + self._shadow.setOffset(0, 2) + self._shadow.setColor(QColor(0, 0, 0, 120)) + self._shadow.setEnabled(False) + self.setGraphicsEffect(self._shadow) + + self._header = _BeamlineStatePillHeader(self) + self._header.setObjectName("beamline_state_header") + self._header.setCursor(Qt.CursorShape.PointingHandCursor) + self._header.clicked.connect(self._toggle_expanded) + + self._stripe = QWidget(self) + self._stripe.setObjectName("beamline_state_stripe") + self._stripe.setFixedWidth(4) + self._stripe.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Expanding) + self._stripe.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + + self._icon_label = QLabel(self) + self._icon_label.setObjectName("beamline_state_icon") + self._icon_label.setFixedSize(32, 32) + self._icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + self._icon_label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + self._name_label = QLabel(self) + self._name_label.setObjectName("beamline_state_name") + self._name_label.setTextFormat(Qt.TextFormat.PlainText) + self._name_label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + self._status_label = QLabel(self) + self._status_label.setObjectName("beamline_state_status") + self._status_label.setTextFormat(Qt.TextFormat.PlainText) + self._status_label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + self._detail_label = QLabel(self) + self._detail_label.setObjectName("beamline_state_detail") + self._detail_label.setTextFormat(Qt.TextFormat.PlainText) + self._detail_label.setWordWrap(True) + self._detail_label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + self._expand_button = QToolButton(self) + self._expand_button.setObjectName("beamline_state_expand") + self._expand_button.setAutoRaise(True) + self._expand_button.setCursor(Qt.CursorShape.PointingHandCursor) + self._expand_button.clicked.connect(self._toggle_expanded) + + text_layout = QVBoxLayout() + text_layout.setContentsMargins(0, 0, 0, 0) + text_layout.setSpacing(1) + text_layout.addWidget(self._name_label) + text_layout.addWidget(self._detail_label) + + header_layout = QHBoxLayout(self._header) + header_layout.setContentsMargins(10, 8, 12, 8) + header_layout.setSpacing(10) + header_layout.addWidget(self._stripe) + header_layout.addWidget(self._icon_label) + header_layout.addLayout(text_layout, 1) + header_layout.addWidget(self._status_label, 0, Qt.AlignmentFlag.AlignRight) + header_layout.addWidget(self._expand_button) + + self._settings = QWidget(self) + self._settings.setObjectName("beamline_state_settings") + self._settings.setVisible(False) + self._state_type_value = QLabel(self._settings) + self._name_value = QLabel(self._settings) + self._title_edit = QLineEdit(self._settings) + self._device_edit = DeviceComboBox(parent=self._settings, client=client) + self._signal_edit = SignalComboBox( + parent=self._settings, client=client, require_device=True + ) + self._low_limit_enabled, self._low_limit = self._create_optional_limit_row() + self._high_limit_enabled, self._high_limit = self._create_optional_limit_row() + self._tolerance = BECSpinBox(self._settings) + self._configure_settings_spinbox(self._tolerance) + self._device_edit.device_selected.connect(self._on_settings_device_selected) + self._device_edit.device_reset.connect(self._on_settings_device_reset) + self._device_edit.currentTextChanged.connect(self._on_text_settings_changed) + self._signal_edit.currentTextChanged.connect(self._on_text_settings_changed) + self._title_edit.textChanged.connect(self._on_text_settings_changed) + self._low_limit_enabled.toggled.connect(self._on_low_limit_enabled_changed) + self._high_limit_enabled.toggled.connect(self._on_high_limit_enabled_changed) + self._low_limit.valueChanged.connect(self._on_numeric_settings_changed) + self._high_limit.valueChanged.connect(self._on_numeric_settings_changed) + self._tolerance.valueChanged.connect(self._on_numeric_settings_changed) + + for field in self._settings_input_fields(): + field.setMinimumWidth(self._SETTINGS_FIELD_WIDTH) + field.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + + button_layout = QHBoxLayout() + button_layout.setContentsMargins(0, 0, 0, 0) + button_layout.setSpacing(8) + self._update_button = QPushButton("Update", self._settings) + self._update_button.setIcon(material_icon("save", convert_to_pixmap=False)) + self._revert_button = QPushButton("Revert", self._settings) + self._revert_button.setIcon(material_icon("undo", convert_to_pixmap=False)) + self._remove_button = QPushButton("Remove", self._settings) + self._remove_button.setObjectName("beamline_state_remove_button") + self._remove_button.setIcon(material_icon("delete", convert_to_pixmap=False)) + self._update_button.clicked.connect(self._emit_update_requested) + self._revert_button.clicked.connect(self._revert_settings) + self._remove_button.clicked.connect(self._emit_remove_requested) + button_layout.addWidget(self._update_button) + button_layout.addWidget(self._revert_button) + button_layout.addWidget(self._remove_button) + button_layout.addStretch(1) + + self._settings_form = QFormLayout() + self._settings_form.setContentsMargins(12, 8, 12, 8) + self._settings_form.setHorizontalSpacing(10) + self._settings_form.setVerticalSpacing(8) + self._settings_form.setFieldGrowthPolicy( + QFormLayout.FieldGrowthPolicy.AllNonFixedFieldsGrow + ) + self._settings_form.setLabelAlignment(Qt.AlignmentFlag.AlignRight) + self._settings_form.addRow("Type", self._state_type_value) + self._settings_form.addRow("Name", self._name_value) + self._settings_form.addRow("Title", self._title_edit) + self._settings_form.addRow("Device", self._device_edit) + self._settings_form.addRow("Signal", self._signal_edit) + self._settings_form.addRow("Low limit", self._low_limit.parentWidget()) + self._settings_form.addRow("High limit", self._high_limit.parentWidget()) + self._settings_form.addRow("Tolerance", self._tolerance) + self._settings_form.addRow(button_layout) + + self._limit_widgets = ( + self._low_limit.parentWidget(), + self._high_limit.parentWidget(), + self._tolerance, + ) + + settings_layout = QVBoxLayout(self._settings) + settings_layout.setContentsMargins(0, 0, 0, 0) + settings_layout.setSpacing(0) + settings_layout.addLayout(self._settings_form) + + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + layout.addWidget(self._header) + layout.addWidget(self._settings) + self.setLayout(layout) + + self.setMinimumWidth(0) + self._header.setMinimumWidth(0) + self._settings.setMinimumWidth(0) + self.setMinimumHeight(58) + self.set_state_name(state_name, title=title) + self._mark_settings_clean_from_current() + + @SafeProperty(str, default=None) + def state_name(self) -> str | None: + """Name of the BEC beamline state displayed by this pill.""" + return self._state_name + + @state_name.setter + def state_name(self, state_name: str | None) -> None: + self.set_state_name(state_name) + + def set_state_name(self, state_name: str | None, title: str | None = None) -> None: + """ + Set the BEC beamline state this pill displays. + + Args: + state_name: State name as published by ``AvailableBeamlineStatesMessage``. + title: Optional human-readable title for the state. + """ + if state_name == self._state_name and title == self._title: + return + + if self._state_name is not None: + self.bec_dispatcher.disconnect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + + self._state_name = state_name + self._title = title + self._name_label.setText(title or state_name or "Beamline state") + + if self._state_name is None: + self._set_visual_state("unknown", "No beamline state selected.") + return + + self._set_visual_state("unknown", "No state information available.") + self._refresh_latest_state() + self.bec_dispatcher.connect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + + def set_state_config(self, state_config: dict[str, Any]) -> None: + """Set the editable BEC state configuration displayed by the expanded panel.""" + self._state_config = state_config + self._populate_settings() + self._mark_settings_clean_from_current() + + @SafeProperty(bool, default=False) + def idle_card_background(self) -> bool: + """ + Whether idle collapsed pills keep the status-tinted card background. + """ + return self._idle_card_background + + @idle_card_background.setter + def idle_card_background(self, enabled: bool) -> None: + self._idle_card_background = _coerce_bool(enabled) + self._apply_visual_state() + + def set_idle_card_background(self, enabled: bool) -> None: + """Set whether idle collapsed pills keep the status-tinted card background.""" + self.idle_card_background = enabled + + def _refresh_latest_state(self) -> None: + if self._state_name is None: + return + msg_container = self.client.connector.get_last( + MessageEndpoints.beamline_state(self._state_name) + ) + if not msg_container: + return + data = msg_container.get("data") if isinstance(msg_container, dict) else None + content = getattr(data, "content", data) + if isinstance(content, dict): + self.update_state(content, getattr(data, "metadata", {})) + + @SafeSlot(dict, dict) + def update_state( + self, content: dict[str, Any], _metadata: dict[str, Any] | None = None + ) -> None: + """ + Update this pill from a ``BeamlineStateMessage`` content dictionary. + """ + name = content.get("name") + if self._state_name is not None and name and name != self._state_name: + return + + status = str(content.get("status", "unknown")).lower() + label = str(content.get("label", "No state information available.")) + self._set_visual_state(status, label) + self.state_changed.emit(self._state_name or str(name or ""), status, label) + + @SafeSlot(str) + def apply_theme(self, _theme: str) -> None: + self._apply_visual_state() + + def _set_visual_state(self, status: str, label: str) -> None: + status = status if status in self._STATUS_LABELS else "unknown" + self._status = status + self._label = label + + self._apply_visual_state() + + def _apply_visual_state(self) -> None: + colors = self._state_colors(self._status) + accent = colors["accent"] + on_accent = colors["on_accent"] + active_card = self._expanded + border = colors["border"] if self._idle_card_background else "transparent" + background = colors["background"] if self._idle_card_background else "transparent" + if active_card: + background = self._card_gradient(colors) + border = colors["card_border"] + hover_background = self._card_gradient(colors) + self._shadow.setColor(QColor(colors["shadow"])) + self._shadow.setBlurRadius(int(colors["shadow_blur"])) + self._shadow.setOffset(0, int(colors["shadow_y_offset"])) + self._shadow.setEnabled(active_card) + + icon_name = self._STATUS_ICONS[self._status] + self._icon_label.setPixmap( + material_icon(icon_name, size=(20, 20), color=on_accent, filled=True) + ) + expand_icon = "expand_less" if self._expanded else "expand_more" + self._expand_button.setIcon(material_icon(expand_icon, convert_to_pixmap=False)) + self._status_label.setText(self._STATUS_LABELS[self._status]) + self._detail_label.setText(self._label) + self.setToolTip(self._label) + self.setStyleSheet( + "#BeamlineStatePill {" + f"background: {background};" + f"border: 1px solid {border};" + f"border-radius: {'12px' if active_card else '8px'};" + "}" + "#BeamlineStatePill:hover {" + f"background: {hover_background};" + f"border: 1px solid {colors['card_border']};" + "border-radius: 12px;" + "}" + "QWidget#beamline_state_header {" + "background: transparent;" + "}" + "QWidget#beamline_state_stripe {" + f"background-color: {accent};" + "border-radius: 2px;" + "}" + "QLabel#beamline_state_icon {" + f"background-color: {accent};" + "border-radius: 16px;" + "}" + "QLabel#beamline_state_name {" + f"color: {colors['foreground']};" + "font-weight: 600;" + "}" + "QLabel#beamline_state_status {" + f"color: {accent};" + "font-weight: 700;" + "font-size: 13px;" + "}" + "QLabel#beamline_state_detail {" + f"color: {colors['muted']};" + "font-size: 11px;" + "}" + "QWidget#beamline_state_settings {" + "background: transparent;" + f"border-top: 1px solid {colors['border']};" + "}" + '*[beamlineStateDirty="true"] {' + f"background-color: {colors['dirty_background']};" + f"border: 1px solid {colors['dirty_border']};" + "border-radius: 4px;" + "}" + "QPushButton#beamline_state_remove_button {" + "background-color: #cc181e;" + "border: 1px solid #cc181e;" + "color: white;" + "border-radius: 4px;" + "padding: 4px 10px;" + "}" + "QPushButton#beamline_state_remove_button:hover {" + "background-color: #a91419;" + "border-color: #a91419;" + "}" + ) + + @SafeSlot() + def _toggle_expanded(self) -> None: + self.set_expanded(not self._expanded) + + def is_expanded(self) -> bool: + """Return whether the editable settings panel is expanded.""" + return self._expanded + + def set_expanded(self, expanded: bool) -> None: + """Set the editable settings panel expanded state.""" + expanded = bool(expanded) + if expanded == self._expanded: + return + self._expanded = expanded + self._settings.setVisible(expanded) + self._apply_visual_state() + self.size_hint_changed.emit() + + def _populate_settings(self) -> None: + self._populating_settings = True + try: + state_type = self._state_type() + self._state_type_value.setText(state_type or "-") + self._name_value.setText(self._state_name or "-") + self._title_edit.setText(str(self._state_field("title") or "")) + device = str(self._state_field("device") or "") + signal = str(self._state_field("signal") or "") + self._set_settings_device(device) + self._set_settings_signal(signal) + + show_limits = self._show_limit_settings() + for widget in self._limit_widgets: + self._set_form_row_visible(widget, show_limits) + + if not show_limits: + return + + self._set_optional_limit( + self._low_limit_enabled, self._low_limit, self._state_field("low_limit") + ) + self._set_optional_limit( + self._high_limit_enabled, self._high_limit, self._state_field("high_limit") + ) + tolerance = self._state_field("tolerance") + self._tolerance.setValue(float(tolerance) if tolerance is not None else 0.1) + finally: + self._populating_settings = False + self._update_settings_dirty_state() + + def edited_parameters(self) -> dict[str, Any]: + """Return editable parameters from the expanded settings panel.""" + device = self._device_edit.currentText().strip() + signal = self._optional_signal() + if not device: + raise ValueError("Device is required.") + + params: dict[str, Any] = { + "title": self._optional_text(self._title_edit), + "device": device, + "signal": signal, + } + if self._state_type() == "DeviceWithinLimitsState": + params.update( + { + "low_limit": ( + self._low_limit.value() if self._low_limit_enabled.isChecked() else None + ), + "high_limit": ( + self._high_limit.value() if self._high_limit_enabled.isChecked() else None + ), + "tolerance": self._tolerance.value(), + } + ) + return params + + def mark_current_settings_clean(self, parameters: dict[str, Any] | None = None) -> None: + """Mark the current editor values as saved.""" + if parameters is None: + parameters = self._current_settings_values() + self._update_state_config_parameters(parameters) + self._mark_settings_clean_from_current() + + def _mark_settings_clean_from_current(self) -> None: + self._settings_baseline = self._current_settings_values() + self._update_settings_dirty_state() + + def _show_limit_settings(self) -> bool: + return self._state_type() == "DeviceWithinLimitsState" or any( + self._state_field(key) is not None for key in ("low_limit", "high_limit", "tolerance") + ) + + def _current_settings_values(self) -> dict[str, Any]: + values: dict[str, Any] = { + "title": self._optional_text(self._title_edit), + "device": self._device_edit.currentText().strip(), + "signal": self._optional_signal(), + } + if self._show_limit_settings(): + values.update( + { + "low_limit": ( + self._low_limit.value() if self._low_limit_enabled.isChecked() else None + ), + "high_limit": ( + self._high_limit.value() if self._high_limit_enabled.isChecked() else None + ), + "tolerance": self._tolerance.value(), + } + ) + return values + + def _update_state_config_parameters(self, parameters: dict[str, Any]) -> None: + if not self._state_config: + return + state_parameters = self._state_config.get("parameters") + if isinstance(state_parameters, dict): + state_parameters.update(parameters) + else: + self._state_config.update(parameters) + for key in ("title",): + if key in parameters: + self._state_config[key] = parameters[key] + + @SafeSlot() + def _revert_settings(self) -> None: + self._apply_settings_values(self._settings_baseline) + self._update_settings_dirty_state() + + def _apply_settings_values(self, values: dict[str, Any]) -> None: + self._populating_settings = True + try: + self._title_edit.setText(str(values.get("title") or "")) + device = str(values.get("device") or "") + signal = str(values.get("signal") or "") + self._set_settings_device(device) + self._set_settings_signal(signal) + + if not self._show_limit_settings(): + return + + self._set_optional_limit( + self._low_limit_enabled, self._low_limit, values.get("low_limit") + ) + self._set_optional_limit( + self._high_limit_enabled, self._high_limit, values.get("high_limit") + ) + tolerance = values.get("tolerance") + self._tolerance.setValue(float(tolerance) if tolerance is not None else 0.1) + finally: + self._populating_settings = False + + @SafeSlot(str) + def _on_text_settings_changed(self, _value: str) -> None: + self._update_settings_dirty_state() + + @SafeSlot(float) + def _on_numeric_settings_changed(self, _value: float) -> None: + self._update_settings_dirty_state() + + @SafeSlot(bool) + def _on_low_limit_enabled_changed(self, enabled: bool) -> None: + self._low_limit.setEnabled(enabled) + self._update_settings_dirty_state() + + @SafeSlot(bool) + def _on_high_limit_enabled_changed(self, enabled: bool) -> None: + self._high_limit.setEnabled(enabled) + self._update_settings_dirty_state() + + def _update_settings_dirty_state(self) -> None: + if self._populating_settings: + return + + current_values = self._current_settings_values() + fields = set(current_values) | set(self._settings_baseline) + self._settings_dirty_fields = { + field + for field in fields + if not self._settings_values_equal( + current_values.get(field), self._settings_baseline.get(field) + ) + } + + has_changes = bool(self._settings_dirty_fields) + self._update_button.setEnabled(has_changes) + self._revert_button.setEnabled(has_changes) + self._apply_dirty_field_highlights() + + @staticmethod + def _settings_values_equal(left: Any, right: Any) -> bool: + if left is None or right is None: + return left is None and right is None + if isinstance(left, float) or isinstance(right, float): + return abs(float(left) - float(right)) < 1e-9 + return left == right + + def _apply_dirty_field_highlights(self) -> None: + field_widgets = { + "title": self._title_edit, + "device": self._device_edit, + "signal": self._signal_edit, + "low_limit": self._low_limit.parentWidget(), + "high_limit": self._high_limit.parentWidget(), + "tolerance": self._tolerance, + } + for field, widget in field_widgets.items(): + self._set_dirty_property(widget, field in self._settings_dirty_fields) + + @staticmethod + def _set_dirty_property(widget: QWidget | None, dirty: bool) -> None: + if widget is None or widget.property("beamlineStateDirty") == dirty: + return + widget.setProperty("beamlineStateDirty", dirty) + widget.style().unpolish(widget) + widget.style().polish(widget) + widget.update() + + @SafeSlot() + def _emit_update_requested(self) -> None: + if self._state_name is None: + return + if not self._settings_dirty_fields: + return + try: + parameters = self.edited_parameters() + except ValueError as exc: + QMessageBox.warning(self, "Invalid Beamline State", str(exc)) + return + self.update_requested.emit(self._state_name, parameters) + + @SafeSlot() + def _emit_remove_requested(self) -> None: + if self._state_name is None: + return + self.remove_requested.emit(self._state_name) + + def _state_field(self, name: str) -> Any: + parameters = self._state_config.get("parameters") + if isinstance(parameters, dict) and name in parameters: + return parameters.get(name) + return self._state_config.get(name) + + def _state_type(self) -> str: + return str(self._state_config.get("state_type") or self._state_field("state_type") or "") + + @SafeSlot(str) + def _on_settings_device_selected(self, device: str) -> None: + self._signal_edit.set_device(device) + + @SafeSlot() + def _on_settings_device_reset(self) -> None: + self._signal_edit.set_device(None) + + def _set_settings_device(self, device: str) -> None: + if not device: + self._device_edit.setCurrentText("") + self._signal_edit.set_device(None) + return + self._device_edit.set_device(device) + if self._device_edit.currentText() != device: + self._device_edit.setCurrentText(device) + if self._device_edit.is_valid_input: + self._signal_edit.set_device(device) + + def _set_settings_signal(self, signal: str) -> None: + if not signal: + self._signal_edit.setCurrentText("") + return + self._signal_edit.set_signal(signal) + if ( + self._signal_edit.currentText() != signal + and self._signal_edit.get_signal_name() != signal + ): + self._signal_edit.setCurrentText(signal) + + @staticmethod + def _optional_text(line_edit: QLineEdit) -> str | None: + value = line_edit.text().strip() + return value or None + + def _optional_signal(self) -> str | None: + value = self._signal_edit.get_signal_name().strip() + return value or None + + @staticmethod + def _configure_settings_spinbox(spin_box: BECSpinBox) -> None: + spin_box.setRange(-1_000_000_000, 1_000_000_000) + spin_box.setDecimals(6) + + def _create_optional_limit_row(self) -> tuple[QCheckBox, BECSpinBox]: + container = QWidget(self) + container.setObjectName("beamline_state_limit_row") + container.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True) + checkbox = QCheckBox(container) + spin_box = BECSpinBox(container) + self._configure_settings_spinbox(spin_box) + spin_box.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + layout = QHBoxLayout(container) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(8) + layout.addWidget(checkbox) + layout.addWidget(spin_box, 1) + return checkbox, spin_box + + @staticmethod + def _set_optional_limit(checkbox: QCheckBox, spin_box: BECSpinBox, value: Any) -> None: + enabled = value is not None + checkbox.setChecked(enabled) + spin_box.setEnabled(enabled) + if enabled: + spin_box.setValue(float(value)) + + def _settings_input_fields(self) -> tuple[QWidget, ...]: + return ( + self._state_type_value, + self._name_value, + self._title_edit, + self._device_edit, + self._signal_edit, + self._low_limit.parentWidget(), + self._high_limit.parentWidget(), + self._tolerance, + ) + + def _set_form_row_visible(self, widget: QWidget, visible: bool) -> None: + widget.setVisible(visible) + label = self._settings_form.labelForField(widget) + if label is not None: + label.setVisible(visible) + + @classmethod + def _state_colors(cls, status: str) -> dict[str, str]: + app = QApplication.instance() + palette = app.palette() if app is not None else QPalette() + theme = getattr(app, "theme", None) if app is not None else None + light_theme = cls._is_light_theme(theme, palette) + + foreground = cls._theme_color(theme, "FG", palette.text().color()) + on_primary = cls._theme_color(theme, "ON_PRIMARY", QColor("#ffffff")) + + accents = getattr(theme, "accent_colors", None) + warning = getattr(accents, "warning", QColor("#EAC435")) + accent = { + "valid": getattr(accents, "success", QColor("#2CA58D")), + "invalid": getattr(accents, "emergency", QColor("#CC181E")), + "warning": warning, + "unknown": cls._theme_color(theme, "ACCENT_DEFAULT", QColor("#7a7a7a")), + }.get(status, QColor("#7a7a7a")) + + if not light_theme: + card_bg = cls._theme_color(theme, "CARD_BG", palette.window().color()) + border = cls._theme_color(theme, "BORDER", palette.mid().color()) + return { + "accent": accent.name(), + "on_accent": on_primary.name(), + "card_background": card_bg.name(), + "card_border": cls._blend(border, accent, 0.45).name(), + "gradient_accent": cls._rgba(accent, 62), + "gradient_stop": "0.62", + "background": cls._blend(card_bg, accent, 0.10).name(), + "border": cls._blend(border, accent, 0.35).name(), + "dirty_background": cls._blend(card_bg, warning, 0.18).name(), + "dirty_border": cls._blend(border, warning, 0.70).name(), + "foreground": foreground.name(), + "muted": cls._blend(card_bg, foreground, 0.66).name(), + "shadow": "#00000078", + "shadow_blur": "18", + "shadow_y_offset": "2", + } + + accent = accent.darker(118) + card_bg = cls._theme_color(theme, "CARD_BG", QColor("#ffffff")) + border = cls._theme_color(theme, "BORDER", QColor("#d9e2ec")) + muted = QColor("#667085") + card_border = cls._blend(border, accent, 0.34) + dirty_background = QColor("#fff8db") + dirty_border = cls._blend(border, warning, 0.72) + + return { + "accent": accent.name(), + "on_accent": on_primary.name(), + "card_background": card_bg.name(), + "card_border": card_border.name(), + "gradient_accent": cls._rgba(accent, 18), + "gradient_stop": "0.38", + "background": card_bg.name(), + "border": card_border.name(), + "dirty_background": dirty_background.name(), + "dirty_border": dirty_border.name(), + "foreground": foreground.name(), + "muted": muted.name(), + "shadow": "#00000024", + "shadow_blur": "24", + "shadow_y_offset": "3", + } + + @staticmethod + def _is_light_theme(theme: Any, palette: QPalette) -> bool: + theme_name = str(getattr(theme, "theme", "")).lower() + if theme_name in {"light", "dark"}: + return theme_name == "light" + return palette.window().color().lightness() > 128 + + @staticmethod + def _rgba(color: QColor, alpha: int) -> str: + return f"rgba({color.red()}, {color.green()}, {color.blue()}, {max(0, min(255, alpha))})" + + @staticmethod + def _card_gradient(colors: dict[str, str]) -> str: + return ( + "qlineargradient(" + "x1:0, y1:0, x2:1, y2:0, " + f"stop:0 {colors['gradient_accent']}, " + f"stop:{colors['gradient_stop']} {colors['card_background']}, " + f"stop:1 {colors['card_background']}" + ")" + ) + + @staticmethod + def _theme_color(theme: Any, key: str, fallback: QColor) -> QColor: + if theme is None: + return fallback + color = theme.color(key, fallback.name()) + return color if isinstance(color, QColor) else QColor(str(color)) + + @staticmethod + def _blend(base: QColor, overlay: QColor, overlay_alpha: float) -> QColor: + overlay_alpha = max(0.0, min(1.0, overlay_alpha)) + base_alpha = 1.0 - overlay_alpha + return QColor( + round(base.red() * base_alpha + overlay.red() * overlay_alpha), + round(base.green() * base_alpha + overlay.green() * overlay_alpha), + round(base.blue() * base_alpha + overlay.blue() * overlay_alpha), + ) + + def cleanup(self) -> None: + if self._state_name is not None: + self.bec_dispatcher.disconnect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + super().cleanup() + + +class _BeamlineStateListModel(QAbstractListModel): + """Model owning beamline state row identity and configuration data.""" + + NameRole = Qt.ItemDataRole.UserRole + 1 + ConfigRole = Qt.ItemDataRole.UserRole + 2 + + def __init__(self, parent: QWidget | None = None) -> None: + super().__init__(parent) + self._state_order: list[str] = [] + self._state_rows: dict[str, int] = {} + self._state_configs: dict[str, dict[str, Any]] = {} + + def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: # noqa: N802 + return 0 if parent.isValid() else len(self._state_order) + + def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole) -> Any: + if not index.isValid() or not 0 <= index.row() < len(self._state_order): + return None + name = self._state_order[index.row()] + if role in (Qt.ItemDataRole.DisplayRole, self.NameRole): + return name + if role == self.ConfigRole: + return self._state_configs.get(name, {}) + if role == Qt.ItemDataRole.SizeHintRole: + return QSize(360, 58) + return None + + def flags(self, index: QModelIndex) -> Qt.ItemFlag: + if not index.isValid(): + return Qt.ItemFlag.NoItemFlags + return Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable + + def set_states(self, state_configs: list[dict[str, Any]]) -> None: + self.beginResetModel() + self._state_order = [str(state["name"]) for state in state_configs if state.get("name")] + self._state_rows = {name: row for row, name in enumerate(self._state_order)} + self._state_configs = { + str(state["name"]): state for state in state_configs if state.get("name") + } + self.endResetModel() + + def index_for_name(self, name: str) -> QModelIndex: + row = self._state_rows.get(name) + if row is None: + return QModelIndex() + return self.index(row, 0) + + def name_at(self, row: int) -> str: + return self._state_order[row] + + +class _BeamlineStatePillDelegate(QStyledItemDelegate): + """Delegate that provides BeamlineStatePill persistent editors for list rows.""" + + def __init__(self, manager: "BeamlineStateManager") -> None: + super().__init__(manager) + self._manager = manager + + def paint(self, _painter, _option: QStyleOptionViewItem, _index: QModelIndex) -> None: + return + + def createEditor( # noqa: N802 + self, parent: QWidget, _option: QStyleOptionViewItem, index: QModelIndex + ) -> QWidget: + name = index.data(_BeamlineStateListModel.NameRole) + state_config = index.data(_BeamlineStateListModel.ConfigRole) or {} + title = state_config.get("title") or name + pill = BeamlineStatePill( + parent=parent, state_name=name, title=title, client=self._manager.client + ) + pill.idle_card_background = self._manager.idle_card_background + pill.set_state_config(state_config) + pill.state_changed.connect(self._manager._on_pill_state_changed) + pill.update_requested.connect(self._manager._update_state_parameters) + pill.remove_requested.connect(self._manager._remove_state_requested) + pill.size_hint_changed.connect(lambda name=name: self._manager._sync_pill_item_size(name)) + self._manager._state_pills[str(name)] = pill + return pill + + def setEditorData(self, editor: QWidget, index: QModelIndex) -> None: # noqa: N802 + if not isinstance(editor, BeamlineStatePill): + return + name = index.data(_BeamlineStateListModel.NameRole) + state_config = index.data(_BeamlineStateListModel.ConfigRole) or {} + title = state_config.get("title") or name + editor.set_state_name(str(name), title=str(title)) + editor.idle_card_background = self._manager.idle_card_background + editor.set_state_config(state_config) + + def updateEditorGeometry( # noqa: N802 + self, editor: QWidget, option: QStyleOptionViewItem, _index: QModelIndex + ) -> None: + editor.setGeometry(option.rect) + + def sizeHint(self, option: QStyleOptionViewItem, index: QModelIndex) -> QSize: # noqa: N802 + name = index.data(_BeamlineStateListModel.NameRole) + pill = self._manager._state_pills.get(str(name)) + if pill is not None: + return pill.sizeHint() + return QSize(option.rect.width() if option.rect.width() > 0 else 360, 58) + + def destroyEditor(self, editor: QWidget, index: QModelIndex) -> None: # noqa: N802 + if isinstance(editor, BeamlineStatePill): + name = editor.state_name + if name and self._manager._state_pills.get(name) is editor: + self._manager._state_pills.pop(name, None) + editor.cleanup() + super().destroyEditor(editor, index) + + +class _BeamlineStateListView(QListView): + """List view using persistent pill editors.""" + + def __init__(self, parent: QWidget | None = None) -> None: + super().__init__(parent) + self.setObjectName("beamline_state_pill_view") + self.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) + self.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel) + self.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) + self.setFrameShape(QListView.Shape.NoFrame) + self.setSpacing(6) + self.setStyleSheet( + "QListView#beamline_state_pill_view {" + "background: transparent;" + "border: none;" + "}" + "QListView#beamline_state_pill_view::item {" + "background: transparent;" + "border: none;" + "padding: 0;" + "}" + "QListView#beamline_state_pill_view::item:selected {" + "background: transparent;" + "border: none;" + "}" + ) + + +class BeamlineStateManager(BECWidget, QWidget): + """ + Widget displaying and managing all BEC beamline states. + + The manager subscribes to ``MessageEndpoints.available_beamline_states()`` and creates, + updates, or removes child ``BeamlineStatePill`` widgets as the set of configured states changes. + """ + + PLUGIN = True + ICON_NAME = "format_list_bulleted" + USER_ACCESS = [ + "idle_card_background", + "set_idle_card_background", + "refresh_states", + "clear_filters", + "remove", + "attach", + "detach", + "screenshot", + ] + + def __init__( + self, + parent: QWidget | None = None, + client=None, + config: ConnectionConfig | None = None, + gui_id: str | None = None, + idle_card_background: bool = False, + **kwargs, + ) -> None: + super().__init__( + parent=parent, client=client, config=config, gui_id=gui_id, theme_update=True, **kwargs + ) + self._state_pills: dict[str, BeamlineStatePill] = {} + self._state_configs: dict[str, dict[str, Any]] = {} + self._state_order: list[str] = [] + self._selected_statuses: set[str] | None = None + self._selected_devices: set[str] | None = None + self._device_filter_text = "" + self._hidden_expanded = False + self._idle_card_background = False + self.idle_card_background = idle_card_background + + self._empty_label = QLabel( + "No beamline states available.\n Add new state from toolbar or CLI.", self + ) + self._empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + + self._toolbar = self._create_toolbar() + self._model = _BeamlineStateListModel(self) + self._view = _BeamlineStateListView(self) + self._delegate = _BeamlineStatePillDelegate(self) + self._view.setModel(self._model) + self._view.setItemDelegate(self._delegate) + + layout = QVBoxLayout(self) + layout.setContentsMargins(8, 8, 8, 8) + layout.setSpacing(6) + layout.addWidget(self._toolbar) + layout.addWidget(self._empty_label) + layout.addWidget(self._view, 1) + self._hidden_summary = QToolButton(self) + self._hidden_summary.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextBesideIcon) + self._hidden_summary.setCheckable(True) + self._hidden_summary.clicked.connect(self._toggle_hidden_states) + layout.addWidget(self._hidden_summary) + self.setLayout(layout) + + self.bec_dispatcher.connect_slot( + self.update_available_states, + MessageEndpoints.available_beamline_states(), + from_start=True, + ) + self.refresh_states() + self._refresh_hidden_summary() + + @SafeProperty(bool, default=False) + def idle_card_background(self) -> bool: + """ + Whether idle collapsed pills keep the status-tinted card background. + """ + return self._idle_card_background + + @idle_card_background.setter + def idle_card_background(self, enabled: bool) -> None: + self._idle_card_background = _coerce_bool(enabled) + for pill in self._state_pills.values(): + pill.idle_card_background = self._idle_card_background + + def set_idle_card_background(self, enabled: bool) -> None: + """Set whether idle collapsed pills keep the status-tinted card background.""" + self.idle_card_background = enabled + + def _create_toolbar(self) -> ModularToolBar: + toolbar = ModularToolBar(parent=self) + + add_state = MaterialIconAction("add", "Add beamline state", filled=True, parent=self) + filter_states = MaterialIconAction( + "filter_alt", "Filter displayed state status", filled=True, parent=self + ) + filter_devices = MaterialIconAction( + "devices", "Filter displayed devices", filled=True, parent=self + ) + clear_filters = MaterialIconAction( + "filter_alt_off", "Clear beamline state filters", filled=True, parent=self + ) + refresh = MaterialIconAction( + "restart_alt", "Refresh beamline states", filled=True, parent=self + ) + + add_state.action.triggered.connect(self.open_add_state_dialog) + filter_states.action.triggered.connect(self.open_status_filter_dialog) + filter_devices.action.triggered.connect(self.open_device_filter_dialog) + clear_filters.action.triggered.connect(self.clear_filters) + refresh.action.triggered.connect(self.refresh_states) + + toolbar.components.add_safe("add_state", add_state) + toolbar.components.add_safe("filter_states", filter_states) + toolbar.components.add_safe("filter_devices", filter_devices) + toolbar.components.add_safe("clear_filters", clear_filters) + toolbar.components.add_safe("refresh", refresh) + + bundle = ToolbarBundle("beamline_state_manager", toolbar.components) + bundle.add_action("add_state") + bundle.add_action("filter_states") + bundle.add_action("filter_devices") + bundle.add_action("clear_filters") + bundle.add_action("refresh") + toolbar.add_bundle(bundle) + toolbar.show_bundles(["beamline_state_manager"]) + return toolbar + + @SafeSlot(str) + def apply_theme(self, _theme: str) -> None: + colors = BeamlineStatePill._state_colors("unknown") + self.setStyleSheet( + "BeamlineStateManager { border: none; }" + "QToolButton#hidden_states_summary {" + f"background-color: {colors['background']};" + f"border: 1px solid {colors['border']};" + "border-radius: 6px;" + "padding: 6px;" + "text-align: left;" + "}" + ) + for pill in self._state_pills.values(): + pill.apply_theme(_theme) + self._refresh_hidden_summary() + + @SafeSlot() + def open_add_state_dialog(self) -> None: + dialog = AddBeamlineStateDialog(self, client=self.client) + config = None + try: + accepted = dialog.exec() == QDialog.Accepted + if accepted: + config = dialog.config_result + finally: + dialog.cleanup() + dialog.deleteLater() + + if config is None: + return + beamline_states = getattr(self.client, "beamline_states", None) + if beamline_states is None: + QMessageBox.warning( + self, "Cannot Add State", "BEC client has no beamline state manager." + ) + return + try: + beamline_states.add(config) + except Exception as exc: + QMessageBox.warning(self, "Cannot Add State", str(exc)) + + @SafeSlot() + def open_status_filter_dialog(self) -> None: + dialog = StatusFilterDialog(self._selected_statuses, self) + if dialog.exec() != QDialog.Accepted: + return + self._selected_statuses = dialog.selected_statuses() + self._apply_filters() + + @SafeSlot() + def open_device_filter_dialog(self) -> None: + dialog = DeviceFilterDialog( + self._available_devices(), self._selected_devices, self._device_filter_text, self + ) + if dialog.exec() != QDialog.Accepted: + return + self._selected_devices = dialog.selected_devices() + self._device_filter_text = dialog.filter_text() + self._apply_filters() + + @SafeSlot() + def clear_filters(self) -> None: + self._selected_statuses = None + self._selected_devices = None + self._device_filter_text = "" + self._hidden_expanded = False + self._apply_filters() + + @SafeSlot() + def refresh_states(self) -> None: + """Fetch the latest cached available beamline states and update the list immediately.""" + msg_container = self.client.connector.get_last(MessageEndpoints.available_beamline_states()) + if not msg_container: + return + data = msg_container.get("data") if isinstance(msg_container, dict) else None + content = getattr(data, "content", data) + if isinstance(content, dict): + self.update_available_states(content, getattr(data, "metadata", {})) + + @SafeSlot(dict, dict) + def update_available_states( + self, content: dict[str, Any], _metadata: dict[str, Any] | None = None + ) -> None: + """Update the displayed pills from ``AvailableBeamlineStatesMessage`` content.""" + expanded_names = {name for name, pill in self._state_pills.items() if pill.is_expanded()} + states = content.get("states", []) + state_configs = [self._state_config_to_dict(state) for state in states] + state_configs = [state for state in state_configs if state.get("name")] + self._state_configs = {str(state["name"]): state for state in state_configs} + self._state_order = [str(state["name"]) for state in state_configs] + self._model.set_states(state_configs) + self._open_persistent_editors(expanded_names) + self._apply_filters() + + def _open_persistent_editors(self, expanded_names: set[str] | None = None) -> None: + expanded_names = expanded_names or set() + for row in range(self._model.rowCount()): + index = self._model.index(row, 0) + self._view.openPersistentEditor(index) + name = str(index.data(_BeamlineStateListModel.NameRole)) + pill = self._state_pills.get(name) + if pill is not None: + pill.set_expanded(name in expanded_names) + self._sync_pill_item_size(name) + + def _apply_filters(self) -> None: + visible_names = [] + hidden_names = [] + for name in self._state_order: + if self._is_state_visible(name): + visible_names.append(name) + else: + hidden_names.append(name) + + self._apply_row_visibility(visible_names, hidden_names) + self._empty_label.setVisible( + not visible_names and not (self._hidden_expanded and hidden_names) + ) + self._view.setVisible(bool(visible_names) or (self._hidden_expanded and bool(hidden_names))) + self._refresh_hidden_summary(hidden_count=len(hidden_names)) + + def _apply_row_visibility(self, visible_names: list[str], hidden_names: list[str]) -> None: + visible_set = set(visible_names) + show_hidden = self._hidden_expanded and bool(hidden_names) + for row, name in enumerate(self._state_order): + hidden_by_filter = name not in visible_set + self._view.setRowHidden(row, hidden_by_filter and not show_hidden) + self._sync_pill_item_size(name) + + def _sync_pill_item_size(self, name: str) -> None: + index = self._model.index_for_name(name) + if not index.isValid(): + return + self._model.dataChanged.emit(index, index, [Qt.ItemDataRole.SizeHintRole]) + self._view.update(index) + + def _is_state_visible(self, name: str) -> bool: + pill = self._state_pills.get(name) + if self._selected_statuses is not None and ( + pill is None or pill._status not in self._selected_statuses + ): + return False + + device = self._state_device(self._state_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._device_filter_text.split(",") + if token.strip() + ] + if tokens: + if device is None: + return False + device_lower = device.casefold() + if not any(token in device_lower for token in tokens): + return False + return True + + @SafeSlot(str, str, str) + def _on_pill_state_changed(self, _name: str, _status: str, _label: str) -> None: + if self._selected_statuses is not None: + self._apply_filters() + + @SafeSlot(str, dict) + def _update_state_parameters(self, state_name: str, parameters: dict[str, Any]) -> None: + beamline_states = getattr(self.client, "beamline_states", None) + state_client = getattr(beamline_states, state_name, None) if beamline_states else None + if state_client is None or not hasattr(state_client, "update_parameters"): + QMessageBox.warning( + self, "Cannot Update State", f"Beamline state '{state_name}' is not available." + ) + return + try: + state_client.update_parameters(**parameters) + except Exception as exc: + QMessageBox.warning(self, "Cannot Update State", str(exc)) + return + pill = self._state_pills.get(state_name) + if pill is not None: + pill.mark_current_settings_clean(parameters) + + @SafeSlot(str) + def _remove_state_requested(self, state_name: str) -> None: + reply = QMessageBox.question( + self, + "Remove Beamline State", + f"Remove beamline state '{state_name}'?", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if reply != QMessageBox.StandardButton.Yes: + return + + beamline_states = getattr(self.client, "beamline_states", None) + if beamline_states is None or not hasattr(beamline_states, "delete"): + QMessageBox.warning( + self, "Cannot Remove State", "BEC client has no beamline state manager." + ) + return + try: + beamline_states.delete(state_name) + except Exception as exc: + QMessageBox.warning(self, "Cannot Remove State", str(exc)) + + @SafeSlot(bool) + def _toggle_hidden_states(self, checked: bool | None = None) -> None: + if checked is None: + checked = self._hidden_summary.isChecked() + self._hidden_expanded = bool(checked) + self._apply_filters() + + def _refresh_hidden_summary(self, hidden_count: int | None = None) -> None: + if hidden_count is None: + hidden_count = sum(1 for name in self._state_order if not self._is_state_visible(name)) + self._hidden_summary.setObjectName("hidden_states_summary") + self._hidden_summary.setVisible(hidden_count > 0) + self._hidden_summary.setChecked(self._hidden_expanded and hidden_count > 0) + icon_name = "expand_less" if self._hidden_expanded else "expand_more" + self._hidden_summary.setIcon(material_icon(icon_name, convert_to_pixmap=False)) + suffix = "state is" if hidden_count == 1 else "states are" + action = "Hide" if self._hidden_expanded else "Show" + self._hidden_summary.setText( + f"{hidden_count} {suffix} hidden by filters. {action} hidden states." + ) + + def _available_devices(self) -> list[str]: + devices = { + device + for state in self._state_configs.values() + if (device := self._state_device(state)) is not None + } + return sorted(devices) + + @staticmethod + def _state_device(state: dict[str, Any]) -> str | None: + parameters = state.get("parameters") + if isinstance(parameters, dict): + device = parameters.get("device") + else: + device = state.get("device") + return str(device) if device else None + + @staticmethod + def _state_config_to_dict(state: Any) -> dict[str, Any]: + if isinstance(state, dict): + return state + if hasattr(state, "model_dump"): + state_dict = state.model_dump() + state_type = getattr(state, "state_type", None) + if state_type is not None: + state_dict.setdefault("state_type", state_type) + return state_dict + return {"name": getattr(state, "name", None), "title": getattr(state, "title", None)} + + def cleanup(self) -> None: + self.bec_dispatcher.disconnect_slot( + self.update_available_states, MessageEndpoints.available_beamline_states() + ) + for row in range(self._model.rowCount()): + self._view.closePersistentEditor(self._model.index(row, 0)) + for pill in list(self._state_pills.values()): + pill.cleanup() + pill.deleteLater() + self._state_pills.clear() + self._toolbar.components.cleanup() + super().cleanup() + + +if __name__ == "__main__": # pragma: no cover + app = QApplication(sys.argv) + + from bec_widgets.utils.colors import apply_theme + from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + + apply_theme("dark") + + window = QWidget() + window.setWindowTitle("Beamline States") + layout = QVBoxLayout(window) + layout.setContentsMargins(12, 12, 12, 12) + layout.setSpacing(8) + + theme_row = QHBoxLayout() + theme_row.addStretch(1) + theme_row.addWidget(DarkModeButton(parent=window)) + layout.addLayout(theme_row) + + manager = BeamlineStateManager(parent=window) + idle_background = QCheckBox("Idle card background", window) + idle_background.toggled.connect( + lambda checked: setattr(manager, "idle_card_background", checked) + ) + + options_row = QHBoxLayout() + options_row.addWidget(idle_background) + options_row.addStretch(1) + layout.addLayout(options_row) + layout.addWidget(manager, 1) + + window.resize(760, 480) + window.show() + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.pyproject b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.pyproject new file mode 100644 index 00000000..340bae9a --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.pyproject @@ -0,0 +1 @@ +{'files': ['beamline_state_pill.py']} \ No newline at end of file diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill_plugin.py b/bec_widgets/widgets/services/beamline_states/beamline_state_pill_plugin.py new file mode 100644 index 00000000..537d2279 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill_plugin.py @@ -0,0 +1,57 @@ +# Copyright (C) 2022 The Qt Company Ltd. +# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause + +from qtpy.QtDesigner import QDesignerCustomWidgetInterface +from qtpy.QtWidgets import QWidget + +from bec_widgets.utils.bec_designer import designer_material_icon +from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStatePill + +DOM_XML = """ + + + + +""" + + +class BeamlineStatePillPlugin(QDesignerCustomWidgetInterface): # pragma: no cover + def __init__(self): + super().__init__() + self._form_editor = None + + def createWidget(self, parent): + if parent is None: + return QWidget() + t = BeamlineStatePill(parent) + return t + + def domXml(self): + return DOM_XML + + def group(self): + return "" + + def icon(self): + return designer_material_icon(BeamlineStatePill.ICON_NAME) + + def includeFile(self): + return "beamline_state_pill" + + def initialize(self, form_editor): + self._form_editor = form_editor + + def isContainer(self): + return False + + def isInitialized(self): + return self._form_editor is not None + + def name(self): + return "BeamlineStatePill" + + def toolTip(self): + return "" + + def whatsThis(self): + return self.toolTip() diff --git a/bec_widgets/widgets/services/beamline_states/dialogs.py b/bec_widgets/widgets/services/beamline_states/dialogs.py new file mode 100644 index 00000000..cf9611fb --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/dialogs.py @@ -0,0 +1,327 @@ +from __future__ import annotations + +import re + +from bec_lib import bl_states +from qtpy.QtWidgets import ( + QCheckBox, + QComboBox, + QDialog, + QDialogButtonBox, + QFormLayout, + QGroupBox, + QHBoxLayout, + QLineEdit, + QMessageBox, + QPushButton, + QSizePolicy, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox +from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox +from bec_widgets.widgets.utility.spinbox.decimal_spinbox import BECSpinBox + +BEAMLINE_STATE_STATUS_LABELS = { + "valid": "VALID", + "invalid": "INVALID", + "warning": "WARNING", + "unknown": "UNKNOWN", +} + + +class AddBeamlineStateDialog(QDialog): + """Dialog for creating supported beamline state configurations.""" + + def __init__(self, parent: QWidget | None = None, client=None) -> None: + super().__init__(parent=parent) + self.setWindowTitle("Add Beamline State") + self._cleaned_up = False + self._config: ( + bl_states.DeviceStateConfig | bl_states.DeviceWithinLimitsStateConfig | None + ) = None + self._auto_generated_name: str | None = None + + self._type_combo = QComboBox(self) + self._type_combo.addItem("Device within limits", "device_within_limits") + self._type_combo.addItem("Shutter", "shutter") + self._type_combo.currentIndexChanged.connect(self._update_field_visibility) + + self._name = QLineEdit(self) + self._name.setPlaceholderText("samx_limits") + self._title = QLineEdit(self) + self._device = DeviceComboBox(parent=self, client=client) + self._signal = SignalComboBox(parent=self, client=client, require_device=True) + self._low_limit = BECSpinBox(self) + self._high_limit = BECSpinBox(self) + self._tolerance = BECSpinBox(self) + self._device.device_selected.connect(self._on_valid_device_selected) + self._device.device_reset.connect(self._on_device_reset) + + for spin_box in (self._low_limit, self._high_limit): + spin_box.setRange(-1_000_000_000, 1_000_000_000) + spin_box.setDecimals(6) + self._low_limit.setValue(0.0) + self._high_limit.setValue(10.0) + self._tolerance.setRange(0.0, 1_000_000_000) + self._tolerance.setDecimals(6) + self._tolerance.setValue(0.1) + for field in self._input_fields(): + field.setFixedWidth(280) + for spin_box in (self._low_limit, self._high_limit, self._tolerance): + spin_box.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed) + + self._form = QFormLayout() + self._form.addRow("State type", self._type_combo) + self._form.addRow("Name", self._name) + self._form.addRow("Title", self._title) + self._form.addRow("Device", self._device) + self._form.addRow("Signal", self._signal) + self._form.addRow("Low limit", self._low_limit) + self._form.addRow("High limit", self._high_limit) + self._form.addRow("Tolerance", self._tolerance) + + self._buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self + ) + self._buttons.accepted.connect(self.accept) + self._buttons.rejected.connect(self.reject) + + layout = QVBoxLayout(self) + layout.addLayout(self._form) + layout.addWidget(self._buttons) + self.setLayout(layout) + self._update_field_visibility() + self.setFixedSize(self.sizeHint().expandedTo(self.minimumSizeHint())) + + def config(self) -> bl_states.DeviceStateConfig | bl_states.DeviceWithinLimitsStateConfig: + state_type = self._type_combo.currentData() + name = self._state_name() + title = self._optional_text(self._title) + device = self._selected_device() + signal = self._selected_signal() + + if state_type == "shutter": + return bl_states.DeviceStateConfig(name=name, title=title, device=device, signal=signal) + + return bl_states.DeviceWithinLimitsStateConfig( + name=name, + title=title, + device=device, + signal=signal, + low_limit=self._low_limit.value(), + high_limit=self._high_limit.value(), + tolerance=self._tolerance.value(), + ) + + def accept(self) -> None: + try: + self._config = self.config() + except Exception as exc: + QMessageBox.warning(self, "Invalid Beamline State", str(exc)) + return + super().accept() + + @property + def config_result( + self, + ) -> bl_states.DeviceStateConfig | bl_states.DeviceWithinLimitsStateConfig: + if self._config is None: + raise RuntimeError("Beamline state dialog was not accepted with a valid config.") + return self._config + + def cleanup(self) -> None: + if self._cleaned_up: + return + self._cleaned_up = True + self._device.close() + self._device.deleteLater() + self._signal.close() + self._signal.deleteLater() + + @SafeSlot(str) + def _on_valid_device_selected(self, device: str) -> None: + if self._cleaned_up: + return + self._signal.set_device(device) + current_name = self._name.text().strip() + if current_name and current_name != self._auto_generated_name: + return + generated_name = f"{self._normalize_identifier(device)}_{self._state_name_suffix()}" + self._auto_generated_name = generated_name + self._name.setText(generated_name) + + @SafeSlot() + def _on_device_reset(self) -> None: + if self._cleaned_up: + return + self._signal.set_device(None) + + @SafeSlot(int) + def _update_field_visibility(self, _index: int = 0) -> None: + show_limits = self._type_combo.currentData() == "device_within_limits" + for widget in (self._low_limit, self._high_limit, self._tolerance): + widget.setVisible(show_limits) + label = self._form.labelForField(widget) + if label is not None: + label.setVisible(show_limits) + + def _state_name(self) -> str: + raw_name = self._name.text().strip() + if not raw_name: + raise ValueError("Name is required.") + name = self._normalize_identifier(raw_name) + self._name.setText(name) + return name + + def _selected_device(self) -> str: + device = self._device.currentText().strip() + if not device: + raise ValueError("Device is required.") + if not self._device.is_valid_input: + raise ValueError(f"Device '{device}' is not available.") + return device + + def _selected_signal(self) -> str | None: + signal = self._signal.get_signal_name().strip() + if not signal: + return None + if not self._signal.is_valid_input: + raise ValueError( + f"Signal '{signal}' is not available for device '{self._device.currentText()}'." + ) + return signal + + @staticmethod + def _optional_text(line_edit: QLineEdit) -> str | None: + value = line_edit.text().strip() + return value or None + + @staticmethod + def _normalize_identifier(value: str) -> str: + name = re.sub(r"\W+", "_", value.strip()) + name = re.sub(r"_+", "_", name).strip("_") + if not name: + raise ValueError("Name must contain at least one letter, number, or underscore.") + if name[0].isdigit(): + name = f"state_{name}" + return name + + def _state_name_suffix(self) -> str: + if self._type_combo.currentData() == "device_within_limits": + return "limits" + return "state" + + def _input_fields(self) -> tuple[QWidget, ...]: + return ( + self._type_combo, + self._name, + self._title, + self._device, + self._signal, + self._low_limit, + self._high_limit, + self._tolerance, + ) + + +class StatusFilterDialog(QDialog): + """Dialog for selecting visible beamline state statuses.""" + + def __init__(self, selected_statuses: set[str] | None, parent: QWidget | None = None) -> None: + super().__init__(parent=parent) + self.setWindowTitle("Filter Beamline State Status") + self._checkboxes: dict[str, QCheckBox] = {} + + controls = QHBoxLayout() + select_all = QPushButton("Select all", self) + clear = QPushButton("Clear", self) + select_all.clicked.connect(lambda: self._set_all(True)) + clear.clicked.connect(lambda: self._set_all(False)) + controls.addWidget(select_all) + controls.addWidget(clear) + controls.addStretch(1) + + list_layout = QVBoxLayout() + for status, label in BEAMLINE_STATE_STATUS_LABELS.items(): + checkbox = QCheckBox(label, self) + checkbox.setChecked(selected_statuses is None or status in selected_statuses) + self._checkboxes[status] = checkbox + list_layout.addWidget(checkbox) + list_layout.addStretch(1) + + box = QGroupBox("Displayed status", self) + box.setLayout(list_layout) + + buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self + ) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + + layout = QVBoxLayout(self) + layout.addLayout(controls) + layout.addWidget(box) + layout.addWidget(buttons) + self.setLayout(layout) + + def selected_statuses(self) -> set[str] | None: + selected = {status for status, checkbox in self._checkboxes.items() if checkbox.isChecked()} + if selected == set(self._checkboxes): + return None + return selected + + def _set_all(self, checked: bool) -> None: + for checkbox in self._checkboxes.values(): + checkbox.setChecked(checked) + + +class DeviceFilterDialog(QDialog): + """Dialog for filtering beamline states by configured device.""" + + def __init__( + self, + devices: list[str], + selected_devices: set[str] | None, + device_filter_text: str, + parent: QWidget | None = None, + ) -> None: + super().__init__(parent=parent) + self.setWindowTitle("Filter Beamline State Devices") + self._checkboxes: dict[str, QCheckBox] = {} + + self._device_text = QLineEdit(self) + self._device_text.setPlaceholderText("Device name or comma-separated names") + self._device_text.setText(device_filter_text) + + list_layout = QVBoxLayout() + for device in devices: + checkbox = QCheckBox(device, self) + checkbox.setChecked(selected_devices is not None and device in selected_devices) + self._checkboxes[device] = checkbox + list_layout.addWidget(checkbox) + list_layout.addStretch(1) + + box = QGroupBox("Known devices", self) + box.setLayout(list_layout) + + buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self + ) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + + layout = QVBoxLayout(self) + layout.addWidget(self._device_text) + layout.addWidget(box) + layout.addWidget(buttons) + self.setLayout(layout) + + def selected_devices(self) -> set[str] | None: + selected = {device for device, checkbox in self._checkboxes.items() if checkbox.isChecked()} + return selected or None + + def filter_text(self) -> str: + return self._device_text.text().strip() diff --git a/bec_widgets/widgets/services/beamline_states/register_beamline_state_manager.py b/bec_widgets/widgets/services/beamline_states/register_beamline_state_manager.py new file mode 100644 index 00000000..7695645a --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/register_beamline_state_manager.py @@ -0,0 +1,17 @@ +def main(): # pragma: no cover + from qtpy import PYSIDE6 + + if not PYSIDE6: + print("PYSIDE6 is not available in the environment. Cannot patch designer.") + return + from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection + + from bec_widgets.widgets.services.beamline_states.beamline_state_manager_plugin import ( + BeamlineStateManagerPlugin, + ) + + QPyDesignerCustomWidgetCollection.addCustomWidget(BeamlineStateManagerPlugin()) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/bec_widgets/widgets/services/beamline_states/register_beamline_state_pill.py b/bec_widgets/widgets/services/beamline_states/register_beamline_state_pill.py new file mode 100644 index 00000000..5c032ab5 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/register_beamline_state_pill.py @@ -0,0 +1,17 @@ +def main(): # pragma: no cover + from qtpy import PYSIDE6 + + if not PYSIDE6: + print("PYSIDE6 is not available in the environment. Cannot patch designer.") + return + from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection + + from bec_widgets.widgets.services.beamline_states.beamline_state_pill_plugin import ( + BeamlineStatePillPlugin, + ) + + QPyDesignerCustomWidgetCollection.addCustomWidget(BeamlineStatePillPlugin()) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py new file mode 100644 index 00000000..50efed90 --- /dev/null +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -0,0 +1,641 @@ +from pathlib import Path + +import shiboken6 +from bec_lib import messages +from qtpy.QtCore import QCoreApplication, QEvent, QPoint, Qt +from qtpy.QtGui import QColor +from qtpy.QtWidgets import QApplication, QMessageBox, QSizePolicy + +from bec_widgets.utils.toolbars.toolbar import ModularToolBar +from bec_widgets.widgets.services.beamline_states import beamline_state_pill as pill_module +from bec_widgets.widgets.services.beamline_states.beamline_state_pill import ( + BeamlineStateManager, + BeamlineStatePill, + _BeamlineStatePillHeader, +) +from bec_widgets.widgets.services.beamline_states.dialogs import AddBeamlineStateDialog +from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox +from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox +from bec_widgets.widgets.utility.spinbox.decimal_spinbox import BECSpinBox + +from .client_mocks import mocked_client + + +class _FakeAccentColors: + default = QColor("#0a60ff") + success = QColor("#2CA58D") + emergency = QColor("#CC181E") + warning = QColor("#EAC435") + + +class _FakeTheme: + def __init__(self, theme: str) -> None: + self.theme = theme + self.accent_colors = _FakeAccentColors() + self._colors = { + "CARD_BG": "#ffffff" if theme == "light" else "#171a21", + "FG": "#151924" if theme == "light" else "#e8ebf1", + "BORDER": "#d9dde6" if theme == "light" else "#2a2f3a", + "ON_PRIMARY": "#ffffff", + "ACCENT_DEFAULT": "#0a60ff" if theme == "light" else "#8ab4f7", + } + + def color(self, key: str, fallback: str = "#000000") -> QColor: + return QColor(self._colors.get(key, fallback)) + + +def _gradient_alpha(colors: dict[str, str]) -> int: + return int(colors["gradient_accent"].rsplit(",", 1)[1].strip(" )")) + + +def test_beamline_state_pill_updates_from_message(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="shutter_open", title="Shutter", client=mocked_client) + qtbot.addWidget(widget) + + widget.update_state( + {"name": "shutter_open", "status": "valid", "label": "Shutter is open."}, {} + ) + + assert widget.state_name == "shutter_open" + assert widget._name_label.text() == "Shutter" + assert widget._status_label.text() == "VALID" + assert widget._detail_label.text() == "Shutter is open." + assert not widget._icon_label.pixmap().isNull() + assert widget.toolTip() == "Shutter is open." + + +def test_beamline_state_pill_ignores_other_states(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="shutter_open", client=mocked_client) + qtbot.addWidget(widget) + + widget.update_state( + {"name": "other_state", "status": "invalid", "label": "Should be ignored."}, {} + ) + + assert widget._status_label.text() == "UNKNOWN" + assert widget.toolTip() == "No state information available." + + +def test_beamline_states_init_is_empty(): + assert Path(pill_module.__file__).with_name("__init__.py").read_text() == "" + + +def test_beamline_state_pill_expands_and_emits_updated_limits(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + widget.set_state_config( + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": { + "name": "limits", + "title": "Limits", + "device": "samx", + "signal": "samx", + "low_limit": 0.0, + "high_limit": 10.0, + "tolerance": 0.1, + }, + } + ) + + assert widget._settings.isHidden() + assert not widget._update_button.isEnabled() + assert not widget._revert_button.isEnabled() + + qtbot.mouseClick(widget._header, Qt.MouseButton.LeftButton) + widget._high_limit.setValue(20.0) + + assert not widget._settings.isHidden() + assert widget._update_button.isEnabled() + assert widget._revert_button.isEnabled() + assert widget._high_limit.parentWidget().property("beamlineStateDirty") is True + assert isinstance(widget._device_edit, DeviceComboBox) + assert isinstance(widget._signal_edit, SignalComboBox) + assert widget._device_edit.currentText() == "samx" + for field in widget._settings_input_fields(): + assert field.minimumWidth() == widget._SETTINGS_FIELD_WIDTH + assert field.sizePolicy().horizontalPolicy() == QSizePolicy.Policy.Expanding + assert widget.edited_parameters()["high_limit"] == 20.0 + + with qtbot.waitSignal(widget.update_requested) as signal: + widget._update_button.click() + + assert signal.args[0] == "limits" + assert signal.args[1]["device"] == "samx" + assert signal.args[1]["signal"] == "samx" + assert signal.args[1]["low_limit"] == 0.0 + assert signal.args[1]["high_limit"] == 20.0 + assert signal.args[1]["tolerance"] == 0.1 + assert not widget._settings.isHidden() + + +def test_beamline_state_pill_reverts_changed_settings(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + widget.set_state_config( + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": { + "device": "samx", + "signal": "samx", + "low_limit": 0.0, + "high_limit": 10.0, + "tolerance": 0.1, + }, + } + ) + + widget._low_limit.setValue(-5.0) + + assert widget._update_button.isEnabled() + assert widget._low_limit.parentWidget().property("beamlineStateDirty") is True + + widget._revert_button.click() + + assert widget._low_limit.value() == 0.0 + assert not widget._update_button.isEnabled() + assert not widget._revert_button.isEnabled() + assert widget._low_limit.parentWidget().property("beamlineStateDirty") is False + + +def test_beamline_state_pill_uses_card_style_when_expanded(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + + assert "#BeamlineStatePill {background: transparent" in widget.styleSheet() + assert "#BeamlineStatePill:hover {background: qlineargradient" in widget.styleSheet() + + widget._toggle_expanded() + + assert "#BeamlineStatePill {background: qlineargradient" in widget.styleSheet() + assert widget._shadow.isEnabled() + + +def test_beamline_state_pill_can_keep_idle_background(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + + assert "#BeamlineStatePill {background: transparent" in widget.styleSheet() + + widget.idle_card_background = True + + assert "#BeamlineStatePill {background: transparent" not in widget.styleSheet() + + +def test_beamline_state_pill_declares_card_style_for_hover(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + + assert "#BeamlineStatePill:hover {background: qlineargradient" in widget.styleSheet() + assert not widget._shadow.isEnabled() + + +def test_beamline_state_pill_light_mode_uses_neutral_card_with_subtle_left_gradient( + qtbot, monkeypatch +): + app = QApplication.instance() + monkeypatch.setattr(app, "theme", _FakeTheme("light"), raising=False) + + for status in ("valid", "invalid", "warning"): + colors = BeamlineStatePill._state_colors(status) + + assert _gradient_alpha(colors) == 18 + assert colors["gradient_stop"] == "0.38" + assert colors["card_background"] == "#ffffff" + assert colors["background"] == "#ffffff" + + +def test_beamline_state_pill_does_not_override_themed_input_controls(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client) + qtbot.addWidget(widget) + widget.set_expanded(True) + + stylesheet = widget.styleSheet() + + assert "QAbstractSpinBox" not in stylesheet + assert "QComboBox" not in stylesheet + assert "QCheckBox::indicator" not in stylesheet + + +def test_beamline_state_pill_dark_mode_keeps_existing_gradient_strength(qtbot, monkeypatch): + app = QApplication.instance() + monkeypatch.setattr(app, "theme", _FakeTheme("dark"), raising=False) + + colors = BeamlineStatePill._state_colors("warning") + + assert _gradient_alpha(colors) == 62 + assert colors["gradient_stop"] == "0.62" + + +def test_beamline_state_pill_header_emits_click_without_pointer_move(qtbot): + header = _BeamlineStatePillHeader() + header.resize(120, 32) + qtbot.addWidget(header) + + clicked = [] + header.clicked.connect(lambda: clicked.append(True)) + + qtbot.mousePress(header, Qt.MouseButton.LeftButton, pos=QPoint(8, 8)) + qtbot.mouseRelease(header, Qt.MouseButton.LeftButton, pos=QPoint(8, 8)) + + assert clicked == [True] + + +def test_beamline_state_pill_header_suppresses_click_after_release_outside(qtbot): + header = _BeamlineStatePillHeader() + header.resize(120, 32) + qtbot.addWidget(header) + + clicked = [] + header.clicked.connect(lambda: clicked.append(True)) + + qtbot.mousePress(header, Qt.MouseButton.LeftButton, pos=QPoint(8, 8)) + qtbot.mouseRelease(header, Qt.MouseButton.LeftButton, pos=QPoint(140, 8)) + + assert clicked == [] + + +def test_beamline_state_manager_adds_and_removes_pills(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + + widget.update_available_states( + { + "states": [ + messages.BeamlineStateConfig( + name="shutter_open", title="Shutter", state_type="ShutterState", parameters={} + ), + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {}, + }, + ] + }, + {}, + ) + + assert sorted(widget._state_pills) == ["limits", "shutter_open"] + assert widget._model.rowCount() == 2 + assert widget._state_pills["shutter_open"]._name_label.text() == "Shutter" + assert not widget._empty_label.isVisible() + + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {}, + } + ] + }, + {}, + ) + + assert sorted(widget._state_pills) == ["limits"] + assert widget._model.rowCount() == 1 + + +def test_beamline_state_manager_items_are_not_draggable(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {}, + } + ] + }, + {}, + ) + + flags = widget._model.flags(widget._model.index_for_name("limits")) + + assert not flags & Qt.ItemFlag.ItemIsDragEnabled + + +def test_beamline_state_manager_header_click_expands_pill_once(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx"}, + } + ] + }, + {}, + ) + + pill = widget._state_pills["limits"] + assert pill._settings.isHidden() + + qtbot.mouseClick(pill._header, Qt.MouseButton.LeftButton) + + assert not pill._settings.isHidden() + + +def test_beamline_state_manager_preserves_expanded_pill_on_refresh(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + state = { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx", "high_limit": 10.0}, + } + widget.update_available_states({"states": [state]}, {}) + + widget._state_pills["limits"].set_expanded(True) + widget.update_available_states({"states": [state]}, {}) + + assert widget._state_pills["limits"].is_expanded() + assert not widget._state_pills["limits"]._settings.isHidden() + + +def test_beamline_state_manager_propagates_idle_card_background(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client, idle_card_background=True) + qtbot.addWidget(widget) + + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx"}, + } + ] + }, + {}, + ) + + assert widget._state_pills["limits"].idle_card_background is True + + widget.idle_card_background = False + + assert widget._state_pills["limits"].idle_card_background is False + + +def test_beamline_state_manager_filters_status(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + + widget.update_available_states( + { + "states": [ + { + "name": "shutter_open", + "title": "Shutter", + "state_type": "ShutterState", + "parameters": {"device": "samy"}, + }, + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx"}, + }, + ] + }, + {}, + ) + + assert isinstance(widget._toolbar, ModularToolBar) + + widget._state_pills["limits"].update_state( + {"name": "limits", "status": "valid", "label": "Within limits."}, {} + ) + widget._state_pills["shutter_open"].update_state( + {"name": "shutter_open", "status": "invalid", "label": "Closed."}, {} + ) + widget._selected_statuses = {"valid"} + widget._apply_filters() + + assert not widget._hidden_summary.isHidden() + assert "1 state is hidden" in widget._hidden_summary.text() + assert not widget._view.isRowHidden(widget._model.index_for_name("limits").row()) + assert widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row()) + + widget._hidden_summary.click() + + assert not widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row()) + assert shiboken6.isValid(widget._state_pills["shutter_open"]) + + widget._hidden_summary.click() + + assert widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row()) + assert shiboken6.isValid(widget._state_pills["shutter_open"]) + + +def test_beamline_state_manager_status_filter_reacts_to_state_changes(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx"}, + } + ] + }, + {}, + ) + + widget._selected_statuses = {"valid"} + widget._state_pills["limits"].update_state( + {"name": "limits", "status": "valid", "label": "Within limits."}, {} + ) + + assert widget._hidden_summary.isHidden() + + widget._state_pills["limits"].update_state( + {"name": "limits", "status": "invalid", "label": "Out of limits."}, {} + ) + + assert not widget._hidden_summary.isHidden() + assert widget._view.isRowHidden(widget._model.index_for_name("limits").row()) + + +def test_beamline_state_manager_filters_devices(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + + widget.update_available_states( + { + "states": [ + { + "name": "samx_limits", + "title": "samx", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samx"}, + }, + { + "name": "samy_limits", + "title": "samy", + "state_type": "DeviceWithinLimitsState", + "parameters": {"device": "samy"}, + }, + ] + }, + {}, + ) + + widget._device_filter_text = "samx" + widget._apply_filters() + + assert not widget._hidden_summary.isHidden() + assert "1 state is hidden" in widget._hidden_summary.text() + assert widget._available_devices() == ["samx", "samy"] + + +def test_beamline_state_manager_updates_state_parameters(qtbot, mocked_client): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + widget.update_available_states( + { + "states": [ + { + "name": "limits", + "title": "Limits", + "state_type": "DeviceWithinLimitsState", + "parameters": { + "device": "samx", + "signal": "samx", + "low_limit": 0.0, + "high_limit": 10.0, + "tolerance": 0.1, + }, + } + ] + }, + {}, + ) + + class StateClient: + def __init__(self): + self.parameters = None + + def update_parameters(self, **kwargs): + self.parameters = kwargs + + class StateManager: + def __init__(self): + self.limits = StateClient() + + mocked_client.beamline_states = StateManager() + pill = widget._state_pills["limits"] + pill._high_limit.setValue(20.0) + + assert pill._update_button.isEnabled() + + widget._update_state_parameters("limits", pill.edited_parameters()) + + assert mocked_client.beamline_states.limits.parameters == { + "title": "Limits", + "device": "samx", + "signal": "samx", + "low_limit": 0.0, + "high_limit": 20.0, + "tolerance": 0.1, + } + assert not pill._update_button.isEnabled() + assert pill._high_limit.parentWidget().property("beamlineStateDirty") is False + + +def test_beamline_state_manager_removes_state(qtbot, mocked_client, monkeypatch): + widget = BeamlineStateManager(client=mocked_client) + qtbot.addWidget(widget) + + class StateManager: + def __init__(self): + self.deleted = None + + def delete(self, state_name): + self.deleted = state_name + + mocked_client.beamline_states = StateManager() + monkeypatch.setattr( + QMessageBox, "question", lambda *args, **kwargs: QMessageBox.StandardButton.Yes + ) + + widget._remove_state_requested("limits") + + assert mocked_client.beamline_states.deleted == "limits" + + +def test_add_beamline_state_dialog_uses_device_signal_widgets_and_normalizes_name( + qtbot, mocked_client +): + dialog = AddBeamlineStateDialog(client=mocked_client) + qtbot.addWidget(dialog) + + dialog._name.setText("samx-limits") + dialog._title.setText("samx-limits-15") + dialog._device.set_device("samx") + dialog._signal.set_signal("samx") + dialog._high_limit.setValue(15.0) + + config = dialog.config() + + assert config.name == "samx_limits" + assert config.title == "samx-limits-15" + assert config.device == "samx" + assert config.signal == "samx" + assert config.low_limit == 0.0 + assert config.high_limit == 15.0 + assert isinstance(dialog._low_limit, BECSpinBox) + assert isinstance(dialog._high_limit, BECSpinBox) + assert dialog._low_limit.width() == dialog._device.width() + + +def test_add_beamline_state_dialog_generates_name_only_after_valid_device_selection( + qtbot, mocked_client +): + dialog = AddBeamlineStateDialog(client=mocked_client) + qtbot.addWidget(dialog) + + dialog._device.setCurrentText("s") + + assert dialog._name.text() == "" + + dialog._device.set_device("samx") + + assert dialog._name.text() == "samx_limits" + + +def test_add_beamline_state_dialog_cleanup_deletes_device_widgets(qtbot, mocked_client): + dialog = AddBeamlineStateDialog(client=mocked_client) + qtbot.addWidget(dialog) + device = dialog._device + signal = dialog._signal + + dialog.reject() + assert shiboken6.isValid(device) + assert shiboken6.isValid(signal) + + dialog.cleanup() + QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete) + + assert not shiboken6.isValid(device) + assert not shiboken6.isValid(signal) diff --git a/tests/unit_tests/test_dock_area.py b/tests/unit_tests/test_dock_area.py index 35c1900b..10334ede 100644 --- a/tests/unit_tests/test_dock_area.py +++ b/tests/unit_tests/test_dock_area.py @@ -869,7 +869,14 @@ class TestToolbarFunctionality: def test_toolbar_utils_actions(self, advanced_dock_area): """Test utils toolbar actions trigger widget creation.""" - utils_actions = ["queue", "terminal", "status", "progress_bar", "sbb_monitor"] + utils_actions = [ + "queue", + "terminal", + "status", + "progress_bar", + "sbb_monitor", + "beamline_state_manager", + ] for action_name in utils_actions: with patch.object(advanced_dock_area, "new") as mock_new: @@ -2428,6 +2435,7 @@ class TestFlatToolbarActions: "flat_terminal", "flat_bec_shell", "flat_sbb_monitor", + "flat_beamline_state_manager", ] for action_name in utils_actions: @@ -2472,6 +2480,7 @@ class TestFlatToolbarActions: "flat_terminal": "BecConsole", "flat_bec_shell": "BECShell", "flat_sbb_monitor": "SBBMonitor", + "flat_beamline_state_manager": "BeamlineStateManager", } for action_name, widget_type in utils_action_mapping.items(): diff --git a/tests/unit_tests/test_main_widnow.py b/tests/unit_tests/test_main_widnow.py index 2b0fd40e..6c4261c3 100644 --- a/tests/unit_tests/test_main_widnow.py +++ b/tests/unit_tests/test_main_widnow.py @@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch import pytest from qtpy.QtCore import QEvent, QPoint, QPointF -from qtpy.QtGui import QEnterEvent +from qtpy.QtGui import QEnterEvent, QStandardItem from qtpy.QtWidgets import QApplication, QFrame, QLabel from bec_widgets.widgets.containers.main_window.addons.hover_widget import ( @@ -74,6 +74,10 @@ def test_event_consumes_status_tip(bec_main_window): assert bec_main_window.event(status_tip_event) is True +def test_event_ignores_non_qevent(bec_main_window): + assert bec_main_window.event(QStandardItem()) is False + + def test_get_launcher_from_qapp_returns_none_when_absent(bec_main_window): with patch.object( QApplication, "instance", return_value=SimpleNamespace(topLevelWidgets=lambda: [])