mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-06-05 04:48:40 +02:00
wip basic pill implementation
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import (
|
||||
BeamlineStateList,
|
||||
BeamlineStatePill,
|
||||
)
|
||||
|
||||
__all__ = ["BeamlineStateList", "BeamlineStatePill"]
|
||||
@@ -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())
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user