bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-06-28 23:59:46 +02:00
Code Activity

fix(beamline_states): support multiple accepted statuses in the scan interlock

Browse Source
This commit is contained in:
wyzula_j
2026-06-15 22:55:11 +02:00
committed by Klaus Wakonig
parent 1fac492b0b
commit 2084edb576
3 changed files with 39 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/widgets/services/beamline_states/beamline_state_manager.py
+5 -4
View File
@@ -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."""
bec_widgets/widgets/services/beamline_states/beamline_state_pill.py
+14 -13
View File
@@ -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:
tests/unit_tests/test_beamline_state_pill.py
+20 -12
View File
@@ -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()