diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py index 7500635e..dc93b8ea 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py @@ -6,7 +6,8 @@ from typing import Any from bec_lib import bl_states, messages from bec_lib.endpoints import MessageEndpoints from bec_qthemes import material_icon -from qtpy.QtCore import QAbstractListModel, QModelIndex, QSize, Qt +from qtpy.QtCore import QAbstractListModel, QModelIndex, QRect, QSize, Qt +from qtpy.QtGui import QColor, QPainter, QPen from qtpy.QtWidgets import ( QAbstractItemView, QApplication, @@ -25,8 +26,9 @@ from qtpy.QtWidgets import ( from bec_widgets.utils.bec_connector import ConnectionConfig from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.colors import get_accent_colors from bec_widgets.utils.error_popups import SafeProperty, SafeSlot -from bec_widgets.utils.toolbars.actions import MaterialIconAction +from bec_widgets.utils.toolbars.actions import MaterialIconAction, WidgetAction from bec_widgets.utils.toolbars.bundles import ToolbarBundle from bec_widgets.utils.toolbars.toolbar import ModularToolBar from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStatePill @@ -38,91 +40,167 @@ from bec_widgets.widgets.services.beamline_states.dialogs import ( class _BeamlineStateListModel(QAbstractListModel): - """Model owning beamline state row identity and configuration data.""" + """ + Model owning beamline state row identity, configuration data, and section headers. + + Rows are identified by ``("state", name)`` or ``("header", kind)`` keys so state rows and + section header rows share one diff-based update path. + """ NameRole = Qt.ItemDataRole.UserRole + 1 ConfigRole = Qt.ItemDataRole.UserRole + 2 + HeaderRole = Qt.ItemDataRole.UserRole + 3 + + INTERLOCK_HEADER = "interlock" + OTHERS_HEADER = "others" + HEADER_LABELS = { + INTERLOCK_HEADER: "Scan interlock states", + OTHERS_HEADER: "Not included in scan interlock", + } def __init__(self, parent: QWidget | None = None) -> None: super().__init__(parent) - self._state_order: list[str] = [] - self._state_rows: dict[str, int] = {} + self._row_keys: list[tuple[str, str]] = [] + self._row_indices: dict[tuple[str, str], int] = {} self._state_configs: dict[str, messages.BeamlineStateConfig] = {} def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: # noqa: N802 - return 0 if parent.isValid() else len(self._state_order) + return 0 if parent.isValid() else len(self._row_keys) def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole) -> Any: - if not index.isValid() or not 0 <= index.row() < len(self._state_order): + if not index.isValid() or not 0 <= index.row() < len(self._row_keys): + return None + kind, value = self._row_keys[index.row()] + if kind == "header": + if role == Qt.ItemDataRole.DisplayRole: + return self.HEADER_LABELS[value] + if role == self.HeaderRole: + return value return None - name = self._state_order[index.row()] if role in (Qt.ItemDataRole.DisplayRole, self.NameRole): - return name + return value if role == self.ConfigRole: - return self._state_configs.get(name) + return self._state_configs.get(value) return None def flags(self, index: QModelIndex) -> Qt.ItemFlag: - if not index.isValid(): + if not index.isValid() or not 0 <= index.row() < len(self._row_keys): + return Qt.ItemFlag.NoItemFlags + if self._row_keys[index.row()][0] == "header": return Qt.ItemFlag.NoItemFlags return Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable - def set_states(self, state_configs: list[messages.BeamlineStateConfig]) -> None: - new_order = [state.name for state in state_configs] + def set_states( + self, state_configs: list[messages.BeamlineStateConfig], interlock_names: set[str] + ) -> None: + interlock = [state for state in state_configs if state.name in interlock_names] + others = [state for state in state_configs if state.name not in interlock_names] + new_keys: list[tuple[str, str]] = [] + if interlock: + new_keys.append(("header", self.INTERLOCK_HEADER)) + new_keys.extend(("state", state.name) for state in interlock) + if others: + new_keys.append(("header", self.OTHERS_HEADER)) + new_keys.extend(("state", state.name) for state in others) + new_configs = {state.name: state for state in state_configs} + new_key_set = set(new_keys) for row in reversed( - [row for row, name in enumerate(self._state_order) if name not in new_configs] + [row for row, key in enumerate(self._row_keys) if key not in new_key_set] ): self.beginRemoveRows(QModelIndex(), row, row) - name = self._state_order.pop(row) - self._state_configs.pop(name, None) + kind, value = self._row_keys.pop(row) + if kind == "state": + self._state_configs.pop(value, None) self.endRemoveRows() self._rebuild_rows() - for target_row, name in enumerate(new_order): - if name not in self._state_rows: + for target_row, key in enumerate(new_keys): + if key not in self._row_indices: self.beginInsertRows(QModelIndex(), target_row, target_row) - self._state_order.insert(target_row, name) - self._state_configs[name] = new_configs[name] + self._row_keys.insert(target_row, key) + if key[0] == "state": + self._state_configs[key[1]] = new_configs[key[1]] self.endInsertRows() self._rebuild_rows() continue - current_row = self._state_rows[name] + current_row = self._row_indices[key] if current_row != target_row: destination_row = target_row if current_row > target_row else target_row + 1 self.beginMoveRows( QModelIndex(), current_row, current_row, QModelIndex(), destination_row ) - self._state_order.insert(target_row, self._state_order.pop(current_row)) + self._row_keys.insert(target_row, self._row_keys.pop(current_row)) self.endMoveRows() self._rebuild_rows() - if self._state_configs.get(name) != new_configs[name]: - self._state_configs[name] = new_configs[name] - index = self.index(self._state_rows[name], 0) + if key[0] == "state" and self._state_configs.get(key[1]) != new_configs[key[1]]: + self._state_configs[key[1]] = new_configs[key[1]] + index = self.index(self._row_indices[key], 0) self.dataChanged.emit(index, index, [self.ConfigRole]) def _rebuild_rows(self) -> None: - self._state_rows = {name: row for row, name in enumerate(self._state_order)} + self._row_indices = {key: row for row, key in enumerate(self._row_keys)} def index_for_name(self, name: str) -> QModelIndex: - row = self._state_rows.get(name) + row = self._row_indices.get(("state", name)) if row is None: return QModelIndex() return self.index(row, 0) class _BeamlineStatePillDelegate(QStyledItemDelegate): - """Delegate that provides BeamlineStatePill persistent editors for list rows.""" + """Delegate painting section headers and providing BeamlineStatePill persistent editors.""" + + HEADER_HEIGHT = 26 def __init__(self, manager: "BeamlineStateManager") -> None: super().__init__(manager) self._manager = manager - def paint(self, _painter, _option: QStyleOptionViewItem, _index: QModelIndex) -> None: - return + def paint(self, painter, option: QStyleOptionViewItem, index: QModelIndex) -> None: + kind = index.data(_BeamlineStateListModel.HeaderRole) + if kind is None: + return + + colors = BeamlineStatePill._state_colors("unknown") + armed = ( + kind == _BeamlineStateListModel.INTERLOCK_HEADER and self._manager._interlock_enabled + ) + label_color = QColor(colors["foreground"] if armed else colors["muted"]) + icon_name = ( + "lock" if kind == _BeamlineStateListModel.INTERLOCK_HEADER else "lock_open_right" + ) + pixmap = material_icon(icon_name, size=(14, 14), color=label_color, filled=armed) + text = str(index.data(Qt.ItemDataRole.DisplayRole)) + + painter.save() + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + rect = option.rect.adjusted(8, 0, -8, 0) + + font = painter.font() + font.setPointSizeF(max(7.0, font.pointSizeF() * 0.85)) + font.setBold(True) + painter.setFont(font) + + icon_width = int(pixmap.width() / pixmap.devicePixelRatio()) + icon_height = int(pixmap.height() / pixmap.devicePixelRatio()) + painter.drawPixmap(rect.left(), rect.center().y() - icon_height // 2, pixmap) + + text_left = rect.left() + icon_width + 6 + painter.setPen(label_color) + text_rect = QRect(text_left, rect.top(), rect.width() - icon_width - 6, rect.height()) + painter.drawText( + text_rect, Qt.AlignmentFlag.AlignVCenter | Qt.AlignmentFlag.AlignLeft, text + ) + + rule_left = text_left + painter.fontMetrics().horizontalAdvance(text) + 8 + if rule_left < rect.right(): + painter.setPen(QPen(QColor(colors["border"]), 1)) + painter.drawLine(rule_left, rect.center().y(), rect.right(), rect.center().y()) + painter.restore() def createEditor( # noqa: N802 self, parent: QWidget, _option: QStyleOptionViewItem, index: QModelIndex @@ -135,6 +213,7 @@ class _BeamlineStatePillDelegate(QStyledItemDelegate): 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.scan_interlock_toggle_requested.connect(self._manager._on_interlock_toggle_requested) pill.row_height_changed.connect(lambda name=name: self._manager._sync_pill_item_size(name)) self._manager._state_pills[str(name)] = pill return pill @@ -154,6 +233,8 @@ class _BeamlineStatePillDelegate(QStyledItemDelegate): editor.setGeometry(option.rect) def sizeHint(self, option: QStyleOptionViewItem, index: QModelIndex) -> QSize: # noqa: N802 + if index.data(_BeamlineStateListModel.HeaderRole) is not None: + return QSize(120, self.HEADER_HEIGHT) name = index.data(_BeamlineStateListModel.NameRole) pill = self._manager._state_pills.get(str(name)) if pill is not None: @@ -240,6 +321,11 @@ class BeamlineStateManager(BECWidget, QWidget): self._hidden_expanded = False self._idle_card_background = False self.idle_card_background = idle_card_background + self._scan_interlock = self.client.builtin_actors.scan_interlock + self._interlock_enabled = False + self._interlock_states: dict[str, str] = {} + self._updating_interlock_action = False + self._interlock_action_armed: bool | None = None self._empty_label = QLabel( "No beamline states available.\n Add new state from toolbar or CLI.", self @@ -269,6 +355,11 @@ class BeamlineStateManager(BECWidget, QWidget): self.bec_dispatcher.connect_slot( self.update_available_states, MessageEndpoints.available_beamline_states() ) + self.bec_dispatcher.connect_slot( + self._refresh_scan_interlock, + MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"), + ) + self._refresh_scan_interlock() self.refresh_states() self._refresh_hidden_summary() @@ -305,18 +396,33 @@ class BeamlineStateManager(BECWidget, QWidget): collapse_all = MaterialIconAction( "collapse_all", "Collapse all states", filled=True, parent=self ) + spacer = QWidget(self) + spacer.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred) + spacer_action = WidgetAction(widget=spacer, adjust_size=False, parent=self) + scan_interlock = MaterialIconAction( + "no_encryption", + "Scan interlock", + checkable=True, + filled=True, + label_text="Scan interlock", + text_position="beside", + 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) collapse_all.action.triggered.connect(self.collapse_all) + scan_interlock.action.toggled.connect(self._on_interlock_action_toggled) 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("collapse_all", collapse_all) + toolbar.components.add_safe("scan_interlock_spacer", spacer_action) + toolbar.components.add_safe("scan_interlock", scan_interlock) bundle = ToolbarBundle("beamline_state_manager", toolbar.components) bundle.add_action("add_state") @@ -324,8 +430,15 @@ class BeamlineStateManager(BECWidget, QWidget): bundle.add_action("filter_devices") bundle.add_action("clear_filters") bundle.add_action("collapse_all") + bundle.add_action("scan_interlock_spacer") + bundle.add_separator() + bundle.add_action("scan_interlock") toolbar.add_bundle(bundle) toolbar.show_bundles(["beamline_state_manager"]) + if spacer_action.container is not None: + spacer_action.container.setSizePolicy( + QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred + ) return toolbar @SafeSlot(str) @@ -343,6 +456,9 @@ class BeamlineStateManager(BECWidget, QWidget): ) for pill in self._state_pills.values(): pill.apply_theme(_theme) + self._interlock_action_armed = None + self._sync_interlock_action() + self._view.viewport().update() self._refresh_hidden_summary() @SafeSlot() @@ -429,21 +545,92 @@ class BeamlineStateManager(BECWidget, QWidget): 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()} state_configs: list[messages.BeamlineStateConfig] = content.get("states", []) - if state_configs == list(self._state_configs.values()): - self._apply_filters() - return self._state_configs = {state.name: state for state in state_configs} self._state_order = [state.name for state in state_configs] - self._model.set_states(state_configs) + self._refresh_view() + + @SafeSlot(dict, dict) + def _refresh_scan_interlock( + self, _content: dict[str, Any] | None = None, _metadata: dict[str, Any] | None = None + ) -> None: + """Re-read the scan-interlock state from BEC and refresh the displayed pills.""" + try: + self._interlock_enabled = bool(self._scan_interlock.enabled) + self._interlock_states = dict(self._scan_interlock.states_watched) + except Exception as exc: + QMessageBox.warning(self, "Scan Interlock Unavailable", str(exc)) + return + self._refresh_view() + + def _refresh_view(self) -> None: + """Render the current state and scan-interlock bookkeeping in one pass.""" + self._sync_interlock_action() + expanded_names = {name for name, pill in self._state_pills.items() if pill.is_expanded()} + state_configs = [self._state_configs[name] for name in self._state_order] + self._model.set_states(state_configs, set(self._interlock_states)) self._open_persistent_editors(expanded_names) + for name, pill in self._state_pills.items(): + self._apply_interlock_to_pill(name, pill) self._apply_filters() + self._view.viewport().update() + + def _sync_interlock_action(self) -> None: + action = self._toolbar.components.get_action("scan_interlock").action + self._updating_interlock_action = True + try: + action.setChecked(self._interlock_enabled) + finally: + self._updating_interlock_action = False + if self._interlock_enabled == self._interlock_action_armed: + return + self._interlock_action_armed = self._interlock_enabled + if self._interlock_enabled: + action.setIcon( + material_icon( + "lock", size=(20, 20), filled=True, color=get_accent_colors().success.name() + ) + ) + action.setToolTip("Scan interlock is armed. Click to disable it.") + else: + action.setIcon(material_icon("no_encryption", size=(20, 20), filled=True)) + action.setToolTip("Scan interlock is disabled. Click to arm it.") + + def _apply_interlock_to_pill(self, name: str, pill: BeamlineStatePill) -> None: + required_status = self._interlock_states.get(name) + triggered = ( + self._interlock_enabled + and required_status is not None + and pill._status != required_status + ) + pill.set_scan_interlock(required_status, triggered) + + @SafeSlot(bool) + def _on_interlock_action_toggled(self, checked: bool) -> None: + if self._updating_interlock_action: + return + try: + self._scan_interlock.enabled = bool(checked) + except Exception as exc: + QMessageBox.warning(self, "Cannot Toggle Scan Interlock", str(exc)) + self._refresh_scan_interlock() + + @SafeSlot(str, bool) + def _on_interlock_toggle_requested(self, state_name: str, include: bool) -> None: + try: + if include: + self._scan_interlock.add_state_to_interlock(state_name, "valid") + else: + self._scan_interlock.remove_state_from_interlock(state_name) + except Exception as exc: + QMessageBox.warning(self, "Cannot Update Scan Interlock", str(exc)) 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) + if index.data(_BeamlineStateListModel.HeaderRole) is not None: + continue self._view.openPersistentEditor(index) name = str(index.data(_BeamlineStateListModel.NameRole)) pill = self._state_pills.get(name) @@ -462,10 +649,32 @@ class BeamlineStateManager(BECWidget, QWidget): 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) + shown_interlock = 0 + shown_others = 0 + for row in range(self._model.rowCount()): + index = self._model.index(row, 0) + if index.data(_BeamlineStateListModel.HeaderRole) is not None: + continue + name = str(index.data(_BeamlineStateListModel.NameRole)) + shown = name in visible_set or show_hidden + self._view.setRowHidden(row, not shown) + if shown: + if name in self._interlock_states: + shown_interlock += 1 + else: + shown_others += 1 self._sync_pill_item_size(name) + for row in range(self._model.rowCount()): + index = self._model.index(row, 0) + kind = index.data(_BeamlineStateListModel.HeaderRole) + if kind is None: + continue + shown_count = ( + shown_interlock + if kind == _BeamlineStateListModel.INTERLOCK_HEADER + else shown_others + ) + self._view.setRowHidden(row, shown_count == 0) self._empty_label.setVisible( not visible_names and not (self._hidden_expanded and hidden_names) ) @@ -504,7 +713,10 @@ class BeamlineStateManager(BECWidget, QWidget): return True @SafeSlot(str, str, str) - def _on_pill_state_changed(self, _name: str, _status: str, _label: str) -> None: + def _on_pill_state_changed(self, name: str, _status: str, _label: str) -> None: + pill = self._state_pills.get(name) + if pill is not None: + self._apply_interlock_to_pill(name, pill) if self._selected_statuses is not None: self._apply_filters() @@ -572,6 +784,10 @@ class BeamlineStateManager(BECWidget, QWidget): self.bec_dispatcher.disconnect_slot( self.update_available_states, MessageEndpoints.available_beamline_states() ) + self.bec_dispatcher.disconnect_slot( + self._refresh_scan_interlock, + MessageEndpoints.builtin_actor_update_notif("ScanInterlockActor"), + ) for row in range(self._model.rowCount()): self._view.closePersistentEditor(self._model.index(row, 0)) for pill in list(self._state_pills.values()): diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py index 7a4e74a2..19654e2b 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py @@ -5,7 +5,7 @@ from typing import Any from bec_lib import bl_states, messages from bec_lib.endpoints import MessageEndpoints from bec_qthemes import material_icon -from qtpy.QtCore import Qt, Signal +from qtpy.QtCore import Property, QEasingCurve, QPropertyAnimation, Qt, Signal from qtpy.QtGui import QColor, QMouseEvent, QPalette from qtpy.QtWidgets import ( QApplication, @@ -62,6 +62,7 @@ class BeamlineStatePill(BECWidget, QWidget): state_changed = Signal(str, str, str) update_requested = Signal(str, object) remove_requested = Signal(str) + scan_interlock_toggle_requested = Signal(str, bool) row_height_changed = Signal() _STATUS_LABELS = BEAMLINE_STATE_STATUS_LABELS @@ -93,6 +94,10 @@ class BeamlineStatePill(BECWidget, QWidget): self._label = "No state information available." self._expanded = False self._idle_card_background = False + self._interlock_required_status: str | None = None + self._interlock_triggered = False + self._interlock_pulse = 0.0 + self._header_icon_cache_key: tuple | None = None self._populating_settings = False self._settings_baseline: dict[str, Any] = {} self._settings_dirty_fields: set[str] = set() @@ -137,12 +142,24 @@ class BeamlineStatePill(BECWidget, QWidget): self._detail_label.setTextFormat(Qt.TextFormat.PlainText) self._detail_label.setWordWrap(True) self._detail_label.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True) + self._interlock_button = QToolButton(self) + self._interlock_button.setObjectName("beamline_state_interlock") + self._interlock_button.setAutoRaise(True) + self._interlock_button.setCursor(Qt.CursorShape.PointingHandCursor) + self._interlock_button.clicked.connect(self._emit_interlock_toggle_requested) 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) + self._interlock_animation = QPropertyAnimation(self, b"interlock_pulse", self) + self._interlock_animation.setDuration(1400) + self._interlock_animation.setStartValue(0.0) + self._interlock_animation.setEndValue(1.0) + self._interlock_animation.setEasingCurve(QEasingCurve.Type.Linear) + self._interlock_animation.setLoopCount(-1) + text_layout = QVBoxLayout() text_layout.setContentsMargins(0, 0, 0, 0) text_layout.setSpacing(1) @@ -156,6 +173,7 @@ class BeamlineStatePill(BECWidget, QWidget): 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._interlock_button) header_layout.addWidget(self._expand_button) self._settings = QWidget(self) @@ -275,6 +293,49 @@ class BeamlineStatePill(BECWidget, QWidget): """Set whether idle collapsed pills keep the status-tinted card background.""" self.idle_card_background = enabled + @Property(float) + def interlock_pulse(self) -> float: + """Animation phase in [0, 1] driving the triggered scan-interlock highlight.""" + return self._interlock_pulse + + @interlock_pulse.setter + def interlock_pulse(self, phase: float) -> None: + self._interlock_pulse = float(phase) + if self._interlock_triggered: + self._apply_visual_state() + + def set_scan_interlock(self, required_status: str | None, triggered: bool) -> None: + """ + Set the scan-interlock participation of this pill. + + Args: + required_status: Status the scan interlock requires for this state, or ``None`` + if the state is not included in the scan interlock. + triggered: Whether the armed scan interlock is currently tripped by this state. + """ + triggered = bool(triggered) and required_status is not None + if (required_status, triggered) == ( + self._interlock_required_status, + self._interlock_triggered, + ): + return + self._interlock_required_status = required_status + self._interlock_triggered = triggered + if triggered: + if self._interlock_animation.state() != QPropertyAnimation.State.Running: + self._interlock_animation.start() + else: + self._interlock_animation.stop() + self._interlock_pulse = 0.0 + self._apply_visual_state() + + @SafeSlot() + def _emit_interlock_toggle_requested(self) -> None: + if self._state_name is None: + return + include = self._interlock_required_status is None + self.scan_interlock_toggle_requested.emit(self._state_name, include) + def _refresh_latest_state(self) -> None: if self._state_name is None: return @@ -314,8 +375,8 @@ class BeamlineStatePill(BECWidget, QWidget): def _apply_visual_state(self) -> None: colors = self._state_colors(self._status) accent = colors["accent"] - on_accent = colors["on_accent"] - active_card = self._expanded + included = self._interlock_required_status is not None + active_card = self._expanded or included border = colors["border"] if self._idle_card_background else "transparent" background = colors["background"] if self._idle_card_background else "transparent" card_gradient = ( @@ -330,29 +391,42 @@ class BeamlineStatePill(BECWidget, QWidget): background = card_gradient border = colors["card_border"] hover_background = card_gradient - self._shadow.setColor(QColor(colors["shadow"])) - self._shadow.setBlurRadius(int(colors["shadow_blur"])) + hover_border = colors["card_border"] + border_width = 1 + shadow_color = QColor(colors["shadow"]) + shadow_blur = int(colors["shadow_blur"]) + shadow_enabled = active_card + if self._interlock_triggered: + flash = 1.0 - abs(2.0 * self._interlock_pulse - 1.0) + background = self._traveling_gradient( + colors["card_background"], colors["interlock_band"], self._interlock_pulse + ) + hover_background = background + border = rgba(QColor(colors["interlock_trigger"]), 110 + int(145 * flash)) + hover_border = border + border_width = 2 + shadow_color = QColor(colors["interlock_trigger"]) + shadow_color.setAlpha(150) + shadow_blur = int(colors["shadow_blur"]) + 10 + shadow_enabled = True + self._shadow.setColor(shadow_color) + self._shadow.setBlurRadius(shadow_blur) self._shadow.setOffset(0, int(colors["shadow_y_offset"])) - self._shadow.setEnabled(active_card) + self._shadow.setEnabled(shadow_enabled) - 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._update_header_icons(colors) 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: {border_width}px solid {border};" f"border-radius: {'12px' if active_card else '8px'};" "}" "#BeamlineStatePill:hover {" f"background: {hover_background};" - f"border: 1px solid {colors['card_border']};" + f"border: {border_width}px solid {hover_border};" "border-radius: 12px;" "}" "QWidget#beamline_state_header {" @@ -379,6 +453,15 @@ class BeamlineStatePill(BECWidget, QWidget): f"color: {colors['muted']};" "font-size: 11px;" "}" + "QToolButton#beamline_state_interlock {" + "background: transparent;" + "border: none;" + "border-radius: 4px;" + "padding: 2px;" + "}" + "QToolButton#beamline_state_interlock:hover {" + f"background-color: {colors['button_hover']};" + "}" "QWidget#beamline_state_settings {" "background: transparent;" f"border-top: 1px solid {colors['border']};" @@ -401,6 +484,64 @@ class BeamlineStatePill(BECWidget, QWidget): "}" ) + def _update_header_icons(self, colors: dict[str, str]) -> None: + cache_key = ( + self._status, + self._expanded, + self._interlock_required_status, + self._interlock_triggered, + get_theme_name(), + ) + if cache_key == self._header_icon_cache_key: + return + self._header_icon_cache_key = cache_key + + self._icon_label.setPixmap( + material_icon( + self._STATUS_ICONS[self._status], + size=(20, 20), + color=colors["on_accent"], + filled=True, + ) + ) + expand_icon = "expand_less" if self._expanded else "expand_more" + self._expand_button.setIcon(material_icon(expand_icon, size=(20, 20))) + if self._interlock_required_status is not None: + lock_color = ( + colors["interlock_trigger"] if self._interlock_triggered else colors["foreground"] + ) + self._interlock_button.setIcon( + material_icon("lock", size=(18, 18), color=lock_color, filled=True) + ) + self._interlock_button.setToolTip( + f"Watched by the scan interlock (required status: " + f"{self._interlock_required_status}).\n" + "Click to remove this state from the scan interlock." + ) + else: + self._interlock_button.setIcon( + material_icon("lock_open_right", size=(18, 18), color=colors["muted"]) + ) + self._interlock_button.setToolTip( + "Not watched by the scan interlock.\n" + "Click to add this state to the scan interlock." + ) + + @staticmethod + def _traveling_gradient(base: str, highlight: str, phase: float) -> str: + """Return a horizontal QSS gradient with a highlight band centered at ``phase``.""" + span = 0.3 + center = phase * (1.0 + 2.0 * span) - span + stops: list[tuple[float, str]] = [(0.0, base)] + for position, color in ((center - span, base), (center, highlight), (center + span, base)): + position = min(1.0, max(0.0, position)) + if position - stops[-1][0] > 0.001: + stops.append((position, color)) + if 1.0 - stops[-1][0] > 0.001: + stops.append((1.0, base)) + body = ", ".join(f"stop:{position:.4f} {color}" for position, color in stops) + return f"qlineargradient(x1:0, y1:0, x2:1, y2:0, {body})" + @SafeSlot() def _toggle_expanded(self) -> None: self.set_expanded(not self._expanded) @@ -610,12 +751,16 @@ class BeamlineStatePill(BECWidget, QWidget): "dirty_border": Colors._blend(border, warning, 0.70).name(), "foreground": foreground.name(), "muted": Colors._blend(card_bg, foreground, 0.66).name(), + "interlock_trigger": accents.emergency.name(), + "interlock_band": rgba(accents.emergency, 64 if light_theme else 110), + "button_hover": rgba(accent, 28 if light_theme else 48), "shadow": "#00000024" if light_theme else "#00000078", "shadow_blur": "24" if light_theme else "18", "shadow_y_offset": "3" if light_theme else "2", } def cleanup(self) -> None: + self._interlock_animation.stop() if self._state_name is not None: self.bec_dispatcher.disconnect_slot( self.update_state, MessageEndpoints.beamline_state(self._state_name) diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py index 4ce717be..f2d10e88 100644 --- a/tests/unit_tests/test_beamline_state_pill.py +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -1,7 +1,8 @@ import shiboken6 from bec_lib import bl_states, messages -from qtpy.QtCore import QCoreApplication, QEvent, Qt -from qtpy.QtWidgets import QMessageBox +from qtpy.QtCore import QCoreApplication, QEvent, QRect, Qt +from qtpy.QtGui import QPainter, QPixmap +from qtpy.QtWidgets import QMessageBox, QStyleOptionViewItem from bec_widgets.utils.toolbars.toolbar import ModularToolBar from bec_widgets.utils.widget_io import WidgetIO @@ -51,6 +52,33 @@ def _shutter_state( return _wire_state(bl_states.ShutterState, config) +class _FakeScanInterlock: + def __init__(self, enabled: bool = False, states_watched: dict[str, str] | None = None): + self.enabled = enabled + self._states = dict(states_watched or {}) + self.added: list[tuple[str, str]] = [] + self.removed: list[str] = [] + + @property + def states_watched(self) -> dict[str, str]: + return dict(self._states) + + def add_state_to_interlock(self, state_name: str, required_value: str = "valid") -> None: + self.added.append((state_name, required_value)) + self._states[state_name] = required_value + + def remove_state_from_interlock(self, state_name: str) -> None: + self.removed.append(state_name) + self._states.pop(state_name, None) + + +def _install_fake_scan_interlock( + manager: BeamlineStateManager, fake_interlock: _FakeScanInterlock +) -> None: + manager._scan_interlock = fake_interlock + manager._refresh_scan_interlock() + + def test_beamline_state_pill_updates_from_message(qtbot, mocked_client): pill = create_widget(qtbot, BeamlineStatePill, state_name="shutter_open", client=mocked_client) pill.update_state({"name": "shutter_open", "status": "valid", "label": "Shutter is open."}, {}) @@ -500,6 +528,254 @@ def test_beamline_state_manager_removes_state(qtbot, mocked_client, monkeypatch) assert mocked_client.beamline_states.deleted == "limits" +def test_beamline_state_pill_emits_interlock_toggle_request(qtbot, mocked_client): + pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client) + + with qtbot.waitSignal(pill.scan_interlock_toggle_requested) as include_signal: + pill._interlock_button.click() + + assert include_signal.args == ["limits", True] + + pill.set_scan_interlock("valid", False) + + with qtbot.waitSignal(pill.scan_interlock_toggle_requested) as exclude_signal: + pill._interlock_button.click() + + assert exclude_signal.args == ["limits", False] + + +def test_beamline_state_pill_included_state_forces_card_background(qtbot, mocked_client): + pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client) + colors = BeamlineStatePill._state_colors("unknown") + + assert f"border: 1px solid {colors['card_border']}" not in pill.styleSheet().split(":hover")[0] + assert not pill._shadow.isEnabled() + + pill.set_scan_interlock("valid", False) + + assert f"border: 1px solid {colors['card_border']}" in pill.styleSheet() + assert pill._shadow.isEnabled() + assert "Watched by the scan interlock" in pill._interlock_button.toolTip() + + pill.set_scan_interlock(None, False) + + assert not pill._shadow.isEnabled() + assert "Not watched by the scan interlock" in pill._interlock_button.toolTip() + + +def test_beamline_state_pill_triggered_interlock_animates(qtbot, mocked_client): + pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client) + + pill.set_scan_interlock("valid", True) + + assert pill._interlock_animation.state() == pill._interlock_animation.State.Running + pill.interlock_pulse = 0.5 + + stylesheet = pill.styleSheet() + assert "border: 2px solid" in stylesheet + assert "qlineargradient" in stylesheet + + pill.set_scan_interlock("valid", False) + + assert pill._interlock_animation.state() == pill._interlock_animation.State.Stopped + assert pill._interlock_pulse == 0.0 + assert "border: 2px solid" not in pill.styleSheet() + assert "border: 1px solid" in pill.styleSheet() + + +def test_beamline_state_pill_traveling_gradient_keeps_stops_sorted(): + for phase in (0.0, 0.2, 0.5, 0.8, 1.0): + gradient = BeamlineStatePill._traveling_gradient("#101010", "#ff0000", phase) + positions = [ + float(stop.split()[0].removeprefix("stop:")) + for stop in gradient.split(", ") + if stop.startswith("stop:") + ] + assert positions == sorted(positions) + assert positions[0] == 0.0 + assert positions[-1] == 1.0 + + +def test_beamline_state_manager_toolbar_scan_interlock_on_right(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + + bundle = beamline_state_manager._toolbar.bundles["beamline_state_manager"] + assert list(bundle.bundle_actions)[-3:] == [ + "scan_interlock_spacer", + "separator", + "scan_interlock", + ] + + spacer_action = beamline_state_manager._toolbar.components.get_action("scan_interlock_spacer") + assert spacer_action.container.sizePolicy().horizontalPolicy() == ( + spacer_action.container.sizePolicy().Policy.Expanding + ) + + interlock_action = beamline_state_manager._toolbar.components.get_action("scan_interlock") + assert interlock_action.action.isCheckable() + assert not interlock_action.action.isChecked() + assert "disabled" in interlock_action.action.toolTip() + + +def test_beamline_state_manager_groups_states_under_headers(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states( + {"states": [_limits_state(), _shutter_state()]}, {} + ) + model = beamline_state_manager._model + + assert model.rowCount() == 2 + assert model.data(model.index(0, 0), model.HeaderRole) is None + + _install_fake_scan_interlock( + beamline_state_manager, _FakeScanInterlock(states_watched={"shutter_open": "valid"}) + ) + + assert model.rowCount() == 4 + assert model.data(model.index(0, 0), model.HeaderRole) == model.INTERLOCK_HEADER + assert model.data(model.index(0, 0), Qt.ItemDataRole.DisplayRole) == "Scan interlock states" + assert model.data(model.index(1, 0), model.NameRole) == "shutter_open" + assert model.data(model.index(2, 0), model.HeaderRole) == model.OTHERS_HEADER + assert ( + model.data(model.index(2, 0), Qt.ItemDataRole.DisplayRole) + == "Not included in scan interlock" + ) + assert model.data(model.index(3, 0), model.NameRole) == "limits" + assert sorted(beamline_state_manager._state_pills) == ["limits", "shutter_open"] + + _install_fake_scan_interlock( + beamline_state_manager, + _FakeScanInterlock(states_watched={"shutter_open": "valid", "limits": "valid"}), + ) + + assert model.rowCount() == 3 + assert model.data(model.index(0, 0), model.HeaderRole) == model.INTERLOCK_HEADER + assert model.data(model.index(1, 0), model.NameRole) == "limits" + assert model.data(model.index(2, 0), model.NameRole) == "shutter_open" + + _install_fake_scan_interlock(beamline_state_manager, _FakeScanInterlock()) + + assert model.rowCount() == 2 + assert model.data(model.index(0, 0), model.HeaderRole) is None + + +def test_beamline_state_manager_header_visibility_follows_filters(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states( + {"states": [_limits_state(), _shutter_state()]}, {} + ) + _install_fake_scan_interlock( + beamline_state_manager, _FakeScanInterlock(states_watched={"shutter_open": "valid"}) + ) + + beamline_state_manager._state_pills["limits"].update_state( + {"name": "limits", "status": "valid", "label": "Within limits."}, {} + ) + beamline_state_manager._state_pills["shutter_open"].update_state( + {"name": "shutter_open", "status": "invalid", "label": "Closed."}, {} + ) + beamline_state_manager._selected_statuses = {"valid"} + beamline_state_manager._apply_filters() + + model = beamline_state_manager._model + assert beamline_state_manager._view.isRowHidden(0) + assert beamline_state_manager._view.isRowHidden(model.index_for_name("shutter_open").row()) + assert not beamline_state_manager._view.isRowHidden(2) + assert not beamline_state_manager._view.isRowHidden(model.index_for_name("limits").row()) + + beamline_state_manager.clear_filters() + + assert not beamline_state_manager._view.isRowHidden(0) + + +def test_beamline_state_manager_paints_section_headers(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states( + {"states": [_limits_state(), _shutter_state()]}, {} + ) + _install_fake_scan_interlock( + beamline_state_manager, + _FakeScanInterlock(enabled=True, states_watched={"shutter_open": "valid"}), + ) + + model = beamline_state_manager._model + delegate = beamline_state_manager._delegate + header_index = model.index(0, 0) + + assert delegate.sizeHint(QStyleOptionViewItem(), header_index).height() == ( + delegate.HEADER_HEIGHT + ) + assert model.flags(header_index) == Qt.ItemFlag.NoItemFlags + + target = QPixmap(400, delegate.HEADER_HEIGHT) + target.fill(Qt.GlobalColor.transparent) + painter = QPainter(target) + option = QStyleOptionViewItem() + option.rect = QRect(0, 0, 400, delegate.HEADER_HEIGHT) + delegate.paint(painter, option, header_index) + delegate.paint(painter, option, model.index(2, 0)) + painter.end() + + +def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states({"states": [_shutter_state()]}, {}) + fake_interlock = _FakeScanInterlock(enabled=True, states_watched={"shutter_open": "valid"}) + _install_fake_scan_interlock(beamline_state_manager, fake_interlock) + + pill = beamline_state_manager._state_pills["shutter_open"] + assert pill._interlock_required_status == "valid" + assert pill._interlock_triggered + + pill.update_state({"name": "shutter_open", "status": "valid", "label": "Open."}, {}) + + assert not pill._interlock_triggered + + pill.update_state({"name": "shutter_open", "status": "invalid", "label": "Closed."}, {}) + + assert pill._interlock_triggered + + fake_interlock.enabled = False + beamline_state_manager._refresh_scan_interlock() + + assert not pill._interlock_triggered + assert pill._interlock_required_status == "valid" + + +def test_beamline_state_manager_toolbar_toggle_writes_backend(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + fake_interlock = _FakeScanInterlock() + _install_fake_scan_interlock(beamline_state_manager, fake_interlock) + interlock_action = beamline_state_manager._toolbar.components.get_action("scan_interlock") + + interlock_action.action.setChecked(True) + + assert fake_interlock.enabled is True + assert beamline_state_manager._interlock_enabled is True + assert "armed" in interlock_action.action.toolTip() + + interlock_action.action.setChecked(False) + + assert fake_interlock.enabled is False + assert "disabled" in interlock_action.action.toolTip() + + +def test_beamline_state_manager_pill_toggle_calls_backend(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states({"states": [_limits_state()]}, {}) + fake_interlock = _FakeScanInterlock() + _install_fake_scan_interlock(beamline_state_manager, fake_interlock) + + beamline_state_manager._state_pills["limits"]._interlock_button.click() + + assert fake_interlock.added == [("limits", "valid")] + + beamline_state_manager._refresh_scan_interlock() + beamline_state_manager._state_pills["limits"]._interlock_button.click() + + assert fake_interlock.removed == ["limits"] + + def test_add_beamline_state_dialog_uses_generated_widgets_and_normalizes_name(qtbot, mocked_client): add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client) limits_index = add_state_dialog._type_combo.findText(bl_states.DeviceWithinLimitsState.__name__)