feat: add aggregated state for state manager widget

This commit is contained in:
2026-07-20 13:46:27 +02:00
parent 951fe5fbcc
commit 9a73ae2196
4 changed files with 455 additions and 8 deletions
@@ -0,0 +1,267 @@
from __future__ import annotations
import json
from typing import Any
from bec_lib import bl_states
from bec_qthemes import material_icon
from qtpy.QtCore import Qt, Signal
from qtpy.QtGui import QBrush
from qtpy.QtWidgets import (
QAbstractItemView,
QComboBox,
QHBoxLayout,
QHeaderView,
QLabel,
QSizePolicy,
QToolButton,
QTreeWidget,
QTreeWidgetItem,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.colors import get_accent_colors
class AggregatedStateConfigEditor(QWidget):
"""Compact editor and rule inspector for an aggregated beamline state."""
changed = Signal()
_EVALUATION_METHODS = (
("Any — one or more labels", "any"),
("All — every label", "all"),
("Exclusive — exactly one label", "exclusive"),
("Disabled — no validation", None),
)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent=parent)
self._model = bl_states.AggregatedStateConfig
self._config: bl_states.AggregatedStateConfig | None = None
self._baseline: dict[str, Any] = {}
self._active_labels: set[str] = set()
self._label_items: dict[str, QTreeWidgetItem] = {}
self._evaluation_method = QComboBox(self)
self._evaluation_method.setObjectName("aggregated_state_evaluation_method")
for label, value in self._EVALUATION_METHODS:
self._evaluation_method.addItem(label, value)
self._evaluation_method.currentIndexChanged.connect(lambda _index: self.changed.emit())
self._summary = QLabel(self)
self._summary.setObjectName("aggregated_state_summary")
controls = QHBoxLayout()
controls.setContentsMargins(0, 0, 0, 0)
controls.setSpacing(8)
controls.addWidget(QLabel("Evaluation", self))
controls.addWidget(self._evaluation_method, 1)
controls.addWidget(self._summary)
self._tree = QTreeWidget(self)
self._tree.setObjectName("aggregated_state_tree")
self._tree.setColumnCount(3)
self._tree.setHeaderLabels(["Rule", "Expected", "Tolerance / details"])
self._tree.setAlternatingRowColors(True)
self._tree.setUniformRowHeights(True)
self._tree.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
self._tree.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self._tree.setMinimumHeight(160)
self._tree.setMaximumHeight(360)
self._tree.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
header = self._tree.header()
header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
header.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
expand_button = QToolButton(self)
expand_button.setIcon(material_icon("unfold_more", convert_to_pixmap=False))
expand_button.setToolTip("Expand all rules")
expand_button.clicked.connect(self._tree.expandAll)
collapse_button = QToolButton(self)
collapse_button.setIcon(material_icon("unfold_less", convert_to_pixmap=False))
collapse_button.setToolTip("Collapse all rules")
collapse_button.clicked.connect(self._tree.collapseAll)
tree_controls = QHBoxLayout()
tree_controls.setContentsMargins(0, 0, 0, 0)
tree_controls.addWidget(QLabel("Configured rules", self))
tree_controls.addStretch(1)
tree_controls.addWidget(expand_button)
tree_controls.addWidget(collapse_button)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(6)
layout.addLayout(controls)
layout.addLayout(tree_controls)
layout.addWidget(self._tree)
@property
def model(self) -> type[bl_states.AggregatedStateConfig]:
return self._model
@property
def widgets(self) -> dict[str, QWidget]:
return {"evaluation_method": self._evaluation_method}
@property
def tree(self) -> QTreeWidget:
return self._tree
def input_widget(self, name: str) -> QWidget:
if name != "evaluation_method":
raise KeyError(name)
return self._evaluation_method
def set_partial_data(self, data: dict[str, Any]) -> None:
merged_data = self._config.model_dump() if self._config is not None else {}
merged_data.update(data)
config = self._model.model_validate(merged_data)
self._config = config
self._set_evaluation_method(config.evaluation_method)
self._populate_tree(config)
self.changed.emit()
def raw_data(self) -> dict[str, Any]:
if self._config is None:
return {}
data = self._config.model_dump()
data["evaluation_method"] = self._evaluation_method.currentData()
return data
def raw_editable_data(self) -> dict[str, Any]:
return {key: value for key, value in self.raw_data().items() if key != "name"}
def model_instance(self) -> bl_states.AggregatedStateConfig:
return self._model.model_validate(self.raw_data())
def dirty_fields(self) -> set[str]:
current = self.raw_data()
fields = set(current) | set(self._baseline)
return {field for field in fields if current.get(field) != self._baseline.get(field)}
def mark_clean(self) -> None:
self._baseline = self.raw_data()
def set_active_label_text(self, label_text: str) -> None:
configured_labels = set(self._label_items)
self._active_labels = set(label_text.split("|")) & configured_labels
self._apply_label_markers()
def cleanup(self) -> None:
self._tree.clear()
self._label_items.clear()
def _set_evaluation_method(self, value: str | None) -> None:
self._evaluation_method.blockSignals(True)
try:
for index in range(self._evaluation_method.count()):
if self._evaluation_method.itemData(index) == value:
self._evaluation_method.setCurrentIndex(index)
return
raise ValueError(f"Unsupported evaluation method: {value!r}")
finally:
self._evaluation_method.blockSignals(False)
def _populate_tree(self, config: bl_states.AggregatedStateConfig) -> None:
self._tree.clear()
self._label_items.clear()
device_names: set[str] = set()
requirement_count = 0
for label, state_config in config.states.items():
label_item = QTreeWidgetItem([label, "", ""])
label_item.setData(0, Qt.ItemDataRole.UserRole, label)
label_item.setFirstColumnSpanned(False)
self._tree.addTopLevelItem(label_item)
self._label_items[label] = label_item
for device_name, device_config in state_config.devices.items():
device_names.add(device_name)
device_item = QTreeWidgetItem([device_name, "", ""])
device_item.setIcon(0, material_icon("memory", convert_to_pixmap=False))
label_item.addChild(device_item)
for target, signal_config in self._requirements(device_config):
requirement_count += 1
expected = (
f"at: {signal_config.at}"
if signal_config.at is not None
else self._format_value(signal_config.value)
)
tolerance = f"± {signal_config.abs_tol:g}"
device_item.addChild(QTreeWidgetItem([target, expected, tolerance]))
if state_config.transition_metadata:
metadata = json.dumps(
state_config.transition_metadata,
ensure_ascii=False,
sort_keys=True,
default=str,
)
display = metadata if len(metadata) <= 100 else f"{metadata[:99]}"
transition_item = QTreeWidgetItem(["Transition metadata", "", display])
transition_item.setToolTip(2, metadata)
transition_item.setForeground(0, QBrush(self.palette().placeholderText().color()))
transition_item.setForeground(2, QBrush(self.palette().placeholderText().color()))
label_item.addChild(transition_item)
self._summary.setText(
f"{len(config.states)} labels · {len(device_names)} devices · "
f"{requirement_count} requirements"
)
self._tree.collapseAll()
self._apply_label_markers()
@staticmethod
def _requirements(
device_config: bl_states.DeviceConfig | bl_states.SignalConfig,
) -> list[tuple[str, bl_states.SignalConfig]]:
if isinstance(device_config, bl_states.SignalConfig):
return [("readback", device_config)]
requirements: list[tuple[str, bl_states.SignalConfig]] = []
if device_config.value is not None or device_config.at is not None:
requirements.append(
(
"readback",
bl_states.SignalConfig(
value=device_config.value,
at=device_config.at,
abs_tol=device_config.abs_tol,
),
)
)
if device_config.low_limit is not None:
requirements.append(("low limit", device_config.low_limit))
if device_config.high_limit is not None:
requirements.append(("high limit", device_config.high_limit))
requirements.extend(
(name, signal) for name, signal in (device_config.signals or {}).items()
)
return requirements
@staticmethod
def _format_value(value: Any) -> str:
return repr(value) if isinstance(value, str) else str(value)
def _apply_label_markers(self) -> None:
active_color = get_accent_colors().success.name()
for label, item in self._label_items.items():
active = label in self._active_labels
item.setIcon(
0,
material_icon(
"check_circle" if active else "radio_button_unchecked",
filled=active,
color=active_color if active else None,
convert_to_pixmap=False,
),
)
font = item.font(0)
font.setBold(active)
item.setFont(0, font)
if active:
item.setExpanded(True)
@@ -11,6 +11,7 @@ from qtpy.QtWidgets import (
QAbstractItemView,
QApplication,
QDialog,
QFileDialog,
QHBoxLayout,
QLabel,
QListView,
@@ -387,6 +388,9 @@ class BeamlineStateManager(BECWidget, QWidget):
def _create_toolbar(self) -> ModularToolBar:
toolbar = ModularToolBar(parent=self)
load_states = MaterialIconAction(
"file_open", "Load beamline states from config", filled=True, parent=self
)
add_state = MaterialIconAction("add", "Add beamline state", filled=True, parent=self)
filter_states = MaterialIconAction(
"filter_alt", "Filter displayed state status", filled=True, parent=self
@@ -413,6 +417,7 @@ class BeamlineStateManager(BECWidget, QWidget):
parent=self,
)
load_states.action.triggered.connect(self.load_states_from_config)
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)
@@ -420,6 +425,7 @@ class BeamlineStateManager(BECWidget, QWidget):
collapse_all.action.triggered.connect(self.collapse_all)
scan_interlock.action.toggled.connect(self._on_interlock_action_toggled)
toolbar.components.add_safe("load_states", load_states)
toolbar.components.add_safe("add_state", add_state)
toolbar.components.add_safe("filter_states", filter_states)
toolbar.components.add_safe("filter_devices", filter_devices)
@@ -429,6 +435,7 @@ class BeamlineStateManager(BECWidget, QWidget):
toolbar.components.add_safe("scan_interlock", scan_interlock)
bundle = ToolbarBundle("beamline_state_manager", toolbar.components)
bundle.add_action("load_states")
bundle.add_action("add_state")
bundle.add_action("filter_states")
bundle.add_action("filter_devices")
@@ -465,6 +472,14 @@ class BeamlineStateManager(BECWidget, QWidget):
self._refresh_section_headers()
self._refresh_hidden_summary()
@SafeSlot()
def load_states_from_config(self) -> None:
path, _ = QFileDialog.getOpenFileName(
self, "Load beamline states from config", "", "YAML files (*.yaml *.yml);;All files (*)"
)
if path:
self.bec_dispatcher.client.state_machine.load_from_config(config_path=path)
@SafeSlot()
def open_add_state_dialog(self) -> None:
dialog = AddBeamlineStateDialog(self, client=self.client)
@@ -6,7 +6,7 @@ from bec_lib import bl_states, messages
from bec_lib.endpoints import MessageEndpoints
from bec_qthemes import material_icon
from qtpy.QtCore import Property, QEasingCurve, QPropertyAnimation, Qt, Signal
from qtpy.QtGui import QColor, QMouseEvent, QPalette
from qtpy.QtGui import QColor, QKeyEvent, QMouseEvent, QPalette
from qtpy.QtWidgets import (
QApplication,
QCheckBox,
@@ -31,6 +31,9 @@ from bec_widgets.utils.forms_from_types.pydantic_widget_form import (
OptionalValueWidget,
PydanticWidgetForm,
)
from bec_widgets.widgets.services.beamline_states.aggregated_state_editor import (
AggregatedStateConfigEditor,
)
from bec_widgets.widgets.services.beamline_states.dialogs import (
BEAMLINE_STATE_STATUS_LABELS,
SUPPORTED_BEAMLINE_STATES,
@@ -91,6 +94,7 @@ class BeamlineStatePill(BECWidget, QWidget):
self.setObjectName("BeamlineStatePill")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum)
self.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
# Floor below which the pill keeps its structure; the title/detail elide rather than
# pushing the pill wider or taller, so collapsed rows stay a consistent size.
self.setMinimumWidth(200)
@@ -186,7 +190,7 @@ class BeamlineStatePill(BECWidget, QWidget):
self._settings.setObjectName("beamline_state_settings")
self._settings.setVisible(False)
self._state_type_value = QLabel(self._settings)
self._config_form: PydanticWidgetForm | None = None
self._config_form: PydanticWidgetForm | AggregatedStateConfigEditor | None = None
self._config_form_host = QVBoxLayout()
self._config_form_host.setContentsMargins(0, 0, 0, 0)
self._config_form_host.setSpacing(0)
@@ -404,6 +408,8 @@ class BeamlineStatePill(BECWidget, QWidget):
status = str(content.get("status", "unknown")).lower()
label = str(content.get("label", "No state information available."))
self._set_visual_state(status, label)
if isinstance(self._config_form, AggregatedStateConfigEditor):
self._config_form.set_active_label_text(label)
self.state_changed.emit(self._state_name or str(name or ""), status, label)
@SafeSlot(str)
@@ -596,8 +602,16 @@ class BeamlineStatePill(BECWidget, QWidget):
@SafeSlot()
def _toggle_expanded(self) -> None:
self.setFocus(Qt.FocusReason.MouseFocusReason)
self.set_expanded(not self._expanded)
def keyPressEvent(self, event: QKeyEvent) -> None: # noqa: N802
if event.key() in (Qt.Key.Key_Delete, Qt.Key.Key_Backspace):
self._emit_remove_requested()
event.accept()
return
super().keyPressEvent(event)
def is_expanded(self) -> bool:
"""Return whether the editable settings panel is expanded."""
return self._expanded
@@ -625,16 +639,22 @@ class BeamlineStatePill(BECWidget, QWidget):
def _ensure_config_form(
self, config_class: type[bl_states.BeamlineStateConfig] = bl_states.DeviceStateConfig
) -> PydanticWidgetForm:
) -> PydanticWidgetForm | AggregatedStateConfigEditor:
if self._config_form is None:
self._config_form = PydanticWidgetForm(
config_class, parent=self._settings, client=self.client, read_only_fields={"name"}
)
if config_class is bl_states.AggregatedStateConfig:
self._config_form = AggregatedStateConfigEditor(parent=self._settings)
else:
self._config_form = PydanticWidgetForm(
config_class,
parent=self._settings,
client=self.client,
read_only_fields={"name"},
)
self._config_form.changed.connect(self._update_settings_dirty_state)
self._config_form_host.addWidget(self._config_form)
return self._config_form
def _ensure_settings_form_current(self) -> PydanticWidgetForm:
def _ensure_settings_form_current(self) -> PydanticWidgetForm | AggregatedStateConfigEditor:
if self._settings_form_stale:
self._populate_settings()
self.mark_current_settings_clean()
@@ -657,6 +677,11 @@ class BeamlineStatePill(BECWidget, QWidget):
try:
state_type = self._state_config.state_type if self._state_config is not None else ""
config_class = None
if state_type in {
bl_states.AggregatedState.__name__,
bl_states.AggregatedState.CONFIG_CLASS.state_type,
}:
config_class = bl_states.AggregatedState.CONFIG_CLASS
for state_class in SUPPORTED_BEAMLINE_STATES:
if state_type in {state_class.__name__, state_class.CONFIG_CLASS.state_type}:
config_class = state_class.CONFIG_CLASS
@@ -665,9 +690,13 @@ class BeamlineStatePill(BECWidget, QWidget):
raise ValueError(f"Unsupported beamline state type '{state_type}'.")
config_form = self._ensure_config_form(config_class)
if config_form.model is not config_class:
if isinstance(config_form, AggregatedStateConfigEditor):
raise ValueError(f"Unsupported beamline state type change to '{state_type}'.")
config_form.set_model(config_class)
self._state_type_value.setText(state_type or "-")
config_form.set_partial_data(self._state_data_for_form(config_class))
if isinstance(config_form, AggregatedStateConfigEditor):
config_form.set_active_label_text(self._label)
self._settings_form_stale = False
finally:
self._populating_settings = False
+137 -1
View File
@@ -1,13 +1,16 @@
import shiboken6
from bec_lib import bl_states, messages
from qtpy.QtCore import QCoreApplication, QEvent, Qt
from qtpy.QtWidgets import QDialog, QMessageBox, QStyleOptionViewItem
from qtpy.QtWidgets import QComboBox, QDialog, QMessageBox, QStyleOptionViewItem, QTreeWidget
from bec_widgets.utils.eliding_label import ElidingLabel
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.services.beamline_states import beamline_state_manager as manager_module
from bec_widgets.widgets.services.beamline_states import beamline_state_pill as pill_module
from bec_widgets.widgets.services.beamline_states.aggregated_state_editor import (
AggregatedStateConfigEditor,
)
from bec_widgets.widgets.services.beamline_states.beamline_state_manager import BeamlineStateManager
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStatePill
from bec_widgets.widgets.services.beamline_states.dialogs import AddBeamlineStateDialog
@@ -45,6 +48,50 @@ def _limits_state(name: str = "limits", **overrides) -> messages.BeamlineStateCo
return _wire_state(bl_states.DeviceWithinLimitsState, config)
def _aggregated_state(
name: str = "machine_mode", evaluation_method: str | None = "any"
) -> messages.BeamlineStateConfig:
config = bl_states.AggregatedStateConfig(
name=name,
evaluation_method=evaluation_method,
states={
"alignment": {
"devices": {
"samx": {
"value": 0,
"abs_tol": 0.1,
"signals": {"velocity": {"value": 5, "abs_tol": 0.2}},
},
"samy": {"at": "in", "abs_tol": 0.1},
},
"transition_metadata": {"description": "Prepare alignment"},
},
"parked": {
"devices": {
"samx": {
"low_limit": {"at": "low", "abs_tol": 0.3},
"high_limit": {"value": 10, "abs_tol": 0.4},
}
}
},
},
)
return _wire_state(bl_states.AggregatedState, config)
def _tree_rows(tree: QTreeWidget) -> list[list[str]]:
rows: list[list[str]] = []
def collect(item) -> None:
rows.append([item.text(column) for column in range(tree.columnCount())])
for index in range(item.childCount()):
collect(item.child(index))
for index in range(tree.topLevelItemCount()):
collect(tree.topLevelItem(index))
return rows
def _as_status_list(value: str | list[str]) -> list[str]:
return [value] if isinstance(value, str) else list(value)
@@ -138,6 +185,95 @@ def test_beamline_state_pill_expands_and_emits_updated_limits(qtbot, mocked_clie
assert not limits_pill._settings.isHidden()
def test_aggregated_state_pill_displays_rule_inspector(qtbot, mocked_client):
pill = create_widget(qtbot, BeamlineStatePill, state_name="machine_mode", client=mocked_client)
pill.set_state_config(_aggregated_state())
pill.update_state({"name": "machine_mode", "status": "valid", "label": "alignment"}, {})
pill.set_expanded(True)
editor = pill._config_form
assert isinstance(editor, AggregatedStateConfigEditor)
assert editor.input_widget("evaluation_method").currentData() == "any"
assert editor._summary.text() == "2 labels · 2 devices · 5 requirements"
assert editor.tree.maximumHeight() == 360
assert editor.tree.topLevelItemCount() == 2
rows = _tree_rows(editor.tree)
assert ["alignment", "", ""] in rows
assert ["samx", "", ""] in rows
assert ["readback", "0", "± 0.1"] in rows
assert ["velocity", "5", "± 0.2"] in rows
assert ["readback", "at: in", "± 0.1"] in rows
assert ["low limit", "at: low", "± 0.3"] in rows
assert ["high limit", "10", "± 0.4"] in rows
assert any(row[0] == "Transition metadata" and "Prepare alignment" in row[2] for row in rows)
alignment = editor.tree.topLevelItem(0)
parked = editor.tree.topLevelItem(1)
assert alignment.text(0) == "alignment"
assert alignment.font(0).bold()
assert alignment.isExpanded()
assert not parked.font(0).bold()
def test_aggregated_state_pill_updates_evaluation_method(qtbot, mocked_client):
pill = create_widget(qtbot, BeamlineStatePill, state_name="machine_mode", client=mocked_client)
pill.set_state_config(_aggregated_state())
pill.set_expanded(True)
editor = pill._config_form
assert isinstance(editor, AggregatedStateConfigEditor)
method_combo = editor.input_widget("evaluation_method")
assert isinstance(method_combo, QComboBox)
exclusive_index = next(
index
for index in range(method_combo.count())
if method_combo.itemData(index) == "exclusive"
)
method_combo.setCurrentIndex(exclusive_index)
assert pill._update_button.isEnabled()
assert pill._revert_button.isEnabled()
assert pill._settings_dirty_fields == {"evaluation_method"}
edited = pill.edited_config()
assert isinstance(edited, bl_states.AggregatedStateConfig)
assert edited.evaluation_method == "exclusive"
assert set(edited.states) == {"alignment", "parked"}
with qtbot.waitSignal(pill.update_requested) as signal:
pill._update_button.click()
assert signal.args[0] == "machine_mode"
assert signal.args[1].evaluation_method == "exclusive"
assert set(signal.args[1].states) == {"alignment", "parked"}
def test_aggregated_state_pill_round_trips_disabled_evaluation_and_reverts(qtbot, mocked_client):
pill = create_widget(qtbot, BeamlineStatePill, state_name="machine_mode", client=mocked_client)
pill.set_state_config(_aggregated_state(evaluation_method=None))
pill.set_expanded(True)
editor = pill._config_form
assert isinstance(editor, AggregatedStateConfigEditor)
method_combo = editor.input_widget("evaluation_method")
assert isinstance(method_combo, QComboBox)
assert method_combo.currentData() is None
assert pill.edited_config().evaluation_method is None
any_index = next(
index for index in range(method_combo.count()) if method_combo.itemData(index) == "any"
)
method_combo.setCurrentIndex(any_index)
assert pill._update_button.isEnabled()
pill._revert_button.click()
assert method_combo.currentData() is None
assert pill.edited_config().evaluation_method is None
assert not pill._update_button.isEnabled()
assert not pill._revert_button.isEnabled()
def test_beamline_state_pill_first_expand_uses_config_class_without_rebuild(
qtbot, mocked_client, monkeypatch
):