From 90084db58085ef777c0c673152b99055c2e98f85 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Fri, 12 Jun 2026 17:58:11 +0200 Subject: [PATCH] feat(beamline_states): order states by status severity within each section --- .../beamline_states/beamline_state_manager.py | 63 ++++++++++----- tests/unit_tests/test_beamline_state_pill.py | 79 +++++++++++++++++++ 2 files changed, 123 insertions(+), 19 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 4eb4d933..e73ae86e 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py @@ -93,17 +93,18 @@ class _BeamlineStateListModel(QAbstractListModel): def set_states( self, state_configs: list[messages.BeamlineStateConfig], interlock_names: set[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] + # Ordering comes from ``state_configs``; ``interlock_names`` is only a membership set. + new_configs = {state.name: state for state in state_configs} + interlock = [name for name in new_configs if name in interlock_names] + others = [name for name in new_configs if name not in interlock_names] + new_keys: list[tuple[str, str]] = [] if interlock: new_keys.append(("header", self.INTERLOCK_HEADER)) - new_keys.extend(("state", state.name) for state in interlock) + new_keys.extend(("state", name) for name in interlock) if others: new_keys.append(("header", self.OTHERS_HEADER)) - new_keys.extend(("state", state.name) for state in others) - - new_configs = {state.name: state for state in state_configs} + new_keys.extend(("state", name) for name in others) new_key_set = set(new_keys) for row in reversed( @@ -287,6 +288,7 @@ class BeamlineStateManager(BECWidget, QWidget): PLUGIN = True ICON_NAME = "format_list_bulleted" + _STATUS_PRIORITY = {"invalid": 0, "warning": 1, "valid": 2, "unknown": 3} USER_ACCESS = [ "clear_filters", "collapse_all", @@ -545,7 +547,10 @@ class BeamlineStateManager(BECWidget, QWidget): """Render the current state and scan-interlock bookkeeping in one pass.""" self._sync_interlock_action() expanded_names = {name for name, pill in self._state_pills.items() if pill.is_expanded()} - state_configs = [self._state_configs[name] for name in self._state_order] + # Both sections are ordered by status severity (invalid, warning, unknown, valid). The + # sort is stable, so the configured order is kept within each status. + ordered_names = sorted(self._state_order, key=self._status_rank) + state_configs = [self._state_configs[name] for name in ordered_names] self._model.set_states(state_configs, set(self._interlock_states)) self._open_persistent_editors(expanded_names) for name, pill in self._state_pills.items(): @@ -581,13 +586,35 @@ 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 + return self._state_status(name) not in (None, required_status) + + def _status_rank(self, name: str) -> int: + """Sort rank of a state by status severity: invalid < warning < unknown < valid.""" + return self._STATUS_PRIORITY.get( + self._state_status(name) or "unknown", self._STATUS_PRIORITY["unknown"] + ) + + def _state_status(self, name: str) -> str | None: + """Latest status of a state, from its live pill or the connector cache. + + The connector fallback lets the sections be ordered by status severity even on the + first render, before the pills (and their statuses) have been created. + """ + pill = self._state_pills.get(name) + if pill is not None: + return pill._status + msg = self.client.connector.get_last(MessageEndpoints.beamline_state(name), key="data") + if msg is None: + return None + return str(msg.content.get("status", "unknown")).lower() @SafeSlot(bool) def _on_interlock_action_toggled(self, checked: bool) -> None: @@ -699,12 +726,10 @@ class BeamlineStateManager(BECWidget, QWidget): return True @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 self._selected_statuses is not None: - self._apply_filters() + def _on_pill_state_changed(self, _name: str, _status: str, _label: str) -> None: + # Status drives the per-section severity ordering (and the interlock triggered flag), + # so any change re-renders the list. + self._refresh_view() @SafeSlot(str, object) def _update_state_parameters( diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py index 4b268efb..bbbef714 100644 --- a/tests/unit_tests/test_beamline_state_pill.py +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -727,6 +727,85 @@ def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client): assert pill._interlock_required_status == "valid" +def test_beamline_state_manager_orders_both_sections_by_status_severity(qtbot, mocked_client): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + beamline_state_manager.update_available_states( + { + "states": [ + _limits_state(name="watched_valid"), + _limits_state(name="watched_invalid"), + _limits_state(name="free_warning"), + _limits_state(name="free_invalid"), + ] + }, + {}, + ) + _install_fake_scan_interlock( + beamline_state_manager, + _FakeScanInterlock( + enabled=True, states_watched={"watched_valid": "valid", "watched_invalid": "valid"} + ), + ) + pills = beamline_state_manager._state_pills + pills["watched_valid"].update_state( + {"name": "watched_valid", "status": "valid", "label": ""}, {} + ) + pills["watched_invalid"].update_state( + {"name": "watched_invalid", "status": "invalid", "label": ""}, {} + ) + pills["free_warning"].update_state( + {"name": "free_warning", "status": "warning", "label": ""}, {} + ) + pills["free_invalid"].update_state( + {"name": "free_invalid", "status": "invalid", "label": ""}, {} + ) + + model = beamline_state_manager._model + names = [model.data(model.index(row, 0), model.NameRole) for row in range(model.rowCount())] + + # Interlock section: invalid before valid. Non-interlock section: invalid before warning. + assert names == [ + None, # "Scan interlock states" header + "watched_invalid", + "watched_valid", + None, # "Not included in scan interlock" header + "free_invalid", + "free_warning", + ] + assert pills["watched_invalid"]._interlock_triggered + assert not pills["watched_valid"]._interlock_triggered + + +def test_beamline_state_manager_orders_by_status_severity_on_initial_render( + qtbot, mocked_client, monkeypatch +): + beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client) + # The statuses are visible in the connector cache when the pills are first created, so the + # severity order must hold from the very first render. ``get_last`` is patched (rather than + # writing to the session-shared fake connector) so the seeded statuses can't leak into and + # pollute other tests. + seeded = {"limits": "valid", "shutter_open": "invalid"} + real_get_last = mocked_client.connector.get_last + + def fake_get_last(endpoint, *args, **kwargs): + topic = str(getattr(endpoint, "endpoint", endpoint)) + for name, status in seeded.items(): + if topic.endswith(f"/beamline_state/{name}"): + return messages.BeamlineStateMessage(name=name, status=status, label="x") + return real_get_last(endpoint, *args, **kwargs) + + monkeypatch.setattr(mocked_client.connector, "get_last", fake_get_last) + + beamline_state_manager.update_available_states( + {"states": [_limits_state(), _shutter_state()]}, {} + ) + + model = beamline_state_manager._model + names = [model.data(model.index(row, 0), model.NameRole) for row in range(model.rowCount())] + assert names == ["shutter_open", "limits"] + assert beamline_state_manager._state_pills["shutter_open"]._status == "invalid" + + 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()