1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 16:02:51 +01:00

feat(status_bar): added StatusToolBar with status actions and Broker

This commit is contained in:
2025-12-01 13:13:38 +01:00
committed by wakonig_k
parent 1534118f21
commit 48b333fd73
5 changed files with 671 additions and 2 deletions

View File

@@ -1,3 +1,5 @@
from __future__ import annotations
from qtpy.QtWidgets import QWidget
from bec_widgets.applications.views.developer_view.developer_widget import DeveloperWidget

View File

@@ -15,6 +15,7 @@ from qtpy.QtWidgets import (
)
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.toolbars.status_bar import StatusToolBar
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox
from bec_widgets.widgets.plots.waveform.waveform import Waveform
@@ -30,6 +31,7 @@ class ViewBase(QWidget):
parent (QWidget | None): Parent widget.
id (str | None): Optional view id, useful for debugging or introspection.
title (str | None): Optional human-readable title.
show_status (bool): Whether to show a status toolbar at the top of the view.
"""
def __init__(
@@ -39,6 +41,8 @@ class ViewBase(QWidget):
*,
id: str | None = None,
title: str | None = None,
show_status: bool = False,
status_names: list[str] | None = None,
):
super().__init__(parent=parent)
self.content: QWidget | None = None
@@ -49,15 +53,48 @@ class ViewBase(QWidget):
lay.setContentsMargins(0, 0, 0, 0)
lay.setSpacing(0)
self.status_bar: StatusToolBar | None = None
if show_status:
# If explicit status names are provided, default to showing only those.
show_all = status_names is None
self.setup_status_bar(show_all_status=show_all, status_names=status_names)
if content is not None:
self.set_content(content)
def set_content(self, content: QWidget) -> None:
"""Replace the current content widget with a new one."""
if self.content is not None:
self.layout().removeWidget(self.content)
self.content.setParent(None)
self.content.close()
self.content.deleteLater()
self.content = content
self.layout().addWidget(content)
if self.status_bar is not None:
insert_at = self.layout().indexOf(self.status_bar) + 1
self.layout().insertWidget(insert_at, content)
else:
self.layout().addWidget(content)
def setup_status_bar(
self, *, show_all_status: bool = True, status_names: list[str] | None = None
) -> None:
"""Create and attach a status toolbar managed by the status broker."""
if self.status_bar is not None:
return
names_arg = None if show_all_status else status_names
self.status_bar = StatusToolBar(parent=self, names=names_arg)
self.layout().addWidget(self.status_bar)
def set_status(
self, name: str = "main", *, state=None, text: str | None = None, tooltip: str | None = None
) -> None:
"""Manually set a status item on the status bar."""
if self.status_bar is None:
self.setup_status_bar(show_all_status=True)
if self.status_bar is None:
return
self.status_bar.set_status(name=name, state=state, text=text, tooltip=tooltip)
@SafeSlot()
def on_enter(self) -> None:

View File

@@ -5,7 +5,8 @@ import os
import weakref
from abc import ABC, abstractmethod
from contextlib import contextmanager
from typing import Dict, Literal
from enum import Enum
from typing import Dict, Literal, Union
from bec_lib.device import ReadoutPriority
from bec_lib.logger import bec_logger
@@ -15,6 +16,7 @@ from qtpy.QtGui import QAction, QColor, QIcon # type: ignore
from qtpy.QtWidgets import (
QApplication,
QComboBox,
QGraphicsDropShadowEffect,
QHBoxLayout,
QLabel,
QMenu,
@@ -26,6 +28,7 @@ from qtpy.QtWidgets import (
)
import bec_widgets
from bec_widgets.utils.colors import AccentColors, get_accent_colors
from bec_widgets.utils.toolbars.splitter import ResizableSpacer
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
@@ -102,6 +105,205 @@ class LongPressToolButton(QToolButton):
self.showMenu()
class StatusState(str, Enum):
DEFAULT = "default"
HIGHLIGHT = "highlight"
WARNING = "warning"
EMERGENCY = "emergency"
SUCCESS = "success"
class StatusIndicatorWidget(QWidget):
"""Pill-shaped status indicator with icon + label using accent colors."""
def __init__(
self, parent=None, text: str = "Ready", state: StatusState | str = StatusState.DEFAULT
):
super().__init__(parent)
self.setObjectName("StatusIndicatorWidget")
self._text = text
self._state = self._normalize_state(state)
self._theme_connected = False
layout = QHBoxLayout(self)
layout.setContentsMargins(6, 2, 8, 2)
layout.setSpacing(6)
self._icon_label = QLabel(self)
self._icon_label.setFixedSize(18, 18)
self._text_label = QLabel(self)
self._text_label.setText(self._text)
layout.addWidget(self._icon_label)
layout.addWidget(self._text_label)
# Give it a consistent pill height
self.setMinimumHeight(24)
self.setSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Fixed)
# Soft shadow similar to notification banners
self._shadow = QGraphicsDropShadowEffect(self)
self._shadow.setBlurRadius(18)
self._shadow.setOffset(0, 2)
self.setGraphicsEffect(self._shadow)
self._apply_state(self._state)
self._connect_theme_change()
def set_state(self, state: Union[StatusState, str]):
"""Update state and refresh visuals."""
self._state = self._normalize_state(state)
self._apply_state(self._state)
def set_text(self, text: str):
"""Update the displayed text."""
self._text = text
self._text_label.setText(text)
def _apply_state(self, state: StatusState):
palette = self._resolve_accent_colors()
color_attr = {
StatusState.DEFAULT: "default",
StatusState.HIGHLIGHT: "highlight",
StatusState.WARNING: "warning",
StatusState.EMERGENCY: "emergency",
StatusState.SUCCESS: "success",
}.get(state, "default")
base_color = getattr(palette, color_attr, None) or getattr(
palette, "default", QColor("gray")
)
# Apply style first (returns text color for label)
text_color = self._update_style(base_color, self._theme_fg_color())
theme_name = self._theme_name()
# Choose icon per state
icon_name_map = {
StatusState.DEFAULT: "check_circle",
StatusState.HIGHLIGHT: "check_circle",
StatusState.SUCCESS: "check_circle",
StatusState.WARNING: "warning",
StatusState.EMERGENCY: "dangerous",
}
icon_name = icon_name_map.get(state, "check_circle")
# Icon color:
# - Dark mode: follow text color (usually white) for high contrast.
# - Light mode: use a stronger version of the accent color for a colored glyph
# that stands out on the pastel pill background.
if theme_name == "light":
icon_q = QColor(base_color)
icon_color = icon_q.name(QColor.HexRgb)
else:
icon_color = text_color
icon = material_icon(
icon_name, size=(18, 18), convert_to_pixmap=False, filled=True, color=icon_color
)
if not icon.isNull():
self._icon_label.setPixmap(icon.pixmap(18, 18))
def _update_style(self, color: QColor, fg_color: QColor) -> str:
# Ensure the widget actually paints its own background
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
fg = QColor(fg_color)
text_color = fg.name(QColor.HexRgb)
theme_name = self._theme_name()
base = QColor(color)
start = QColor(base)
end = QColor(base)
border = QColor(base)
if theme_name == "light":
start.setAlphaF(0.20)
end.setAlphaF(0.06)
else:
start.setAlphaF(0.35)
end.setAlphaF(0.12)
border = border.darker(120)
# shadow color tuned per theme to match notification banners
if hasattr(self, "_shadow"):
if theme_name == "light":
shadow_color = QColor(15, 23, 42, 60) # softer shadow on light bg
else:
shadow_color = QColor(0, 0, 0, 160)
self._shadow.setColor(shadow_color)
# Use a fixed radius for a stable pill look inside toolbars
radius = 10
self.setStyleSheet(
f"""
#StatusIndicatorWidget {{
background-color: qlineargradient(spread:pad, x1:0, y1:0, x2:1, y2:1,
stop:0 {start.name(QColor.HexArgb)}, stop:1 {end.name(QColor.HexArgb)});
border: 1px solid {border.name(QColor.HexRgb)};
border-radius: {radius}px;
padding: 2px 8px;
}}
#StatusIndicatorWidget QLabel {{
color: {text_color};
background: transparent;
}}
"""
)
return text_color
def _theme_fg_color(self) -> QColor:
app = QApplication.instance()
theme = getattr(app, "theme", None)
if theme is not None and hasattr(theme, "color"):
try:
fg = theme.color("FG")
if isinstance(fg, QColor):
return fg
except Exception:
pass
palette = self._resolve_accent_colors()
base = getattr(palette, "default", QColor("white"))
luminance = (0.299 * base.red() + 0.587 * base.green() + 0.114 * base.blue()) / 255
return QColor("#000000") if luminance > 0.65 else QColor("#ffffff")
def _theme_name(self) -> str:
app = QApplication.instance()
theme = getattr(app, "theme", None)
name = getattr(theme, "theme", None)
if isinstance(name, str):
return name.lower()
return "dark"
def _connect_theme_change(self):
if self._theme_connected:
return
app = QApplication.instance()
theme = getattr(app, "theme", None)
if theme is not None and hasattr(theme, "theme_changed"):
try:
theme.theme_changed.connect(lambda _: self._apply_state(self._state))
self._theme_connected = True
except Exception:
pass
@staticmethod
def _normalize_state(state: Union[StatusState, str]) -> StatusState:
if isinstance(state, StatusState):
return state
try:
return StatusState(state)
except ValueError:
return StatusState.DEFAULT
@staticmethod
def _resolve_accent_colors() -> AccentColors:
return get_accent_colors()
class ToolBarAction(ABC):
"""
Abstract base class for toolbar actions.
@@ -148,6 +350,54 @@ class SeparatorAction(ToolBarAction):
toolbar.addSeparator()
class StatusIndicatorAction(ToolBarAction):
"""Toolbar action hosting a LED indicator and status text."""
def __init__(
self,
*,
text: str = "Ready",
state: Union[StatusState, str] = StatusState.DEFAULT,
tooltip: str | None = None,
):
super().__init__(icon_path=None, tooltip=tooltip or "View status", checkable=False)
self._text = text
self._state: StatusState = StatusIndicatorWidget._normalize_state(state)
self.widget: StatusIndicatorWidget | None = None
self.tooltip = tooltip or ""
def add_to_toolbar(self, toolbar: QToolBar, target: QWidget):
if (
self.widget is None
or self.widget.parent() is None
or self.widget.parent() is not toolbar
):
self.widget = StatusIndicatorWidget(parent=toolbar, text=self._text, state=self._state)
self.action = toolbar.addWidget(self.widget)
self.action.setText(self._text)
self.set_tooltip(self.tooltip)
def set_state(self, state: Union[StatusState, str]):
self._state = StatusIndicatorWidget._normalize_state(state)
if self.widget is not None:
self.widget.set_state(self._state)
def set_text(self, text: str):
self._text = text
if self.widget is not None:
self.widget.set_text(text)
if hasattr(self, "action") and self.action is not None:
self.action.setText(text)
def set_tooltip(self, tooltip: str | None):
"""Set tooltip on both the underlying widget and the QWidgetAction."""
self.tooltip = tooltip or ""
if self.widget is not None:
self.widget.setToolTip(self.tooltip)
if hasattr(self, "action") and self.action is not None:
self.action.setToolTip(self.tooltip)
class QtIconAction(IconAction):
def __init__(
self,

View File

@@ -0,0 +1,283 @@
from __future__ import annotations
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import BeamlineConditionUpdateEntry
from qtpy.QtCore import QObject, QTimer, Signal
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.toolbars.actions import StatusIndicatorAction, StatusState
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
logger = bec_logger.logger
class BECStatusBroker(BECConnector, QObject):
"""Listen to BEC beamline condition endpoints and emit structured signals."""
_instance: "BECStatusBroker | None" = None
_initialized: bool = False
available_updated = Signal(list) # list of conditions available
status_updated = Signal(str, dict) # name, status update
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, parent=None, gui_id: str | None = None, client=None, **kwargs):
if self._initialized:
return
super().__init__(parent=parent, gui_id=gui_id, client=client, **kwargs)
self._watched: set[str] = set()
self.bec_dispatcher.connect_slot(
self.on_available, MessageEndpoints.available_beamline_conditions()
)
self._initialized = True
self.refresh_available()
def refresh_available(self):
"""Fetch the current set of beamline conditions once."""
try:
msg = self.client.connector.get_last(MessageEndpoints.available_beamline_conditions())
logger.info(f"StatusBroker: fetched available conditions payload: {msg}")
if msg:
self.on_available(msg.get("data").content, None)
except Exception as exc: # pragma: no cover - runtime env
logger.debug(f"Could not fetch available conditions: {exc}")
@SafeSlot(dict, dict)
def on_available(self, data: dict, meta: dict | None = None):
condition_list = data.get("conditions") # latest one from the stream
self.available_updated.emit(condition_list)
for condition in condition_list:
name = condition.name
if name:
self.watch_condition(name)
def watch_condition(self, name: str):
"""Subscribe to updates for a single beamline condition."""
if name in self._watched:
return
self._watched.add(name)
endpoint = MessageEndpoints.beamline_condition(name)
logger.info(f"StatusBroker: watching condition '{name}' on {endpoint.endpoint}")
self.bec_dispatcher.connect_slot(self.on_condition, endpoint)
self.fetch_condition(name)
def fetch_condition(self, name: str):
"""Fetch the current value of a beamline condition once."""
endpoint = MessageEndpoints.beamline_condition(name)
try:
msg = self.client.connector.get_last(endpoint)
logger.info(f"StatusBroker: fetched condition '{name}' payload: {msg}")
if msg:
self.on_condition(msg.get("data").content, None)
except Exception as exc: # pragma: no cover - runtime env
logger.debug(f"Could not fetch condition {name}: {exc}")
@SafeSlot(dict, dict)
def on_condition(self, data: dict, meta: dict | None = None):
name = data.get("name")
if not name:
return
logger.info(f"StatusBroker: condition update for '{name}' -> {data}")
self.status_updated.emit(str(name), data)
@classmethod
def reset_singleton(cls):
"""
Reset the singleton instance of the BECStatusBroker.
"""
cls._instance = None
cls._initialized = False
class StatusToolBar(ModularToolBar):
"""Status toolbar that auto-manages beamline condition indicators."""
STATUS_MAP: dict[str, StatusState] = {
"normal": StatusState.SUCCESS,
"warning": StatusState.WARNING,
"alarm": StatusState.EMERGENCY,
}
def __init__(self, parent=None, names: list[str] | None = None, **kwargs):
super().__init__(parent=parent, orientation="horizontal", **kwargs)
self.setObjectName("StatusToolbar")
self._status_bundle = self.new_bundle("status")
self.show_bundles(["status"])
self._apply_status_toolbar_style()
self.allowed_names: set[str] | None = set(names) if names is not None else None
logger.info(f"StatusToolbar init allowed_names={self.allowed_names}")
self.broker = BECStatusBroker()
self.broker.available_updated.connect(self.on_available_updated)
self.broker.status_updated.connect(self.on_status_updated)
QTimer.singleShot(0, self.refresh_from_broker)
def refresh_from_broker(self) -> None:
if self.allowed_names is None:
self.broker.refresh_available()
else:
for name in self.allowed_names:
if not self.components.exists(name):
# Pre-create a placeholder pill so it is visible even before data arrives.
self.add_status_item(
name=name, text=name, state=StatusState.DEFAULT, tooltip=None
)
self.broker.watch_condition(name)
def _apply_status_toolbar_style(self) -> None:
self.setStyleSheet(
"QToolBar#StatusToolbar {"
f" background-color: {self.background_color};"
" border: none;"
" border-bottom: 1px solid palette(mid);"
"}"
)
# -------- Slots for updates --------
@SafeSlot(list)
def on_available_updated(self, available_conditions: list):
"""Process the available conditions stream and start watching them."""
# Keep track of current names from the broker to remove stale ones.
current_names: set[str] = set()
for condition in available_conditions:
if not isinstance(condition, BeamlineConditionUpdateEntry):
continue
name = condition.name
title = condition.title or name
if not name:
continue
current_names.add(name)
logger.info(f"StatusToolbar: discovered condition '{name}' title='{title}'")
# auto-add unless filtered out
if self.allowed_names is None or name in self.allowed_names:
self.add_status_item(name=name, text=title, state=StatusState.DEFAULT, tooltip=None)
else:
# keep hidden but present for context menu toggling
self.add_status_item(name=name, text=title, state=StatusState.DEFAULT, tooltip=None)
act = self.components.get_action(name)
if act and act.action:
act.action.setVisible(False)
# Remove actions that are no longer present in available_conditions.
known_actions = [
n for n in self.components._components.keys() if n not in ("separator",)
] # direct access used for clean-up
for name in known_actions:
if name not in current_names:
logger.info(f"StatusToolbar: removing stale condition '{name}'")
try:
self.components.remove_action(name)
except Exception as exc:
logger.warning(f"Failed to remove stale condition '{name}': {exc}")
self.refresh()
@SafeSlot(str, dict)
def on_status_updated(self, name: str, payload: dict): # TODO finish update logic
"""Update a status pill when a condition update arrives."""
state = self.STATUS_MAP.get(str(payload.get("status", "")).lower(), StatusState.DEFAULT)
action = self.components.get_action(name) if self.components.exists(name) else None
# Only update the label when a title is explicitly provided; otherwise keep current text.
title = payload.get("title") or None
text = title
if text is None and action is None:
text = payload.get("name") or name
if "message" in payload:
tooltip = payload.get("message") or ""
else:
tooltip = None
logger.info(
f"StatusToolbar: update condition '{name}' -> state={state} text='{text}' tooltip='{tooltip}'"
)
self.set_status(name=name, text=text, state=state, tooltip=tooltip)
# -------- Items Management --------
def add_status_item(
self,
name: str,
*,
text: str = "Ready",
state: StatusState | str = StatusState.DEFAULT,
tooltip: str | None = None,
) -> StatusIndicatorAction | None:
"""
Add or update a named status item in the toolbar.
After you added all actions, call `toolbar.refresh()` to update the display.
Args:
name(str): Unique name for the status item.
text(str): Text to display in the status item.
state(StatusState | str): State of the status item.
tooltip(str | None): Optional tooltip for the status item.
Returns:
StatusIndicatorAction | None: The created or updated status action, or None if toolbar is not initialized.
"""
if self._status_bundle is None:
return
if self.components.exists(name):
return
action = StatusIndicatorAction(text=text, state=state, tooltip=tooltip)
return self.add_status_action(name, action)
def add_status_action(
self, name: str, action: StatusIndicatorAction
) -> StatusIndicatorAction | None:
"""
Attach an existing StatusIndicatorAction to the status toolbar.
After you added all actions, call `toolbar.refresh()` to update the display.
Args:
name(str): Unique name for the status item.
action(StatusIndicatorAction): The status action to add.
Returns:
StatusIndicatorAction | None: The added status action, or None if toolbar is not initialized.
"""
self.components.add_safe(name, action)
self.get_bundle("status").add_action(name)
self.refresh()
self.broker.fetch_condition(name)
return action
def set_status(
self,
name: str = "main",
*,
state: StatusState | str | None = None,
text: str | None = None,
tooltip: str | None = None,
) -> None:
"""
Update the status item with the given name, creating it if necessary.
Args:
name(str): Unique name for the status item.
state(StatusState | str | None): New state for the status item.
text(str | None): New text for the status item.
"""
action = self.components.get_action(name) if self.components.exists(name) else None
if action is None:
action = self.add_status_item(
name, text=text or "Ready", state=state or "default", tooltip=tooltip
)
if action is None:
return
if state is not None:
action.set_state(state)
if text is not None:
action.set_text(text)
if tooltip is not None and hasattr(action, "set_tooltip"):
action.set_tooltip(tooltip)

