wip adapting for the additional signals

bec_widgets/widgets/plots/image/image.py

@@ -98,6 +98,7 @@ class Image(ImageBase):
         "remove_roi",
         "rois",
     ]
+    SUPPORTED_SIGNALS = ["AsyncSignal", "AsyncMultiSignal", "DynamicSignal"]
 
     def __init__(
         self,
@@ -208,15 +209,15 @@ class Image(ImageBase):
         self.device_combo_box.insertItem(base_count, "", None)
 
         preview_signals = self.client.device_manager.get_bec_signals("PreviewSignal")
-        async_signals = self.client.device_manager.get_bec_signals("AsyncSignal")
+        async_signals = self.client.device_manager.get_bec_signals(self.SUPPORTED_SIGNALS)
         all_signals = preview_signals + async_signals
         for device, signal, signal_config in all_signals:
             describe = signal_config.get("describe") or {}
             signal_info = describe.get("signal_info") or {}
-            ndim = signal_info.get("ndim")
+            ndim = signal_info.get("ndim", 0)
             if ndim == 0:
                 continue
-            label = signal_config.get("obj_name", f"{device}_{signal}")
+            label = signal_config.get("storage_name", f"{device}_{signal}")
             self.device_combo_box.addItem(label, (device, signal, signal_config))
         self.device_combo_box.setCurrentText("")
         self.device_combo_box.blockSignals(False)
@@ -457,7 +458,8 @@ class Image(ImageBase):
             except KeyError:
                 logger.warning(f"Device '{monitor[0]}' not found; cannot connect monitor.")
                 return
-            signal = monitor[1]
+            # signal = monitor[1]
+            signal = self._check_async_signal_found(monitor[0], monitor[1])
             if len(monitor) == 3:
                 signal_config = monitor[2]
             else:
@@ -467,8 +469,11 @@ class Image(ImageBase):
                     logger.warning(f"Signal '{signal}' not found on device '{device.name}'.")
                     return
             signal_class = signal_config.get("signal_class", None)
-            if signal_class not in ["PreviewSignal", "AsyncSignal"]:
-                logger.warning(f"Signal '{monitor}' is not a PreviewSignal or AsyncSignal.")
+            allowed_signal_classes = ["PreviewSignal"] + self.SUPPORTED_SIGNALS
+            if signal_class not in allowed_signal_classes:
+                logger.warning(
+                    f"Signal `{monitor}` is not a PreviewSignal or a supported async signal."
+                )
                 return
 
             describe = signal_config.get("describe") or {}
@@ -487,7 +492,7 @@ class Image(ImageBase):
                         self.on_image_update_1d,
                         MessageEndpoints.device_preview(device.name, signal),
                     )
-                elif signal_class == "AsyncSignal":
+                elif signal_class in self.SUPPORTED_SIGNALS:
                     self.async_update = True
                     needs_async_setup = True
                     config.async_signal_name = signal_config.get(
@@ -504,7 +509,7 @@ class Image(ImageBase):
                         self.on_image_update_2d,
                         MessageEndpoints.device_preview(device.name, signal),
                     )
