feat(beamline-states): add state manager widget

This commit is contained in:
2026-06-09 11:02:57 +02:00
parent 7916a79c53
commit d2bfe92aae
14 changed files with 2532 additions and 0 deletions
+106
View File
@@ -32,6 +32,8 @@ _Widgets = {
"BECQueue": "BECQueue",
"BECShell": "BECShell",
"BECStatusBox": "BECStatusBox",
"BeamlineStateManager": "BeamlineStateManager",
"BeamlineStatePill": "BeamlineStatePill",
"BecConsole": "BecConsole",
"DapComboBox": "DapComboBox",
"DeviceBrowser": "DeviceBrowser",
@@ -717,6 +719,110 @@ class BaseROI(RPCBase):
"""
class BeamlineStateManager(RPCBase):
"""Widget displaying and managing all BEC beamline states."""
_IMPORT_MODULE = "bec_widgets.widgets.services.beamline_states.beamline_state_pill"
@property
@rpc_call
def idle_card_background(self) -> "bool":
"""
Whether idle collapsed pills keep the status-tinted card background.
"""
@rpc_call
def set_idle_card_background(self, enabled: "bool") -> "None":
"""
Set whether idle collapsed pills keep the status-tinted card background.
"""
@rpc_call
def refresh_states(self) -> "None":
"""
Fetch the latest cached available beamline states and update the list immediately.
"""
@rpc_call
def clear_filters(self) -> "None":
"""
None
"""
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
@rpc_timeout(None)
@rpc_call
def screenshot(self, file_name: "str | None" = None):
"""
Take a screenshot of the dock area and save it to a file.
"""
class BeamlineStatePill(RPCBase):
"""Compact widget showing one BEC beamline state."""
_IMPORT_MODULE = "bec_widgets.widgets.services.beamline_states.beamline_state_pill"
@property
@rpc_call
def state_name(self) -> "str | None":
"""
Name of the BEC beamline state displayed by this pill.
"""
@rpc_call
def set_state_name(self, state_name: "str | None", title: "str | None" = None) -> "None":
"""
Set the BEC beamline state this pill displays.
Args:
state_name: State name as published by ``AvailableBeamlineStatesMessage``.
title: Optional human-readable title for the state.
"""
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
@rpc_timeout(None)
@rpc_call
def screenshot(self, file_name: "str | None" = None):
"""
Take a screenshot of the dock area and save it to a file.
"""
class BecConsole(RPCBase):
"""A console widget with access to a shared registry of terminals, such that instances can be moved around."""
+10
View File
@@ -19,6 +19,14 @@ designer_plugins = {
"BECShell": ("bec_widgets.widgets.editors.bec_console.bec_console", "BECShell"),
"BECSpinBox": ("bec_widgets.widgets.utility.spinbox.decimal_spinbox", "BECSpinBox"),
"BECStatusBox": ("bec_widgets.widgets.services.bec_status_box.bec_status_box", "BECStatusBox"),
"BeamlineStateManager": (
"bec_widgets.widgets.services.beamline_states.beamline_state_pill",
"BeamlineStateManager",
),
"BeamlineStatePill": (
"bec_widgets.widgets.services.beamline_states.beamline_state_pill",
"BeamlineStatePill",
),
"BecConsole": ("bec_widgets.widgets.editors.bec_console.bec_console", "BecConsole"),
"ColorButton": ("bec_widgets.widgets.utility.visual.color_button.color_button", "ColorButton"),
"ColorButtonNative": (
@@ -118,6 +126,8 @@ widget_icons = {
"BECShell": "hub",
"BECSpinBox": "123",
"BECStatusBox": "widgets",
"BeamlineStateManager": "format_list_bulleted",
"BeamlineStatePill": "info",
"BecConsole": "terminal",
"ColorButton": "colors",
"ColorButtonNative": "colors",
@@ -0,0 +1 @@
{'files': ['beamline_state_pill.py']}
@@ -0,0 +1,57 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStateManager
DOM_XML = """
<ui language='c++'>
<widget class='BeamlineStateManager' name='beamline_state_manager'>
</widget>
</ui>
"""
class BeamlineStateManagerPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = BeamlineStateManager(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Services"
def icon(self):
return designer_material_icon(BeamlineStateManager.ICON_NAME)
def includeFile(self):
return "beamline_state_manager"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "BeamlineStateManager"
def toolTip(self):
return ""
def whatsThis(self):
return self.toolTip()
File diff suppressed because it is too large Load Diff
@@ -0,0 +1 @@
{'files': ['beamline_state_pill.py']}
@@ -0,0 +1,57 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStatePill
DOM_XML = """
<ui language='c++'>
<widget class='BeamlineStatePill' name='beamline_state_pill'>
</widget>
</ui>
"""
class BeamlineStatePillPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = BeamlineStatePill(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(BeamlineStatePill.ICON_NAME)
def includeFile(self):
return "beamline_state_pill"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "BeamlineStatePill"
def toolTip(self):
return ""
def whatsThis(self):
return self.toolTip()
@@ -0,0 +1,267 @@
from __future__ import annotations
import slugify
from bec_lib import bl_states
from qtpy.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QDialogButtonBox,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLineEdit,
QMessageBox,
QPushButton,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.forms_from_types.pydantic_widget_form import PydanticWidgetForm
from bec_widgets.utils.name_utils import pascal_to_snake
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
BEAMLINE_STATE_STATUS_LABELS = {
"valid": "VALID",
"invalid": "INVALID",
"warning": "WARNING",
"unknown": "UNKNOWN",
}
SUPPORTED_BEAMLINE_STATES: tuple[type[bl_states.BeamlineState], ...] = (
bl_states.DeviceWithinLimitsState,
bl_states.ShutterState,
)
class AddBeamlineStateDialog(QDialog):
"""Dialog for creating supported beamline state configurations."""
def __init__(self, parent: QWidget | None = None, client=None) -> None:
super().__init__(parent=parent)
self.setWindowTitle("Add Beamline State")
self._cleaned_up = False
self._client = client
self._config: bl_states.BeamlineStateConfig | None = None
self._auto_generated_name: str | None = None
self._type_combo = QComboBox(self)
for state_class in SUPPORTED_BEAMLINE_STATES:
self._type_combo.addItem(state_class.__name__, state_class)
self._type_combo.currentIndexChanged.connect(self._update_config_form)
self._form = QFormLayout()
self._form.addRow("State type", self._type_combo)
self._config_form_host = QVBoxLayout()
self._config_form: PydanticWidgetForm | None = None
self._buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self
)
self._buttons.accepted.connect(self.accept)
self._buttons.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addLayout(self._form)
layout.addLayout(self._config_form_host)
layout.addWidget(self._buttons)
self.setLayout(layout)
self.setMinimumWidth(280)
self._update_config_form()
self._fit_height_to_contents()
def config(self) -> bl_states.BeamlineStateConfig:
state_class = self._selected_state_class()
config_class = state_class.CONFIG_CLASS
name = self._state_name()
data = self._config_form.get_data()
data["name"] = name
return config_class.model_validate(data)
def accept(self) -> None:
try:
self._config = self.config()
except Exception as exc:
QMessageBox.warning(self, "Invalid Beamline State", str(exc))
return
super().accept()
@property
def config_result(self) -> bl_states.BeamlineStateConfig:
if self._config is None:
raise RuntimeError("Beamline state dialog was not accepted with a valid config.")
return self._config
def cleanup(self) -> None:
if self._cleaned_up:
return
self._cleaned_up = True
if self._config_form is not None:
self._config_form.cleanup()
self._config_form.close()
self._config_form.deleteLater()
def closeEvent(self, event) -> None: # noqa: N802
self.cleanup()
super().closeEvent(event)
@SafeSlot(str)
def _on_valid_device_selected(self, device: str) -> None:
if self._cleaned_up:
return
name_widget = self._config_form.input_widget("name")
current_name = name_widget.text().strip()
if current_name and current_name != self._auto_generated_name:
return
suffix = slugify.slugify(
pascal_to_snake(self._selected_state_class().__name__), separator="_"
)
generated_name = f"{slugify.slugify(device, separator='_')}_{suffix}"
self._auto_generated_name = generated_name
name_widget.setText(generated_name)
@SafeSlot(int)
def _update_config_form(self, _index: int = 0) -> None:
previous_data = self._config_form.raw_data() if self._config_form is not None else {}
if self._config_form is not None:
self._config_form_host.removeWidget(self._config_form)
self._config_form.cleanup()
self._config_form.setParent(None)
self._config_form.deleteLater()
config_class = self._selected_state_class().CONFIG_CLASS
data = {
key: value
for key, value in previous_data.items()
if key in config_class.model_fields and value is not None
}
self._config_form = PydanticWidgetForm(config_class, parent=self, client=self._client)
self._config_form.set_partial_data(data)
self._config_form_host.addWidget(self._config_form)
for device_widget in self._config_form.input_widgets_by_type(DeviceComboBox):
device_widget.device_selected.connect(self._on_valid_device_selected)
self._fit_height_to_contents()
def _fit_height_to_contents(self) -> None:
self.setMinimumHeight(0)
self.setMaximumHeight(16777215)
self.layout().activate()
self.adjustSize()
height = self.sizeHint().expandedTo(self.minimumSizeHint()).height()
self.setMinimumHeight(height)
self.setMaximumHeight(height)
def _selected_state_class(self) -> type[bl_states.BeamlineState]:
state_class = self._type_combo.currentData()
if state_class is None:
raise RuntimeError("No beamline state class selected.")
return state_class
def _state_name(self) -> str:
name_widget = self._config_form.input_widget("name")
raw_name = name_widget.text().strip()
if not raw_name:
raise ValueError("Name is required.")
name = slugify.slugify(raw_name, separator="_")
name_widget.setText(name)
return name
class StatusFilterDialog(QDialog):
"""Dialog for selecting visible beamline state statuses."""
def __init__(self, selected_statuses: set[str] | None, parent: QWidget | None = None) -> None:
super().__init__(parent=parent)
self.setWindowTitle("Filter Beamline State Status")
self._checkboxes: dict[str, QCheckBox] = {}
controls = QHBoxLayout()
select_all = QPushButton("Select all", self)
clear = QPushButton("Clear", self)
select_all.clicked.connect(lambda: self._set_all(True))
clear.clicked.connect(lambda: self._set_all(False))
controls.addWidget(select_all)
controls.addWidget(clear)
controls.addStretch(1)
list_layout = QVBoxLayout()
for status, label in BEAMLINE_STATE_STATUS_LABELS.items():
checkbox = QCheckBox(label, self)
checkbox.setChecked(selected_statuses is None or status in selected_statuses)
self._checkboxes[status] = checkbox
list_layout.addWidget(checkbox)
list_layout.addStretch(1)
box = QGroupBox("Displayed status", self)
box.setLayout(list_layout)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self
)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addLayout(controls)
layout.addWidget(box)
layout.addWidget(buttons)
self.setLayout(layout)
def selected_statuses(self) -> set[str] | None:
selected = {status for status, checkbox in self._checkboxes.items() if checkbox.isChecked()}
if selected == set(self._checkboxes):
return None
return selected
def _set_all(self, checked: bool) -> None:
for checkbox in self._checkboxes.values():
checkbox.setChecked(checked)
class DeviceFilterDialog(QDialog):
"""Dialog for filtering beamline states by configured device."""
def __init__(
self,
devices: list[str],
selected_devices: set[str] | None,
device_filter_text: str,
parent: QWidget | None = None,
) -> None:
super().__init__(parent=parent)
self.setWindowTitle("Filter Beamline State Devices")
self._checkboxes: dict[str, QCheckBox] = {}
self._device_text = QLineEdit(self)
self._device_text.setPlaceholderText("Device name or comma-separated names")
self._device_text.setText(device_filter_text)
list_layout = QVBoxLayout()
for device in devices:
checkbox = QCheckBox(device, self)
checkbox.setChecked(selected_devices is not None and device in selected_devices)
self._checkboxes[device] = checkbox
list_layout.addWidget(checkbox)
list_layout.addStretch(1)
box = QGroupBox("Known devices", self)
box.setLayout(list_layout)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self
)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addWidget(self._device_text)
layout.addWidget(box)
layout.addWidget(buttons)
self.setLayout(layout)
def selected_devices(self) -> set[str] | None:
selected = {device for device, checkbox in self._checkboxes.items() if checkbox.isChecked()}
return selected or None
def filter_text(self) -> str:
return self._device_text.text().strip()
@@ -0,0 +1,17 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.services.beamline_states.beamline_state_manager_plugin import (
BeamlineStateManagerPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(BeamlineStateManagerPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -0,0 +1,17 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.services.beamline_states.beamline_state_pill_plugin import (
BeamlineStatePillPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(BeamlineStatePillPlugin())
if __name__ == "__main__": # pragma: no cover
main()
+1
View File
@@ -24,6 +24,7 @@ dependencies = [
"pydantic~=2.0",
"pylsp-bec~=1.2",
"pyqtgraph==0.13.7",
"python-slugify~=8.0",
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtmonaco~=0.8, >=0.8.1",
"qtpy~=2.4",
@@ -0,0 +1,696 @@
from typing import Any, Generator
import pytest
import shiboken6
from bec_lib import bl_states
from qtpy.QtCore import QCoreApplication, QEvent, Qt
from qtpy.QtWidgets import QMessageBox, QStyleOptionViewItem
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.services.beamline_states import beamline_state_pill as pill_module
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import (
BeamlineStateManager,
BeamlineStatePill,
)
from bec_widgets.widgets.services.beamline_states.dialogs import AddBeamlineStateDialog
from .client_mocks import mocked_client
@pytest.fixture
def pill(qtbot, mocked_client) -> Generator[BeamlineStatePill, Any, None]:
widget = BeamlineStatePill(state_name="shutter_open", title="Shutter", client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_beamline_state_pill_updates_from_message(pill):
pill.update_state({"name": "shutter_open", "status": "valid", "label": "Shutter is open."}, {})
assert pill._state_name == "shutter_open"
assert pill._name_label.text() == "Shutter"
assert pill._status_label.text() == "VALID"
assert pill._detail_label.text() == "Shutter is open."
assert not pill._icon_label.pixmap().isNull()
assert pill.toolTip() == "Shutter is open."
def test_beamline_state_pill_ignores_other_states(pill):
pill.update_state(
{"name": "other_state", "status": "invalid", "label": "Should be ignored."}, {}
)
assert pill._status_label.text() == "UNKNOWN"
assert pill.toolTip() == "No state information available."
def test_beamline_state_pill_expands_and_emits_updated_limits(qtbot, mocked_client):
widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
widget.set_state_config(
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"name": "limits",
"title": "Limits",
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
)
assert widget._settings.isHidden()
assert widget._config_form is None
assert not widget._update_button.isEnabled()
assert not widget._revert_button.isEnabled()
qtbot.mouseClick(widget._header, Qt.MouseButton.LeftButton)
assert widget._config_form is not None
high_limit = widget._config_form.input_widget("high_limit")
high_limit.setValue(20.0)
assert not widget._settings.isHidden()
assert widget._update_button.isEnabled()
assert widget._revert_button.isEnabled()
assert widget._config_form.field_widget("high_limit").property("beamlineStateDirty") is True
assert widget._config_form.get_data()["device"] == "samx"
assert widget.edited_config().high_limit == 20.0
with qtbot.waitSignal(widget.update_requested) as signal:
widget._update_button.click()
assert signal.args[0] == "limits"
assert isinstance(signal.args[1], bl_states.DeviceWithinLimitsState.CONFIG_CLASS)
assert signal.args[1].device == "samx"
assert signal.args[1].signal == "samx"
assert signal.args[1].low_limit == 0.0
assert signal.args[1].high_limit == 20.0
assert signal.args[1].tolerance == 0.1
assert not widget._settings.isHidden()
def test_beamline_state_pill_first_expand_uses_config_class_without_rebuild(
qtbot, mocked_client, monkeypatch
):
set_model_calls = []
original_set_model = pill_module.PydanticWidgetForm.set_model
def set_model_spy(self, model, data=None):
set_model_calls.append(model)
return original_set_model(self, model, data=data)
monkeypatch.setattr(pill_module.PydanticWidgetForm, "set_model", set_model_spy)
widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client)
qtbot.addWidget(widget)
widget.set_state_config(
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
)
widget.set_expanded(True)
assert widget._config_form is not None
assert set_model_calls == []
def test_beamline_state_pill_reverts_changed_settings(qtbot, mocked_client):
widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client)
qtbot.addWidget(widget)
widget.set_state_config(
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
)
widget.set_expanded(True)
assert widget._config_form is not None
low_limit = widget._config_form.input_widget("low_limit")
low_limit.setValue(-5.0)
assert widget._update_button.isEnabled()
assert widget._config_form.field_widget("low_limit").property("beamlineStateDirty") is True
widget._revert_button.click()
assert low_limit.value() == 0.0
assert not widget._update_button.isEnabled()
assert not widget._revert_button.isEnabled()
assert widget._config_form.field_widget("low_limit").property("beamlineStateDirty") is False
def test_beamline_state_pill_does_not_override_themed_input_controls(qtbot, mocked_client):
widget = BeamlineStatePill(state_name="limits", title="Limits", client=mocked_client)
qtbot.addWidget(widget)
widget.set_expanded(True)
stylesheet = widget.styleSheet()
assert "QAbstractSpinBox" not in stylesheet
assert "QComboBox" not in stylesheet
assert "QCheckBox::indicator" not in stylesheet
def test_beamline_state_manager_adds_and_removes_pills(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
messages.BeamlineStateConfig(
name="shutter_open", title="Shutter", state_type="ShutterState", parameters={}
),
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {},
},
]
},
{},
)
assert sorted(widget._state_pills) == ["limits", "shutter_open"]
assert widget._model.rowCount() == 2
assert widget._state_pills["shutter_open"]._name_label.text() == "Shutter"
assert not widget._empty_label.isVisible()
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {},
}
]
},
{},
)
assert sorted(widget._state_pills) == ["limits"]
assert widget._model.rowCount() == 1
def test_beamline_state_manager_ignores_unchanged_available_states(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
content = {
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
]
}
widget.update_available_states(content, {})
pill = widget._state_pills["limits"]
widget.update_available_states(content, {})
assert widget._state_pills["limits"] is pill
assert pill._config_form is None
def test_beamline_state_manager_adds_state_without_recreating_existing_pills(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
limits_state = {
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
shutter_state = {
"name": "shutter_open",
"title": "Shutter",
"state_type": "ShutterState",
"parameters": {},
}
widget.update_available_states({"states": [limits_state]}, {})
pill = widget._state_pills["limits"]
pill.set_expanded(True)
config_form = pill._config_form
widget.update_available_states({"states": [limits_state, shutter_state]}, {})
assert widget._state_pills["limits"] is pill
assert pill._config_form is config_form
assert pill.is_expanded()
assert sorted(widget._state_pills) == ["limits", "shutter_open"]
def test_beamline_state_manager_does_not_force_horizontal_minimum(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
]
},
{},
)
index = widget._model.index_for_name("limits")
hint = widget._delegate.sizeHint(QStyleOptionViewItem(), index)
assert widget.minimumWidth() == 0
assert widget.minimumSizeHint().width() == 0
assert widget._view.minimumWidth() == 0
assert widget._view.minimumSizeHint().width() == 0
assert hint.width() == 0
def test_beamline_state_manager_header_click_expands_pill_once(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx"},
}
]
},
{},
)
pill = widget._state_pills["limits"]
assert pill._settings.isHidden()
qtbot.mouseClick(pill._header, Qt.MouseButton.LeftButton)
assert not pill._settings.isHidden()
def test_beamline_state_manager_preserves_expanded_pill_on_refresh(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
state = {
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx", "high_limit": 10.0},
}
widget.update_available_states({"states": [state]}, {})
widget._state_pills["limits"].set_expanded(True)
widget.update_available_states({"states": [state]}, {})
assert widget._state_pills["limits"].is_expanded()
assert not widget._state_pills["limits"]._settings.isHidden()
def test_beamline_state_manager_propagates_idle_card_background(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client, idle_card_background=True)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx"},
}
]
},
{},
)
assert widget._state_pills["limits"]._idle_card_background is True
widget.idle_card_background = False
assert widget._state_pills["limits"]._idle_card_background is False
def test_beamline_state_manager_filters_status(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "shutter_open",
"title": "Shutter",
"state_type": "ShutterState",
"parameters": {"device": "samy"},
},
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx"},
},
]
},
{},
)
assert isinstance(widget._toolbar, ModularToolBar)
widget._state_pills["limits"].update_state(
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
)
widget._state_pills["shutter_open"].update_state(
{"name": "shutter_open", "status": "invalid", "label": "Closed."}, {}
)
widget._selected_statuses = {"valid"}
widget._apply_filters()
assert not widget._hidden_summary.isHidden()
assert "1 state is hidden" in widget._hidden_summary.text()
assert not widget._view.isRowHidden(widget._model.index_for_name("limits").row())
assert widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row())
widget._hidden_summary.click()
assert not widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row())
assert shiboken6.isValid(widget._state_pills["shutter_open"])
widget._hidden_summary.click()
assert widget._view.isRowHidden(widget._model.index_for_name("shutter_open").row())
assert shiboken6.isValid(widget._state_pills["shutter_open"])
def test_beamline_state_manager_status_filter_reacts_to_state_changes(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx"},
}
]
},
{},
)
widget._selected_statuses = {"valid"}
widget._state_pills["limits"].update_state(
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
)
assert widget._hidden_summary.isHidden()
widget._state_pills["limits"].update_state(
{"name": "limits", "status": "invalid", "label": "Out of limits."}, {}
)
assert not widget._hidden_summary.isHidden()
assert widget._view.isRowHidden(widget._model.index_for_name("limits").row())
def test_beamline_state_manager_filters_devices(qtbot, mocked_client, monkeypatch):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "samx_limits",
"title": "samx",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samx"},
},
{
"name": "samy_limits",
"title": "samy",
"state_type": "DeviceWithinLimitsState",
"parameters": {"device": "samy"},
},
]
},
{},
)
widget._device_filter_text = "samx"
widget._apply_filters()
assert not widget._hidden_summary.isHidden()
assert "1 state is hidden" in widget._hidden_summary.text()
captured = {}
class FakeDeviceFilterDialog:
def __init__(self, devices, selected_devices, device_filter_text, parent):
captured["devices"] = devices
captured["selected_devices"] = selected_devices
captured["device_filter_text"] = device_filter_text
captured["parent"] = parent
def exec(self):
return 0
monkeypatch.setattr(pill_module, "DeviceFilterDialog", FakeDeviceFilterDialog)
widget.open_device_filter_dialog()
assert captured["devices"] == ["samx", "samy"]
assert captured["device_filter_text"] == "samx"
assert captured["parent"] is widget
def test_beamline_state_manager_updates_state_parameters(qtbot, mocked_client):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
widget.update_available_states(
{
"states": [
{
"name": "limits",
"title": "Limits",
"state_type": "DeviceWithinLimitsState",
"parameters": {
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 10.0,
"tolerance": 0.1,
},
}
]
},
{},
)
class StateClient:
def __init__(self):
self.parameters = None
def update_parameters(self, **kwargs):
self.parameters = kwargs
class StateManager:
def __init__(self):
self.limits = StateClient()
mocked_client.beamline_states = StateManager()
pill = widget._state_pills["limits"]
pill.set_expanded(True)
high_limit = pill._config_form.input_widget("high_limit")
high_limit.setValue(20.0)
assert pill._update_button.isEnabled()
widget._update_state_parameters("limits", pill.edited_config())
assert mocked_client.beamline_states.limits.parameters == {
"title": "Limits",
"device": "samx",
"signal": "samx",
"low_limit": 0.0,
"high_limit": 20.0,
"tolerance": 0.1,
}
assert not pill._update_button.isEnabled()
assert pill._config_form.field_widget("high_limit").property("beamlineStateDirty") is False
def test_beamline_state_manager_removes_state(qtbot, mocked_client, monkeypatch):
widget = BeamlineStateManager(client=mocked_client)
qtbot.addWidget(widget)
class StateManager:
def __init__(self):
self.deleted = None
def delete(self, state_name):
self.deleted = state_name
mocked_client.beamline_states = StateManager()
monkeypatch.setattr(
QMessageBox, "question", lambda *args, **kwargs: QMessageBox.StandardButton.Yes
)
widget._remove_state_requested("limits")
assert mocked_client.beamline_states.deleted == "limits"
def test_add_beamline_state_dialog_uses_generated_widgets_and_normalizes_name(qtbot, mocked_client):
dialog = AddBeamlineStateDialog(client=mocked_client)
qtbot.addWidget(dialog)
limits_index = dialog._type_combo.findText(bl_states.DeviceWithinLimitsState.__name__)
assert limits_index >= 0
dialog._type_combo.setCurrentIndex(limits_index)
assert dialog._config_form.model is bl_states.DeviceWithinLimitsState.CONFIG_CLASS
name = dialog._config_form.input_widget("name")
title = dialog._config_form.input_widget("title")
device = dialog._config_form.input_widget("device")
signal = dialog._config_form.input_widget("signal")
low_limit = dialog._config_form.field_widget("low_limit")
high_limit = dialog._config_form.field_widget("high_limit")
name.setText("samx-limits")
title.setText("samx-limits-15")
WidgetIO.set_value(device, "samx")
WidgetIO.set_value(signal, "samx")
low_limit.checkbox.setChecked(True)
high_limit.checkbox.setChecked(True)
high_limit.value_widget.setValue(15.0)
config = dialog.config()
assert config.name == "samx_limits"
assert config.title == "samx-limits-15"
assert config.device == "samx"
assert config.signal == "samx"
assert config.low_limit == 0.0
assert config.high_limit == 15.0
def test_add_beamline_state_dialog_generates_name_only_after_valid_device_selection(
qtbot, mocked_client
):
dialog = AddBeamlineStateDialog(client=mocked_client)
qtbot.addWidget(dialog)
name = dialog._config_form.input_widget("name")
device = dialog._config_form.input_widget("device")
device.setCurrentText("s")
assert name.text() == ""
device.set_device("samx")
assert name.text() == "samx_device_within_limits_state"
def test_add_beamline_state_dialog_switches_state_type_without_collapsing(qtbot, mocked_client):
dialog = AddBeamlineStateDialog(client=mocked_client)
qtbot.addWidget(dialog)
initial_height = dialog.height()
limits_index = dialog._type_combo.findText("DeviceWithinLimitsState")
assert limits_index >= 0
shutter_index = dialog._type_combo.findText("ShutterState")
assert shutter_index >= 0
dialog._type_combo.setCurrentIndex(shutter_index)
qtbot.wait(0)
assert dialog._config_form.model is bl_states.DeviceStateConfig
assert dialog._config_form_host.count() == 1
assert not dialog._config_form.isHidden()
assert not dialog._buttons.isHidden()
assert dialog.sizeHint().height() > dialog._buttons.sizeHint().height()
assert dialog.minimumWidth() == 280
assert dialog.maximumWidth() > dialog.minimumWidth()
assert dialog.minimumHeight() == dialog.maximumHeight()
dialog._type_combo.setCurrentIndex(limits_index)
qtbot.wait(0)
assert dialog._config_form.model is bl_states.DeviceWithinLimitsState.CONFIG_CLASS
assert dialog.height() >= initial_height
assert dialog.minimumHeight() == dialog.maximumHeight()
def test_add_beamline_state_dialog_cleanup_deletes_device_widgets(qtbot, mocked_client):
dialog = AddBeamlineStateDialog(client=mocked_client)
qtbot.addWidget(dialog)
device = dialog._config_form.input_widget("device")
signal = dialog._config_form.input_widget("signal")
dialog.reject()
assert shiboken6.isValid(device)
assert shiboken6.isValid(signal)
dialog.cleanup()
QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete)
assert not shiboken6.isValid(device)
assert not shiboken6.isValid(signal)
+13
View File
@@ -0,0 +1,13 @@
from bec_widgets.utils.name_utils import pascal_to_snake, sanitize_namespace
def test_pascal_to_snake():
assert pascal_to_snake("DeviceWithinLimitsState") == "device_within_limits_state"
assert pascal_to_snake("BECStatusWidget") == "bec_status_widget"
def test_sanitize_namespace():
assert sanitize_namespace("scan 1 / user") == "scan_1_user"
assert sanitize_namespace(" beamline.state-1 ") == "beamline.state-1"
assert sanitize_namespace(" ") is None
assert sanitize_namespace(None) is None