From 2084edb576497124b2d4d2aa1a4fa66ec8952956 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Mon, 15 Jun 2026 22:55:11 +0200 Subject: [PATCH] fix(beamline_states): support multiple accepted statuses in the scan interlock --- .../beamline_states/beamline_state_manager.py | 9 +++--- .../beamline_states/beamline_state_pill.py | 27 ++++++++-------- tests/unit_tests/test_beamline_state_pill.py | 32 ++++++++++++------- 3 files changed, 39 insertions(+), 29 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 dd925ebf..4c63a3e7 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_manager.py @@ -334,7 +334,7 @@ class BeamlineStateManager(BECWidget, QWidget): self._hidden_expanded = False self._scan_interlock = self.client.builtin_actors.scan_interlock self._interlock_enabled = False - self._interlock_states: dict[str, str] = {} + self._interlock_states: dict[str, list[str]] = {} self._updating_interlock_action = False self._interlock_action_armed: bool | None = None @@ -622,10 +622,11 @@ class BeamlineStateManager(BECWidget, QWidget): self._apply_section_header(header, kind) 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: + accepted_statuses = self._interlock_states.get(name) + if not accepted_statuses or not self._interlock_enabled: return False - return self._state_status(name) not in (None, required_status) + status = self._state_status(name) + return status is not None and status not in accepted_statuses def _status_rank(self, name: str) -> int: """Sort rank of a state by status severity: invalid < warning < unknown < valid.""" diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py index 6cccd015..4e4c5907 100644 --- a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py @@ -98,7 +98,7 @@ class BeamlineStatePill(BECWidget, QWidget): self._label = "No state information available." self._expanded = False self._idle_card_background = False - self._interlock_required_status: str | None = None + self._interlock_required_statuses: list[str] | None = None self._interlock_triggered = False self._interlock_pulse = 0.0 self._header_icon_cache_key: tuple | None = None @@ -307,22 +307,22 @@ class BeamlineStatePill(BECWidget, QWidget): if self._interlock_triggered: self._apply_visual_state() - def set_scan_interlock(self, required_status: str | None, triggered: bool) -> None: + def set_scan_interlock(self, required_statuses: list[str] | None, triggered: bool) -> None: """ Set the scan-interlock participation of this pill. Args: - required_status: Status the scan interlock requires for this state, or ``None`` + required_statuses: Statuses the scan interlock accepts for this state, or ``None`` if the state is not included in the scan interlock. triggered: Whether the armed scan interlock is currently tripped by this state. """ - triggered = bool(triggered) and required_status is not None - if (required_status, triggered) == ( - self._interlock_required_status, + triggered = bool(triggered) and required_statuses is not None + if (required_statuses, triggered) == ( + self._interlock_required_statuses, self._interlock_triggered, ): return - self._interlock_required_status = required_status + self._interlock_required_statuses = required_statuses self._interlock_triggered = triggered if triggered: if self._interlock_animation.state() != QPropertyAnimation.State.Running: @@ -336,7 +336,7 @@ class BeamlineStatePill(BECWidget, QWidget): def _emit_interlock_toggle_requested(self) -> None: if self._state_name is None: return - include = self._interlock_required_status is None + include = self._interlock_required_statuses is None self.scan_interlock_toggle_requested.emit(self._state_name, include) def _refresh_latest_state(self) -> None: @@ -378,7 +378,7 @@ class BeamlineStatePill(BECWidget, QWidget): def _apply_visual_state(self) -> None: colors = self._state_colors(self._status) accent = colors["accent"] - included = self._interlock_required_status is not None + included = self._interlock_required_statuses is not None active_card = self._expanded or included border = colors["border"] if self._idle_card_background else "transparent" background = colors["background"] if self._idle_card_background else "transparent" @@ -491,7 +491,8 @@ class BeamlineStatePill(BECWidget, QWidget): cache_key = ( self._status, self._expanded, - self._interlock_required_status, + tuple(self._interlock_required_statuses or ()), + self._interlock_required_statuses is not None, self._interlock_triggered, get_theme_name(), ) @@ -511,7 +512,7 @@ class BeamlineStatePill(BECWidget, QWidget): self._expand_button.setIcon( material_icon(expand_icon, size=(20, 20), convert_to_pixmap=False) ) - if self._interlock_required_status is not None: + if self._interlock_required_statuses is not None: lock_color = ( colors["interlock_trigger"] if self._interlock_triggered else colors["foreground"] ) @@ -521,8 +522,8 @@ class BeamlineStatePill(BECWidget, QWidget): ) ) self._interlock_button.setToolTip( - f"Watched by the scan interlock (required status: " - f"{self._interlock_required_status}).\n" + "Watched by the scan interlock (accepted statuses: " + f"{', '.join(self._interlock_required_statuses)}).\n" "Click to remove this state from the scan interlock." ) else: diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py index 4088d42a..21bb1a06 100644 --- a/tests/unit_tests/test_beamline_state_pill.py +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -52,20 +52,28 @@ def _shutter_state( return _wire_state(bl_states.ShutterState, config) +def _as_status_list(value: str | list[str]) -> list[str]: + return [value] if isinstance(value, str) else list(value) + + class _FakeScanInterlock: - def __init__(self, enabled: bool = False, states_watched: dict[str, str] | None = None): + def __init__( + self, enabled: bool = False, states_watched: dict[str, str | list[str]] | None = None + ): self.enabled = enabled - self._states = dict(states_watched or {}) + self._states = { + name: _as_status_list(value) for name, value in (states_watched or {}).items() + } self.added: list[tuple[str, str]] = [] self.removed: list[str] = [] @property - def states_watched(self) -> dict[str, str]: - return dict(self._states) + def states_watched(self) -> dict[str, list[str]]: + return {name: list(statuses) for name, statuses in self._states.items()} def add_state_to_interlock(self, state_name: str, required_value: str = "valid") -> None: self.added.append((state_name, required_value)) - self._states[state_name] = required_value + self._states[state_name] = _as_status_list(required_value) def remove_state_from_interlock(self, state_name: str) -> None: self.removed.append(state_name) @@ -536,7 +544,7 @@ def test_beamline_state_pill_emits_interlock_toggle_request(qtbot, mocked_client assert include_signal.args == ["limits", True] - pill.set_scan_interlock("valid", False) + pill.set_scan_interlock(["valid"], False) with qtbot.waitSignal(pill.scan_interlock_toggle_requested) as exclude_signal: pill._interlock_button.click() @@ -551,7 +559,7 @@ def test_beamline_state_pill_included_state_forces_card_background(qtbot, mocked assert f"border: 1px solid {colors['card_border']}" not in pill.styleSheet().split(":hover")[0] assert not pill._shadow.isEnabled() - pill.set_scan_interlock("valid", False) + pill.set_scan_interlock(["valid"], False) assert f"border: 1px solid {colors['card_border']}" in pill.styleSheet() assert pill._shadow.isEnabled() @@ -566,7 +574,7 @@ def test_beamline_state_pill_included_state_forces_card_background(qtbot, mocked def test_beamline_state_pill_triggered_interlock_animates(qtbot, mocked_client): pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client) - pill.set_scan_interlock("valid", True) + pill.set_scan_interlock(["valid"], True) assert pill._interlock_animation.state() == pill._interlock_animation.State.Running pill.interlock_pulse = 0.5 @@ -575,7 +583,7 @@ def test_beamline_state_pill_triggered_interlock_animates(qtbot, mocked_client): assert "border: 2px solid" in stylesheet assert "qlineargradient" in stylesheet - pill.set_scan_interlock("valid", False) + pill.set_scan_interlock(["valid"], False) assert pill._interlock_animation.state() == pill._interlock_animation.State.Stopped assert pill._interlock_pulse == 0.0 @@ -751,7 +759,7 @@ def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client): _install_fake_scan_interlock(beamline_state_manager, fake_interlock) pill = beamline_state_manager._state_pills["shutter_open"] - assert pill._interlock_required_status == "valid" + assert pill._interlock_required_statuses == ["valid"] assert pill._interlock_triggered pill.update_state({"name": "shutter_open", "status": "valid", "label": "Open."}, {}) @@ -766,7 +774,7 @@ def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client): beamline_state_manager._refresh_scan_interlock() assert not pill._interlock_triggered - assert pill._interlock_required_status == "valid" + assert pill._interlock_required_statuses == ["valid"] def test_beamline_state_manager_orders_both_sections_by_status_severity(qtbot, mocked_client): @@ -876,7 +884,7 @@ def test_beamline_state_manager_pill_toggle_calls_backend(qtbot, mocked_client): assert fake_interlock.added == [("limits", "valid")] # 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"]} beamline_state_manager._state_pills["limits"]._interlock_button.click()