Compare commits

...

4 Commits

3 changed files with 769 additions and 94 deletions
@@ -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.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
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,169 @@ 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: list[str]
) -> None:
configs_by_name = {state.name: state for state in state_configs}
interlock = [configs_by_name[name] for name in interlock_names if name in configs_by_name]
interlock_set = set(interlock_names)
others = [state for state in state_configs if state.name not in interlock_set]
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
@@ -130,11 +210,11 @@ class _BeamlineStatePillDelegate(QStyledItemDelegate):
name = index.data(_BeamlineStateListModel.NameRole)
state_config = index.data(_BeamlineStateListModel.ConfigRole)
pill = BeamlineStatePill(parent=parent, state_name=name, 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.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
@@ -145,7 +225,6 @@ class _BeamlineStatePillDelegate(QStyledItemDelegate):
name = index.data(_BeamlineStateListModel.NameRole)
state_config = index.data(_BeamlineStateListModel.ConfigRole)
editor.set_state_name(str(name))
editor.idle_card_background = self._manager.idle_card_background
editor.set_state_config(state_config)
def updateEditorGeometry( # noqa: N802
@@ -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:
@@ -224,7 +305,6 @@ class BeamlineStateManager(BECWidget, QWidget):
client=None,
config: ConnectionConfig | None = None,
gui_id: str | None = None,
idle_card_background: bool = False,
**kwargs,
) -> None:
super().__init__(
@@ -238,8 +318,11 @@ class BeamlineStateManager(BECWidget, QWidget):
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._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,26 +352,14 @@ 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()
@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 = 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)
@@ -305,18 +376,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 +410,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 +436,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 +525,101 @@ 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]
# Triggered states sort to the top of the interlock section; the sort is stable, so
# the configured state order is kept within each group.
interlock_order = sorted(
(name for name in self._state_order if name in self._interlock_states),
key=lambda name: not self._is_interlock_triggered(name),
)
self._model.set_states(state_configs, interlock_order)
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:
pill.set_scan_interlock(
self._interlock_states.get(name), self._is_interlock_triggered(name)
)
def _is_interlock_triggered(self, name: str) -> bool:
required_status = self._interlock_states.get(name)
if required_status is None or not self._interlock_enabled:
return False
pill = self._state_pills.get(name)
return pill is not None and pill._status != required_status
@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)
@@ -455,17 +631,40 @@ class BeamlineStateManager(BECWidget, QWidget):
visible_names = []
hidden_names = []
for name in self._state_order:
if self._is_state_visible(name):
# States watched by the scan interlock are exempt from filtering.
if name in self._interlock_states or self._is_state_visible(name):
visible_names.append(name)
else:
hidden_names.append(name)
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 +703,12 @@ 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:
if name in self._interlock_states:
# The triggered flag and the triggered-first ordering of the interlock section
# both follow the pill status.
self._refresh_view()
return
if self._selected_statuses is not None:
self._apply_filters()
@@ -572,6 +776,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()):
@@ -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)
+339 -17
View File
@@ -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."}, {})
@@ -289,21 +317,6 @@ def test_beamline_state_manager_preserves_expanded_pill_on_refresh(qtbot, mocked
assert not beamline_state_manager._state_pills["limits"]._settings.isHidden()
def test_beamline_state_manager_propagates_idle_card_background(qtbot, mocked_client):
idle_card_manager = create_widget(
qtbot, BeamlineStateManager, client=mocked_client, idle_card_background=True
)
idle_card_manager.update_available_states(
{"states": [_state("limits", "DeviceWithinLimitsState", {"device": "samx"})]}, {}
)
assert idle_card_manager._state_pills["limits"]._idle_card_background is True
idle_card_manager.idle_card_background = False
assert idle_card_manager._state_pills["limits"]._idle_card_background is False
def test_beamline_state_manager_filters_status(qtbot, mocked_client):
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
beamline_state_manager.update_available_states(
@@ -500,6 +513,315 @@ 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": "invalid", "label": "Out of limits."}, {}
)
beamline_state_manager._state_pills["shutter_open"].update_state(
{"name": "shutter_open", "status": "valid", "label": "Open."}, {}
)
beamline_state_manager._selected_statuses = {"valid"}
beamline_state_manager._apply_filters()
model = beamline_state_manager._model
assert not beamline_state_manager._view.isRowHidden(0)
assert not beamline_state_manager._view.isRowHidden(model.index_for_name("shutter_open").row())
assert beamline_state_manager._view.isRowHidden(2)
assert beamline_state_manager._view.isRowHidden(model.index_for_name("limits").row())
beamline_state_manager.clear_filters()
assert not beamline_state_manager._view.isRowHidden(2)
def test_beamline_state_manager_interlock_states_bypass_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": "invalid", "label": "Out of 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 not beamline_state_manager._view.isRowHidden(model.index_for_name("shutter_open").row())
assert beamline_state_manager._view.isRowHidden(model.index_for_name("limits").row())
assert "1 state is hidden" in beamline_state_manager._hidden_summary.text()
beamline_state_manager._device_filter_text = "nonexistent_device"
beamline_state_manager._apply_filters()
assert not beamline_state_manager._view.isRowHidden(model.index_for_name("shutter_open").row())
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_sorts_triggered_interlock_states_first(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={"limits": "valid", "shutter_open": "valid"}
),
)
model = beamline_state_manager._model
limits_pill = beamline_state_manager._state_pills["limits"]
def interlock_section():
return [model.data(model.index(row, 0), model.NameRole) for row in (1, 2)]
assert model.rowCount() == 3
assert interlock_section() == ["limits", "shutter_open"]
limits_pill.update_state({"name": "limits", "status": "valid", "label": "Within limits."}, {})
assert interlock_section() == ["shutter_open", "limits"]
assert beamline_state_manager._state_pills["limits"] is limits_pill
assert not limits_pill._interlock_triggered
limits_pill.update_state({"name": "limits", "status": "invalid", "label": "Out of limits."}, {})
assert interlock_section() == ["limits", "shutter_open"]
assert limits_pill._interlock_triggered
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__)