1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-19 06:45:36 +02:00

Compare commits

...

12 Commits

Author SHA1 Message Date
semantic-release
03474cf7f7 2.12.4
Automatically generated by python-semantic-release
2025-06-10 14:42:40 +00:00
9ef418bf55 fix(image_roi): coordinates are emitted correctly when handles are inverted; closes #672 2025-06-10 16:41:59 +02:00
b3ce68070d ci: add stale issue job 2025-06-06 14:48:10 +02:00
semantic-release
784b54af6e 2.12.3
Automatically generated by python-semantic-release
2025-06-05 19:07:20 +00:00
3740ac8e32 build: update min dependency of bec to 3.38 2025-06-05 21:06:32 +02:00
edfac87868 fix(crosshair): use objectName instead of config for retrieving the monitor name 2025-06-05 21:06:32 +02:00
271116453d fix(image): preview signals can be used in Image widget; update logic adjusted; closes #683 2025-06-05 21:06:32 +02:00
12f5233745 fix(device_combobox): tuple entries of preview signals are checked in DeviceComboBoxes just for the relevant device 2025-06-05 21:06:32 +02:00
semantic-release
392ddf9d1a 2.12.2
Automatically generated by python-semantic-release
2025-06-05 13:27:05 +00:00
85705383e4 fix(waveform): safeguard for history data access, closes #571; removed return values "none" 2025-06-05 15:26:19 +02:00
semantic-release
224863569f 2.12.1
Automatically generated by python-semantic-release
2025-06-05 12:07:35 +00:00
3e2544e52a fix(crosshair): emitted name from crosshair 2D is objectName of image or its id 2025-06-05 14:04:44 +02:00
12 changed files with 466 additions and 94 deletions

15
.github/workflows/stale-issues.yml vendored Normal file
View File

@@ -0,0 +1,15 @@
name: 'Close stale issues and PRs'
on:
schedule:
- cron: '00 10 * * *'
jobs:
stale:
runs-on: ubuntu-latest
steps:
- uses: actions/stale@v9
with:
stale-issue-message: 'This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 7 days.'
stale-pr-message: 'This PR is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 7 days.'
days-before-stale: 60
days-before-close: 7

View File

