csaxs_bec/devices/epics/allied_vision_camera.py

@@ -0,0 +1,161 @@
+"""Module for the EPICS integration of the AlliedVision Camera via Vimba SDK."""
+
+import threading
+import traceback
+from enum import IntEnum
+
+import numpy as np
+from bec_lib.logger import bec_logger
+from ophyd import Component as Cpt, Kind, Signal
+from ophyd.areadetector import ADComponent as ADCpt
+from ophyd.areadetector import DetectorBase
+from ophyd_devices import PreviewSignal
+from ophyd_devices.devices.areadetector.cam import VimbaDetectorCam
+from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35 as ImagePlugin
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+from typeguard import typechecked
+
+logger = bec_logger.logger
+
+
+class ACQUIRE_MODES(IntEnum):
+    """Acquiring enums for Allied Vision Camera"""
+
+    ACQUIRING = 1
+    DONE = 0
+
+
+class AlliedVisionCamera(PSIDeviceBase, DetectorBase):
+    """
+    Epics Area Detector interface for the Allied Vision Alvium G1-507m camera via Vimba SDK.
+    The IOC runs with under the prefix: 'X12SA-GIGECAM-AV1:'.
+    """
+
+    USER_ACCESS = ["start_live_mode", "stop_live_mode"]
+
+    cam = ADCpt(VimbaDetectorCam, "cam1:")
+    image = ADCpt(ImagePlugin, "image1:")
+
+    preview = Cpt(
+        PreviewSignal,
+        name="preview",
+        ndim=2,
+        num_rotation_90=0,
+        transpose=False,
+        doc="Preview signal of the AlliedVision camera.",
+    )
+
+    live_mode_enabled = Cpt(
+        Signal,
+        name="live_mode_enabled",
+        value=False,
+        doc="Enable or disable live mode.",
+        kind=Kind.config,
+    )
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        prefix: str,
+        poll_rate: int = 5,
+        num_rotation_90: int = 0,
+        transpose: bool = False,
+        scan_info=None,
+        device_manager=None,
+        **kwargs,
+    ):
+        super().__init__(
+            name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
+        )
+        self._poll_thread = threading.Thread(
+            target=self._poll_array_data, daemon=True, name=f"{self.name}_poll_thread"
+        )
+        self._poll_thread_kill_event = threading.Event()
+        self._poll_start_event = threading.Event()
+        if poll_rate <= 0:
+            logger.warning(
+                f"Poll rate must be positive for Camera {self.name} and non-zero, setting to 1 Hz."
+            )
+            poll_rate = 1
+            self.stop_live_mode()
+        elif poll_rate > 10:
+            logger.warning(f"Poll rate too high for Camera {self.name}, setting to 10 Hz max.")
+            poll_rate = 10
+        self._poll_rate = poll_rate
+        self._unique_array_id = 0
+        self._pv_timeout = 2.0
+        self.image: ImagePlugin
+        self.preview.num_rotation_90 = num_rotation_90
+        self.preview.transpose = transpose
+        self._live_mode_lock = threading.RLock()
+        self.live_mode_enabled.subscribe(self._on_live_mode_enabled_changed, run=False)
+        self.cam.acquire.subscribe(self._on_live_mode_enabled_changed, run=False)
+
+    def start_live_mode(self) -> None:
+        """Start live mode."""
+        self.live_mode_enabled.put(True)
+
+    def stop_live_mode(self) -> None:
+        """Stop live mode."""
+        self.live_mode_enabled.put(False)
+
+    def _on_live_mode_enabled_changed(self, *args, value, **kwargs) -> None:
+        self._apply_live_mode(bool(value))
+
+    def _apply_live_mode(self, enabled: bool) -> None:
+        with self._live_mode_lock:
+            if enabled:
+                if not self._poll_start_event.is_set():
+                    self._poll_start_event.set()
+                    self.cam.acquire.put(ACQUIRE_MODES.ACQUIRING.value)  # Start acquisition
+                else:
+                    logger.info(f"Live mode already started for {self.name}.")
+                return
+
+            if self._poll_start_event.is_set():
+                self._poll_start_event.clear()
+                self.cam.acquire.put(ACQUIRE_MODES.DONE.value)  # Stop acquisition
+            else:
+                logger.info(f"Live mode already stopped for {self.name}.")
+
+    def on_connected(self):
+        """Reset the unique array ID on connection."""
+        self.cam.array_counter.set(0).wait(timeout=self._pv_timeout)
+        self.cam.array_callbacks.set(1).wait(timeout=self._pv_timeout)
+        self._poll_thread.start()
+
+    def _poll_array_data(self):
+        """Poll the array data for preview updates."""
+        while self._poll_start_event.wait():
+            while not self._poll_thread_kill_event.wait(1 / self._poll_rate):
+                try:
+                    # First check if there is a new image
+                    if self.image.unique_id.get() != self._unique_array_id:
+                        self._unique_array_id = self.image.unique_id.get()
+                    else:
+                        continue  # No new image, skip update
+                    # Get new image data
+                    value = self.image.array_data.get()
+                    if value is None:
+                        logger.info(f"No image data available for preview of {self.name}")
+                        continue
+
+                    array_size = self.image.array_size.get()
+                    if array_size[0] == 0:  # 2D image, not color image
+                        array_size = array_size[1:]
+                    # Geometry correction for the image
+                    data = np.reshape(value, array_size)
+                    self.preview.put(data)
+                except Exception:  # pylint: disable=broad-except
+                    content = traceback.format_exc()
+                    logger.error(
+                        f"Error while polling array data for preview of {self.name}: {content}"
+                    )
+
+    def on_destroy(self):
+        """Stop the polling thread on destruction."""
+        self._poll_start_event.set()
+        self._poll_thread_kill_event.set()
+        if self._poll_thread.is_alive():
+            self._poll_thread.join(timeout=2)

