From 8b7017cda412accf23cd9905302bb5493b11b105 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Fri, 12 Jun 2026 17:58:11 +0200 Subject: [PATCH] feat(beamline_states): sort triggered states to the top of the interlock section --- .../beamline_states/beamline_state_manager.py | 39 ++++++++++++------- tests/unit_tests/test_beamline_state_pill.py | 32 +++++++++++++++ 2 files changed, 58 insertions(+), 13 deletions(-) diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py index 0835aabe..ec442714 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py @@ -91,10 +91,12 @@ class _BeamlineStateListModel(QAbstractListModel): return Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable def set_states( - self, state_configs: list[messages.BeamlineStateConfig], interlock_names: set[str] + self, state_configs: list[messages.BeamlineStateConfig], interlock_names: list[str] ) -> None: - interlock = [state for state in state_configs if state.name in interlock_names] - others = [state for state in state_configs if state.name not in interlock_names] + configs_by_name = {state.name: state for state in state_configs} + interlock = [configs_by_name[name] for name in interlock_names if name in configs_by_name] + interlock_set = set(interlock_names) + others = [state for state in state_configs if state.name not in interlock_set] new_keys: list[tuple[str, str]] = [] if interlock: new_keys.append(("header", self.INTERLOCK_HEADER)) @@ -546,7 +548,13 @@ class BeamlineStateManager(BECWidget, QWidget): self._sync_interlock_action() expanded_names = {name for name, pill in self._state_pills.items() if pill.is_expanded()} state_configs = [self._state_configs[name] for name in self._state_order] - self._model.set_states(state_configs, set(self._interlock_states)) + # Triggered states sort to the top of the interlock section; the sort is stable, so + # the configured state order is kept within each group. + interlock_order = sorted( + (name for name in self._state_order if name in self._interlock_states), + key=lambda name: not self._is_interlock_triggered(name), + ) + self._model.set_states(state_configs, interlock_order) self._open_persistent_editors(expanded_names) for name, pill in self._state_pills.items(): self._apply_interlock_to_pill(name, pill) @@ -575,13 +583,16 @@ 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: - required_status = self._interlock_states.get(name) - triggered = ( - self._interlock_enabled - and required_status is not None - and pill._status != required_status + pill.set_scan_interlock( + self._interlock_states.get(name), self._is_interlock_triggered(name) ) - pill.set_scan_interlock(required_status, triggered) + + def _is_interlock_triggered(self, name: str) -> bool: + required_status = self._interlock_states.get(name) + if required_status is None or not self._interlock_enabled: + return False + pill = self._state_pills.get(name) + return pill is not None and pill._status != required_status @SafeSlot(bool) def _on_interlock_action_toggled(self, checked: bool) -> None: @@ -692,9 +703,11 @@ class BeamlineStateManager(BECWidget, QWidget): @SafeSlot(str, str, str) def _on_pill_state_changed(self, name: str, _status: str, _label: str) -> None: - pill = self._state_pills.get(name) - if pill is not None: - self._apply_interlock_to_pill(name, pill) + if name in self._interlock_states: + # The triggered flag and the triggered-first ordering of the interlock section + # both follow the pill status. + self._refresh_view() + return if self._selected_statuses is not None: self._apply_filters() diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py index ed436517..e3c11e73 100644 --- a/tests/unit_tests/test_beamline_state_pill.py +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -727,6 +727,38 @@ def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client): assert pill._interlock_required_status == "valid" +def test_beamline_state_manager_sorts_triggered_interlock_states_first(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states( + {"states": [_limits_state(), _shutter_state()]}, {} + ) + _install_fake_scan_interlock( + beamline_state_manager, + _FakeScanInterlock( + enabled=True, states_watched={"limits": "valid", "shutter_open": "valid"} + ), + ) + model = beamline_state_manager._model + limits_pill = beamline_state_manager._state_pills["limits"] + + def interlock_section(): + return [model.data(model.index(row, 0), model.NameRole) for row in (1, 2)] + + assert model.rowCount() == 3 + assert interlock_section() == ["limits", "shutter_open"] + + limits_pill.update_state({"name": "limits", "status": "valid", "label": "Within limits."}, {}) + + assert interlock_section() == ["shutter_open", "limits"] + assert beamline_state_manager._state_pills["limits"] is limits_pill + assert not limits_pill._interlock_triggered + + limits_pill.update_state({"name": "limits", "status": "invalid", "label": "Out of limits."}, {}) + + assert interlock_section() == ["limits", "shutter_open"] + assert limits_pill._interlock_triggered + + def test_beamline_state_manager_toolbar_toggle_writes_backend(qtbot, mocked_client): beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) fake_interlock = _FakeScanInterlock()