View File

@@ -0,0 +1,97 @@
from __future__ import annotations
import pytest
from bec_lib.messages import BeamlineConditionUpdateEntry
from qtpy.QtWidgets import QToolBar
from bec_widgets.utils.toolbars.actions import StatusIndicatorAction, StatusIndicatorWidget
from bec_widgets.utils.toolbars.status_bar import BECStatusBroker, StatusToolBar
from .client_mocks import mocked_client
from .conftest import create_widget
class TestStatusIndicators:
"""Widget/action level tests independent of broker wiring."""
def test_indicator_widget_state_and_text(self, qtbot):
widget = StatusIndicatorWidget(text="Ready", state="success")
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
widget.set_state("warning")
widget.set_text("Alert")
assert widget._state.value == "warning"
assert widget._text_label.text() == "Alert"
def test_indicator_action_updates_widget_and_action(self, qtbot):
qt_toolbar = QToolBar()
qtbot.addWidget(qt_toolbar)
action = StatusIndicatorAction(text="Ready", tooltip="Initial")
action.add_to_toolbar(qt_toolbar, qt_toolbar)
action.set_tooltip("Updated tooltip")
action.set_text("Running")
assert action.action.toolTip() == "Updated tooltip"
assert action.widget.toolTip() == "Updated tooltip" # type: ignore[union-attr]
assert action.widget._text_label.text() == "Running" # type: ignore[union-attr]
class TestStatusBar:
"""Status bar + broker integration using fake redis client (mocked_client)."""
@pytest.fixture(params=[{}, {"names": ["alpha"]}])
def status_toolbar(self, qtbot, mocked_client, request):
broker = BECStatusBroker(client=mocked_client)
toolbar = create_widget(qtbot, StatusToolBar, **request.param)
yield toolbar
broker.reset_singleton()
def test_allowed_names_precreates_placeholder(self, status_toolbar):
status_toolbar.broker.refresh_available = lambda: None
status_toolbar.refresh_from_broker()
# We parametrize the fixture so one invocation has allowed_names set.
if status_toolbar.allowed_names:
name = next(iter(status_toolbar.allowed_names))
assert status_toolbar.components.exists(name)
act = status_toolbar.components.get_action(name)
assert isinstance(act, StatusIndicatorAction)
assert act.widget._text_label.text() == name # type: ignore[union-attr]
def test_on_available_adds_and_removes(self, status_toolbar):
conditions = [
BeamlineConditionUpdateEntry(name="c1", title="Cond 1", condition_type="test"),
BeamlineConditionUpdateEntry(name="c2", title="Cond 2", condition_type="test"),
]
status_toolbar.on_available_updated(conditions)
assert status_toolbar.components.exists("c1")
assert status_toolbar.components.exists("c2")
conditions2 = [
BeamlineConditionUpdateEntry(name="c1", title="Cond 1", condition_type="test")
]
status_toolbar.on_available_updated(conditions2)
assert status_toolbar.components.exists("c1")
assert not status_toolbar.components.exists("c2")
def test_on_status_updated_sets_title_and_message(self, status_toolbar):
status_toolbar.add_status_item("beam", text="Initial", state="default", tooltip=None)
payload = {"name": "beam", "status": "warning", "title": "New Title", "message": "Detail"}
status_toolbar.on_status_updated("beam", payload)
action = status_toolbar.components.get_action("beam")
assert isinstance(action, StatusIndicatorAction)
assert action.widget._text_label.text() == "New Title" # type: ignore[union-attr]
assert action.action.toolTip() == "Detail"
def test_on_status_updated_keeps_existing_text_when_no_title(self, status_toolbar):
status_toolbar.add_status_item("beam", text="Keep Me", state="default", tooltip=None)
payload = {"name": "beam", "status": "normal", "message": "Note"}
status_toolbar.on_status_updated("beam", payload)
action = status_toolbar.components.get_action("beam")
assert isinstance(action, StatusIndicatorAction)
assert action.widget._text_label.text() == "Keep Me" # type: ignore[union-attr]
assert action.action.toolTip() == "Note"