-                elif signal_class == "AsyncSignal":
+                elif signal_class in self.SUPPORTED_SIGNALS:
                     self.async_update = True
                     needs_async_setup = True
                     config.async_signal_name = signal_config.get(
@@ -603,9 +608,29 @@ class Image(ImageBase):
         if monitor is None or not isinstance(monitor, (list, tuple)) or len(monitor) < 2:
             return None
         device_name = monitor[0]
-        async_signal = config.async_signal_name or monitor[1]
+        async_signal = self._check_async_signal_found(
+            name=device_name, signal=config.async_signal_name or monitor[1]
+        )
         return device_name, async_signal
 
+    def _check_async_signal_found(self, name: str, signal: str) -> str:
+        """
+        Check if the async signal is found in the BEC device manager.
+
+        Args:
+            name(str): The name of the async signal.
+            signal(str): The entry of the async signal.
+
+        Returns:
+            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(self.SUPPORTED_SIGNALS)
+        for entry_name, _, entry_data in bec_async_signals:
+            if entry_name == name and entry_data.get("obj_name") == signal:
+                return entry_data.get("storage_name")
+        return signal
+
     def _setup_async_image(self, scan_id: str | None):
         """
         (Re)connect async image readback for the current scan.

tests/unit_tests/test_image_view_next_gen.py

@@ -716,17 +716,50 @@ def test_monitor_selection_populate_signals(qtbot, mocked_client, monkeypatch):
     """
     view = create_widget(qtbot, Image, client=mocked_client)
 
+    signal_configs = {
+        "PreviewSignal": [
+            ("eiger", "img", {"obj_name": "eiger_img", "describe": {"signal_info": {"ndim": 2}}}),
+            (
+                "eiger2",
+                "img2",
+                {"obj_name": "eiger_img2", "describe": {"signal_info": {"ndim": 2}}},
+            ),
+        ],
+        "AsyncSignal": [
+            (
+                "async_device",
+                "img_async",
+                {"obj_name": "async_device_img_async", "describe": {"signal_info": {"ndim": 2}}},
+            )
+        ],
+        "AsyncMultiSignal": [
+            (
+                "multi_device",
+                "img_multi",
+                {"obj_name": "multi_device_img_multi", "describe": {"signal_info": {"ndim": 2}}},
+            )
+        ],
+        "DynamicSignal": [
+            (
+                "dynamic_device",
+                "img_dyn",
+                {"obj_name": "dynamic_device_img_dyn", "describe": {"signal_info": {"ndim": 2}}},
+            )
+        ],
+    }
+
     # Provide a deterministic fake device_manager with get_bec_signals
     class _FakeDM:
         def get_bec_signals(self, _filter):
-            if _filter == "PreviewSignal":
-                return [
-                    ("eiger", "img", {"obj_name": "eiger_img"}),
-                    ("eiger2", "img2", {"obj_name": "eiger_img2"}),
-                ]
-            if _filter == "AsyncSignal":
-                return [("async_device", "img_async", {"obj_name": "async_device_img_async"})]
-            return []
+            if isinstance(_filter, str):
+                filters = [_filter]
+            else:
+                filters = list(_filter)
+
+            signals = []
+            for filt in filters:
+                signals.extend(signal_configs.get(filt, []))
+            return signals
 
     monkeypatch.setattr(view.client, "device_manager", _FakeDM())
 
@@ -748,14 +781,27 @@ def test_monitor_selection_populate_signals(qtbot, mocked_client, monkeypatch):
         assert isinstance(data, tuple)
         signal_texts.append(text)
 
-    assert {"eiger_img", "eiger_img2", "async_device_img_async"}.issubset(set(signal_texts))
+    expected_labels = {
+        "eiger_img",
+        "eiger_img2",
+        "async_device_img_async",
+        "multi_device_img_multi",
+        "dynamic_device_img_dyn",
+    }
+    assert expected_labels.issubset(set(signal_texts))
     first_signal_idx = next(
         i
         for i in range(view.device_combo_box.count())
         if isinstance(view.device_combo_box.itemData(i), tuple)
     )
     data = view.device_combo_box.itemData(first_signal_idx)
-    assert isinstance(data, tuple) and data[0] in ["eiger", "eiger2", "async_device"]
+    assert isinstance(data, tuple) and data[0] in [
+        "eiger",
+        "eiger2",
+        "async_device",
+        "multi_device",
+        "dynamic_device",
+    ]
 
 
 def test_monitor_selection_adjust_and_connect(qtbot, mocked_client, monkeypatch):
@@ -770,9 +816,26 @@ def test_monitor_selection_adjust_and_connect(qtbot, mocked_client, monkeypatch)
     # Deterministic fake device_manager
     class _FakeDM:
         def get_bec_signals(self, _filter):
-            if _filter == "PreviewSignal":
-                return [("eiger", "img", {"obj_name": "eiger_img"})]
-            return []
+            if isinstance(_filter, str):
+                filters = [_filter]
+            else:
+                filters = list(_filter)
+
+            signals = []
+            for filt in filters:
+                if filt == "PreviewSignal":
+                    signals.extend(
+                        [
+                            (
+                                "eiger",
+                                "img",
+                                {"obj_name": "eiger_img", "describe": {"signal_info": {"ndim": 2}}},
+                            )
+                        ]
+                    )
+                else:
+                    signals.extend([])
+            return signals
 
     monkeypatch.setattr(view.client, "device_manager", _FakeDM())