bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 19:21:50 +02:00
Code Activity

fix(image): preview signals can be used in Image widget; update logic adjusted; closes #683

Browse Source
This commit is contained in:
wyzula_j
2025-06-04 15:30:18 +02:00
committed by Jan Wyzula
parent 12f5233745
commit 271116453d
4 changed files with 320 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bec_widgets/cli/client.py
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.

98
bec_widgets/widgets/plots/image/image.py
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)
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.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"
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,14 +348,51 @@ class Image(ImageBase):
"""
# TODO consider moving connecting and disconnecting logic to Image itself if multiple images
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)
@ -365,16 +400,37 @@ class Image(ImageBase):
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.
"""
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)
)

57
bec_widgets/widgets/plots/image/toolbar_bundles/image_selection.py
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 preview‑signal 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()
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)

166
tests/unit_tests/test_image_view_next_gen.py
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
##############################################
# Preview‑signal 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 1‑D PreviewSignal connects using the 1‑D 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 1‑D 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 2‑D PreviewSignal connects using the 2‑D 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 2‑D 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 2‑D 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 combo‑box 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 preview‑signal devices to the combo‑box
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 set‑up:
‑ fills the combo‑box 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() == ""