1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-09 02:00:56 +02:00

Compare commits

...

29 Commits

Author SHA1 Message Date
edd3038e10 feat(curve, waveform): add dap_parameters for lmfit customization in DAP requests 2025-12-16 14:05:21 +01:00
7993cf9c1d wip cleanup 2025-12-16 13:58:53 +01:00
40e186cb13 fix(colors): more benevolent fetching of colormap names, avoid hardcoded wrong colormap mapping from GradientWidget from pg 2025-12-16 13:58:53 +01:00
semantic-release
ca6f355aac 2.45.12
Automatically generated by python-semantic-release
2025-12-16 12:37:06 +00:00
d876ca72bc fix(heatmap): more robust logic for fast and slow axis in grid scan 2025-12-16 13:36:17 +01:00
e0fd97616d fix(heatmap): flush image if config changes during scan 2025-12-16 13:36:17 +01:00
6af8a5cbfe fix(heatmap): grid scan image correctly map to scan positions 2025-12-16 13:36:17 +01:00
semantic-release
944e2cedf8 2.45.11
Automatically generated by python-semantic-release
2025-12-15 14:48:20 +00:00
cd11a6cce3 fix(waveform): support for AsyncMultiSignal 2025-12-15 15:47:26 +01:00
semantic-release
c98106e594 2.45.10
Automatically generated by python-semantic-release
2025-12-10 11:21:40 +00:00
04f1ff4fe7 fix(devices): minor fix to comply with new config helper in bec_lib 2025-12-10 12:20:48 +01:00
semantic-release
45ed92494c 2.45.9
Automatically generated by python-semantic-release
2025-12-09 14:21:27 +00:00
5fc96bd299 fix(rpc): add expiration to GUI registry state updates 2025-12-09 15:20:42 +01:00
semantic-release
1ad5df57fe 2.45.8
Automatically generated by python-semantic-release
2025-12-08 14:36:13 +00:00
440e778162 fix(notification_banner): backwards compatibility to push messages from Broker to Centre as dict 2025-12-08 15:35:23 +01:00
fdeb8fcb0f fix(notification_banner): formatted error messages fetched directly from BECMessage; do not repreat notifications ids 2025-12-08 15:35:23 +01:00
5c90983dd4 fix(notification_banner): better contrast in light mode 2025-12-08 15:35:23 +01:00
4171de1e45 fix(notification_banner): expired messages are hidden in notification center but still accessible 2025-12-08 15:35:23 +01:00
semantic-release
f12339e6f9 2.45.7
Automatically generated by python-semantic-release
2025-12-08 14:04:02 +00:00
ce8e5f0bec fix: handle none in literal combobox 2025-12-08 15:03:17 +01:00
semantic-release
7ea9ab5175 2.45.6
Automatically generated by python-semantic-release
2025-11-27 09:43:46 +00:00
b72f0dc6e8 fix(curve): update dap curves if data are set manually 2025-11-27 10:42:58 +01:00
semantic-release
cb9d429884 2.45.5
Automatically generated by python-semantic-release
2025-11-26 13:25:28 +00:00
0a80bd0a92 fix: remove ghost widgets in scan metadata 2025-11-26 14:24:36 +01:00
semantic-release
9bc9d355e2 2.45.4
Automatically generated by python-semantic-release
2025-11-24 13:30:46 +00:00
7d5e702a11 fix(web_links): fixed link to bec widget issues from gitlab to github 2025-11-24 14:29:56 +01:00
40cbf7fe4f fix(main_window): removed hiding scan progressbar animation 2025-11-24 14:29:56 +01:00
semantic-release
7b287c45f2 2.45.3
Automatically generated by python-semantic-release
2025-11-17 19:27:38 +00:00
c9455672b5 fix(fakeredis): add support for additional args 2025-11-17 20:24:44 +01:00
25 changed files with 831 additions and 280 deletions

View File