csaxs_bec/devices/ids_cameras/base_integration/camera.py

@@ -156,7 +156,6 @@ class Camera:
         camera_id (int): The ID of the camera device.
         m_n_colormode (Literal[0, 1, 2, 3]): Color mode for the camera.
         bits_per_pixel (Literal[8, 24]): Number of bits per pixel for the camera.
-        live_mode (bool): Whether to enable live mode for the camera.
     """
 
     def __init__(

csaxs_bec/devices/ids_cameras/ids_camera.py

@@ -3,21 +3,19 @@
 from __future__ import annotations
 
 import threading
-import time
-from typing import TYPE_CHECKING, Literal, Tuple, TypedDict
+from typing import TYPE_CHECKING, Literal
 
 import numpy as np
+from ophyd import Component as Cpt, Signal, Kind
+
 from bec_lib import messages
 from bec_lib.logger import bec_logger
-from ophyd import Component as Cpt
+from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 from ophyd_devices.utils.bec_signals import AsyncSignal, PreviewSignal
 
-from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera
-
 if TYPE_CHECKING:
     from bec_lib.devicemanager import ScanInfo
-    from pydantic import ValidationInfo
 
 logger = bec_logger.logger
 
@@ -45,8 +43,15 @@ class IDSCamera(PSIDeviceBase):
         doc="Signal for the region of interest (ROI).",
         async_update={"type": "add", "max_shape": [None]},
     )
+    live_mode_enabled = Cpt(
+        Signal,
+        name="live_mode_enabled",
+        value=False,
+        doc="Enable or disable live mode.",
+        kind=Kind.config,
+    )
 
-    USER_ACCESS = ["live_mode", "mask", "set_rect_roi", "get_last_image"]
+    USER_ACCESS = ["start_live_mode", "stop_live_mode", "mask", "set_rect_roi", "get_last_image"]
 
     def __init__(
         self,
@@ -83,15 +88,22 @@ class IDSCamera(PSIDeviceBase):
             bits_per_pixel=bits_per_pixel,
             connect=False,
         )
-        self._live_mode = False
         self._inputs = {"live_mode": live_mode}
         self._mask = np.zeros((1, 1), dtype=np.uint8)
         self.image.num_rotation_90 = num_rotation_90
         self.image.transpose = transpose
         self._force_monochrome = force_monochrome
+        self.live_mode_enabled.subscribe(self._on_live_mode_enabled_changed, run=False)
+        self.live_mode_enabled.put(bool(live_mode))
 
     ############## Live Mode Methods ##############
 
+    def start_live_mode(self) -> None:
+        self.live_mode_enabled.put(True)
+
+    def stop_live_mode(self) -> None:
+        self.live_mode_enabled.put(False)
+
     @property
     def mask(self) -> np.ndarray:
         """Return the current region of interest (ROI) for the camera."""
@@ -114,22 +126,15 @@ class IDSCamera(PSIDeviceBase):
             )
         self._mask = value
 
-    @property
-    def live_mode(self) -> bool:
-        """Return whether the camera is in live mode."""
-        return self._live_mode
-
-    @live_mode.setter
-    def live_mode(self, value: bool):
-        """Set the live mode for the camera."""
-        if value != self._live_mode:
-            if self.cam._connected is False:  # $ pylint: disable=protected-access
-                self.cam.on_connect()
-            self._live_mode = value
-            if value:
-                self._start_live()
-            else:
-                self._stop_live()
+    def _on_live_mode_enabled_changed(self, *args, value, **kwargs):
+        """Callback for when live mode is changed."""
+        enabled = bool(value)
+        if enabled and self.cam._connected is False:  # pylint: disable=protected-access
+            self.cam.on_connect()
+        if enabled:
+            self._start_live()
+        else:
+            self._stop_live()
 
     def set_rect_roi(self, x: int, y: int, width: int, height: int):
         """Set the rectangular region of interest (ROI) for the camera."""
@@ -196,7 +201,7 @@ class IDSCamera(PSIDeviceBase):
         """Connect to the camera."""
         self.cam.force_monochrome = self._force_monochrome
         self.cam.on_connect()
-        self.live_mode = self._inputs.get("live_mode", False)
+        self.live_mode_enabled.put(bool(self._inputs.get("live_mode", False)))
         self.set_rect_roi(0, 0, self.cam.cam.width.value, self.cam.cam.height.value)
 
     def on_destroy(self):
@@ -206,7 +211,7 @@ class IDSCamera(PSIDeviceBase):
 
     def on_trigger(self):
         """Handle the trigger event."""
-        if not self.live_mode:
+        if not bool(self.live_mode_enabled.get()):
             return
         image = self.image.get()
         if image is not None:

csaxs_bec/devices/omny/webcam_viewer.py

@@ -2,7 +2,7 @@ import requests
 import threading
 import cv2
 import numpy as np
-from ophyd import Device, Component as Cpt
+from ophyd import Device, Component as Cpt, Kind, Signal
 from ophyd_devices import PreviewSignal
 import traceback 
 
@@ -13,6 +13,13 @@ logger = bec_logger.logger
 class WebcamViewer(Device):
     USER_ACCESS = ["start_live_mode", "stop_live_mode"]
     preview = Cpt(PreviewSignal, ndim=2, num_rotation_90=0, transpose=False)
+    live_mode_enabled = Cpt(
+        Signal,
+        name="live_mode_enabled",
+        value=False,
+        doc="Enable or disable live mode.",
+        kind=Kind.config,
+    )
 
     def __init__(self, url:str, name:str, num_rotation_90=0, transpose=False, **kwargs) -> None:
         super().__init__(name=name, **kwargs)
@@ -21,20 +28,54 @@ class WebcamViewer(Device):
         self._update_thread = None
         self._buffer = b""
         self._shutdown_event = threading.Event()
+        self._live_mode_lock = threading.RLock()
         self.preview.num_rotation_90 = num_rotation_90
         self.preview.transpose = transpose
+        self.live_mode_enabled.subscribe(self._on_live_mode_enabled_changed, run=False)
 
     def start_live_mode(self) -> None:
-        if self._connection is not None:
-            return
-        self._update_thread = threading.Thread(target=self._update_loop, daemon=True)
-        self._update_thread.start()
+        self.live_mode_enabled.put(True)
+
+    def stop_live_mode(self) -> None:
+        self.live_mode_enabled.put(False)
+
+    def _on_live_mode_enabled_changed(self, *args, value, **kwargs) -> None:
+        self._apply_live_mode(bool(value))
+
+    def _apply_live_mode(self, enabled: bool) -> None:
+        with self._live_mode_lock:
+            if enabled:
+                if self._update_thread is not None and self._update_thread.is_alive():
+                    return
+                self._shutdown_event.clear()
+                self._update_thread = threading.Thread(target=self._update_loop, daemon=True)
+                self._update_thread.start()
+                return
+
+            if self._update_thread is None:
+                return
+            self._shutdown_event.set()
+            if self._connection is not None:
+                try:
+                    self._connection.close()
+                except Exception:  # pylint: disable=broad-except
+                    pass
+                self._connection = None
+            self._update_thread.join(timeout=2)
+            if self._update_thread.is_alive():
+                logger.warning("Webcam live mode thread did not stop within timeout.")
+                return
+            self._update_thread = None
+            self._buffer = b""
+            self._shutdown_event.clear()
 
     def _update_loop(self) -> None:
         while not self._shutdown_event.is_set():
             try:
-                self._connection = requests.get(self.url, stream=True)
+                self._connection = requests.get(self.url, stream=True, timeout=5)
                 for chunk in self._connection.iter_content(chunk_size=1024):
+                    if self._shutdown_event.is_set():
+                        break
                     self._buffer += chunk
                     start = self._buffer.find(b'\xff\xd8') # JPEG start
                     end = self._buffer.find(b'\xff\xd9') # JPEG end
@@ -50,16 +91,3 @@ class WebcamViewer(Device):
             except Exception as exc:
                 content = traceback.format_exc()
                 logger.error(f"Image update loop failed: {content}")
-
-    def stop_live_mode(self) -> None:
-        if self._connection is None:
-            return
-        self._shutdown_event.set()
-        if self._connection is not None:
-            self._connection.close()
-            self._connection = None
-        if self._update_thread is not None:
-            self._update_thread.join()
-            self._update_thread = None
-
-        self._shutdown_event.clear()

tests/tests_devices/test_ids_camera.py

@@ -54,7 +54,7 @@ def test_on_connected_sets_mask_and_live_mode(ids_camera):
 
 def test_on_trigger_roi_signal(ids_camera):
     """Test the on_trigger method to ensure it processes the ROI signal correctly."""
-    ids_camera.live_mode = True
+    ids_camera.start_live_mode()
     test_image = np.array([[2, 4], [6, 8]])
     test_mask = np.array([[1, 0], [0, 1]], dtype=np.uint8)
     ids_camera.mask = test_mask