bec/bec_widgets
1
0 Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-10-28 16:11:27 +01:00
Code Activity

feat(beamline_states): add QML and QWidget prototypes with modernized UX for comparison

Browse Source
This commit is contained in:
Claude committed 2026-10-09 20:31:05 +00:00
1 parent 703513b175
commit cba33152e8
9 files changed
+2615
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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