feat(beamline_states): add scan-interlock options to the add-state dialog

This commit is contained in:
2026-06-17 10:00:54 +02:00
committed by wakonig_k
co-authored by wakonig_k
parent 2084edb576
commit 3cd37fea26
4 changed files with 186 additions and 7 deletions
@@ -335,6 +335,7 @@ class BeamlineStateManager(BECWidget, QWidget):
self._scan_interlock = self.client.builtin_actors.scan_interlock
self._interlock_enabled = False
self._interlock_states: dict[str, list[str]] = {}
self._pending_interlock_statuses: dict[str, list[str]] = {}
self._updating_interlock_action = False
self._interlock_action_armed: bool | None = None
@@ -459,10 +460,14 @@ class BeamlineStateManager(BECWidget, QWidget):
def open_add_state_dialog(self) -> None:
dialog = AddBeamlineStateDialog(self, client=self.client)
config = None
add_to_interlock = False
interlock_statuses: list[str] = ["valid", "warning"]
try:
accepted = dialog.exec() == QDialog.Accepted
if accepted:
config = dialog.config_result
add_to_interlock = dialog.add_to_interlock()
interlock_statuses = dialog.interlock_statuses()
finally:
dialog.cleanup()
dialog.deleteLater()
@@ -473,6 +478,14 @@ class BeamlineStateManager(BECWidget, QWidget):
self.client.beamline_states.add(config)
except Exception as exc:
QMessageBox.warning(self, "Cannot Add State", str(exc))
return
if add_to_interlock:
try:
self._scan_interlock.add_state_to_interlock(config.name, interlock_statuses)
except Exception as exc:
QMessageBox.warning(self, "Cannot Update Scan Interlock", str(exc))
else:
self._pending_interlock_statuses[config.name] = interlock_statuses
@SafeSlot()
def open_status_filter_dialog(self) -> None:
@@ -600,9 +613,10 @@ class BeamlineStateManager(BECWidget, QWidget):
action.setToolTip("Scan interlock is disabled. Click to arm it.")
def _apply_interlock_to_pill(self, name: str, pill: BeamlineStatePill) -> None:
pill.set_scan_interlock(
self._interlock_states.get(name), self._is_interlock_triggered(name)
)
required_statuses = self._interlock_states.get(name)
if required_statuses is None and name in self._pending_interlock_statuses:
pill.set_interlock_statuses(self._pending_interlock_statuses.pop(name))
pill.set_scan_interlock(required_statuses, self._is_interlock_triggered(name))
def _apply_section_header(self, header: _BeamlineStateSectionHeader, kind: str) -> None:
colors = BeamlineStatePill._state_colors("unknown")
@@ -662,7 +676,9 @@ class BeamlineStateManager(BECWidget, QWidget):
def _on_interlock_toggle_requested(self, state_name: str, include: bool) -> None:
try:
if include:
self._scan_interlock.add_state_to_interlock(state_name, "valid")
pill = self._state_pills.get(state_name)
statuses = pill.interlock_statuses if pill is not None else ["valid", "warning"]
self._scan_interlock.add_state_to_interlock(state_name, statuses)
else:
self._scan_interlock.remove_state_from_interlock(state_name)
except Exception as exc:
@@ -99,6 +99,7 @@ class BeamlineStatePill(BECWidget, QWidget):
self._expanded = False
self._idle_card_background = False
self._interlock_required_statuses: list[str] | None = None
self._interlock_statuses: list[str] = ["valid", "warning"]
self._interlock_triggered = False
self._interlock_pulse = 0.0
self._header_icon_cache_key: tuple | None = None
@@ -317,6 +318,8 @@ class BeamlineStatePill(BECWidget, QWidget):
triggered: Whether the armed scan interlock is currently tripped by this state.
"""
triggered = bool(triggered) and required_statuses is not None
if required_statuses is not None:
self._interlock_statuses = list(required_statuses)
if (required_statuses, triggered) == (
self._interlock_required_statuses,
self._interlock_triggered,
@@ -332,6 +335,15 @@ class BeamlineStatePill(BECWidget, QWidget):
self._interlock_pulse = 0.0
self._apply_visual_state()
@property
def interlock_statuses(self) -> list[str]:
"""Accepted statuses to enroll this state with when it joins the scan interlock."""
return list(self._interlock_statuses)
def set_interlock_statuses(self, statuses: list[str]) -> None:
"""Configure the accepted scan-interlock statuses for this state."""
self._interlock_statuses = list(statuses)
@SafeSlot()
def _emit_interlock_toggle_requested(self) -> None:
if self._state_name is None:
@@ -56,6 +56,19 @@ class AddBeamlineStateDialog(QDialog):
self._config_form_host = QVBoxLayout()
self._config_form: PydanticWidgetForm | None = None
self._trigger_on_warning_checkbox = QCheckBox(
"Trigger ScanInterlock on WARNING state", self
)
self._trigger_on_warning_checkbox.setToolTip(
"By default both VALID and WARNING are accepted. Enable this so a WARNING status also "
"trips the scan interlock (only VALID accepted)."
)
self._add_to_interlock_checkbox = QCheckBox("Add new state to ScanInterlock", self)
self._add_to_interlock_checkbox.setToolTip(
"Watch this state in the scan interlock right away. Leave unchecked to add it later "
"with the lock button on the state."
)
self._buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self
)
@@ -65,6 +78,8 @@ class AddBeamlineStateDialog(QDialog):
layout = QVBoxLayout(self)
layout.addLayout(self._form)
layout.addLayout(self._config_form_host)
layout.addWidget(self._trigger_on_warning_checkbox)
layout.addWidget(self._add_to_interlock_checkbox)
layout.addWidget(self._buttons)
self.setLayout(layout)
self._update_config_form()
@@ -78,6 +93,20 @@ class AddBeamlineStateDialog(QDialog):
data["name"] = name
return config_class.model_validate(data)
def add_to_interlock(self) -> bool:
"""Whether the new state should be enrolled in the scan interlock immediately."""
return self._add_to_interlock_checkbox.isChecked()
def interlock_statuses(self) -> list[str]:
"""Accepted scan-interlock statuses for the state.
VALID and WARNING are both accepted by default; triggering on WARNING accepts only VALID.
Only applied when the state is actually enrolled (see :meth:`add_to_interlock`).
"""
if self._trigger_on_warning_checkbox.isChecked():
return ["valid"]
return ["valid", "warning"]
def accept(self) -> None:
try:
self._config = self.config()
+125 -3
View File
@@ -1,7 +1,7 @@
import shiboken6
from bec_lib import bl_states, messages
from qtpy.QtCore import QCoreApplication, QEvent, Qt
from qtpy.QtWidgets import QMessageBox, QStyleOptionViewItem
from qtpy.QtWidgets import QDialog, QMessageBox, QStyleOptionViewItem
from bec_widgets.utils.eliding_label import ElidingLabel
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
@@ -882,9 +882,9 @@ def test_beamline_state_manager_pill_toggle_calls_backend(qtbot, mocked_client):
beamline_state_manager._state_pills["limits"]._interlock_button.click()
assert fake_interlock.added == [("limits", "valid")]
assert fake_interlock.added == [("limits", ["valid", "warning"])]
# The toggle refreshes immediately, so the state is reflected without a backend notification.
assert beamline_state_manager._interlock_states == {"limits": ["valid"]}
assert beamline_state_manager._interlock_states == {"limits": ["valid", "warning"]}
beamline_state_manager._state_pills["limits"]._interlock_button.click()
@@ -892,6 +892,34 @@ def test_beamline_state_manager_pill_toggle_calls_backend(qtbot, mocked_client):
assert beamline_state_manager._interlock_states == {}
def test_beamline_state_manager_lock_uses_dialog_status_preference(
qtbot, mocked_client, monkeypatch
):
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
fake_interlock = _FakeScanInterlock()
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
_stub_beamline_states_add(mocked_client)
# Create a state that triggers on WARNING (statuses = ["valid"]) but is NOT enrolled now.
_install_fake_add_dialog(
monkeypatch,
config=_limits_state(name="limits"),
add_to_interlock=False,
interlock_statuses=["valid"],
)
beamline_state_manager.open_add_state_dialog()
assert fake_interlock.added == []
# When the state's pill appears, it is seeded with the configured statuses.
beamline_state_manager.update_available_states({"states": [_limits_state(name="limits")]}, {})
pill = beamline_state_manager._state_pills["limits"]
assert pill.interlock_statuses == ["valid"]
# Clicking the lock later enrolls with the remembered preference, not the default.
pill._interlock_button.click()
assert fake_interlock.added == [("limits", ["valid"])]
def test_add_beamline_state_dialog_uses_generated_widgets_and_normalizes_name(qtbot, mocked_client):
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
limits_index = add_state_dialog._type_combo.findText(bl_states.DeviceWithinLimitsState.__name__)
@@ -922,6 +950,100 @@ def test_add_beamline_state_dialog_uses_generated_widgets_and_normalizes_name(qt
assert config.high_limit == 15.0
def test_add_beamline_state_dialog_interlock_checkboxes(qtbot, mocked_client):
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
# Not enrolled by default; VALID and WARNING are both accepted (WARNING does not trip it).
assert add_state_dialog.add_to_interlock() is False
assert add_state_dialog.interlock_statuses() == ["valid", "warning"]
# The warning condition is independent of enrollment (settable without "Add to interlock").
assert add_state_dialog._trigger_on_warning_checkbox.isEnabled()
add_state_dialog._trigger_on_warning_checkbox.setChecked(True)
assert add_state_dialog.interlock_statuses() == ["valid"]
assert add_state_dialog.add_to_interlock() is False
add_state_dialog._add_to_interlock_checkbox.setChecked(True)
assert add_state_dialog.add_to_interlock() is True
def _install_fake_add_dialog(monkeypatch, *, config, add_to_interlock, interlock_statuses):
class FakeAddDialog:
def __init__(self, parent, client=None):
pass
def exec(self):
return QDialog.Accepted
@property
def config_result(self):
return config
def add_to_interlock(self):
return add_to_interlock
def interlock_statuses(self):
return interlock_statuses
def cleanup(self):
pass
def deleteLater(self): # noqa: N802
pass
monkeypatch.setattr(manager_module, "AddBeamlineStateDialog", FakeAddDialog)
def _stub_beamline_states_add(mocked_client):
added_configs = []
class StateManager:
def add(self, config):
added_configs.append(config)
mocked_client.beamline_states = StateManager()
return added_configs
def test_beamline_state_manager_add_dialog_enrolls_new_state_in_interlock(
qtbot, mocked_client, monkeypatch
):
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
fake_interlock = _FakeScanInterlock()
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
added_configs = _stub_beamline_states_add(mocked_client)
new_config = _limits_state(name="new_state")
_install_fake_add_dialog(
monkeypatch,
config=new_config,
add_to_interlock=True,
interlock_statuses=["valid", "warning"],
)
beamline_state_manager.open_add_state_dialog()
assert added_configs == [new_config]
assert fake_interlock.added == [("new_state", ["valid", "warning"])]
def test_beamline_state_manager_add_dialog_skips_interlock_when_not_requested(
qtbot, mocked_client, monkeypatch
):
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
fake_interlock = _FakeScanInterlock()
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
added_configs = _stub_beamline_states_add(mocked_client)
new_config = _limits_state(name="new_state")
_install_fake_add_dialog(
monkeypatch, config=new_config, add_to_interlock=False, interlock_statuses=["valid"]
)
beamline_state_manager.open_add_state_dialog()
assert added_configs == [new_config]
assert fake_interlock.added == []
def test_add_beamline_state_dialog_generates_name_only_after_valid_device_selection(
qtbot, mocked_client
):