mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-04-13 20:20:55 +02:00
Compare commits
11 Commits
fix/ide_vi
...
feat/statu
| Author | SHA1 | Date | |
|---|---|---|---|
| d6f5c0e4f9 | |||
| 55146665d2 | |||
| 6836036507 | |||
| c60cb31eaa | |||
| 12103724a1 | |||
| 1fbaa517a2 | |||
| be132ad823 | |||
| 5763830ef1 | |||
| 4c9fa27450 | |||
| c991faeb35 | |||
| ac9edd10d9 |
@@ -549,7 +549,7 @@ class LaunchWindow(BECMainWindow):
|
||||
remaining_connections = [
|
||||
connection for connection in connections.values() if connection.parent_id != self.gui_id
|
||||
]
|
||||
return len(remaining_connections) <= 4
|
||||
return len(remaining_connections) <= 5
|
||||
|
||||
def _turn_off_the_lights(self, connections: dict):
|
||||
"""
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.applications.views.developer_view.developer_widget import DeveloperWidget
|
||||
|
||||
@@ -18,6 +18,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
|
||||
@@ -69,6 +70,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__(
|
||||
@@ -78,6 +80,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
|
||||
@@ -88,15 +92,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:
|
||||
|
||||
@@ -3480,6 +3480,34 @@ class MotorMap(RPCBase):
|
||||
Set Y-axis to log scale if True, linear if False.
|
||||
"""
|
||||
|
||||
@property
|
||||
@rpc_call
|
||||
def x_motor(self) -> "str":
|
||||
"""
|
||||
Name of the motor shown on the X axis.
|
||||
"""
|
||||
|
||||
@x_motor.setter
|
||||
@rpc_call
|
||||
def x_motor(self) -> "str":
|
||||
"""
|
||||
Name of the motor shown on the X axis.
|
||||
"""
|
||||
|
||||
@property
|
||||
@rpc_call
|
||||
def y_motor(self) -> "str":
|
||||
"""
|
||||
Name of the motor shown on the Y axis.
|
||||
"""
|
||||
|
||||
@y_motor.setter
|
||||
@rpc_call
|
||||
def y_motor(self) -> "str":
|
||||
"""
|
||||
Name of the motor shown on the Y axis.
|
||||
"""
|
||||
|
||||
@property
|
||||
@rpc_call
|
||||
def legend_label_size(self) -> "int":
|
||||
@@ -3604,7 +3632,9 @@ class MotorMap(RPCBase):
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def map(self, x_name: "str", y_name: "str", validate_bec: "bool" = True) -> "None":
|
||||
def map(
|
||||
self, x_name: "str", y_name: "str", validate_bec: "bool" = True, suppress_errors=False
|
||||
) -> "None":
|
||||
"""
|
||||
Set the x and y motor names.
|
||||
|
||||
@@ -3612,6 +3642,7 @@ class MotorMap(RPCBase):
|
||||
x_name(str): The name of the x motor.
|
||||
y_name(str): The name of the y motor.
|
||||
validate_bec(bool, optional): If True, validate the signal with BEC. Defaults to True.
|
||||
suppress_errors(bool, optional): If True, suppress errors during validation. Defaults to False. Used for properties setting. If the validation fails, the changes are not applied.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
|
||||
@@ -392,7 +392,8 @@ class BECGuiClient(RPCBase):
|
||||
timeout = 60
|
||||
# Wait for 'bec' gui to be registered, this may take some time
|
||||
# After 60s timeout. Should this raise an exception on timeout?
|
||||
while time.time() < time.time() + timeout:
|
||||
start = time.monotonic()
|
||||
while time.monotonic() < start + timeout:
|
||||
if len(list(self._server_registry.keys())) < 2 or not hasattr(
|
||||
self, self._anchor_widget
|
||||
):
|
||||
|
||||
@@ -25,7 +25,6 @@ from qtpy.QtWidgets import (
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets import BECWidget
|
||||
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
|
||||
from bec_widgets.utils.colors import apply_theme
|
||||
from bec_widgets.utils.widget_io import WidgetHierarchy as wh
|
||||
@@ -48,6 +47,7 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
|
||||
|
||||
self._widgets_by_name: Dict[str, QWidget] = {}
|
||||
self._init_ui()
|
||||
self.app = QApplication.instance()
|
||||
|
||||
# expose helper API and basics in the inprocess console
|
||||
if self.console.inprocess is True:
|
||||
@@ -94,7 +94,7 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
|
||||
return list(widget_handler.widget_classes.keys())
|
||||
|
||||
self.jc = _ConsoleAPI(self)
|
||||
self._push_to_console({"jc": self.jc, "np": np, "pg": pg, "wh": wh})
|
||||
self._push_to_console({"jc": self.jc, "np": np, "pg": pg, "wh": wh, "app": self.app})
|
||||
|
||||
def _init_ui(self):
|
||||
self.layout = QHBoxLayout(self)
|
||||
|
||||
@@ -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.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
|
||||
|
||||
@@ -101,6 +104,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.
|
||||
@@ -147,6 +349,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,
|
||||
|
||||
283
bec_widgets/utils/toolbars/status_bar.py
Normal file
283
bec_widgets/utils/toolbars/status_bar.py
Normal 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)
|
||||
@@ -21,7 +21,17 @@ from bec_widgets.utils.widget_io import WidgetHierarchy
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
PROPERTY_TO_SKIP = ["palette", "font", "windowIcon", "windowIconText", "locale", "styleSheet"]
|
||||
PROPERTY_TO_SKIP = [
|
||||
"palette",
|
||||
"font",
|
||||
"windowIcon",
|
||||
"windowIconText",
|
||||
"locale",
|
||||
"styleSheet",
|
||||
"updatesEnabled",
|
||||
"objectName",
|
||||
"visible",
|
||||
]
|
||||
|
||||
|
||||
class WidgetStateManager:
|
||||
@@ -110,16 +120,8 @@ class WidgetStateManager:
|
||||
prop = meta.property(i)
|
||||
name = prop.name()
|
||||
|
||||
# Skip persisting QWidget visibility because container widgets (e.g. tab
|
||||
# stacks, dock managers) manage that state themselves. Restoring a saved
|
||||
# False can permanently hide a widget, while forcing True makes hidden
|
||||
# tabs show on top. Leave the property to the parent widget instead.
|
||||
if name == "visible":
|
||||
continue
|
||||
|
||||
if (
|
||||
name == "objectName"
|
||||
or name in PROPERTY_TO_SKIP
|
||||
name in PROPERTY_TO_SKIP
|
||||
or not prop.isReadable()
|
||||
or not prop.isWritable()
|
||||
or not prop.isStored() # can be extended to fine filter
|
||||
@@ -176,7 +178,7 @@ class WidgetStateManager:
|
||||
for i in range(meta.propertyCount()):
|
||||
prop = meta.property(i)
|
||||
name = prop.name()
|
||||
if name == "visible":
|
||||
if name in PROPERTY_TO_SKIP:
|
||||
continue
|
||||
if settings.contains(name):
|
||||
value = settings.value(name)
|
||||
|
||||
@@ -941,30 +941,14 @@ class AdvancedDockArea(DockAreaWidget):
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import sys
|
||||
|
||||
from qtpy.QtWidgets import QTabWidget
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
apply_theme("dark")
|
||||
dispatcher = BECDispatcher(gui_id="ads")
|
||||
window = BECMainWindowNoRPC()
|
||||
central = QWidget()
|
||||
layout = QVBoxLayout(central)
|
||||
window.setCentralWidget(central)
|
||||
|
||||
# two dock areas stacked vertically no instance ids
|
||||
ads = AdvancedDockArea(mode="creator", enable_profile_management=True)
|
||||
ads2 = AdvancedDockArea(mode="creator", enable_profile_management=True)
|
||||
layout.addWidget(ads, 1)
|
||||
layout.addWidget(ads2, 1)
|
||||
|
||||
# two dock areas inside a tab widget
|
||||
tabs = QTabWidget(parent=central)
|
||||
ads3 = AdvancedDockArea(mode="creator", enable_profile_management=True, instance_id="AdsTab3")
|
||||
ads4 = AdvancedDockArea(mode="creator", enable_profile_management=True, instance_id="AdsTab4")
|
||||
tabs.addTab(ads3, "Workspace 3")
|
||||
tabs.addTab(ads4, "Workspace 4")
|
||||
layout.addWidget(tabs, 1)
|
||||
ads = AdvancedDockArea(mode="creator", enable_profile_management=True, root_widget=True)
|
||||
|
||||
window.setCentralWidget(ads)
|
||||
window.show()
|
||||
window.resize(800, 1000)
|
||||
|
||||
|
||||
@@ -5,7 +5,6 @@ from qtpy.QtCore import QMimeData, Qt, Signal
|
||||
from qtpy.QtGui import QDrag
|
||||
from qtpy.QtWidgets import QHBoxLayout, QPushButton, QSizePolicy, QToolButton, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.colors import get_theme_palette
|
||||
from bec_widgets.utils.error_popups import SafeProperty
|
||||
|
||||
|
||||
@@ -49,6 +48,8 @@ class CollapsibleSection(QWidget):
|
||||
|
||||
# Create header button
|
||||
self.header_button = QPushButton()
|
||||
# Apply theme variant for title styling
|
||||
self.header_button.setProperty("variant", "title")
|
||||
self.header_button.clicked.connect(self.toggle_expanded)
|
||||
|
||||
# Enable drag and drop for reordering
|
||||
@@ -105,23 +106,6 @@ class CollapsibleSection(QWidget):
|
||||
self.header_button.setIcon(icon)
|
||||
self.header_button.setText(self.title)
|
||||
|
||||
# Get theme colors
|
||||
palette = get_theme_palette()
|
||||
|
||||
self.header_button.setStyleSheet(
|
||||
"""
|
||||
QPushButton {
|
||||
font-weight: bold;
|
||||
text-align: left;
|
||||
margin: 0;
|
||||
padding: 0px;
|
||||
border: none;
|
||||
background: transparent;
|
||||
icon-size: 20px 20px;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
def toggle_expanded(self):
|
||||
"""Toggle the expanded state and update size policy"""
|
||||
self.expanded = not self.expanded
|
||||
|
||||
@@ -22,6 +22,7 @@ from bec_widgets.utils import UILoader
|
||||
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,
|
||||
@@ -114,14 +115,11 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
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)
|
||||
@@ -394,13 +392,17 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
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.setEnabled(False)
|
||||
help_menu.addAction(self._app_id_action)
|
||||
|
||||
################################################################################
|
||||
# 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"
|
||||
@@ -408,7 +410,8 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
# 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):
|
||||
|
||||
@@ -17,7 +17,9 @@ from bec_widgets.utils.settings_dialog import SettingsDialog
|
||||
from bec_widgets.utils.toolbars.toolbar import MaterialIconAction
|
||||
from bec_widgets.widgets.plots.motor_map.settings.motor_map_settings import MotorMapSettings
|
||||
from bec_widgets.widgets.plots.motor_map.toolbar_components.motor_selection import (
|
||||
MotorSelectionAction,
|
||||
MotorSelection,
|
||||
MotorSelectionConnection,
|
||||
motor_selection_bundle,
|
||||
)
|
||||
from bec_widgets.widgets.plots.plot_base import PlotBase, UIMode
|
||||
|
||||
@@ -126,6 +128,10 @@ class MotorMap(PlotBase):
|
||||
"x_log.setter",
|
||||
"y_log",
|
||||
"y_log.setter",
|
||||
"x_motor",
|
||||
"x_motor.setter",
|
||||
"y_motor",
|
||||
"y_motor.setter",
|
||||
"legend_label_size",
|
||||
"legend_label_size.setter",
|
||||
"attach",
|
||||
@@ -195,11 +201,10 @@ class MotorMap(PlotBase):
|
||||
"""
|
||||
Initialize the toolbar for the motor map widget.
|
||||
"""
|
||||
motor_selection = MotorSelectionAction(parent=self)
|
||||
self.toolbar.add_action("motor_selection", motor_selection)
|
||||
|
||||
motor_selection.motor_x.currentTextChanged.connect(self.on_motor_selection_changed)
|
||||
motor_selection.motor_y.currentTextChanged.connect(self.on_motor_selection_changed)
|
||||
self.toolbar.add_bundle(motor_selection_bundle(self.toolbar.components))
|
||||
self.toolbar.connect_bundle(
|
||||
"motor_selection", MotorSelectionConnection(self.toolbar.components, target_widget=self)
|
||||
)
|
||||
|
||||
self.toolbar.components.get_action("reset_legend").action.setVisible(False)
|
||||
|
||||
@@ -228,12 +233,19 @@ class MotorMap(PlotBase):
|
||||
if self.ui_mode == UIMode.POPUP:
|
||||
bundles.append("axis_popup")
|
||||
self.toolbar.show_bundles(bundles)
|
||||
self._sync_motor_map_selection_toolbar()
|
||||
|
||||
@SafeSlot()
|
||||
def on_motor_selection_changed(self, _):
|
||||
action: MotorSelectionAction = self.toolbar.components.get_action("motor_selection")
|
||||
motor_x = action.motor_x.currentText()
|
||||
motor_y = action.motor_y.currentText()
|
||||
action = self.toolbar.components.get_action("motor_selection")
|
||||
motor_selection: MotorSelection = action.widget
|
||||
motor_x = motor_selection.motor_x.currentText()
|
||||
motor_y = motor_selection.motor_y.currentText()
|
||||
|
||||
if motor_x and not self._validate_motor_name(motor_x):
|
||||
return
|
||||
if motor_y and not self._validate_motor_name(motor_y):
|
||||
return
|
||||
|
||||
if motor_x != "" and motor_y != "":
|
||||
if motor_x != self.config.x_motor.name or motor_y != self.config.y_motor.name:
|
||||
@@ -286,6 +298,36 @@ class MotorMap(PlotBase):
|
||||
# Widget Specific Properties
|
||||
################################################################################
|
||||
|
||||
@SafeProperty(str)
|
||||
def x_motor(self) -> str:
|
||||
"""Name of the motor shown on the X axis."""
|
||||
return self.config.x_motor.name or ""
|
||||
|
||||
@x_motor.setter
|
||||
def x_motor(self, motor_name: str) -> None:
|
||||
motor_name = motor_name or ""
|
||||
if motor_name == (self.config.x_motor.name or ""):
|
||||
return
|
||||
if motor_name and self.y_motor:
|
||||
self.map(motor_name, self.y_motor, suppress_errors=True)
|
||||
return
|
||||
self._set_motor_name(axis="x", motor_name=motor_name)
|
||||
|
||||
@SafeProperty(str)
|
||||
def y_motor(self) -> str:
|
||||
"""Name of the motor shown on the Y axis."""
|
||||
return self.config.y_motor.name or ""
|
||||
|
||||
@y_motor.setter
|
||||
def y_motor(self, motor_name: str) -> None:
|
||||
motor_name = motor_name or ""
|
||||
if motor_name == (self.config.y_motor.name or ""):
|
||||
return
|
||||
if motor_name and self.x_motor:
|
||||
self.map(self.x_motor, motor_name, suppress_errors=True)
|
||||
return
|
||||
self._set_motor_name(axis="y", motor_name=motor_name)
|
||||
|
||||
# color_scatter for designer, color for CLI to not bother users with QColor
|
||||
@SafeProperty("QColor")
|
||||
def color_scatter(self) -> QtGui.QColor:
|
||||
@@ -427,11 +469,47 @@ class MotorMap(PlotBase):
|
||||
self.update_signal.emit()
|
||||
self.property_changed.emit("scatter_size", scatter_size)
|
||||
|
||||
def _validate_motor_name(self, motor_name: str) -> bool:
|
||||
"""
|
||||
Check motor validity against BEC without raising.
|
||||
|
||||
Args:
|
||||
motor_name(str): Name of the motor to validate.
|
||||
|
||||
Returns:
|
||||
bool: True if motor is valid, False otherwise.
|
||||
"""
|
||||
if not motor_name:
|
||||
return False
|
||||
try:
|
||||
self.entry_validator.validate_signal(motor_name, None)
|
||||
return True
|
||||
except Exception: # noqa: BLE001 - validator can raise multiple error types
|
||||
return False
|
||||
|
||||
def _set_motor_name(self, axis: str, motor_name: str, *, sync_toolbar: bool = True) -> None:
|
||||
"""
|
||||
Update stored motor name for given axis and optionally refresh the toolbar selection.
|
||||
"""
|
||||
motor_name = motor_name or ""
|
||||
motor_config = self.config.x_motor if axis == "x" else self.config.y_motor
|
||||
|
||||
if motor_config.name == motor_name:
|
||||
return
|
||||
|
||||
motor_config.name = motor_name
|
||||
self.property_changed.emit(f"{axis}_motor", motor_name)
|
||||
|
||||
if sync_toolbar:
|
||||
self._sync_motor_map_selection_toolbar()
|
||||
|
||||
################################################################################
|
||||
# High Level methods for API
|
||||
################################################################################
|
||||
@SafeSlot()
|
||||
def map(self, x_name: str, y_name: str, validate_bec: bool = True) -> None:
|
||||
def map(
|
||||
self, x_name: str, y_name: str, validate_bec: bool = True, suppress_errors=False
|
||||
) -> None:
|
||||
"""
|
||||
Set the x and y motor names.
|
||||
|
||||
@@ -439,15 +517,23 @@ class MotorMap(PlotBase):
|
||||
x_name(str): The name of the x motor.
|
||||
y_name(str): The name of the y motor.
|
||||
validate_bec(bool, optional): If True, validate the signal with BEC. Defaults to True.
|
||||
suppress_errors(bool, optional): If True, suppress errors during validation. Defaults to False. Used for properties setting. If the validation fails, the changes are not applied.
|
||||
"""
|
||||
self.plot_item.clear()
|
||||
|
||||
if validate_bec:
|
||||
self.entry_validator.validate_signal(x_name, None)
|
||||
self.entry_validator.validate_signal(y_name, None)
|
||||
if suppress_errors:
|
||||
try:
|
||||
self.entry_validator.validate_signal(x_name, None)
|
||||
self.entry_validator.validate_signal(y_name, None)
|
||||
except Exception:
|
||||
return
|
||||
else:
|
||||
self.entry_validator.validate_signal(x_name, None)
|
||||
self.entry_validator.validate_signal(y_name, None)
|
||||
|
||||
self.config.x_motor.name = x_name
|
||||
self.config.y_motor.name = y_name
|
||||
self._set_motor_name(axis="x", motor_name=x_name, sync_toolbar=False)
|
||||
self._set_motor_name(axis="y", motor_name=y_name, sync_toolbar=False)
|
||||
|
||||
motor_x_limit = self._get_motor_limit(self.config.x_motor.name)
|
||||
motor_y_limit = self._get_motor_limit(self.config.y_motor.name)
|
||||
@@ -774,21 +860,24 @@ class MotorMap(PlotBase):
|
||||
"""
|
||||
Sync the motor map selection toolbar with the current motor map.
|
||||
"""
|
||||
motor_selection = self.toolbar.components.get_action("motor_selection")
|
||||
try:
|
||||
motor_selection_action = self.toolbar.components.get_action("motor_selection")
|
||||
except Exception: # noqa: BLE001 - toolbar might not be ready during early init
|
||||
logger.warning(f"MotorMap ({self.object_name}) toolbar was not ready during init.")
|
||||
return
|
||||
if motor_selection_action is None:
|
||||
return
|
||||
motor_selection: MotorSelection = motor_selection_action.widget
|
||||
target_x = self.config.x_motor.name or ""
|
||||
target_y = self.config.y_motor.name or ""
|
||||
|
||||
motor_x = motor_selection.motor_x.currentText()
|
||||
motor_y = motor_selection.motor_y.currentText()
|
||||
if (
|
||||
motor_selection.motor_x.currentText() == target_x
|
||||
and motor_selection.motor_y.currentText() == target_y
|
||||
):
|
||||
return
|
||||
|
||||
if motor_x != self.config.x_motor.name:
|
||||
motor_selection.motor_x.blockSignals(True)
|
||||
motor_selection.motor_x.set_device(self.config.x_motor.name)
|
||||
motor_selection.motor_x.check_validity(self.config.x_motor.name)
|
||||
motor_selection.motor_x.blockSignals(False)
|
||||
if motor_y != self.config.y_motor.name:
|
||||
motor_selection.motor_y.blockSignals(True)
|
||||
motor_selection.motor_y.set_device(self.config.y_motor.name)
|
||||
motor_selection.motor_y.check_validity(self.config.y_motor.name)
|
||||
motor_selection.motor_y.blockSignals(False)
|
||||
motor_selection.set_motors(target_x, target_y)
|
||||
|
||||
################################################################################
|
||||
# Export Methods
|
||||
|
||||
@@ -1,43 +1,55 @@
|
||||
from qtpy.QtWidgets import QHBoxLayout, QToolBar, QWidget
|
||||
from qtpy.QtWidgets import QHBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.toolbars.actions import NoCheckDelegate, ToolBarAction
|
||||
from bec_widgets.utils.toolbars.actions import NoCheckDelegate, WidgetAction
|
||||
from bec_widgets.utils.toolbars.bundles import ToolbarBundle, ToolbarComponents
|
||||
from bec_widgets.utils.toolbars.connections import BundleConnection
|
||||
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
|
||||
|
||||
|
||||
class MotorSelectionAction(ToolBarAction):
|
||||
class MotorSelection(QWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(icon_path=None, tooltip=None, checkable=False)
|
||||
self.motor_x = DeviceComboBox(parent=parent, device_filter=[BECDeviceFilter.POSITIONER])
|
||||
super().__init__(parent=parent)
|
||||
|
||||
self.motor_x = DeviceComboBox(parent=self, device_filter=[BECDeviceFilter.POSITIONER])
|
||||
self.motor_x.addItem("", None)
|
||||
self.motor_x.setCurrentText("")
|
||||
self.motor_x.setToolTip("Select Motor X")
|
||||
self.motor_x.setItemDelegate(NoCheckDelegate(self.motor_x))
|
||||
self.motor_y = DeviceComboBox(parent=parent, device_filter=[BECDeviceFilter.POSITIONER])
|
||||
self.motor_x.setEditable(True)
|
||||
self.motor_y = DeviceComboBox(parent=self, device_filter=[BECDeviceFilter.POSITIONER])
|
||||
self.motor_y.addItem("", None)
|
||||
self.motor_y.setCurrentText("")
|
||||
self.motor_y.setToolTip("Select Motor Y")
|
||||
self.motor_y.setItemDelegate(NoCheckDelegate(self.motor_y))
|
||||
self.motor_y.setEditable(True)
|
||||
|
||||
self.container = QWidget(parent)
|
||||
layout = QHBoxLayout(self.container)
|
||||
layout = QHBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
layout.addWidget(self.motor_x)
|
||||
layout.addWidget(self.motor_y)
|
||||
self.container.setLayout(layout)
|
||||
self.action = self.container
|
||||
|
||||
def add_to_toolbar(self, toolbar: QToolBar, target: QWidget):
|
||||
"""
|
||||
Adds the widget to the toolbar.
|
||||
|
||||
Args:
|
||||
toolbar (QToolBar): The toolbar to add the widget to.
|
||||
target (QWidget): The target widget for the action.
|
||||
"""
|
||||
|
||||
toolbar.addWidget(self.container)
|
||||
def set_motors(self, motor_x: str | None, motor_y: str | None) -> None:
|
||||
"""Set the displayed motors without emitting selection signals."""
|
||||
motor_x = motor_x or ""
|
||||
motor_y = motor_y or ""
|
||||
self.motor_x.blockSignals(True)
|
||||
self.motor_y.blockSignals(True)
|
||||
try:
|
||||
if motor_x:
|
||||
self.motor_x.set_device(motor_x)
|
||||
self.motor_x.check_validity(motor_x)
|
||||
else:
|
||||
self.motor_x.setCurrentText("")
|
||||
if motor_y:
|
||||
self.motor_y.set_device(motor_y)
|
||||
self.motor_y.check_validity(motor_y)
|
||||
else:
|
||||
self.motor_y.setCurrentText("")
|
||||
finally:
|
||||
self.motor_x.blockSignals(False)
|
||||
self.motor_y.blockSignals(False)
|
||||
|
||||
def cleanup(self):
|
||||
"""
|
||||
@@ -47,5 +59,57 @@ class MotorSelectionAction(ToolBarAction):
|
||||
self.motor_x.deleteLater()
|
||||
self.motor_y.close()
|
||||
self.motor_y.deleteLater()
|
||||
self.container.close()
|
||||
self.container.deleteLater()
|
||||
|
||||
|
||||
def motor_selection_bundle(components: ToolbarComponents) -> ToolbarBundle:
|
||||
"""
|
||||
Creates a workspace toolbar bundle for MotorMap.
|
||||
|
||||
Args:
|
||||
components (ToolbarComponents): The components to be added to the bundle.
|
||||
|
||||
Returns:
|
||||
ToolbarBundle: The workspace toolbar bundle.
|
||||
"""
|
||||
|
||||
motor_selection_widget = MotorSelection(parent=components.toolbar)
|
||||
components.add_safe(
|
||||
"motor_selection", WidgetAction(widget=motor_selection_widget, adjust_size=False)
|
||||
)
|
||||
|
||||
bundle = ToolbarBundle("motor_selection", components)
|
||||
bundle.add_action("motor_selection")
|
||||
return bundle
|
||||
|
||||
|
||||
class MotorSelectionConnection(BundleConnection):
|
||||
"""
|
||||
Connection helper for the motor selection bundle.
|
||||
"""
|
||||
|
||||
def __init__(self, components: ToolbarComponents, target_widget=None):
|
||||
super().__init__(parent=components.toolbar)
|
||||
self.bundle_name = "motor_selection"
|
||||
self.components = components
|
||||
self.target_widget = target_widget
|
||||
self._connected = False
|
||||
|
||||
def _widget(self) -> MotorSelection:
|
||||
return self.components.get_action("motor_selection").widget
|
||||
|
||||
def connect(self):
|
||||
if self._connected:
|
||||
return
|
||||
widget = self._widget()
|
||||
widget.motor_x.currentTextChanged.connect(self.target_widget.on_motor_selection_changed)
|
||||
widget.motor_y.currentTextChanged.connect(self.target_widget.on_motor_selection_changed)
|
||||
self._connected = True
|
||||
|
||||
def disconnect(self):
|
||||
if not self._connected:
|
||||
return
|
||||
widget = self._widget()
|
||||
widget.motor_x.currentTextChanged.disconnect(self.target_widget.on_motor_selection_changed)
|
||||
widget.motor_y.currentTextChanged.disconnect(self.target_widget.on_motor_selection_changed)
|
||||
self._connected = False
|
||||
widget.cleanup()
|
||||
|
||||
@@ -78,15 +78,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__:
|
||||
@@ -122,7 +122,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 #######
|
||||
@@ -154,10 +154,10 @@ def test_available_widgets(qtbot, connected_client_gui_obj):
|
||||
f"is {len(gui._server_registry)} instead of {top_level_widgets_count}. The following "
|
||||
f"widgets are not cleaned up: {set(gui._server_registry.keys()) - names}"
|
||||
) from exc
|
||||
# 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
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
from qtpy.QtTest import QSignalSpy
|
||||
|
||||
from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
|
||||
from tests.unit_tests.client_mocks import mocked_client
|
||||
@@ -274,18 +273,74 @@ def test_motor_map_toolbar_selection(qtbot, mocked_client):
|
||||
# Verify toolbar bundle was created during initialization
|
||||
motor_selection = mm.toolbar.components.get_action("motor_selection")
|
||||
|
||||
motor_selection.motor_x.setCurrentText("samx")
|
||||
motor_selection.motor_y.setCurrentText("samy")
|
||||
motor_selection.widget.motor_x.setCurrentText("samx")
|
||||
motor_selection.widget.motor_y.setCurrentText("samy")
|
||||
|
||||
assert mm.config.x_motor.name == "samx"
|
||||
assert mm.config.y_motor.name == "samy"
|
||||
|
||||
motor_selection.motor_y.setCurrentText("samz")
|
||||
motor_selection.widget.motor_y.setCurrentText("samz")
|
||||
|
||||
assert mm.config.x_motor.name == "samx"
|
||||
assert mm.config.y_motor.name == "samz"
|
||||
|
||||
|
||||
def test_motor_selection_set_motors_blocks_signals(qtbot, mocked_client):
|
||||
"""Ensure set_motors updates both comboboxes without emitting change signals."""
|
||||
mm = create_widget(qtbot, MotorMap, client=mocked_client)
|
||||
motor_selection = mm.toolbar.components.get_action("motor_selection").widget
|
||||
|
||||
spy_x = QSignalSpy(motor_selection.motor_x.currentTextChanged)
|
||||
spy_y = QSignalSpy(motor_selection.motor_y.currentTextChanged)
|
||||
|
||||
motor_selection.set_motors("samx", "samy")
|
||||
|
||||
assert motor_selection.motor_x.currentText() == "samx"
|
||||
assert motor_selection.motor_y.currentText() == "samy"
|
||||
assert spy_x.count() == 0
|
||||
assert spy_y.count() == 0
|
||||
|
||||
|
||||
def test_motor_properties_partial_then_complete_map(qtbot, mocked_client):
|
||||
"""Setting x then y via properties should map once both are valid."""
|
||||
mm = create_widget(qtbot, MotorMap, client=mocked_client)
|
||||
|
||||
spy = QSignalSpy(mm.property_changed)
|
||||
mm.x_motor = "samx"
|
||||
|
||||
assert mm.config.x_motor.name == "samx"
|
||||
assert mm.config.y_motor.name is None
|
||||
assert mm._trace is None # map not triggered yet
|
||||
assert spy.at(0) == ["x_motor", "samx"]
|
||||
|
||||
mm.y_motor = "samy"
|
||||
|
||||
assert mm.config.x_motor.name == "samx"
|
||||
assert mm.config.y_motor.name == "samy"
|
||||
assert mm._trace is not None # map called once both valid
|
||||
assert spy.at(1) == ["y_motor", "samy"]
|
||||
assert len(mm._buffer["x"]) == 1
|
||||
assert len(mm._buffer["y"]) == 1
|
||||
|
||||
|
||||
def test_set_motor_name_emits_and_syncs_toolbar(qtbot, mocked_client):
|
||||
"""_set_motor_name should emit property changes and sync toolbar widgets."""
|
||||
mm = create_widget(qtbot, MotorMap, client=mocked_client)
|
||||
motor_selection = mm.toolbar.components.get_action("motor_selection").widget
|
||||
|
||||
spy = QSignalSpy(mm.property_changed)
|
||||
mm._set_motor_name("x", "samx")
|
||||
|
||||
assert mm.config.x_motor.name == "samx"
|
||||
assert motor_selection.motor_x.currentText() == "samx"
|
||||
assert spy.at(0) == ["x_motor", "samx"]
|
||||
|
||||
# Calling with same name should be a no-op
|
||||
initial_count = spy.count()
|
||||
mm._set_motor_name("x", "samx")
|
||||
assert spy.count() == initial_count
|
||||
|
||||
|
||||
def test_motor_map_settings_dialog(qtbot, mocked_client):
|
||||
"""Test the settings dialog for the motor map."""
|
||||
mm = create_widget(qtbot, MotorMap, client=mocked_client, popups=True)
|
||||
|
||||
97
tests/unit_tests/test_status_bar.py
Normal file
97
tests/unit_tests/test_status_bar.py
Normal 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"
|
||||
Reference in New Issue
Block a user