@@ -1,6 +1,55 @@
# CHANGELOG
## v2.12.4 (2025-06-10)
### Bug Fixes
- **image_roi**: Coordinates are emitted correctly when handles are inverted; closes #672
([`9ef418b`](https://github.com/bec-project/bec_widgets/commit/9ef418bf5597d4be77adc3c0c88c1c1619c9aa2f))
### Continuous Integration
- Add stale issue job
([`b3ce680`](https://github.com/bec-project/bec_widgets/commit/b3ce68070d58cdd76559cbd7db04cdbcc6c1f075))
## v2.12.3 (2025-06-05)
### Bug Fixes
- **crosshair**: Use objectName instead of config for retrieving the monitor name
([`edfac87`](https://github.com/bec-project/bec_widgets/commit/edfac87868605b4b755f7732b2841673de53bc3f))
- **device_combobox**: Tuple entries of preview signals are checked in DeviceComboBoxes just for the
relevant device
([`12f5233`](https://github.com/bec-project/bec_widgets/commit/12f523374586d55499f80baf56a50b6ef486cd43))
- **image**: Preview signals can be used in Image widget; update logic adjusted; closes #683
([`2711164`](https://github.com/bec-project/bec_widgets/commit/271116453d1ef5316b19457d04613b6ddc939cdb))
### Build System
- Update min dependency of bec to 3.38
([`3740ac8`](https://github.com/bec-project/bec_widgets/commit/3740ac8e325a489d59faca648896ffcea29e1a02))
## v2.12.2 (2025-06-05)
### Bug Fixes
- **waveform**: Safeguard for history data access, closes #571; removed return values "none"
([`8570538`](https://github.com/bec-project/bec_widgets/commit/85705383e4aff2f83f76d342db0a13380aeca42f))
## v2.12.1 (2025-06-05)
### Bug Fixes
- **crosshair**: Emitted name from crosshair 2D is objectName of image or its id
([`3e2544e`](https://github.com/bec-project/bec_widgets/commit/3e2544e52a84b30a5acb4a7874025fa359a3c58d))
## v2.12.0 (2025-06-04)
### Bug Fixes

View File

@@ -1459,12 +1459,12 @@ class Image(RPCBase):
@rpc_call
def image(
self,
monitor: "str | None" = None,
monitor: "str | tuple | None" = None,
monitor_type: "Literal['auto', '1d', '2d']" = "auto",
color_map: "str | None" = None,
color_bar: "Literal['simple', 'full'] | None" = None,
vrange: "tuple[int, int] | None" = None,
) -> "ImageItem":
) -> "ImageItem | None":
"""
Set the image source and update the image.

View File

@@ -312,7 +312,7 @@ class Crosshair(QObject):
y_values[name] = closest_y
x_values[name] = closest_x
elif isinstance(item, pg.ImageItem): # 2D plot
name = item.config.monitor or str(id(item))
name = item.objectName() or str(id(item))
image_2d = item.image
if image_2d is None:
continue
@@ -400,7 +400,7 @@ class Crosshair(QObject):
)
self.coordinatesChanged1D.emit(coordinate_to_emit)
elif isinstance(item, pg.ImageItem):
name = item.config.monitor or str(id(item))
name = item.objectName() or str(id(item))
x, y = x_snap_values[name], y_snap_values[name]
if x is None or y is None:
continue
@@ -458,7 +458,7 @@ class Crosshair(QObject):
)
self.coordinatesClicked1D.emit(coordinate_to_emit)
elif isinstance(item, pg.ImageItem):
name = item.config.monitor or str(id(item))
name = item.objectName() or str(id(item))
x, y = x_snap_values[name], y_snap_values[name]
if x is None or y is None:
continue

View File

@@ -149,6 +149,25 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
self._is_valid_input = False
self.update()
def validate_device(self, device: str) -> bool: # type: ignore[override]
"""
Extend validation so that previewsignal pseudodevices (labels like
``"eiger_preview"``) are accepted as valid choices.
The validation run only on device not on the previewsignal.
Args:
device: The text currently entered/selected.
Returns:
True if the device is a genuine BEC device *or* one of the
whitelisted previewsignal entries.
"""
idx = self.findText(device)
if idx >= 0 and isinstance(self.itemData(idx), tuple):
device = self.itemData(idx)[0] # type: ignore[assignment]
return super().validate_device(device)
if __name__ == "__main__": # pragma: no cover
# pylint: disable=import-outside-toplevel

View File

@@ -35,7 +35,7 @@ class ImageConfig(ConnectionConfig):
class ImageLayerConfig(BaseModel):
monitor: str | None = Field(None, description="The name of the monitor.")
monitor: str | tuple | None = Field(None, description="The name of the monitor.")
monitor_type: Literal["1d", "2d", "auto"] = Field("auto", description="The type of monitor.")
source: Literal["device_monitor_1d", "device_monitor_2d", "auto"] = Field(
"auto", description="The source of the image data."
@@ -179,12 +179,12 @@ class Image(ImageBase):
@SafeSlot(popup_error=True)
def image(
self,
monitor: str | None = None,
monitor: str | tuple | None = None,
monitor_type: Literal["auto", "1d", "2d"] = "auto",
color_map: str | None = None,
color_bar: Literal["simple", "full"] | None = None,
vrange: tuple[int, int] | None = None,
) -> ImageItem:
) -> ImageItem | None:
"""
Set the image source and update the image.
@@ -201,21 +201,13 @@ class Image(ImageBase):
if self.subscriptions["main"].monitor:
self.disconnect_monitor(self.subscriptions["main"].monitor)
self.entry_validator.validate_monitor(monitor)
self.subscriptions["main"].monitor = monitor
if monitor_type == "1d":
self.subscriptions["main"].source = "device_monitor_1d"
self.subscriptions["main"].monitor_type = "1d"
elif monitor_type == "2d":
self.subscriptions["main"].source = "device_monitor_2d"
self.subscriptions["main"].monitor_type = "2d"
elif monitor_type == "auto":
self.subscriptions["main"].source = "auto"
logger.warning(
f"Updates for '{monitor}' will be fetch from both 1D and 2D monitor endpoints."
)
self.subscriptions["main"].monitor_type = "auto"
if monitor is None or monitor == "":
logger.warning(f"No monitor specified, cannot set image, old monitor is unsubscribed")
return None
if isinstance(monitor, tuple):
self.entry_validator.validate_monitor(monitor[0])
else:
self.entry_validator.validate_monitor(monitor)
self.set_image_update(monitor=monitor, type=monitor_type)
if color_map is not None:
@@ -240,7 +232,12 @@ class Image(ImageBase):
self.selection_bundle.dim_combo_box,
):
combo.blockSignals(True)
self.selection_bundle.device_combo_box.set_device(config.monitor)
if isinstance(config.monitor, tuple):
self.selection_bundle.device_combo_box.setCurrentText(
f"{config.monitor[0]}_{config.monitor[1]}"
)
else:
self.selection_bundle.device_combo_box.setCurrentText(config.monitor)
self.selection_bundle.dim_combo_box.setCurrentText(config.monitor_type)
for combo in (
self.selection_bundle.device_combo_box,
@@ -340,7 +337,8 @@ class Image(ImageBase):
########################################
# Connections
def set_image_update(self, monitor: str, type: Literal["1d", "2d", "auto"]):
@SafeSlot()
def set_image_update(self, monitor: str | tuple, type: Literal["1d", "2d", "auto"]):
"""
Set the image update method for the given monitor.
@@ -350,37 +348,95 @@ class Image(ImageBase):
"""
# TODO consider moving connecting and disconnecting logic to Image itself if multiple images
if type == "1d":
self.bec_dispatcher.connect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
elif type == "2d":
self.bec_dispatcher.connect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
elif type == "auto":
self.bec_dispatcher.connect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
self.bec_dispatcher.connect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
if isinstance(monitor, tuple):
device = self.dev[monitor[0]]
signal = monitor[1]
if len(monitor) == 3:
signal_config = monitor[2]
else:
signal_config = device._info["signals"][signal]
signal_class = signal_config.get("signal_class", None)
if signal_class != "PreviewSignal":
logger.warning(f"Signal '{monitor}' is not a PreviewSignal.")
return
ndim = signal_config.get("describe", None).get("signal_info", None).get("ndim", None)
if ndim is None:
logger.warning(
f"Signal '{monitor}' does not have a valid 'ndim' in its signal_info."
)
return
if ndim == 1:
self.bec_dispatcher.connect_slot(
self.on_image_update_1d, MessageEndpoints.device_preview(device.name, signal)
)
self.subscriptions["main"].source = "device_monitor_1d"
self.subscriptions["main"].monitor_type = "1d"
elif ndim == 2:
self.bec_dispatcher.connect_slot(
self.on_image_update_2d, MessageEndpoints.device_preview(device.name, signal)
)
self.subscriptions["main"].source = "device_monitor_2d"
self.subscriptions["main"].monitor_type = "2d"
else: # FIXME old monitor 1d/2d endpoint handling, present for backwards compatibility, will be removed in future versions
if type == "1d":
self.bec_dispatcher.connect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
self.subscriptions["main"].source = "device_monitor_1d"
self.subscriptions["main"].monitor_type = "1d"
elif type == "2d":
self.bec_dispatcher.connect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
self.subscriptions["main"].source = "device_monitor_2d"
self.subscriptions["main"].monitor_type = "2d"
elif type == "auto":
self.bec_dispatcher.connect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
self.bec_dispatcher.connect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
self.subscriptions["main"].source = "auto"
logger.warning(
f"Updates for '{monitor}' will be fetch from both 1D and 2D monitor endpoints."
)
self.subscriptions["main"].monitor_type = "auto"
logger.info(f"Connected to {monitor} with type {type}")
self.subscriptions["main"].monitor = monitor
def disconnect_monitor(self, monitor: str):
def disconnect_monitor(self, monitor: str | tuple):
"""
Disconnect the monitor from the image update signals, both 1D and 2D.
Args:
monitor(str): The name of the monitor to disconnect.
monitor(str|tuple): The name of the monitor to disconnect, or a tuple of (device, signal) for preview signals.
"""
self.bec_dispatcher.disconnect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
self.bec_dispatcher.disconnect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
if isinstance(monitor, tuple):
if self.subscriptions["main"].source == "device_monitor_1d":
self.bec_dispatcher.disconnect_slot(
self.on_image_update_1d, MessageEndpoints.device_preview(monitor[0], monitor[1])
)
elif self.subscriptions["main"].source == "device_monitor_2d":
self.bec_dispatcher.disconnect_slot(
self.on_image_update_2d, MessageEndpoints.device_preview(monitor[0], monitor[1])
)
else:
logger.warning(
f"Cannot disconnect monitor {monitor} with source {self.subscriptions['main'].source}"
)
return
else: # FIXME old monitor 1d/2d endpoint handling, present for backwards compatibility, will be removed in future versions
self.bec_dispatcher.disconnect_slot(
self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor)
)
self.bec_dispatcher.disconnect_slot(
self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor)
)
self.subscriptions["main"].monitor = None
self._sync_device_selection()

View File

@@ -1,5 +1,5 @@
from bec_lib.device import ReadoutPriority
from qtpy.QtCore import Qt
from qtpy.QtCore import Qt, QTimer
from qtpy.QtWidgets import QComboBox, QStyledItemDelegate
from bec_widgets.utils.error_popups import SafeSlot
@@ -50,11 +50,58 @@ class MonitorSelectionToolbarBundle(ToolbarBundle):
self.add_action("dim_combo", WidgetAction(widget=self.dim_combo_box, adjust_size=False))
# Connect slots, a device will be connected upon change of any combobox
self.device_combo_box.currentTextChanged.connect(lambda: self.connect_monitor())
self.dim_combo_box.currentTextChanged.connect(lambda: self.connect_monitor())
self.device_combo_box.currentTextChanged.connect(self.connect_monitor)
self.dim_combo_box.currentTextChanged.connect(self.connect_monitor)
QTimer.singleShot(0, self._adjust_and_connect)
def _adjust_and_connect(self):
"""
Adjust the size of the device combo box and populate it with preview signals.
Has to be done with QTimer.singleShot to ensure the UI is fully initialized, needed for testing.
"""
self._populate_preview_signals()
self._reverse_device_items()
self.device_combo_box.setCurrentText("") # set again default to empty string
def _populate_preview_signals(self) -> None:
"""
Populate the device combo box with previewsignal devices in the
format '<device>_<signal>' and store the tuple(device, signal) in
the item's userData for later use.
"""
preview_signals = self.target_widget.client.device_manager.get_bec_signals("PreviewSignal")
for device, signal, signal_config in preview_signals:
label = signal_config.get("obj_name", f"{device}_{signal}")
self.device_combo_box.addItem(label, (device, signal, signal_config))
def _reverse_device_items(self) -> None:
"""
Reverse the current order of items in the device combo box while
keeping their userData and restoring the previous selection.
"""
current_text = self.device_combo_box.currentText()
items = [
(self.device_combo_box.itemText(i), self.device_combo_box.itemData(i))
for i in range(self.device_combo_box.count())
]
self.device_combo_box.clear()
for text, data in reversed(items):
self.device_combo_box.addItem(text, data)
if current_text:
self.device_combo_box.setCurrentText(current_text)
@SafeSlot()
def connect_monitor(self):
def connect_monitor(self, *args, **kwargs):
"""
Connect the target widget to the selected monitor based on the current device and dimension.
If the selected device is a preview-signal device, it will use the tuple (device, signal) as the monitor.
"""
dim = self.dim_combo_box.currentText()
self.target_widget.image(monitor=self.device_combo_box.currentText(), monitor_type=dim)
data = self.device_combo_box.currentData()
if isinstance(data, tuple):
self.target_widget.image(monitor=data, monitor_type="auto")
else:
self.target_widget.image(monitor=self.device_combo_box.currentText(), monitor_type=dim)

View File

@@ -437,6 +437,23 @@ class RectangularROI(BaseROI, pg.RectROI):
self.hoverPen = fn.mkPen(color=(255, 0, 0), width=3, style=QtCore.Qt.DashLine)
self.handleHoverPen = fn.mkPen("lime", width=4)
def _normalized_edges(self) -> tuple[float, float, float, float]:
"""
Return rectangle edges as (left, bottom, right, top) with consistent
ordering even when the ROI has been inverted by its scale handles.
Returns:
tuple: A tuple containing the left, bottom, right, and top edges
of the ROI rectangle in normalized coordinates.
"""
x0, y0 = self.pos().x(), self.pos().y()
w, h = self.state["size"]
x_left = min(x0, x0 + w)
x_right = max(x0, x0 + w)
y_bottom = min(y0, y0 + h)
y_top = max(y0, y0 + h)
return x_left, y_bottom, x_right, y_top
def add_scale_handle(self):
"""
Add scale handles at every corner and edge of the ROI.
@@ -465,17 +482,15 @@ class RectangularROI(BaseROI, pg.RectROI):
def _on_region_changed(self):
"""
Handles ROI region change events.
Handles changes to the ROI's region.
This method is called whenever the ROI's position or size changes.
It calculates the new corner coordinates and emits the edgesChanged signal
with the updated coordinates.
"""
x0, y0 = self.pos().x(), self.pos().y()
w, h = self.state["size"]
self.edgesChanged.emit(x0, y0, x0 + w, y0 + h)
viewBox = self.parent_plot_item.vb
viewBox.update()
x_left, y_bottom, x_right, y_top = self._normalized_edges()
self.edgesChanged.emit(x_left, y_bottom, x_right, y_top)
self.parent_plot_item.vb.update()
def mouseDragEvent(self, ev):
"""
@@ -489,9 +504,8 @@ class RectangularROI(BaseROI, pg.RectROI):
"""
super().mouseDragEvent(ev)
if ev.isFinish():
x0, y0 = self.pos().x(), self.pos().y()
w, h = self.state["size"]
self.edgesReleased.emit(x0, y0, x0 + w, y0 + h)
x_left, y_bottom, x_right, y_top = self._normalized_edges()
self.edgesReleased.emit(x_left, y_bottom, x_right, y_top)
def get_coordinates(self, typed: bool | None = None) -> dict | tuple:
"""
@@ -510,17 +524,16 @@ class RectangularROI(BaseROI, pg.RectROI):
if typed is None:
typed = self.description
x0, y0 = self.pos().x(), self.pos().y()
w, h = self.state["size"]
x1, y1 = x0 + w, y0 + h
x_left, y_bottom, x_right, y_top = self._normalized_edges()
if typed:
return {
"bottom_left": (x0, y0),
"bottom_right": (x1, y0),
"top_left": (x0, y1),
"top_right": (x1, y1),
"bottom_left": (x_left, y_bottom),
"bottom_right": (x_right, y_bottom),
"top_left": (x_left, y_top),
"top_right": (x_right, y_top),
}
return ((x0, y0), (x1, y0), (x0, y1), (x1, y1))
return (x_left, y_bottom), (x_right, y_bottom), (x_left, y_top), (x_right, y_top)
def _lookup_scene_image(self):
"""
@@ -654,7 +667,7 @@ class CircularROI(BaseROI, pg.CircleROI):
if typed is None:
typed = self.description
d = self.state["size"][0]
d = abs(self.state["size"][0])
cx = self.pos().x() + d / 2
cy = self.pos().y() + d / 2

View File

@@ -1,7 +1,7 @@
from __future__ import annotations
import json
from typing import Literal
from typing import Any, Literal
import lmfit
import numpy as np
@@ -163,7 +163,7 @@ class Waveform(PlotBase):
self._async_curves = []
self._slice_index = None
self._dap_curves = []
self._mode: Literal["none", "sync", "async", "mixed"] = "none"
self._mode = None
# Scan data
self._scan_done = True # means scan is not running
@@ -1139,7 +1139,7 @@ class Waveform(PlotBase):
QTimer.singleShot(100, self.update_sync_curves)
QTimer.singleShot(300, self.update_sync_curves)
def _fetch_scan_data_and_access(self):
def _fetch_scan_data_and_access(self) -> tuple[dict, str] | tuple[None, None]:
"""
Decide whether the widget is in live or historical mode
and return the appropriate data dict and access key.
@@ -1153,7 +1153,7 @@ class Waveform(PlotBase):
self.update_with_scan_history(-1)
if self.scan_item is None:
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none", "none"
return None, None
if hasattr(self.scan_item, "live_data"):
# Live scan
@@ -1169,7 +1169,7 @@ class Waveform(PlotBase):
"""
if self.scan_item is None:
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none"
return
data, access_key = self._fetch_scan_data_and_access()
for curve in self._sync_curves:
device_name = curve.config.signal.name
@@ -1177,9 +1177,8 @@ class Waveform(PlotBase):
if access_key == "val":
device_data = data.get(device_name, {}).get(device_entry, {}).get(access_key, None)
else:
device_data = (
data.get(device_name, {}).get(device_entry, {}).read().get("value", None)
)
entry_obj = data.get(device_name, {}).get(device_entry)
device_data = entry_obj.read()["value"] if entry_obj else None
x_data = self._get_x_data(device_name, device_entry)
if x_data is not None:
if len(x_data) == 1:
@@ -1217,7 +1216,8 @@ class Waveform(PlotBase):
if self._skip_large_dataset_check is False:
if not self._check_dataset_size_and_confirm(dataset_obj, device_entry):
continue # user declined to load; skip this curve
device_data = dataset_obj.get(device_entry, {}).read().get("value", None)
entry_obj = dataset_obj.get(device_entry, None)
device_data = entry_obj.read()["value"] if entry_obj else None
# if shape is 2D cast it into 1D and take the last waveform
if len(np.shape(device_data)) > 1:
@@ -1549,15 +1549,21 @@ class Waveform(PlotBase):
if access_key == "val": # live data
x_data = data.get(x_name, {}).get(x_entry, {}).get(access_key, [0])
else: # history data
x_data = data.get(x_name, {}).get(x_entry, {}).read().get("value", [0])
entry_obj = data.get(x_name, {}).get(x_entry)
x_data = entry_obj.read()["value"] if entry_obj else [0]
new_suffix = f" (custom: {x_name}-{x_entry})"
# 2 User wants timestamp
if self.x_axis_mode["name"] == "timestamp":
if access_key == "val": # live
timestamps = data[device_name][device_entry].timestamps
x_data = data.get(device_name, {}).get(device_entry, None)
if x_data is None:
return None
else:
timestamps = x_data.timestamps
else: # history data
timestamps = data[device_name][device_entry].read().get("timestamp", [0])
entry_obj = data.get(device_name, {}).get(device_entry)
timestamps = entry_obj.read()["timestamp"] if entry_obj else [0]
x_data = timestamps
new_suffix = " (timestamp)"
@@ -1584,7 +1590,8 @@ class Waveform(PlotBase):
if access_key == "val":
x_data = data.get(x_name, {}).get(x_entry, {}).get(access_key, None)
else:
x_data = data.get(x_name, {}).get(x_entry, {}).read().get("value", None)
entry_obj = data.get(x_name, {}).get(x_entry)
x_data = entry_obj.read()["value"] if entry_obj else None
new_suffix = f" (auto: {x_name}-{x_entry})"
self._update_x_label_suffix(new_suffix)
return x_data
@@ -1637,7 +1644,7 @@ class Waveform(PlotBase):
self.update_with_scan_history(-1)
if self.scan_item is None:
logger.info("No scan executed so far; skipping device curves categorisation.")
return "none"
return None
if hasattr(self.scan_item, "live_data"):
readout_priority = self.scan_item.status_message.info["readout_priority"] # live data

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "2.12.0"
version = "2.12.4"
description = "BEC Widgets"
requires-python = ">=3.10"
classifiers = [
@@ -13,15 +13,15 @@ classifiers = [
"Topic :: Scientific/Engineering",
]
dependencies = [
"bec_ipython_client>=2.21.4, <=4.0", # needed for jupyter console
"bec_lib>=3.29, <=4.0",
"bec_ipython_client>=3.38, <=4.0", # needed for jupyter console
"bec_lib>=3.38, <=4.0",
"bec_qthemes~=0.7, >=0.7",
"black~=25.0", # needed for bw-generate-cli
"isort~=5.13, >=5.13.2", # needed for bw-generate-cli
"black~=25.0", # needed for bw-generate-cli
"isort~=5.13, >=5.13.2", # needed for bw-generate-cli
"pydantic~=2.0",
"pyqtgraph~=0.13",
"PySide6~=6.8.2",
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtpy~=2.4",
]

View File

@@ -29,7 +29,6 @@ def image_widget_with_crosshair(qtbot):
image_item = pg.ImageItem()
image_item.setImage(np.random.rand(100, 100))
image_item.config = type("obj", (object,), {"monitor": "test"})
widget.addItem(image_item)
plot_item = widget.getPlotItem()
@@ -99,6 +98,7 @@ def test_mouse_moved_signals_outside(plot_widget_with_crosshair):
def test_mouse_moved_signals_2D(image_widget_with_crosshair):
crosshair, plot_item = image_widget_with_crosshair
image_item = plot_item.items[0]
emitted_values_2D = []
@@ -113,7 +113,7 @@ def test_mouse_moved_signals_2D(image_widget_with_crosshair):
crosshair.mouse_moved(event_mock)
assert emitted_values_2D == [("test", 21, 55)]
assert emitted_values_2D == [(str(id(image_item)), 21, 55)]
def test_mouse_moved_signals_2D_outside(image_widget_with_crosshair):

View File

@@ -113,6 +113,75 @@ def test_enable_colorbar_with_vrange(qtbot, mocked_client, colorbar_type):
assert bec_image_view._color_bar is not None
##############################################
# Previewsignal update mechanism
def test_image_setup_preview_signal_1d(qtbot, mocked_client, monkeypatch):
"""
Ensure that calling .image() with a (device, signal, config) tuple representing
a 1D PreviewSignal connects using the 1D path and updates correctly.
"""
import numpy as np
view = create_widget(qtbot, Image, client=mocked_client)
signal_config = {
"obj_name": "waveform1d_img",
"signal_class": "PreviewSignal",
"describe": {"signal_info": {"ndim": 1}},
}
# Set the image monitor to the preview signal
view.image(monitor=("waveform1d", "img", signal_config))
# Subscriptions should indicate 1D preview connection
sub = view.subscriptions["main"]
assert sub.source == "device_monitor_1d"
assert sub.monitor_type == "1d"
assert sub.monitor == ("waveform1d", "img", signal_config)
# Simulate a waveform update from the dispatcher
waveform = np.arange(25, dtype=float)
view.on_image_update_1d({"data": waveform}, {"scan_id": "scan_test"})
assert view.main_image.raw_data.shape == (1, 25)
np.testing.assert_array_equal(view.main_image.raw_data[0], waveform)
def test_image_setup_preview_signal_2d(qtbot, mocked_client, monkeypatch):
"""
Ensure that calling .image() with a (device, signal, config) tuple representing
a 2D PreviewSignal connects using the 2D path and updates correctly.
"""
import numpy as np
view = create_widget(qtbot, Image, client=mocked_client)
signal_config = {
"obj_name": "eiger_img2d",
"signal_class": "PreviewSignal",
"describe": {"signal_info": {"ndim": 2}},
}
# Set the image monitor to the preview signal
view.image(monitor=("eiger", "img2d", signal_config))
# Subscriptions should indicate 2D preview connection
sub = view.subscriptions["main"]
assert sub.source == "device_monitor_2d"
assert sub.monitor_type == "2d"
assert sub.monitor == ("eiger", "img2d", signal_config)
# Simulate a 2D image update
test_data = np.arange(16, dtype=float).reshape(4, 4)
view.on_image_update_2d({"data": test_data}, {})
np.testing.assert_array_equal(view.main_image.image, test_data)
##############################################
# Device monitor endpoint update mechanism
def test_image_setup_image_2d(qtbot, mocked_client):
bec_image_view = create_widget(qtbot, Image, client=mocked_client)
bec_image_view.image(monitor="eiger", monitor_type="2d")
@@ -167,6 +236,10 @@ def test_image_data_update_1d(qtbot, mocked_client):
assert bec_image_view.main_image.raw_data.shape == (2, 60)
##############################################
# Toolbar and Actions Tests
def test_toolbar_actions_presence(qtbot, mocked_client):
bec_image_view = create_widget(qtbot, Image, client=mocked_client)
assert "autorange_image" in bec_image_view.toolbar.widgets
@@ -484,3 +557,96 @@ def test_roi_plot_data_from_image(qtbot, mocked_client):
# Horizontal slice (row)
h_slice, _ = y_items[0].getData()
np.testing.assert_array_equal(h_slice, test_data[2])
##############################################
# MonitorSelectionToolbarBundle specific tests
##############################################
def test_monitor_selection_reverse_device_items(qtbot, mocked_client):
"""
Verify that _reverse_device_items correctly reverses the order of items in the
device combobox while preserving the current selection.
"""
view = create_widget(qtbot, Image, client=mocked_client)
bundle = view.selection_bundle
combo = bundle.device_combo_box
# Replace existing items with a deterministic list
combo.clear()
combo.addItem("samx", 1)
combo.addItem("samy", 2)
combo.addItem("samz", 3)
combo.setCurrentText("samy")
# Reverse the items
bundle._reverse_device_items()
# Order should be reversed and selection preserved
assert [combo.itemText(i) for i in range(combo.count())] == ["samz", "samy", "samx"]
assert combo.currentText() == "samy"
def test_monitor_selection_populate_preview_signals(qtbot, mocked_client, monkeypatch):
"""
Verify that _populate_preview_signals adds previewsignal devices to the combobox
with the correct userData.
"""
view = create_widget(qtbot, Image, client=mocked_client)
bundle = view.selection_bundle
# Provide a deterministic fake device_manager with get_bec_signals
class _FakeDM:
def get_bec_signals(self, _filter):
return [
("eiger", "img", {"obj_name": "eiger_img"}),
("async_device", "img2", {"obj_name": "async_device_img2"}),
]
monkeypatch.setattr(view.client, "device_manager", _FakeDM())
initial_count = bundle.device_combo_box.count()
bundle._populate_preview_signals()
# Two new entries should have been added
assert bundle.device_combo_box.count() == initial_count + 2
# The first newly added item should carry tuple userData describing the device/signal
data = bundle.device_combo_box.itemData(initial_count)
assert isinstance(data, tuple) and data[0] == "eiger"
def test_monitor_selection_adjust_and_connect(qtbot, mocked_client, monkeypatch):
"""
Verify that _adjust_and_connect performs the full setup:
fills the combobox with preview signals,
reverses their order,
and resets the currentText to an empty string.
"""
view = create_widget(qtbot, Image, client=mocked_client)
bundle = view.selection_bundle
# Deterministic fake device_manager
class _FakeDM:
def get_bec_signals(self, _filter):
return [("eiger", "img", {"obj_name": "eiger_img"})]
monkeypatch.setattr(view.client, "device_manager", _FakeDM())
combo = bundle.device_combo_box
# Start from a clean state
combo.clear()
combo.addItem("", None)
combo.setCurrentText("")
# Execute the method under test
bundle._adjust_and_connect()
# Expect exactly two items: preview label followed by the empty default
assert combo.count() == 2
# Because of the reversal, the preview label comes first
assert combo.itemText(0) == "eiger_img"
# Current selection remains empty
assert combo.currentText() == ""