diff --git a/bec_widgets/widgets/services/beamline_states/__init__.py b/bec_widgets/widgets/services/beamline_states/__init__.py new file mode 100644 index 00000000..37a87e41 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/__init__.py @@ -0,0 +1,6 @@ +from bec_widgets.widgets.services.beamline_states.beamline_state_pill import ( + BeamlineStateList, + BeamlineStatePill, +) + +__all__ = ["BeamlineStateList", "BeamlineStatePill"] diff --git a/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py new file mode 100644 index 00000000..77c38450 --- /dev/null +++ b/bec_widgets/widgets/services/beamline_states/beamline_state_pill.py @@ -0,0 +1,291 @@ +from __future__ import annotations + +import sys +from typing import Any + +from bec_lib.endpoints import MessageEndpoints +from qtpy.QtCore import Qt, Signal, Slot +from qtpy.QtGui import QColor +from qtpy.QtWidgets import QApplication, QHBoxLayout, QLabel, QScrollArea, QVBoxLayout, QWidget + +from bec_widgets.utils.bec_connector import ConnectionConfig +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.colors import get_accent_colors + + +class BeamlineStatePill(BECWidget, QWidget): + """ + Compact widget showing one BEC beamline state. + + The pill subscribes to ``MessageEndpoints.beamline_state(state_name)`` and updates whenever + a ``BeamlineStateMessage`` is published for that state. + """ + + PLUGIN = True + ICON_NAME = "info" + USER_ACCESS = ["state_name", "set_state_name", "remove", "attach", "detach", "screenshot"] + + state_changed = Signal(str, str, str) + + _STATUS_LABELS = { + "valid": "VALID", + "invalid": "INVALID", + "warning": "WARNING", + "unknown": "UNKNOWN", + } + + def __init__( + self, + parent: QWidget | None = None, + state_name: str | None = None, + title: str | None = None, + client=None, + config: ConnectionConfig | None = None, + gui_id: str | None = None, + **kwargs, + ) -> None: + super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs) + self._state_name: str | None = None + self._title: str | None = None + + self._name_label = QLabel(self) + self._name_label.setObjectName("beamline_state_name") + self._status_label = QLabel(self) + self._status_label.setObjectName("beamline_state_status") + + layout = QHBoxLayout(self) + layout.setContentsMargins(10, 4, 10, 4) + layout.setSpacing(8) + layout.addWidget(self._name_label) + layout.addWidget(self._status_label, 0, Qt.AlignmentFlag.AlignRight) + self.setLayout(layout) + + self.set_state_name(state_name, title=title) + + @property + def state_name(self) -> str | None: + """Name of the BEC beamline state displayed by this pill.""" + return self._state_name + + 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. + """ + if state_name == self._state_name and title == self._title: + return + + if self._state_name is not None: + self.bec_dispatcher.disconnect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + + self._state_name = state_name + self._title = title + self._name_label.setText(title or state_name or "Beamline state") + + if self._state_name is None: + self._set_visual_state("unknown", "No beamline state selected.") + return + + self._set_visual_state("unknown", "No state information available.") + self._refresh_latest_state() + self.bec_dispatcher.connect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + + def _refresh_latest_state(self) -> None: + if self._state_name is None: + return + msg_container = self.client.connector.get_last( + MessageEndpoints.beamline_state(self._state_name) + ) + if not msg_container: + return + data = msg_container.get("data") if isinstance(msg_container, dict) else None + content = getattr(data, "content", data) + if isinstance(content, dict): + self.update_state(content, getattr(data, "metadata", {})) + + @Slot(dict, dict) + def update_state( + self, content: dict[str, Any], _metadata: dict[str, Any] | None = None + ) -> None: + """ + Update this pill from a ``BeamlineStateMessage`` content dictionary. + """ + name = content.get("name") + if self._state_name is not None and name and name != self._state_name: + return + + status = str(content.get("status", "unknown")).lower() + label = str(content.get("label", "No state information available.")) + self._set_visual_state(status, label) + self.state_changed.emit(self._state_name or str(name or ""), status, label) + + def _set_visual_state(self, status: str, label: str) -> None: + status = status if status in self._STATUS_LABELS else "unknown" + color = self._status_color(status) + self._status_label.setText(self._STATUS_LABELS[status]) + self.setToolTip(label) + self.setStyleSheet( + "BeamlineStatePill {" + f"border: 1px solid {color};" + "border-radius: 10px;" + "}" + "QLabel#beamline_state_name {" + "font-weight: 600;" + "}" + "QLabel#beamline_state_status {" + f"color: {color};" + "font-weight: 700;" + "}" + ) + + @staticmethod + def _status_color(status: str) -> str: + accent_colors = get_accent_colors() + colors = { + "valid": accent_colors.success, + "invalid": accent_colors.emergency, + "warning": accent_colors.warning, + "unknown": "#7a7a7a", + } + color = colors.get(status, colors["unknown"]) + if isinstance(color, QColor): + return color.name() + return str(color) + + def cleanup(self) -> None: + if self._state_name is not None: + self.bec_dispatcher.disconnect_slot( + self.update_state, MessageEndpoints.beamline_state(self._state_name) + ) + super().cleanup() + + +class BeamlineStateList(BECWidget, QWidget): + """ + Widget displaying all BEC beamline states as a vertical list of pills. + + The list subscribes to ``MessageEndpoints.available_beamline_states()`` and creates, updates, + or removes child ``BeamlineStatePill`` widgets as the set of configured states changes. + """ + + PLUGIN = True + ICON_NAME = "format_list_bulleted" + USER_ACCESS = ["refresh_states", "remove", "attach", "detach", "screenshot"] + + def __init__( + self, + parent: QWidget | None = None, + client=None, + config: ConnectionConfig | None = None, + gui_id: str | None = None, + **kwargs, + ) -> None: + super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs) + self._state_pills: dict[str, BeamlineStatePill] = {} + + self._empty_label = QLabel("No beamline states available.", self) + self._empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + + self._content = QWidget(self) + self._content_layout = QVBoxLayout(self._content) + self._content_layout.setContentsMargins(0, 0, 0, 0) + self._content_layout.setSpacing(6) + self._content_layout.addWidget(self._empty_label) + self._content_layout.addStretch(1) + + self._scroll_area = QScrollArea(self) + self._scroll_area.setWidgetResizable(True) + self._scroll_area.setFrameShape(QScrollArea.Shape.NoFrame) + self._scroll_area.setWidget(self._content) + + layout = QVBoxLayout(self) + layout.setContentsMargins(8, 8, 8, 8) + layout.setSpacing(6) + layout.addWidget(self._scroll_area) + self.setLayout(layout) + + self.bec_dispatcher.connect_slot( + self.update_available_states, + MessageEndpoints.available_beamline_states(), + from_start=True, + ) + self.refresh_states() + + def refresh_states(self) -> None: + """Fetch the latest cached available beamline states and update the list immediately.""" + msg_container = self.client.connector.get_last(MessageEndpoints.available_beamline_states()) + if not msg_container: + return + data = msg_container.get("data") if isinstance(msg_container, dict) else None + content = getattr(data, "content", data) + if isinstance(content, dict): + self.update_available_states(content, getattr(data, "metadata", {})) + + @Slot(dict, dict) + def update_available_states( + self, content: dict[str, Any], _metadata: dict[str, Any] | None = None + ) -> None: + """Update the displayed pills from ``AvailableBeamlineStatesMessage`` content.""" + states = content.get("states", []) + state_configs = [self._state_config_to_dict(state) for state in states] + state_configs = [state for state in state_configs if state.get("name")] + state_names = {str(state["name"]) for state in state_configs} + + for removed_name in sorted(set(self._state_pills) - state_names): + self._remove_pill(removed_name) + + for state in state_configs: + name = str(state["name"]) + title = state.get("title") or name + if name in self._state_pills: + self._state_pills[name].set_state_name(name, title=title) + continue + self._add_pill(name, title=title) + + self._empty_label.setVisible(not self._state_pills) + + def _add_pill(self, name: str, title: str) -> None: + pill = BeamlineStatePill( + parent=self._content, state_name=name, title=title, client=self.client + ) + self._state_pills[name] = pill + self._content_layout.insertWidget(max(self._content_layout.count() - 1, 0), pill) + + def _remove_pill(self, name: str) -> None: + pill = self._state_pills.pop(name) + pill.cleanup() + self._content_layout.removeWidget(pill) + pill.setParent(None) + pill.deleteLater() + + @staticmethod + def _state_config_to_dict(state: Any) -> dict[str, Any]: + if isinstance(state, dict): + return state + if hasattr(state, "model_dump"): + return state.model_dump() + return {"name": getattr(state, "name", None), "title": getattr(state, "title", None)} + + def cleanup(self) -> None: + self.bec_dispatcher.disconnect_slot( + self.update_available_states, MessageEndpoints.available_beamline_states() + ) + for name in list(self._state_pills): + self._remove_pill(name) + super().cleanup() + + +if __name__ == "__main__": # pragma: no cover + app = QApplication(sys.argv) + widget = BeamlineStateList() + widget.setWindowTitle("Beamline States") + widget.resize(360, 420) + widget.show() + sys.exit(app.exec()) diff --git a/tests/unit_tests/test_beamline_state_pill.py b/tests/unit_tests/test_beamline_state_pill.py new file mode 100644 index 00000000..16b9195a --- /dev/null +++ b/tests/unit_tests/test_beamline_state_pill.py @@ -0,0 +1,76 @@ +from bec_lib import messages + +from bec_widgets.widgets.services.beamline_states.beamline_state_pill import ( + BeamlineStateList, + BeamlineStatePill, +) + +from .client_mocks import mocked_client + + +def test_beamline_state_pill_updates_from_message(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="shutter_open", title="Shutter", client=mocked_client) + qtbot.addWidget(widget) + + widget.update_state( + {"name": "shutter_open", "status": "valid", "label": "Shutter is open."}, {} + ) + + assert widget.state_name == "shutter_open" + assert widget._name_label.text() == "Shutter" + assert widget._status_label.text() == "VALID" + assert widget.toolTip() == "Shutter is open." + + +def test_beamline_state_pill_ignores_other_states(qtbot, mocked_client): + widget = BeamlineStatePill(state_name="shutter_open", client=mocked_client) + qtbot.addWidget(widget) + + widget.update_state( + {"name": "other_state", "status": "invalid", "label": "Should be ignored."}, {} + ) + + assert widget._status_label.text() == "UNKNOWN" + assert widget.toolTip() == "No state information available." + + +def test_beamline_state_list_adds_and_removes_pills(qtbot, mocked_client): + widget = BeamlineStateList(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._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"]