@@ -1,6 +1,105 @@
# CHANGELOG
## v2.45.12 (2025-12-16)
### Bug Fixes
- **heatmap**: Flush image if config changes during scan
([`e0fd976`](https://github.com/bec-project/bec_widgets/commit/e0fd97616d370722e2ebf12d0f93862ac35cb20d))
- **heatmap**: Grid scan image correctly map to scan positions
([`6af8a5c`](https://github.com/bec-project/bec_widgets/commit/6af8a5cbfe0f97327b31039033d3e6946388347c))
- **heatmap**: More robust logic for fast and slow axis in grid scan
([`d876ca7`](https://github.com/bec-project/bec_widgets/commit/d876ca72bc50f967f0872eb777f2378a3db68ddf))
## v2.45.11 (2025-12-15)
### Bug Fixes
- **waveform**: Support for AsyncMultiSignal
([`cd11a6c`](https://github.com/bec-project/bec_widgets/commit/cd11a6cce33f3c0642984ae6b2d159c7441e22c6))
## v2.45.10 (2025-12-10)
### Bug Fixes
- **devices**: Minor fix to comply with new config helper in bec_lib
([`04f1ff4`](https://github.com/bec-project/bec_widgets/commit/04f1ff4fe7869215f010bf73f7271e063e21f2a2))
## v2.45.9 (2025-12-09)
### Bug Fixes
- **rpc**: Add expiration to GUI registry state updates
([`5fc96bd`](https://github.com/bec-project/bec_widgets/commit/5fc96bd299115c1849240bae3b37112aad8f5a54))
## v2.45.8 (2025-12-08)
### Bug Fixes
- **notification_banner**: Backwards compatibility to push messages from Broker to Centre as dict
([`440e778`](https://github.com/bec-project/bec_widgets/commit/440e778162ebb359fc33be26e3d22f99b4f9dcfe))
- **notification_banner**: Better contrast in light mode
([`5c90983`](https://github.com/bec-project/bec_widgets/commit/5c90983dd4c3ff96e5625ebda0054a1ac1256227))
- **notification_banner**: Expired messages are hidden in notification center but still accessible
([`4171de1`](https://github.com/bec-project/bec_widgets/commit/4171de1e454c4832513ca599c0fd0eaa365c7c32))
- **notification_banner**: Formatted error messages fetched directly from BECMessage; do not repreat
notifications ids
([`fdeb8fc`](https://github.com/bec-project/bec_widgets/commit/fdeb8fcb0f223d64933f2791585756527c2f41ed))
## v2.45.7 (2025-12-08)
### Bug Fixes
- Handle none in literal combobox
([`ce8e5f0`](https://github.com/bec-project/bec_widgets/commit/ce8e5f0bec7643c9f826e06f987775de95abb91d))
## v2.45.6 (2025-11-27)
### Bug Fixes
- **curve**: Update dap curves if data are set manually
([`b72f0dc`](https://github.com/bec-project/bec_widgets/commit/b72f0dc6e8474a65c83f7e2c938fc6356b7b5f3a))
## v2.45.5 (2025-11-26)
### Bug Fixes
- Remove ghost widgets in scan metadata
([`0a80bd0`](https://github.com/bec-project/bec_widgets/commit/0a80bd0a9279cef1136a04c252c97e624ef2e779))
## v2.45.4 (2025-11-24)
### Bug Fixes
- **main_window**: Removed hiding scan progressbar animation
([`40cbf7f`](https://github.com/bec-project/bec_widgets/commit/40cbf7fe4f834a1a65306e54b3882d2c0495f90a))
- **web_links**: Fixed link to bec widget issues from gitlab to github
([`7d5e702`](https://github.com/bec-project/bec_widgets/commit/7d5e702a11043ed96a8cb97fce6b2162681e8fab))
## v2.45.3 (2025-11-17)
### Bug Fixes
- **fakeredis**: Add support for additional args
([`c945567`](https://github.com/bec-project/bec_widgets/commit/c9455672b58b9df101ccd0d80a169bdf6c707f34))
## v2.45.2 (2025-11-17)
### Bug Fixes

View File

@@ -5418,6 +5418,7 @@ class Waveform(RPCBase):
color: "str | None" = None,
label: "str | None" = None,
dap: "str | None" = None,
dap_parameters: "dict | lmfit.Parameters | None | object" = None,
scan_id: "str | None" = None,
scan_number: "int | None" = None,
**kwargs,
@@ -5442,6 +5443,8 @@ class Waveform(RPCBase):
dap(str): The dap model to use for the curve. When provided, a DAP curve is
attached automatically for device, history, or custom data sources. Use
the same string as the LMFit model name.
dap_parameters(dict | lmfit.Parameters | None): Optional lmfit parameter overrides sent to the DAP server.
Values can be numeric (interpreted as fixed parameters) or dicts like`{"value": 1.0, "vary": False}`.
scan_id(str): Optional scan ID. When provided, the curve is treated as a **history** curve and
the ydata (and optional xdata) are fetched from that historical scan. Such curves are
never cleared by livescan resets.
@@ -5458,6 +5461,7 @@ class Waveform(RPCBase):
dap_name: "str",
color: "str | None" = None,
dap_oversample: "int" = 1,
dap_parameters: "dict | lmfit.Parameters | None" = None,
**kwargs,
) -> "Curve":
"""
@@ -5470,6 +5474,7 @@ class Waveform(RPCBase):
dap_name(str): The name of the DAP model to use.
color(str): The color of the curve.
dap_oversample(int): The oversampling factor for the DAP curve.
dap_parameters(dict | lmfit.Parameters | None): Optional lmfit parameter overrides sent to the DAP server.
**kwargs
Returns:

View File

@@ -1,11 +1,13 @@
from __future__ import annotations
from functools import lru_cache
import re
from typing import TYPE_CHECKING, Literal
import bec_qthemes
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger
from bec_qthemes._os_appearance.listener import OSThemeSwitchListener
from pydantic_core import PydanticCustomError
from qtpy.QtGui import QColor
@@ -15,6 +17,9 @@ if TYPE_CHECKING: # pragma: no cover
from bec_qthemes._main import AccentColors
logger = bec_logger.logger
def get_theme_name():
if QApplication.instance() is None or not hasattr(QApplication.instance(), "theme"):
return "dark"
@@ -138,6 +143,83 @@ def apply_theme(theme: Literal["dark", "light"]):
class Colors:
@staticmethod
def list_available_colormaps() -> list[str]:
"""
List colormap names available via the pyqtgraph colormap registry.
Note: This does not include `GradientEditorItem` presets (used by HistogramLUT menus).
"""
def _list(source: str | None = None) -> list[str]:
try:
return pg.colormap.listMaps() if source is None else pg.colormap.listMaps(source)
except Exception: # pragma: no cover - backend may be missing
return []
return [*_list(None), *_list("matplotlib"), *_list("colorcet")]
@staticmethod
def list_available_gradient_presets() -> list[str]:
"""
List `GradientEditorItem` preset names (HistogramLUT right-click menu entries).
"""
from pyqtgraph.graphicsItems.GradientEditorItem import Gradients
return list(Gradients.keys())
@staticmethod
def canonical_colormap_name(color_map: str) -> str:
"""
Return an available colormap/preset name if a case-insensitive match exists.
"""
requested = (color_map or "").strip()
if not requested:
return requested
registry = Colors.list_available_colormaps()
presets = Colors.list_available_gradient_presets()
available = set(registry) | set(presets)
if requested in available:
return requested
# Case-insensitive match.
lower_to_canonical = {name.lower(): name for name in available}
return lower_to_canonical.get(requested.lower(), requested)
@staticmethod
def get_colormap(color_map: str) -> pg.ColorMap:
"""
Resolve a string into a `pg.ColorMap` using either:
- the `pg.colormap` registry (optionally including matplotlib/colorcet backends), or
- `GradientEditorItem` presets (HistogramLUT right-click menu).
"""
name = Colors.canonical_colormap_name(color_map)
if not name:
raise ValueError("Empty colormap name")
return Colors._get_colormap_cached(name)
@staticmethod
@lru_cache(maxsize=256)
def _get_colormap_cached(name: str) -> pg.ColorMap:
# 1) Registry/backends
try:
return pg.colormap.get(name)
except Exception:
pass
for source in ("matplotlib", "colorcet"):
try:
return pg.colormap.get(name, source=source)
except Exception:
continue
# 2) Presets -> ColorMap
ge = pg.GradientEditorItem()
ge.loadPreset(name)
return ge.colorMap()
@staticmethod
def golden_ratio(num: int) -> list:
@@ -219,7 +301,7 @@ class Colors:
if theme_offset < 0 or theme_offset > 1:
raise ValueError("theme_offset must be between 0 and 1")
cmap = pg.colormap.get(colormap)
cmap = Colors.get_colormap(colormap)
min_pos, max_pos = Colors.set_theme_offset(theme, theme_offset)
# Generate positions that are evenly spaced within the acceptable range
@@ -267,7 +349,7 @@ class Colors:
ValueError: If theme_offset is not between 0 and 1.
"""
cmap = pg.colormap.get(colormap)
cmap = Colors.get_colormap(colormap)
phi = (1 + np.sqrt(5)) / 2 # Golden ratio
golden_angle_conjugate = 1 - (1 / phi) # Approximately 0.38196601125
@@ -533,18 +615,21 @@ class Colors:
Raises:
PydanticCustomError: If colormap is invalid.
"""
available_pg_maps = pg.colormap.listMaps()
available_mpl_maps = pg.colormap.listMaps("matplotlib")
available_mpl_colorcet = pg.colormap.listMaps("colorcet")
available_colormaps = available_pg_maps + available_mpl_maps + available_mpl_colorcet
if color_map not in available_colormaps:
normalized = Colors.canonical_colormap_name(color_map)
try:
Colors.get_colormap(normalized)
except Exception as ext:
logger.warning(f"Colormap validation error: {ext}")
if return_error:
available_colormaps = sorted(
set(Colors.list_available_colormaps())
| set(Colors.list_available_gradient_presets())
)
raise PydanticCustomError(
"unsupported colormap",
f"Colormap '{color_map}' not found in the current installation of pyqtgraph. Choose on the following: {available_colormaps}.",
f"Colormap '{color_map}' not found in the current installation of pyqtgraph. Choose from the following: {available_colormaps}.",
{"wrong_value": color_map},
)
else:
return False
return color_map
return normalized

View File

@@ -81,10 +81,11 @@ class TypedForm(BECWidget, QWidget):
self._form_grid_container = QWidget(parent=self)
self._form_grid_container.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Minimum)
self._form_grid_container.setLayout(QVBoxLayout())
self._layout.addWidget(self._form_grid_container)
self._form_grid = QWidget(parent=self._form_grid_container)
self._form_grid.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Minimum)
self._layout.addWidget(self._form_grid_container)
self._form_grid_container.setLayout(QVBoxLayout())
self._form_grid.setLayout(self._new_grid_layout())
self._widget_types: dict | None = None
@@ -105,11 +106,11 @@ class TypedForm(BECWidget, QWidget):
def _add_griditem(self, item: FormItemSpec, row: int):
grid = self._form_grid.layout()
label = QLabel(item.name)
label = QLabel(parent=self._form_grid, text=item.name)
label.setProperty("_model_field_name", item.name)
label.setToolTip(item.info.description or item.name)
grid.addWidget(label, row, 0)
widget = self._widget_from_type(item, self._widget_types)(parent=self, spec=item)
widget = self._widget_from_type(item, self._widget_types)(parent=self._form_grid, spec=item)
widget.valueChanged.connect(self.value_changed)
widget.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Minimum)
grid.addWidget(widget, row, 1)
@@ -128,19 +129,17 @@ class TypedForm(BECWidget, QWidget):
}
def _clear_grid(self):
if (old_layout := self._form_grid.layout()) is not None:
while old_layout.count():
item = old_layout.takeAt(0)
widget = item.widget()
if widget is not None:
widget.deleteLater()
old_layout.deleteLater()
self._form_grid.deleteLater()
gl = self._form_grid.layout()
while w := gl.takeAt(0):
w = w.widget()
if hasattr(w, "teardown"):
w.teardown()
w.deleteLater()
self._form_grid_container.layout().removeWidget(self._form_grid)
self._form_grid.deleteLater()
self._form_grid = QWidget()
self._form_grid.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Minimum)
self._form_grid.setLayout(self._new_grid_layout())
self._form_grid_container.layout().addWidget(self._form_grid)
self.update_size()
def update_size(self):
@@ -149,7 +148,7 @@ class TypedForm(BECWidget, QWidget):
self.adjustSize()
def _new_grid_layout(self):
new_grid = QGridLayout()
new_grid = QGridLayout(self)
new_grid.setContentsMargins(0, 0, 0, 0)
return new_grid

View File

@@ -3,14 +3,16 @@ from __future__ import annotations
import typing
from abc import abstractmethod
from decimal import Decimal
from types import GenericAlias, UnionType
from types import GenericAlias, NoneType, UnionType
from typing import (
Any,
Callable,
Final,
Generic,
Iterable,
Literal,
NamedTuple,
Optional,
OrderedDict,
TypeVar,
get_args,
@@ -71,7 +73,7 @@ class FormItemSpec(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
item_type: type | UnionType | GenericAlias
item_type: type | UnionType | GenericAlias | Optional[Any]
name: str
info: FieldInfo = FieldInfo()
pretty_display: bool = Field(
@@ -188,6 +190,10 @@ class DynamicFormItem(QWidget):
"""Add the main data entry widget to self._main_widget and appply any
constraints from the field info"""
@SafeSlot()
def clear(self, *_):
return
def _set_pretty_display(self):
self.setEnabled(False)
if button := getattr(self, "_clear_button", None):
@@ -204,11 +210,17 @@ class DynamicFormItem(QWidget):
self._layout.addWidget(self._clear_button)
# the widget added in _add_main_widget must implement .clear() if value is not required
self._clear_button.setToolTip("Clear value or reset to default.")
self._clear_button.clicked.connect(self._main_widget.clear) # type: ignore
self._clear_button.clicked.connect(self.clear) # type: ignore
def _value_changed(self, *_, **__):
self.valueChanged.emit()
def teardown(self):
self._layout.deleteLater()
self._layout.removeWidget(self._main_widget)
self._main_widget.deleteLater()
self._main_widget = None
class StrFormItem(DynamicFormItem):
def __init__(self, parent: QWidget | None = None, *, spec: FormItemSpec) -> None:
@@ -542,11 +554,14 @@ class StrLiteralFormItem(DynamicFormItem):
self._layout.addWidget(self._main_widget)
def getValue(self):
if self._main_widget.currentIndex() == -1:
return None
return self._main_widget.currentText()
def setValue(self, value: str | None):
if value is None:
self.clear()
return
for i in range(self._main_widget.count()):
if self._main_widget.itemText(i) == value:
self._main_widget.setCurrentIndex(i)
@@ -557,15 +572,39 @@ class StrLiteralFormItem(DynamicFormItem):
self._main_widget.setCurrentIndex(-1)
class OptionalStrLiteralFormItem(StrLiteralFormItem):
def _add_main_widget(self) -> None:
self._main_widget = QComboBox()
self._options = get_args(get_args(self._spec.info.annotation)[0])
for opt in self._options:
self._main_widget.addItem(opt)
self._layout.addWidget(self._main_widget)
WidgetTypeRegistry = OrderedDict[str, tuple[Callable[[FormItemSpec], bool], type[DynamicFormItem]]]
def _is_string_literal(t: type):
return type(t) is type(Literal[""]) and set(type(arg) for arg in get_args(t)) == {str}
def _is_optional_string_literal(t: type):
if not hasattr(t, "__args__"):
return False
if len(t.__args__) != 2:
return False
if _is_string_literal(t.__args__[0]) and t.__args__[1] is NoneType:
return True
return False
DEFAULT_WIDGET_TYPES: Final[WidgetTypeRegistry] = OrderedDict() | {
# dict literals are ordered already but TypedForm subclasses may modify coppies of this dict
# and delete/insert keys or change the order
"literal_str": (
lambda spec: type(spec.info.annotation) is type(Literal[""])
and set(type(arg) for arg in get_args(spec.info.annotation)) == {str},
StrLiteralFormItem,
"literal_str": (lambda spec: _is_string_literal(spec.info.annotation), StrLiteralFormItem),
"optional_literal_str": (
lambda spec: _is_optional_string_literal(spec.info.annotation),
OptionalStrLiteralFormItem,
),
"str": (lambda spec: spec.item_type in [str, str | None, None], StrFormItem),
"int": (lambda spec: spec.item_type in [int, int | None], IntFormItem),
@@ -616,6 +655,8 @@ if __name__ == "__main__": # pragma: no cover
value5: int | None = Field()
value6: list[int] = Field()
value7: list = Field()
literal: Literal["a", "b", "c"]
nullable_literal: Literal["a", "b", "c"] | None = None
app = QApplication([])
w = QWidget()
@@ -623,7 +664,7 @@ if __name__ == "__main__": # pragma: no cover
w.setLayout(layout)
items = []
for i, (field_name, info) in enumerate(TestModel.model_fields.items()):
spec = spec = FormItemSpec(item_type=info.annotation, name=field_name, info=info)
spec = FormItemSpec(item_type=info.annotation, name=field_name, info=info)
layout.addWidget(QLabel(field_name), i, 0)
widg = widget_from_type(spec)(spec=spec)
items.append(widg)

View File

@@ -229,6 +229,7 @@ class RPCServer:
MessageEndpoints.gui_registry_state(self.gui_id),
msg_dict={"data": messages.GUIRegistryStateMessage(state=data)},
max_size=1,
expire=60,
)
def _serialize_bec_connector(self, connector: BECConnector, wait=False) -> dict:

View File

@@ -14,13 +14,14 @@ from __future__ import annotations
import json
import sys
from datetime import datetime
from enum import Enum, auto
from enum import Enum
from typing import Literal
from uuid import uuid4
import pyqtgraph as pg
from bec_lib.alarm_handler import Alarms # external enum
from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import ErrorInfo
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
from qtpy.QtCore import QObject, QTimer
@@ -28,6 +29,7 @@ from qtpy.QtWidgets import QApplication, QFrame, QMainWindow, QScrollArea, QWidg
from bec_widgets import SafeProperty, SafeSlot
from bec_widgets.utils import BECConnector
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.widget_io import WidgetIO
@@ -53,10 +55,10 @@ DARK_PALETTE = {
}
LIGHT_PALETTE = {
"base": "#e9ecef",
"title": "#212121",
"body": "#424242",
"separator": "rgba(0,0,0,40)",
"base": "#f5f5f7",
"title": "#111827",
"body": "#374151",
"separator": "rgba(15,23,42,40)",
}
@@ -108,6 +110,7 @@ class NotificationToast(QFrame):
self._kind = kind if isinstance(kind, SeverityKind) else SeverityKind(kind)
self._traceback = traceback
self._accent_color = QtGui.QColor(SEVERITY[self._kind.value]["color"])
self._accent_alpha = 50
self.setFrameShape(QtWidgets.QFrame.StyledPanel)
self.created = datetime.now()
@@ -379,22 +382,31 @@ class NotificationToast(QFrame):
# buttons (text colour)
base_btn_color = palette["title"]
card_bg = QtGui.QColor(palette["base"])
# tune card background and hover contrast per theme
if theme == "light":
card_bg.setAlphaF(0.98)
btn_hover = self._accent_color.darker(105).name()
else:
card_bg.setAlphaF(0.88)
btn_hover = self._accent_color.name()
self.setStyleSheet(
"""
#NotificationToast {
background: transparent;
f"""
#NotificationToast {{
background: {card_bg.name(QtGui.QColor.HexArgb)};
border-radius: 12px;
color: %s;
}
#NotificationToast QPushButton {
color: {base_btn_color};
border: 1px solid {palette["separator"]};
}}
#NotificationToast QPushButton {{
background: transparent;
border: none;
color: %s;
color: {base_btn_color};
font-size: 14px;
}
#NotificationToast QPushButton:hover { color: %s; }
}}
#NotificationToast QPushButton:hover {{ color: {btn_hover}; }}
"""
% (base_btn_color, base_btn_color, self._accent_color.name())
)
# traceback panel colours
trace_bg = "#1e1e1e" if theme == "dark" else "#f0f0f0"
@@ -407,6 +419,37 @@ class NotificationToast(QFrame):
"""
)
# icon glyph vs badge background: darker badge, lighter icon in light mode
icon_fg = "#ffffff" if theme == "light" else self._accent_color.name()
icon = material_icon(
icon_name=SEVERITY[self._kind.value]["icon"],
color=icon_fg,
filled=True,
size=(24, 24),
convert_to_pixmap=False,
)
self._icon_btn.setIcon(icon)
badge_bg = QtGui.QColor(self._accent_color)
if theme == "light":
# darken and strengthen the badge on light cards for contrast
badge_bg = badge_bg.darker(115)
badge_bg.setAlphaF(0.70)
else:
badge_bg.setAlphaF(0.30)
icon_bg = badge_bg.name(QtGui.QColor.HexArgb)
self._icon_btn.setStyleSheet(
f"""
QToolButton {{
background: {icon_bg};
border: none;
border-radius: 20px;
}}
"""
)
# stronger accent wash in light mode, slightly stronger in dark too
self._accent_alpha = 110 if theme == "light" else 60
self.update()
########################################
@@ -488,7 +531,9 @@ class NotificationToast(QFrame):
# accent gradient, fades to transparent
grad = QtGui.QLinearGradient(0, 0, self.width() * 0.7, 0)
accent = QtGui.QColor(self._accent_color)
accent.setAlpha(50)
if getattr(self, "_theme", "dark") == "light":
accent = accent.darker(115)
accent.setAlpha(getattr(self, "_accent_alpha", 50))
grad.setColorAt(0.0, accent)
fade = QtGui.QColor(self._accent_color)
fade.setAlpha(0)
@@ -690,7 +735,6 @@ class NotificationCentre(QScrollArea):
toast.notification_id = notification_id
broker = BECNotificationBroker()
toast.closed.connect(lambda nid=notification_id: broker.notification_closed.emit(nid))
toast.expired.connect(lambda nid=notification_id: broker.notification_closed.emit(nid))
toast.closed.connect(lambda: self._hide_notification(toast))
toast.expired.connect(lambda t=toast: self._handle_expire(t))
toast.expanded.connect(self._adjust_height)
@@ -1016,32 +1060,55 @@ class BECNotificationBroker(BECConnector, QObject):
"""
Called when a new alarm arrives. Builds and pushes a toast to each centre
with a shared notification_id, and hooks its close/expire signals.
Args:
msg(dict): The message containing alarm details.
meta(dict): Metadata about the alarm.
"""
msg = msg or {}
meta = meta or {}
centres = WidgetIO.find_widgets(NotificationCentre)
kind = self._banner_kind_from_severity(msg.get("severity", 0))
# Normalise the incoming info payload (can be ErrorInfo, dict or missing entirely)
raw_info = msg.get("info")
if isinstance(raw_info, dict):
try:
raw_info = ErrorInfo(**raw_info)
except Exception:
raw_info = None
notification_id = getattr(raw_info, "id", None) or uuid4().hex
# build title and body
scan_id = meta.get("scan_id")
scan_number = meta.get("scan_number")
formatted_trace = self._err_util.format_traceback(msg.get("msg", ""))
short_msg = self._err_util.parse_error_message(formatted_trace)
title = msg.get("alarm_type", "Alarm")
alarm_type = msg.get("alarm_type") or getattr(raw_info, "exception_type", None) or "Alarm"
title = alarm_type
if scan_number:
title += f" - Scan #{scan_number}"
body_text = short_msg
# build detailed traceback
sections: list[str] = []
if scan_id:
sections.extend(["-------- SCAN_ID --------\n", scan_id])
sections.extend(["-------- TRACEBACK --------", formatted_trace])
source = msg.get("source")
if source:
source_pretty = json.dumps(source, indent=4, default=str)
sections.extend(["", "-------- SOURCE --------", source_pretty])
detailed_trace = "\n".join(sections)
trace_text = getattr(raw_info, "error_message", None) or msg.get("msg") or ""
compact_msg = getattr(raw_info, "compact_error_message", None)
# Prefer the compact message; fall back to parsing the traceback for a humanreadable snippet
body_text = compact_msg or self._err_util.parse_error_message(trace_text)
# build detailed traceback for the expandable panel
detailed_trace: str | None = None
if trace_text:
sections: list[str] = []
if scan_id:
sections.extend(["-------- SCAN_ID --------\n", scan_id])
sections.extend(["-------- TRACEBACK --------", trace_text])
detailed_trace = "\n".join(sections)
lifetime = 0 if kind == SeverityKind.MAJOR else 5_000
# generate one ID for all toasts of this event
notification_id = uuid4().hex
if notification_id in self._active_notifications:
return # already posted
# record this notification for future centres
self._active_notifications[notification_id] = {
"title": title,
@@ -1059,9 +1126,8 @@ class BECNotificationBroker(BECConnector, QObject):
lifetime_ms=lifetime,
notification_id=notification_id,
)
# broadcast any close or expire
# broadcast close events (expiry is handled locally to keep history)
toast.closed.connect(lambda nid=notification_id: self.notification_closed.emit(nid))
toast.expired.connect(lambda nid=notification_id: self.notification_closed.emit(nid))
@SafeSlot(dict, dict)
def on_scan_status(self, msg: dict, meta: dict) -> None:
@@ -1086,6 +1152,13 @@ class BECNotificationBroker(BECConnector, QObject):
Translate an integer severity (0/1/2) into a SeverityKind enum.
Unknown values fall back to SeverityKind.WARNING.
"""
if isinstance(severity, SeverityKind):
return severity
if isinstance(severity, str):
try:
return SeverityKind(severity)
except ValueError:
pass
try:
return SeverityKind[Alarms(severity).name] # e.g. WARNING → SeverityKind.WARNING
except (ValueError, KeyError):
@@ -1164,10 +1237,10 @@ class DemoWindow(QMainWindow): # pragma: no cover
# ----- wiring ------------------------------------------------------------
self._counter = 1
self.info_btn.clicked.connect(lambda: self._post("info"))
self.warning_btn.clicked.connect(lambda: self._post("warning"))
self.minor_btn.clicked.connect(lambda: self._post("minor"))
self.major_btn.clicked.connect(lambda: self._post("major"))
self.info_btn.clicked.connect(lambda: self._post(SeverityKind.INFO))
self.warning_btn.clicked.connect(lambda: self._post(SeverityKind.WARNING))
self.minor_btn.clicked.connect(lambda: self._post(SeverityKind.MINOR))
self.major_btn.clicked.connect(lambda: self._post(SeverityKind.MAJOR))
# Raise buttons simulate alarms
self.raise_warning_btn.clicked.connect(lambda: self._raise_error(Alarms.WARNING))
self.raise_minor_btn.clicked.connect(lambda: self._raise_error(Alarms.MINOR))
@@ -1183,30 +1256,28 @@ class DemoWindow(QMainWindow): # pragma: no cover
indicator.hide_all_requested.connect(self.notification_centre.hide_all)
# ------------------------------------------------------------------
def _post(self, kind):
expire = 0 if kind == "error" else 5000
trace = (
'Traceback (most recent call last):\n File "<stdin>", line 1\nZeroDivisionError: 1/0'
if kind == "error"
else None
)
self.notification_centre.add_notification(
title=f"{kind.capitalize()} #{self._counter}",
body="Lorem ipsum dolor sit amet.",
kind=SeverityKind(kind),
lifetime_ms=expire,
traceback=trace,
)
def _post(self, kind: SeverityKind):
"""
Send a simple notification through the broker (non-error case).
"""
msg = {
"severity": kind.value, # handled by broker for SeverityKind
"alarm_type": f"{kind.value.capitalize()}",
"msg": f"{kind.value.capitalize()} #{self._counter}",
}
self.notification_broker.post_notification(msg, meta={})
self._counter += 1
def _raise_error(self, severity):
"""Simulate an error that would be caught by the notification broker."""
self.notification_broker.client.connector.raise_alarm(
severity=severity,
alarm_type="ValueError",
source={"device": "samx", "source": "async_file_writer"},
msg=f"test alarm",
metadata={"test": 1},
info=ErrorInfo(
id=uuid4().hex,
exception_type="ValueError",
error_message="An example error occurred in DemoWindowApp.",
compact_error_message="An example error occurred.",
),
)
# this part is same as implemented in the BECMainWindow
@@ -1225,6 +1296,7 @@ class DemoWindow(QMainWindow): # pragma: no cover
def main(): # pragma: no cover
app = QtWidgets.QApplication(sys.argv)
apply_theme("dark")
win = DemoWindow()
win.show()
sys.exit(app.exec())

View File

@@ -12,4 +12,4 @@ class BECWebLinksMixin:
@staticmethod
def open_bec_bug_report():
webbrowser.open("https://gitlab.psi.ch/groups/bec/-/issues/")
webbrowser.open("https://github.com/bec-project/bec_widgets/issues")

View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import os
from bec_lib.endpoints import MessageEndpoints
from qtpy.QtCore import QEasingCurve, QEvent, QPropertyAnimation, QSize, Qt, QTimer
from qtpy.QtCore import QEvent, QSize, Qt, QTimer
from qtpy.QtGui import QAction, QActionGroup, QIcon
from qtpy.QtWidgets import (
QApplication,
@@ -42,7 +42,7 @@ class BECMainWindow(BECWidget, QMainWindow):
RPC = True
PLUGIN = True
SCAN_PROGRESS_WIDTH = 100 # px
STATUS_BAR_WIDGETS_EXPIRE_TIME = 60_000 # milliseconds
SCAN_PROGRESS_HEIGHT = 12 # px
def __init__(
self,
@@ -201,8 +201,8 @@ class BECMainWindow(BECWidget, QMainWindow):
self._scan_progress_bar_simple.show_remaining_time = False
self._scan_progress_bar_simple.show_source_label = False
self._scan_progress_bar_simple.progressbar.label_template = ""
self._scan_progress_bar_simple.progressbar.setFixedHeight(8)
self._scan_progress_bar_simple.progressbar.setFixedWidth(80)
self._scan_progress_bar_simple.progressbar.setFixedHeight(self.SCAN_PROGRESS_HEIGHT)
self._scan_progress_bar_simple.progressbar.setFixedWidth(self.SCAN_PROGRESS_WIDTH)
self._scan_progress_bar_full = ScanProgressBar(self)
self._scan_progress_hover = HoverWidget(
self, simple=self._scan_progress_bar_simple, full=self._scan_progress_bar_full
@@ -219,62 +219,8 @@ class BECMainWindow(BECWidget, QMainWindow):
self._scan_progress_bar_with_separator.layout.addWidget(separator)
self._scan_progress_bar_with_separator.layout.addWidget(self._scan_progress_hover)
# Set Size
self._scan_progress_bar_target_width = self.SCAN_PROGRESS_WIDTH
self._scan_progress_bar_with_separator.setMaximumWidth(self._scan_progress_bar_target_width)
self.status_bar.addWidget(self._scan_progress_bar_with_separator)
# Visibility logic
self._scan_progress_bar_with_separator.hide()
self._scan_progress_bar_with_separator.setMaximumWidth(0)
# Timer for hiding logic
self._scan_progress_hide_timer = QTimer(self)
self._scan_progress_hide_timer.setSingleShot(True)
self._scan_progress_hide_timer.setInterval(self.STATUS_BAR_WIDGETS_EXPIRE_TIME)
self._scan_progress_hide_timer.timeout.connect(self._animate_hide_scan_progress_bar)
# Show / hide behaviour
self._scan_progress_bar_simple.progress_started.connect(self._show_scan_progress_bar)
self._scan_progress_bar_simple.progress_finished.connect(self._delay_hide_scan_progress_bar)
def _show_scan_progress_bar(self):
if self._scan_progress_hide_timer.isActive():
self._scan_progress_hide_timer.stop()
if self._scan_progress_bar_with_separator.isVisible():
return
# Make visible and reset width
self._scan_progress_bar_with_separator.show()
self._scan_progress_bar_with_separator.setMaximumWidth(0)
self._show_container_anim = QPropertyAnimation(
self._scan_progress_bar_with_separator, b"maximumWidth", self
)
self._show_container_anim.setDuration(300)
self._show_container_anim.setStartValue(0)
self._show_container_anim.setEndValue(self._scan_progress_bar_target_width)
self._show_container_anim.setEasingCurve(QEasingCurve.OutCubic)
self._show_container_anim.start()
def _delay_hide_scan_progress_bar(self):
"""Start the countdown to hide the scan progress bar."""
if hasattr(self, "_scan_progress_hide_timer"):
self._scan_progress_hide_timer.start()
def _animate_hide_scan_progress_bar(self):
"""Shrink container to the right, then hide."""
self._hide_container_anim = QPropertyAnimation(
self._scan_progress_bar_with_separator, b"maximumWidth", self
)
self._hide_container_anim.setDuration(300)
self._hide_container_anim.setStartValue(self._scan_progress_bar_with_separator.width())
self._hide_container_anim.setEndValue(0)
self._hide_container_anim.setEasingCurve(QEasingCurve.InCubic)
self._hide_container_anim.finished.connect(self._scan_progress_bar_with_separator.hide)
self._hide_container_anim.start()
def _add_separator(self, separate_object: bool = False) -> QWidget | None:
"""
Add a vertically centred separator to the status bar or just return it as a separate object.
@@ -474,8 +420,6 @@ class BECMainWindow(BECWidget, QMainWindow):
# Timer cleanup
if hasattr(self, "_client_info_expire_timer") and self._client_info_expire_timer.isActive():
self._client_info_expire_timer.stop()
if hasattr(self, "_scan_progress_hide_timer") and self._scan_progress_hide_timer.isActive():
self._scan_progress_hide_timer.stop()
########################################
# Status bar widgets cleanup

View File

@@ -85,7 +85,6 @@ class ScanMetadata(PydanticModelForm):
def set_schema_from_scan(self, scan_name: str | None):
self._scan_name = scan_name or ""
self.set_schema(get_metadata_schema_for_scan(self._scan_name))
self.populate()
if __name__ == "__main__": # pragma: no cover

View File

@@ -268,6 +268,20 @@ class Heatmap(ImageBase):
if show_config_label is None:
show_config_label = self._image_config.show_config_label
def _device_key(device: HeatmapDeviceSignal | None) -> tuple[str | None, str | None]:
return (device.name if device else None, device.entry if device else None)
prev_cfg = getattr(self, "_image_config", None)
config_changed = False
if prev_cfg and prev_cfg.x_device and prev_cfg.y_device and prev_cfg.z_device:
config_changed = any(
(
_device_key(prev_cfg.x_device) != (x_name, x_entry),
_device_key(prev_cfg.y_device) != (y_name, y_entry),
_device_key(prev_cfg.z_device) != (z_name, z_entry),
)
)
self._image_config = HeatmapConfig(
parent_id=self.gui_id,
x_device=HeatmapDeviceSignal(name=x_name, entry=x_entry),
@@ -282,7 +296,10 @@ class Heatmap(ImageBase):
show_config_label=show_config_label,
)
self.color_map = color_map
self.reload = reload
self.reload = reload or config_changed
if config_changed:
self._grid_index = None
self.main_image.clear()
self.update_labels()
self._fetch_running_scan()
@@ -603,55 +620,51 @@ class Heatmap(ImageBase):
args = self.arg_bundle_to_dict(4, msg.request_inputs["arg_bundle"])
shape = (
args[self._image_config.x_device.entry][-1],
args[self._image_config.y_device.entry][-1],
)
x_entry = self._image_config.x_device.entry
y_entry = self._image_config.y_device.entry
shape = (args[x_entry][-1], args[y_entry][-1])
data = self.main_image.raw_data
if data is None or data.shape != shape:
data = np.empty(shape)
data.fill(np.nan)
def _get_grid_data(axis, snaked=True):
x_grid, y_grid = np.meshgrid(axis[0], axis[1])
if snaked:
y_grid.T[::2] = np.fliplr(y_grid.T[::2])
x_flat = x_grid.T.ravel()
y_flat = y_grid.T.ravel()
positions = np.vstack((x_flat, y_flat)).T
return positions
elif self.reload:
data.fill(np.nan)
snaked = msg.request_inputs["kwargs"].get("snaked", True)
# If the scan's fast axis is x, we need to swap the x and y axes
swap = bool(msg.request_inputs["arg_bundle"][4] == self._image_config.x_device.entry)
slow_entry, fast_entry = (
msg.request_inputs["arg_bundle"][0],
msg.request_inputs["arg_bundle"][4],
)
# calculate the QTransform to put (0,0) at the axis origin
scan_pos = np.asarray(msg.info["positions"])
x_min = min(scan_pos[:, 0])
x_max = max(scan_pos[:, 0])
y_min = min(scan_pos[:, 1])
y_max = max(scan_pos[:, 1])
scan_pos = np.asarray(msg.info["positions"], dtype=float)
relative = bool(msg.request_inputs["kwargs"].get("relative", False))
x_range = x_max - x_min
y_range = y_max - y_min
def _axis_column(entry: str) -> int:
return 0 if entry == slow_entry else 1
pixel_size_x = x_range / (shape[0] - 1)
pixel_size_y = y_range / (shape[1] - 1)
def _axis_levels(entry: str, npts: int) -> np.ndarray:
start, stop = args[entry][:2]
if relative:
origin = float(scan_pos[0, _axis_column(entry)] - start)
return origin + np.linspace(start, stop, npts)
return np.linspace(start, stop, npts)
x_levels = _axis_levels(x_entry, shape[0])
y_levels = _axis_levels(y_entry, shape[1])
pixel_size_x = (
float(x_levels[-1] - x_levels[0]) / max(shape[0] - 1, 1) if shape[0] > 1 else 1.0
)
pixel_size_y = (
float(y_levels[-1] - y_levels[0]) / max(shape[1] - 1, 1) if shape[1] > 1 else 1.0
)
transform = QTransform()
if swap:
transform.scale(pixel_size_y, pixel_size_x)
transform.translate(y_min / pixel_size_y - 0.5, x_min / pixel_size_x - 0.5)
else:
transform.scale(pixel_size_x, pixel_size_y)
transform.translate(x_min / pixel_size_x - 0.5, y_min / pixel_size_y - 0.5)
target_positions = _get_grid_data(
(np.arange(shape[int(swap)]), np.arange(shape[int(not swap)])), snaked=snaked
)
transform.scale(pixel_size_x, pixel_size_y)
transform.translate(x_levels[0] / pixel_size_x - 0.5, y_levels[0] / pixel_size_y - 0.5)
# Fill the data array with the z values
if self._grid_index is None or self.reload:
@@ -659,7 +672,16 @@ class Heatmap(ImageBase):
self.reload = False
for i in range(self._grid_index, len(z_data)):
data[target_positions[i, int(swap)], target_positions[i, int(not swap)]] = z_data[i]
slow_i, fast_i = divmod(i, args[fast_entry][-1])
if snaked and (slow_i % 2 == 1):
fast_i = args[fast_entry][-1] - 1 - fast_i
if x_entry == fast_entry:
x_i, y_i = fast_i, slow_i
else:
x_i, y_i = slow_i, fast_i
data[x_i, y_i] = z_data[i]
self._grid_index = len(z_data)
return data, transform

View File

@@ -9,6 +9,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationError
from qtpy.QtCore import QPointF, Signal, SignalInstance
from qtpy.QtWidgets import QDialog, QVBoxLayout
from bec_widgets.utils import Colors
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.side_panel import SidePanel
@@ -131,8 +132,7 @@ class ImageLayerManager:
image.setZValue(z_position)
image.removed.connect(self._remove_destroyed_layer)
# FIXME: For now, we hard-code the default color map here. In the future, this should be configurable.
image.color_map = "plasma"
image.color_map = self.parent.config.color_map
self.layers[name] = ImageLayer(name=name, image=image, sync=sync)
self.plot_item.addItem(image)
@@ -249,6 +249,8 @@ class ImageBase(PlotBase):
Base class for the Image widget.
"""
MAX_TICKS_COLORBAR = 10
sync_colorbar_with_autorange = Signal()
image_updated = Signal()
layer_added = Signal(str)
@@ -460,18 +462,20 @@ class ImageBase(PlotBase):
self.setProperty("autorange", False)
if style == "simple":
self._color_bar = pg.ColorBarItem(colorMap=self.config.color_map)
cmap = Colors.get_colormap(self.config.color_map)
self._color_bar = pg.ColorBarItem(colorMap=cmap)
self._color_bar.setImageItem(self.layer_manager["main"].image)
self._color_bar.sigLevelsChangeFinished.connect(disable_autorange)
self.config.color_bar = "simple"
elif style == "full":
self._color_bar = pg.HistogramLUTItem()
self._color_bar.setImageItem(self.layer_manager["main"].image)
self._color_bar.gradient.loadPreset(self.config.color_map)
self.config.color_bar = "full"
self._apply_colormap_to_colorbar(self.config.color_map)
self._color_bar.sigLevelsChanged.connect(disable_autorange)
self.plot_widget.addItem(self._color_bar, row=0, col=1)
self.config.color_bar = style
else:
if self._color_bar:
self.plot_widget.removeItem(self._color_bar)
@@ -484,6 +488,37 @@ class ImageBase(PlotBase):
if vrange: # should be at the end to disable the autorange if defined
self.v_range = vrange
def _apply_colormap_to_colorbar(self, color_map: str) -> None:
if not self._color_bar:
return
cmap = Colors.get_colormap(color_map)
if self.config.color_bar == "simple":
self._color_bar.setColorMap(cmap)
return
if self.config.color_bar != "full":
return
gradient = getattr(self._color_bar, "gradient", None)
if gradient is None:
return
positions = np.linspace(0.0, 1.0, self.MAX_TICKS_COLORBAR)
colors = cmap.map(positions, mode="byte")
colors = np.asarray(colors)
if colors.ndim != 2:
return
if colors.shape[1] == 3: # add alpha
alpha = np.full((colors.shape[0], 1), 255, dtype=colors.dtype)
colors = np.concatenate([colors, alpha], axis=1)
ticks = [(float(p), tuple(int(x) for x in c)) for p, c in zip(positions, colors)]
state = {"mode": "rgb", "ticks": ticks}
gradient.restoreState(state)
################################################################################
# Static rois with roi manager
@@ -754,10 +789,7 @@ class ImageBase(PlotBase):
layer.image.color_map = value
if self._color_bar:
if self.config.color_bar == "simple":
self._color_bar.setColorMap(value)
elif self.config.color_bar == "full":
self._color_bar.gradient.loadPreset(value)
self._apply_colormap_to_colorbar(self.config.color_map)
except ValidationError:
return

View File

@@ -119,7 +119,8 @@ class ImageItem(BECConnector, pg.ImageItem):
"""Set a new color map."""
try:
self.config.color_map = value
self.setColorMap(value)
cmap = Colors.get_colormap(self.config.color_map)
self.setColorMap(cmap)
except ValidationError:
logger.error(f"Invalid colormap '{value}' provided.")

View File

@@ -24,6 +24,7 @@ class DeviceSignal(BaseModel):
entry: str
dap: str | None = None
dap_oversample: int = 1
dap_parameters: dict | None = None
model_config: dict = {"validate_assignment": True}
@@ -206,6 +207,7 @@ class Curve(BECConnector, pg.PlotDataItem):
"""
if self.config.source in ["custom", "history"]:
self.setData(x, y)
self.parent_item.request_dap_update.emit()
else:
raise ValueError(f"Source {self.config.source} do not allow custom data setting.")

View File

@@ -1,13 +1,13 @@
from __future__ import annotations
import json
from typing import Literal
from typing import TYPE_CHECKING, Literal
import lmfit
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger, messages
from bec_lib.endpoints import MessageEndpoints
from bec_lib.lmfit_serializer import serialize_lmfit_params, serialize_param_object
from bec_lib.scan_data_container import ScanDataContainer
from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Qt, QTimer, Signal
@@ -41,6 +41,15 @@ from bec_widgets.widgets.services.scan_history_browser.scan_history_browser impo
)
logger = bec_logger.logger
_DAP_PARAM = object()
if TYPE_CHECKING: # pragma: no cover
import lmfit # type: ignore
else:
try:
import lmfit # type: ignore
except Exception: # pragma: no cover
lmfit = None
# noinspection PyDataclass
@@ -697,6 +706,7 @@ class Waveform(PlotBase):
color: str | None = None,
label: str | None = None,
dap: str | None = None,
dap_parameters: dict | lmfit.Parameters | None | object = None,
scan_id: str | None = None,
scan_number: int | None = None,
**kwargs,
@@ -721,6 +731,8 @@ class Waveform(PlotBase):
dap(str): The dap model to use for the curve. When provided, a DAP curve is
attached automatically for device, history, or custom data sources. Use
the same string as the LMFit model name.
dap_parameters(dict | lmfit.Parameters | None): Optional lmfit parameter overrides sent to the DAP server.
Values can be numeric (interpreted as fixed parameters) or dicts like`{"value": 1.0, "vary": False}`.
scan_id(str): Optional scan ID. When provided, the curve is treated as a **history** curve and
the ydata (and optional xdata) are fetched from that historical scan. Such curves are
never cleared by livescan resets.
@@ -733,6 +745,8 @@ class Waveform(PlotBase):
source = "custom"
x_data = None
y_data = None
if dap_parameters is _DAP_PARAM:
dap_parameters = kwargs.pop("dap_parameters", None) or kwargs.pop("parameters", None)
# 1. Custom curve logic
if x is not None and y is not None:
@@ -810,7 +824,9 @@ class Waveform(PlotBase):
curve = self._add_curve(config=config, x_data=x_data, y_data=y_data)
if dap is not None and curve.config.source in ("device", "history", "custom"):
self.add_dap_curve(device_label=curve.name(), dap_name=dap, **kwargs)
self.add_dap_curve(
device_label=curve.name(), dap_name=dap, dap_parameters=dap_parameters, **kwargs
)
return curve
@@ -823,6 +839,7 @@ class Waveform(PlotBase):
dap_name: str,
color: str | None = None,
dap_oversample: int = 1,
dap_parameters: dict | lmfit.Parameters | None = None,
**kwargs,
) -> Curve:
"""
@@ -835,6 +852,7 @@ class Waveform(PlotBase):
dap_name(str): The name of the DAP model to use.
color(str): The color of the curve.
dap_oversample(int): The oversampling factor for the DAP curve.
dap_parameters(dict | lmfit.Parameters | None): Optional lmfit parameter overrides sent to the DAP server.
**kwargs
Returns:
@@ -882,7 +900,11 @@ class Waveform(PlotBase):
# Attach device signal with DAP
config.signal = DeviceSignal(
name=dev_name, entry=dev_entry, dap=dap_name, dap_oversample=dap_oversample
name=dev_name,
entry=dev_entry,
dap=dap_name,
dap_oversample=dap_oversample,
dap_parameters=self._normalize_dap_parameters(dap_parameters),
)
# 4) Create the DAP curve config using `_add_curve(...)`
@@ -1520,7 +1542,7 @@ class Waveform(PlotBase):
self.request_dap_update.emit()
def _check_async_signal_found(self, name: str, signal: str) -> bool:
def _check_async_signal_found(self, name: str, signal: str) -> tuple[bool, str]:
"""
Check if the async signal is found in the BEC device manager.
@@ -1529,13 +1551,16 @@ class Waveform(PlotBase):
signal(str): The entry of the async signal.
Returns:
bool: True if the async signal is found, False otherwise.
tuple[bool, str]: A tuple where the first element is True if the async signal is found (False otherwise),
and the second element is the signal name (either the original signal or the storage_name for AsyncMultiSignal).
"""
bec_async_signals = self.client.device_manager.get_bec_signals("AsyncSignal")
bec_async_signals = self.client.device_manager.get_bec_signals(
["AsyncSignal", "AsyncMultiSignal"]
)
for entry_name, _, entry_data in bec_async_signals:
if entry_name == name and entry_data.get("obj_name") == signal:
return True
return False
return True, entry_data.get("storage_name")
return False, signal
def _setup_async_curve(self, curve: Curve):
"""
@@ -1546,7 +1571,7 @@ class Waveform(PlotBase):
"""
name = curve.config.signal.name
signal = curve.config.signal.entry
async_signal_found = self._check_async_signal_found(name, signal)
async_signal_found, signal = self._check_async_signal_found(name, signal)
try:
curve.clear_data()
@@ -1759,12 +1784,21 @@ class Waveform(PlotBase):
x_min = None
x_max = None
dap_parameters = getattr(dap_curve.config.signal, "dap_parameters", None)
dap_kwargs = {
"data_x": x_data,
"data_y": y_data,
"oversample": dap_curve.dap_oversample,
}
if dap_parameters:
dap_kwargs["parameters"] = dap_parameters
msg = messages.DAPRequestMessage(
dap_cls="LmfitService1D",
dap_type="on_demand",
config={
"args": [],
"kwargs": {"data_x": x_data, "data_y": y_data},
"kwargs": dap_kwargs,
"class_args": model._plugin_info["class_args"],
"class_kwargs": model._plugin_info["class_kwargs"],
"curve_label": dap_curve.name(),
@@ -1773,6 +1807,49 @@ class Waveform(PlotBase):
)
self.client.connector.set_and_publish(MessageEndpoints.dap_request(), msg)
@staticmethod
def _normalize_dap_parameters(parameters: dict | lmfit.Parameters | None) -> dict | None:
"""
Normalize user-provided lmfit parameters into a JSON-serializable dict suitable for the DAP server.
Supports:
- `lmfit.Parameters`
- `dict[name -> number]` (treated as fixed parameter with `vary=False`)
- `dict[name -> dict]` (lmfit.Parameter fields; defaults to `vary=False` if unspecified)
- `dict[name -> lmfit.Parameter]`
"""
if parameters is None:
return None
if lmfit is not None and isinstance(parameters, lmfit.Parameters):
return serialize_lmfit_params(parameters)
if not isinstance(parameters, dict):
if lmfit is None:
raise TypeError(
"dap_parameters must be a dict when lmfit is not installed on the client."
)
raise TypeError("dap_parameters must be a dict or lmfit.Parameters (or omitted).")
normalized: dict[str, dict] = {}
for name, spec in parameters.items():
if spec is None:
continue
if isinstance(spec, (int, float, np.number)):
normalized[name] = {"name": name, "value": float(spec), "vary": False}
continue
if lmfit is not None and isinstance(spec, lmfit.Parameter):
normalized[name] = serialize_param_object(spec)
continue
if isinstance(spec, dict):
normalized[name] = {"name": name, **spec}
if "vary" not in normalized[name]:
normalized[name]["vary"] = False
continue
raise TypeError(
f"Invalid dap_parameters entry for '{name}': expected number, dict, or lmfit.Parameter."
)
return normalized or None
@SafeSlot(dict, dict)
def update_dap_curves(self, msg, metadata):
"""
@@ -1790,14 +1867,6 @@ class Waveform(PlotBase):
if not curve:
return
# Get data from the parent (device) curve
parent_curve = self._find_curve_by_label(curve.config.parent_label)
if parent_curve is None:
return
x_parent, _ = parent_curve.get_data()
if x_parent is None or len(x_parent) == 0:
return
# Retrieve and store the fit parameters and summary from the DAP server response
try:
curve.dap_params = msg["data"][1]["fit_parameters"]
@@ -1806,19 +1875,13 @@ class Waveform(PlotBase):
logger.warning(f"Failed to retrieve DAP data for curve '{curve.name()}'")
return
# Render model according to the DAP model name and parameters
model_name = curve.config.signal.dap
model_function = getattr(lmfit.models, model_name)()
x_min, x_max = x_parent.min(), x_parent.max()
oversample = curve.dap_oversample
new_x = np.linspace(x_min, x_max, int(len(x_parent) * oversample))
# Evaluate the model with the provided parameters to generate the y values
new_y = model_function.eval(**curve.dap_params, x=new_x)
# Update the curve with the new data
curve.setData(new_x, new_y)
# Plot the fitted curve using the server-provided output to avoid requiring lmfit on the client.
try:
fit_data = msg["data"][0]
curve.setData(np.asarray(fit_data["x"]), np.asarray(fit_data["y"]))
except Exception:
logger.exception(f"Failed to plot DAP result for curve '{curve.name()}'")
return
metadata.update({"curve_id": curve_id})
self.dap_params_update.emit(curve.dap_params, metadata)
@@ -2374,8 +2437,48 @@ class DemoApp(QMainWindow): # pragma: no cover
sigma = 0.8
y = amplitude * np.exp(-((x - center) ** 2) / (2 * sigma**2)) + noise
# 1) No explicit parameters: server will use lmfit defaults/guesses.
self.custom_waveform.plot(x=x, y=y, label="custom-gaussian", dap="GaussianModel")
# 2) Easy dict: numbers mean "fix this parameter to value" (vary=False).
self.custom_waveform.plot(
x=x,
y=y,
label="custom-gaussian-fixed-easy",
dap="GaussianModel",
dap_parameters={"amplitude": 1.0},
dap_oversample=5,
)
# 3) lmfit-style dict: any subset of lmfit.Parameter fields.
# Here `center` is not fixed (vary=True) but its initial value is set.
self.custom_waveform.plot(
x=x,
y=y,
label="custom-gaussian-override-dict",
dap="GaussianModel",
dap_parameters={
"center": {"value": 1.2, "vary": True},
"sigma": {"value": sigma, "vary": False, "min": 0.0},
},
)
# 4) Passing a real `lmfit.Parameters` object (optional: requires lmfit on the client).
if lmfit is not None:
params_gauss = lmfit.models.GaussianModel().make_params()
params_gauss["amplitude"].set(value=amplitude, vary=False)
params_gauss["center"].set(value=center, vary=False)
params_gauss["sigma"].set(value=sigma, vary=False, min=0.0)
self.custom_waveform.plot(
x=x,
y=y,
label="custom-gaussian-fixed-params",
dap="GaussianModel",
dap_parameters=params_gauss,
)
else:
logger.info("Skipping lmfit.Parameters demo (lmfit not installed on client).")
if __name__ == "__main__": # pragma: no cover
import sys

View File

@@ -55,7 +55,9 @@ class DeviceBrowser(BECWidget, QWidget):
) -> None:
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.get_bec_shortcuts()
self._config_helper = ConfigHelper(self.client.connector, self.client._service_name)
self._config_helper = ConfigHelper(
self.client.connector, self.client._service_name, self.client.device_manager
)
self._q_threadpool = QThreadPool()
self.ui = None
self.init_ui()

View File

@@ -5,7 +5,7 @@ from bec_lib.atlas_models import Device as DeviceConfigModel
from bec_lib.config_helper import CONF as DEVICE_CONF_KEYS
from bec_lib.config_helper import ConfigHelper
from bec_lib.logger import bec_logger
from pydantic import ValidationError, field_validator
from pydantic import field_validator
from qtpy.QtCore import QSize, Qt, QThreadPool, Signal
from qtpy.QtWidgets import (
QApplication,
@@ -65,7 +65,7 @@ class DeviceConfigDialog(BECWidget, QDialog):
self._initial_config = {}
super().__init__(parent=parent, **kwargs)
self._config_helper = config_helper or ConfigHelper(
self.client.connector, self.client._service_name
self.client.connector, self.client._service_name, self.client.device_manager
)
self._device = device
self._action: Literal["update", "add"] = action

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "2.45.2"
version = "2.45.12"
description = "BEC Widgets"
requires-python = ">=3.11"
classifiers = [

View File

@@ -75,6 +75,7 @@ def test_rpc_plotting_shortcuts_init_configs(qtbot, connected_client_gui_obj):
assert c1._config_dict["signal"] == {
"dap": None,
"name": "bpm4i",
"dap_parameters": None,
"entry": "bpm4i",
"dap_oversample": 1,
}

View File

@@ -12,7 +12,7 @@ from bec_lib.scan_history import ScanHistory
from bec_widgets.tests.utils import DEVICES, DMMock, FakePositioner, Positioner
def fake_redis_server(host, port):
def fake_redis_server(host, port, **kwargs):
redis = fakeredis.FakeRedis()
return redis

View File

@@ -82,6 +82,18 @@ def test_rgba_to_hex():
assert Colors.rgba_to_hex(255, 87, 51) == "#FF5733FF"
def test_validate_color_map_accepts_gradient_presets_and_greys_alias():
presets = {p.lower() for p in Colors.list_available_gradient_presets()}
candidate = next(
(p for p in ("grey", "gray", "bipolar", "spectrum", "flame") if p in presets), None
)
if candidate is None:
pytest.skip("No known GradientEditorItem presets available in this environment.")
assert Colors.validate_color_map(candidate) != ""
assert Colors.get_colormap(candidate) is not None
@pytest.mark.parametrize("num", [10, 100, 400])
def test_evenly_spaced_colors(num):
colors_qcolor = Colors.evenly_spaced_colors(colormap="magma", num=num, format="QColor")

View File

@@ -134,7 +134,7 @@ def test_update_cycle(update_dialog, qtbot):
"deviceClass": "TestDevice",
"deviceConfig": {"param1": "val1"},
"readoutPriority": "monitored",
"description": None,
"description": "",
"readOnly": False,
"softwareTrigger": False,
"onFailure": "retry",

View File

@@ -4,6 +4,7 @@ import numpy as np
import pytest
from bec_lib import messages
from bec_lib.scan_history import ScanHistory
from qtpy.QtCore import QPointF
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap, HeatmapConfig, HeatmapDeviceSignal
@@ -125,12 +126,16 @@ def test_heatmap_get_image_data_unsupported_scan(heatmap_widget):
def test_heatmap_get_grid_scan_image(heatmap_widget):
x_levels = np.linspace(-5, 5, 10).tolist()
y_levels = np.linspace(-5, 5, 10).tolist()
scan_msg = messages.ScanStatusMessage(
scan_id="123",
status="open",
scan_name="grid_scan",
metadata={},
info={"positions": np.random.rand(100, 2).tolist()},
info={
"positions": _grid_positions(slow_levels=x_levels, fast_levels=y_levels, snaked=True)
},
request_inputs={"arg_bundle": ["samx", -5, 5, 10, "samy", -5, 5, 10], "kwargs": {}},
)
heatmap_widget._image_config = HeatmapConfig(
@@ -145,6 +150,111 @@ def test_heatmap_get_grid_scan_image(heatmap_widget):
assert sorted(np.asarray(img, dtype=int).flatten().tolist()) == list(range(100))
def _grid_positions(
*, slow_levels: list[float], fast_levels: list[float], snaked: bool, slow_is_col0: bool = True
) -> list[list[float]]:
positions: list[list[float]] = []
for slow_i, slow_val in enumerate(slow_levels):
row_fast = fast_levels if (not snaked or slow_i % 2 == 0) else list(reversed(fast_levels))
for fast_val in row_fast:
if slow_is_col0:
positions.append([slow_val, fast_val])
else:
positions.append([fast_val, slow_val])
return positions
def test_heatmap_grid_scan_direction_and_snaking_x_fast(heatmap_widget):
heatmap_widget._image_config = HeatmapConfig(
parent_id="parent_id",
x_device=HeatmapDeviceSignal(name="samx", entry="samx"),
y_device=HeatmapDeviceSignal(name="samy", entry="samy"),
z_device=HeatmapDeviceSignal(name="bpm4i", entry="bpm4i"),
color_map="viridis",
)
# x decreases (relative), y increases (relative), x is fast axis
x0 = 10.0
y0 = -3.0
x_levels = (x0 + np.linspace(1.0, -1.0, 3)).tolist()
y_levels = (y0 + np.linspace(-2.0, 2.0, 2)).tolist()
snaked = True
scan_msg = messages.ScanStatusMessage(
scan_id="123",
status="open",
scan_name="grid_scan",
metadata={},
info={
"positions": _grid_positions(slow_levels=y_levels, fast_levels=x_levels, snaked=snaked)
},
request_inputs={
"arg_bundle": ["samy", -2.0, 2.0, 2, "samx", 1.0, -1.0, 3],
"kwargs": {"snaked": snaked, "relative": True},
},
)
img, transform = heatmap_widget.get_grid_scan_image(list(range(6)), msg=scan_msg)
assert img.shape == (3, 2)
assert img[0, 0] == 0 # first point: (x0,y0) in scan order
assert img[2, 1] == 3 # second row first point due to snaking
assert img[0, 1] == 5 # last point in second row
p0 = transform.map(QPointF(0.5, 0.5))
p1 = transform.map(QPointF(2.5, 1.5))
assert p0.x() == pytest.approx(x_levels[0])
assert p0.y() == pytest.approx(y_levels[0])
assert p1.x() == pytest.approx(x_levels[-1])
assert p1.y() == pytest.approx(y_levels[-1])
def test_heatmap_grid_scan_direction_and_snaking_y_fast(heatmap_widget):
heatmap_widget._image_config = HeatmapConfig(
parent_id="parent_id",
x_device=HeatmapDeviceSignal(name="samx", entry="samx"),
y_device=HeatmapDeviceSignal(name="samy", entry="samy"),
z_device=HeatmapDeviceSignal(name="bpm4i", entry="bpm4i"),
color_map="viridis",
)
# x decreases (relative), y increases (relative), y is fast axis
x0 = 1.5
y0 = 22.0
x_levels = (x0 + np.linspace(1.0, -1.0, 3)).tolist()
y_levels = (y0 + np.linspace(-2.0, 2.0, 2)).tolist()
snaked = True
scan_msg = messages.ScanStatusMessage(
scan_id="123",
status="open",
scan_name="grid_scan",
metadata={},
info={
"positions": _grid_positions(slow_levels=x_levels, fast_levels=y_levels, snaked=snaked)
},
request_inputs={
"arg_bundle": ["samx", 1.0, -1.0, 3, "samy", -2.0, 2.0, 2],
"kwargs": {"snaked": snaked, "relative": True},
},
)
img, transform = heatmap_widget.get_grid_scan_image(list(range(6)), msg=scan_msg)
assert img.shape == (3, 2)
assert img[0, 0] == 0
# For y-fast scans, snaking reverses the y index on every odd x row.
assert img[1, 1] == 2
assert img[1, 0] == 3
p0 = transform.map(QPointF(0.5, 0.5))
p1 = transform.map(QPointF(2.5, 1.5))
assert p0.x() == pytest.approx(x_levels[0])
assert p0.y() == pytest.approx(y_levels[0])
assert p1.x() == pytest.approx(x_levels[-1])
assert p1.y() == pytest.approx(y_levels[-1])
def test_heatmap_get_step_scan_image(heatmap_widget):
scan_msg = messages.ScanStatusMessage(
@@ -193,12 +303,16 @@ def test_heatmap_update_plot(heatmap_widget):
color_map="viridis",
)
heatmap_widget.scan_item = create_dummy_scan_item()
x_levels = np.linspace(-5, 5, 10).tolist()
y_levels = np.linspace(-5, 5, 10).tolist()
heatmap_widget.scan_item.status_message = messages.ScanStatusMessage(
scan_id="123",
status="open",
scan_name="grid_scan",
metadata={},
info={"positions": np.random.rand(100, 2).tolist()},
info={
"positions": _grid_positions(slow_levels=x_levels, fast_levels=y_levels, snaked=True)
},
request_inputs={"arg_bundle": ["samx", -5, 5, 10, "samy", -5, 5, 10], "kwargs": {}},
)
with mock.patch.object(heatmap_widget.main_image, "setImage") as mock_set_image:

View File

@@ -191,51 +191,10 @@ def test_bec_weblinks(monkeypatch):
assert opened_urls == [
"https://beamline-experiment-control.readthedocs.io/en/latest/",
"https://bec.readthedocs.io/projects/bec-widgets/en/latest/",
"https://gitlab.psi.ch/groups/bec/-/issues/",
"https://github.com/bec-project/bec_widgets/issues",
]
#################################################################
# Tests for scanprogress bar animations
def test_scan_progress_bar_show_animation(qtbot, bec_main_window):
"""
_show_scan_progress_bar should animate the container's maximumWidth
from 0 to the configured target width.
"""
container = bec_main_window._scan_progress_bar_with_separator
# Precondition: collapsed
assert container.maximumWidth() == 0
bec_main_window._show_scan_progress_bar()
target = bec_main_window._scan_progress_bar_target_width
qtbot.waitUntil(lambda: container.maximumWidth() == target, timeout=2000)
assert container.maximumWidth() == target
def test_scan_progress_bar_hide_animation(qtbot, bec_main_window):
"""
_animate_hide_scan_progress_bar should collapse the container back to 0 width.
"""
container = bec_main_window._scan_progress_bar_with_separator
# First expand it
bec_main_window._show_scan_progress_bar()
target = bec_main_window._scan_progress_bar_target_width
qtbot.waitUntil(lambda: container.maximumWidth() == target, timeout=2000)
# Trigger hide animation
bec_main_window._animate_hide_scan_progress_bar()
qtbot.waitUntil(lambda: container.maximumWidth() == 0, timeout=2000)
assert container.maximumWidth() == 0
#################################################################
# Tests for hover widget and tooltip behaviour

View File

@@ -496,6 +496,13 @@ def test_add_dap_curve_custom_source(qtbot, mocked_client_with_dap):
assert dap_curve.config.signal.dap == "GaussianModel"
def test_curve_set_data_emits_dap_update(qtbot, mocked_client):
wf = create_widget(qtbot, Waveform, client=mocked_client)
c = wf.plot(x=[1, 2, 3], y=[4, 5, 6], label="test_curve")
with qtbot.waitSignal(wf.request_dap_update):
c.set_data([7, 8, 9], [10, 11, 12])
def test_plot_custom_curve_with_inline_dap(qtbot, mocked_client_with_dap):
"""
Supplying the `dap` kwarg when plotting custom data should auto-create the fit curve.
@@ -509,6 +516,57 @@ def test_plot_custom_curve_with_inline_dap(qtbot, mocked_client_with_dap):
assert dap_curve.config.signal.dap == "GaussianModel"
def test_normalize_dap_parameters_number_dict():
normalized = Waveform._normalize_dap_parameters({"amplitude": 1.0, "center": 2})
assert normalized == {
"amplitude": {"name": "amplitude", "value": 1.0, "vary": False},
"center": {"name": "center", "value": 2.0, "vary": False},
}
def test_normalize_dap_parameters_dict_spec_defaults_vary_false():
normalized = Waveform._normalize_dap_parameters({"sigma": {"value": 0.8, "min": 0.0}})
assert normalized["sigma"]["name"] == "sigma"
assert normalized["sigma"]["value"] == 0.8
assert normalized["sigma"]["min"] == 0.0
assert normalized["sigma"]["vary"] is False
def test_normalize_dap_parameters_invalid_type_raises():
with pytest.raises(TypeError):
Waveform._normalize_dap_parameters(["amplitude", 1.0]) # type: ignore[arg-type]
def test_request_dap_includes_normalized_parameters(qtbot, mocked_client_with_dap, monkeypatch):
wf = create_widget(qtbot, Waveform, client=mocked_client_with_dap)
curve = wf.plot(
x=[0, 1, 2],
y=[1, 2, 3],
label="custom-inline-params",
dap="GaussianModel",
dap_parameters={"amplitude": 1.0},
)
dap_curve = wf.get_curve(f"{curve.name()}-GaussianModel")
assert dap_curve is not None
dap_curve.dap_oversample = 3
captured = {}
def capture(topic, msg, *args, **kwargs): # noqa: ARG001
captured["topic"] = topic
captured["msg"] = msg
monkeypatch.setattr(wf.client.connector, "set_and_publish", capture)
wf.request_dap()
msg = captured["msg"]
dap_kwargs = msg.content["config"]["kwargs"]
assert dap_kwargs["oversample"] == 3
assert dap_kwargs["parameters"] == {
"amplitude": {"name": "amplitude", "value": 1.0, "vary": False}
}
def test_fetch_scan_data_and_access(qtbot, mocked_client, monkeypatch):
"""
Test the _fetch_scan_data_and_access method returns live_data/val if in a live scan,