diff --git a/bec_widgets/applications/views/developer_view/developer_view.py b/bec_widgets/applications/views/developer_view/developer_view.py index 6f177c752..eb3350e34 100644 --- a/bec_widgets/applications/views/developer_view/developer_view.py +++ b/bec_widgets/applications/views/developer_view/developer_view.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from qtpy.QtWidgets import QWidget from bec_widgets.applications.views.developer_view.developer_widget import DeveloperWidget diff --git a/bec_widgets/applications/views/view.py b/bec_widgets/applications/views/view.py index 3b98f7568..5a51411ee 100644 --- a/bec_widgets/applications/views/view.py +++ b/bec_widgets/applications/views/view.py @@ -15,6 +15,7 @@ ) 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 @@ def __init__( *, 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 @@ def __init__( 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: diff --git a/bec_widgets/utils/toolbars/actions.py b/bec_widgets/utils/toolbars/actions.py index f82753ffe..982a77e6f 100644 --- a/bec_widgets/utils/toolbars/actions.py +++ b/bec_widgets/utils/toolbars/actions.py @@ -5,7 +5,8 @@ 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.QtWidgets import ( QApplication, QComboBox, + QGraphicsDropShadowEffect, QHBoxLayout, QLabel, QMenu, @@ -26,6 +28,7 @@ ) 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 @@ def handleLongPress(self): 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 @@ def add_to_toolbar(self, toolbar: QToolBar, target: QWidget): 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, diff --git a/bec_widgets/utils/toolbars/status_bar.py b/bec_widgets/utils/toolbars/status_bar.py new file mode 100644 index 000000000..5de03510d --- /dev/null +++ b/bec_widgets/utils/toolbars/status_bar.py @@ -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 BeamlineStateConfig +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 state endpoints and emit structured signals.""" + + _instance: "BECStatusBroker | None" = None + _initialized: bool = False + + available_updated = Signal(list) # list of states 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_states() + ) + + 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_states()) + 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): + state_list = data.get("states") # latest one from the stream + self.available_updated.emit(state_list) + for state in state_list: + name = state.name + if name: + self.watch_state(name) + + def watch_state(self, name: str): + """Subscribe to updates for a single beamline state.""" + if name in self._watched: + return + self._watched.add(name) + endpoint = MessageEndpoints.beamline_state(name) + logger.info(f"StatusBroker: watching state '{name}' on {endpoint.endpoint}") + self.bec_dispatcher.connect_slot(self.on_state, endpoint) + self.fetch_state(name) + + def fetch_state(self, name: str): + """Fetch the current value of a beamline state once.""" + endpoint = MessageEndpoints.beamline_state(name) + try: + msg = self.client.connector.get_last(endpoint) + logger.info(f"StatusBroker: fetched state '{name}' payload: {msg}") + if msg: + self.on_state(msg.get("data").content, None) + except Exception as exc: # pragma: no cover - runtime env + logger.debug(f"Could not fetch state {name}: {exc}") + + @SafeSlot(dict, dict) + def on_state(self, data: dict, meta: dict | None = None): + name = data.get("name") + if not name: + return + logger.info(f"StatusBroker: state 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 state indicators.""" + + STATUS_MAP: dict[str, StatusState] = { + "valid": StatusState.SUCCESS, + "warning": StatusState.WARNING, + "invalid": 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_state(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_states: list): + """Process the available states stream and start watching them.""" + # Keep track of current names from the broker to remove stale ones. + current_names: set[str] = set() + for state in available_states: + if not isinstance(state, BeamlineStateConfig): + continue + name = state.name + title = state.title or name + if not name: + continue + current_names.add(name) + logger.info(f"StatusToolbar: discovered state '{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_states. + 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 state '{name}'") + try: + self.components.remove_action(name) + except Exception as exc: + logger.warning(f"Failed to remove stale state '{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 state 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 "label" in payload: + tooltip = payload.get("label") or "" + else: + tooltip = None + logger.info( + f"StatusToolbar: update state '{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_state(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) diff --git a/bec_widgets/widgets/containers/main_window/main_window.py b/bec_widgets/widgets/containers/main_window/main_window.py index 55fdf1f1f..28740e9fb 100644 --- a/bec_widgets/widgets/containers/main_window/main_window.py +++ b/bec_widgets/widgets/containers/main_window/main_window.py @@ -1,7 +1,6 @@ from __future__ import annotations import os -from typing import TYPE_CHECKING from bec_lib.endpoints import MessageEndpoints from qtpy.QtCore import QEvent, QSize, Qt, QTimer @@ -22,6 +21,7 @@ from bec_widgets.utils.bec_widget import BECWidget from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.utils.toolbars.status_bar import StatusToolBar from bec_widgets.widgets.containers.main_window.addons.hover_widget import HoverWidget from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import ( BECNotificationBroker, @@ -115,14 +115,11 @@ def _init_status_bar_widgets(self): Prepare the BEC specific widgets in the status bar. """ - # Left: App‑ID label - self._app_id_label = QLabel() - self._app_id_label.setAlignment( - Qt.AlignmentFlag.AlignHCenter | Qt.AlignmentFlag.AlignVCenter - ) - self.status_bar.addWidget(self._app_id_label) + # Left: Beamline condition status toolbar (auto-fetches all conditions) + self._status_toolbar = StatusToolBar(parent=self, names=None) + self.status_bar.addWidget(self._status_toolbar) - # Add a separator after the app ID label + # Add a separator after the status toolbar self._add_separator() # Centre: Client‑info label (stretch=1 so it expands) @@ -341,13 +338,27 @@ def _setup_menu_bar(self): help_menu.addAction(bec_docs) help_menu.addAction(widgets_docs) help_menu.addAction(bug_report) + help_menu.addSeparator() + self._app_id_action = QAction(self) + self._app_id_action.triggered.connect(self._copy_app_id_to_clipboard) + + help_menu.addAction(self._app_id_action) + + def _copy_app_id_to_clipboard(self): + """ + Copy the app ID to the clipboard. + """ + if self.bec_dispatcher.cli_server is not None: + server_id = self.bec_dispatcher.cli_server.gui_id + clipboard = QApplication.clipboard() + clipboard.setText(server_id) ################################################################################ # Status Bar Addons ################################################################################ def display_app_id(self): """ - Display the app ID in the status bar. + Display the app ID in the Help menu. """ if self.bec_dispatcher.cli_server is None: status_message = "Not connected" @@ -355,7 +366,8 @@ def display_app_id(self): # Get the server ID from the dispatcher server_id = self.bec_dispatcher.cli_server.gui_id status_message = f"App ID: {server_id}" - self._app_id_label.setText(status_message) + if hasattr(self, "_app_id_action"): + self._app_id_action.setText(status_message) @SafeSlot(dict, dict) def display_client_message(self, msg: dict, meta: dict): diff --git a/tests/end-2-end/test_rpc_widgets_e2e.py b/tests/end-2-end/test_rpc_widgets_e2e.py index 2a81168f9..f64f33d35 100644 --- a/tests/end-2-end/test_rpc_widgets_e2e.py +++ b/tests/end-2-end/test_rpc_widgets_e2e.py @@ -74,15 +74,15 @@ def test_available_widgets(qtbot, connected_client_gui_obj): """This test checks that all widgets that are available via gui.available_widgets can be created and removed.""" gui = connected_client_gui_obj dock_area = gui.bec - # Number of top level widgets, should be 4 - top_level_widgets_count = 12 + # Number of top level widgets, should be 5 + top_level_widgets_count = 13 assert len(gui._server_registry) == top_level_widgets_count names = set(list(gui._server_registry.keys())) - # Number of widgets with parent_id == None, should be 2 + # Number of widgets with parent_id == None, should be 3 widgets = [ widget for widget in gui._server_registry.values() if widget["config"]["parent_id"] is None ] - assert len(widgets) == 2 + assert len(widgets) == 3 # Test all relevant widgets for object_name in gui.available_widgets.__dict__: @@ -115,7 +115,7 @@ def test_available_widgets(qtbot, connected_client_gui_obj): for widget in gui._server_registry.values() if widget["config"]["parent_id"] is None ] - assert len(widgets) == 2 + assert len(widgets) == 3 ############################# ####### Remove widget ####### diff --git a/tests/unit_tests/test_status_bar.py b/tests/unit_tests/test_status_bar.py new file mode 100644 index 000000000..72e2584f0 --- /dev/null +++ b/tests/unit_tests/test_status_bar.py @@ -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"