csaxs_bec/device_configs/bl_detectors.yaml

@@ -38,3 +38,16 @@ eiger_1_5:
 #   softwareTrigger: True
 #   deviceTags:
 #     - bl_detectors
+
+# allied_vision_cam:
+#   description: AlliedVision Alvium G1-507m camera for live image acquisition (native Aravis/GigE Vision, no EPICS IOC)
+#   deviceClass: csaxs_bec.devices.allied_vision_cameras.AlliedVisionAravisCamera
+#   deviceConfig:
+#     camera_id: "Allied Vision-Alvium G1-507m-06J41"   # Aravis device ID; resolved via discovery, independent of current IP
+#     live_mode: True
+#   onFailure: raise
+#   enabled: true
+#   readoutPriority: async
+#   softwareTrigger: True
+#   deviceTags:
+#     - bl_detectors

csaxs_bec/device_configs/ptycho_omny.yaml

@@ -61,6 +61,19 @@ cam203:
   deviceTags:
     - ptycho_omny
 
+# allied_vision_cam:
+#   description: AlliedVision Alvium G1-507m camera for live image acquisition (native Aravis/GigE Vision, no EPICS IOC)
+#   deviceClass: csaxs_bec.devices.allied_vision_cameras.AlliedVisionAravisCamera
+#   deviceConfig:
+#     camera_id: "Allied Vision-Alvium G1-507m-06J41"   # Aravis device ID; resolved via discovery, independent of current IP
+#     live_mode: True
+#   onFailure: raise
+#   enabled: true
+#   readoutPriority: async
+#   softwareTrigger: True
+#   deviceTags:
+#     - ptycho_omny
+
 ############################################################
 #################### OMNY RT motors ########################
 ############################################################

csaxs_bec/devices/allied_vision_cameras/README.md

@@ -0,0 +1,177 @@
+# Allied Vision camera (native Aravis integration)
+
+`AlliedVisionAravisCamera` is the native-Python replacement for the old EPICS/Vimba-IOC
+`AlliedVisionCamera` (`csaxs_bec/devices/epics/allied_vision_camera.py`, still present but
+unused/unwired). It talks to Allied Vision GigE cameras (e.g. the Alvium G1-507m, mono,
+Sony IMX264) directly via **Aravis**, the open-source GenICam/GigE Vision library — no
+vendor SDK, account, or EPICS IOC/areaDetector driver involved. Aravis implements the GigE
+Vision network protocol (GVCP/GVSP) itself, so it works with any GenICam-compliant GigE
+Vision camera, not just Allied Vision's.
+
+```
+allied_vision_cameras/
+├── __init__.py                     # exports AlliedVisionAravisCamera
+├── allied_vision_aravis_camera.py  # ophyd device (PSIDeviceBase)
+└── base_integration/
+    └── camera.py                    # low-level Aravis wrapper (class Camera)
+```
+
+Simulation counterpart: `SimAlliedVisionAravisCamera` in `csaxs_bec/devices/sim/sim_cameras.py`
+(same device, `self.cam` swapped for a synthetic-frame backend — no hardware or Aravis
+needed).
+
+## Packages required
+
+- **Unit tests / import** (`tests/tests_devices/test_allied_vision_aravis_camera.py`):
+  nothing extra. `Camera.__init__` guards the `import gi` / `gi.require_version("Aravis",
+  "0.8")` and the tests mock `camera.cam` directly, so
+  `pytest tests/tests_devices/test_allied_vision_aravis_camera.py` runs on a plain dev
+  environment, same as the IDS camera tests run without `pyueye`.
+- **Talking to real hardware** — two things, both local to the machine running the device
+  server (or wherever you instantiate `AlliedVisionAravisCamera`/`Camera` directly):
+  1. **PyGObject** (`pygobject`, the `gi` module) — already listed in `pyproject.toml`,
+     pinned to `<3.52`. It's the GObject Introspection bridge, not Aravis itself. The pin
+     matters: PyGObject 3.52+ requires `girepository-2.0` (GLib >=2.80 /
+     gobject-introspection >=1.80), which RHEL9's system packages don't provide (only the
+     older `girepository-1.0` via `gobject-introspection` 1.68) — `pip install pygobject`
+     unpinned fails at build time with `Dependency 'girepository-2.0' is required but not
+     found`. Unpin once deployment hosts move to a newer GLib/gobject-introspection.
+  2. **Aravis** — a native C library with a GObject Introspection typelib. It is *not* a
+     pip package; it comes from the system package manager (or building from source), same
+     class of local, non-pip dependency as the Vimba X SDK would have been, or the DLLs
+     `pyueye` needs for the IDS cameras.
+
+### Installing PyGObject + Aravis (Linux)
+
+`pygobject` alone (`pip install pygobject`, already in `pyproject.toml`) only gets you the
+`gi` bridge — `import gi; gi.require_version("Aravis", "0.8")` will raise until the Aravis
+library and its typelib are installed on the system.
+
+1. **Debian/Ubuntu** — Aravis and its GObject Introspection bindings are packaged directly:
+   ```bash
+   sudo apt install libaravis-0.8-0 gir1.2-aravis-0.8 aravis-tools
+   ```
+   `aravis-tools` provides `arv-tool-0.8` and `arv-viewer-0.8`, useful for discovery/testing
+   without writing any Python. If PyGObject's own system dependencies aren't already
+   present (GLib/GObject introspection headers), add `python3-gi` or
+   `sudo apt install libgirepository1.0-dev` before `pip install pygobject`.
+2. **RHEL/Fedora/CentOS** (relevant if the device-server host runs one of these):
+   ```bash
+   sudo dnf install aravis gobject-introspection-devel python3-gobject
+   ```
+   Package name/availability varies by distro version — if `aravis` isn't in the default
+   repos, enable EPEL or build from source (see below).
+3. **Building Aravis from source** (if no packaged version is new enough — GigE features
+   and camera compatibility improve across Aravis releases):
+   ```bash
+   sudo apt install meson ninja-build libglib2.0-dev libgirepository1.0-dev \
+       gobject-introspection libxml2-dev
+   git clone https://github.com/AravisProject/aravis.git
+   cd aravis && meson setup build && ninja -C build
+   sudo ninja -C build install && sudo ldconfig
+   ```
+   This installs the typelib to the standard GI search path, so no environment variable
+   juggling is normally needed (unlike Vimba X's `GENICAM_GENTL64_PATH` step).
+
+No transport-layer install step is needed beyond this — Aravis's GigE Vision support is
+built into the library itself, not a separate producer plugin.
+
+Verify the install:
+```bash
+python3 -c "import gi; gi.require_version('Aravis', '0.8'); from gi.repository import Aravis; print(Aravis)"
+python3 -c "
+import gi; gi.require_version('Aravis', '0.8')
+from gi.repository import Aravis
+Aravis.update_device_list()
+print([Aravis.get_device_id(i) for i in range(Aravis.get_n_devices())])
+"
+```
+The second command should list any camera currently reachable on the network (empty list
+is fine if none are powered/connected — it just confirms Aravis loaded and can broadcast
+discovery without error). `arv-tool-0.8 --list` (from `aravis-tools`) does the same check
+from the shell.
+
+Reference: [Aravis project (GitHub)](https://github.com/AravisProject/aravis),
+[Aravis reference manual](https://aravisproject.github.io/aravis/aravis-stable/),
+[Aravis.Camera Python bindings](https://lazka.github.io/pgi-docs/Aravis-0.8/classes/Camera.html).
+
+## Finding the camera's ID
+
+`camera_id` (constructor arg / `deviceConfig` key) is the Aravis device ID string — not an
+IP address, and not the same format vmbpy would have used. For GigE Vision devices it's
+typically `"<Vendor>-<Model>-<Serial>"` (e.g.
+`"Allied Vision Technologies-50-0536885929"`). It's resolved via GigE Vision discovery
+broadcast, so it works regardless of the camera's current DHCP-assigned IP.
+
+Once PyGObject + Aravis are installed on a machine on the same network segment as the
+camera, list connected cameras and their IDs with:
+
+```bash
+arv-tool-0.8 --list
+```
+
+or from Python:
+
+```python
+import gi; gi.require_version("Aravis", "0.8")
+from gi.repository import Aravis
+Aravis.update_device_list()
+for i in range(Aravis.get_n_devices()):
+    print(Aravis.get_device_id(i), Aravis.get_device_vendor(i), Aravis.get_device_model(i))
+```
+
+`arv-viewer-0.8` (also from `aravis-tools`) is the GUI equivalent — lists every camera it
+discovers on the network along with its ID, the quickest way to check without writing any
+code. The ID is derived from the camera's MAC address and doesn't change when the IP does,
+which is why it's the right identifier here given the camera has no fixed IP.
+
+## Testing without hardware
+
+- Run the unit tests: `pytest tests/tests_devices/test_allied_vision_aravis_camera.py`.
+  These mock `camera.cam`, so they cover the ophyd-level logic (mask/ROI, live-mode
+  threading, `on_trigger` ROI-signal math, exposure/gain delegation) without touching
+  Aravis.
+- Exercise the full signal/preview/live-mode path with no camera at all via
+  `SimAlliedVisionAravisCamera` (`csaxs_bec/devices/sim/sim_cameras.py`) — it serves a
+  static synthetic test-pattern frame (with optional gaussian noise via `sim_noise_std`)
+  through the same `image`/`roi_signal` signals and `start_live_mode`/`get_last_image`/
+  `set_rect_roi` USER_ACCESS methods as the real device.
+
+## Testing with real hardware
+
+1. Install PyGObject + Aravis as above.
+2. Find the camera's ID (see above).
+3. Uncomment and adjust the `allied_vision_cam` block in
+   `csaxs_bec/device_configs/bl_detectors.yaml` with the real `camera_id`, or instantiate
+   directly:
+   ```python
+   from csaxs_bec.devices.allied_vision_cameras import AlliedVisionAravisCamera
+   cam = AlliedVisionAravisCamera(
+       name="allied_vision_cam", camera_id="Allied Vision Technologies-50-0536885929"
+   )
+   cam.on_connected()          # opens the camera via Aravis
+   cam.start_live_mode()       # begins the 5 Hz polling thread, updates `image` preview signal
+   cam.get_last_image()        # last captured frame as a numpy array
+   cam.get_exposure_time() / cam.set_exposure_time(us)
+   cam.get_gain() / cam.set_gain(value)
+   cam.stop_live_mode()
+   cam.on_destroy()            # drops the Aravis camera reference
+   ```
+4. Once wired into a running BEC deployment via `bl_detectors.yaml`, verify from an
+   ipython client session that the device shows up in `dm.devices` and that
+   `dev.allied_vision_cam.start_live_mode()` produces a live preview.
+
+## Notes / open items
+
+- `Camera.get_image_data()` sets the camera into single-frame acquisition mode on every
+  call (`Aravis.Camera.acquisition()`'s documented behavior) rather than running a
+  continuous stream — matches the polling architecture of `_live_mode_loop` (5 Hz) but is
+  slightly less efficient per-frame than a persistent stream would be. Fine at this poll
+  rate; would need revisiting for high-frame-rate use cases.
+- The exact Python method names used here (`Camera.acquisition`, `get_region`,
+  `get_width_increment`, `set_pixel_format_from_string`, `BufferStatus.SUCCESS`, ...) were
+  taken from Aravis's published GObject Introspection API docs, not verified against a
+  running install — double-check against the actual installed Aravis version.
+- The EPICS-based `AlliedVisionCamera` (`csaxs_bec/devices/epics/allied_vision_camera.py`)
+  is left in place and untouched for now. It stays dead code until the native path above is
+  validated on real hardware, at which point it can be retired.

csaxs_bec/devices/allied_vision_cameras/__init__.py

@@ -0,0 +1 @@
+from .allied_vision_aravis_camera import AlliedVisionAravisCamera

csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py

@@ -0,0 +1,276 @@
+"""Native (non-EPICS) Allied Vision camera class for cSAXS."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections import deque
+from typing import TYPE_CHECKING
+
+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 csaxs_bec.devices.allied_vision_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
+
+if TYPE_CHECKING:
+    from bec_lib.devicemanager import ScanInfo
+
+logger = bec_logger.logger
+
+
+class AlliedVisionAravisCamera(PSIDeviceBase):
+    """Allied Vision GigE camera class for cSAXS (e.g. Alvium G1-507m, mono).
+
+    This class inherits from PSIDeviceBase and implements the necessary methods
+    to interact with the camera using Aravis, the open-source GenICam/GigE Vision
+    library, directly over GigE Vision - no vendor SDK or EPICS IOC is involved.
+    """
+
+    image = Cpt(
+        PreviewSignal,
+        name="image",
+        ndim=2,
+        doc="Preview signal for the camera.",
+        num_rotation_90=0,
+        transpose=False,
+    )
+    roi_signal = Cpt(
+        AsyncSignal,
+        name="roi_signal",
+        ndim=0,
+        max_size=1000,
+        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 = [
+        "start_live_mode",
+        "stop_live_mode",
+        "mask",
+        "set_rect_roi",
+        "get_last_image",
+        "get_exposure_time",
+        "set_exposure_time",
+        "get_gain",
+        "set_gain",
+        "get_live_fps",
+    ]
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        camera_id: str,
+        prefix: str = "",
+        scan_info: ScanInfo | None = None,
+        live_mode: bool = False,
+        num_rotation_90: int = 0,
+        transpose: bool = False,
+        live_mode_poll_interval_s: float = 0.2,
+        **kwargs,
+    ):
+        """Initialize the Allied Vision Camera.
+
+        Args:
+            name (str): Name of the device.
+            camera_id (str): The Aravis device ID string of the camera (e.g. as reported
+                by `arv-tool-0.8` or `Aravis.get_device_id()`).
+            prefix (str): Prefix for the device.
+            scan_info (ScanInfo | None): Scan information for the device.
+            live_mode (bool): Whether to enable live mode for the camera.
+            live_mode_poll_interval_s (float): Delay between frame grabs in
+                the live-mode loop. Lower this for cameras/use cases that
+                need a higher live-mode push rate; the achievable rate is
+                still bounded by the camera's own exposure/acquisition time.
+        """
+        super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
+        self._live_mode_thread: threading.Thread | None = None
+        self._stop_live_mode_event: threading.Event = threading.Event()
+        self._live_mode_poll_interval_s = live_mode_poll_interval_s
+        # Rolling buffer of push timestamps from _live_mode_loop, used to
+        # measure the actual live-mode frame rate (see get_live_fps()).
+        self._live_frame_times: deque[float] = deque(maxlen=10)
+        self.cam = Camera(camera_id=camera_id, connect=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.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."""
+        return self._mask
+
+    @mask.setter
+    def mask(self, value: np.ndarray):
+        """
+        Set the region of interest (ROI) for the camera.
+
+        Args:
+            value (np.ndarray): The mask to set as the ROI.
+        """
+        if value.ndim != 2:
+            raise ValueError("ROI mask must be a 2D array.")
+        img_shape = (self.cam.height, self.cam.width)
+        if value.shape[0] != img_shape[0] or value.shape[1] != img_shape[1]:
+            raise ValueError(
+                f"ROI mask shape {value.shape} does not match image shape {img_shape}."
+            )
+        self._mask = value
+
+    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."""
+        if x < 0 or y < 0 or width <= 0 or height <= 0:
+            raise ValueError("ROI coordinates and dimensions must be positive integers.")
+        img_shape = (self.cam.height, self.cam.width)
+        if x + width > img_shape[1] or y + height > img_shape[0]:
+            raise ValueError("ROI exceeds camera dimensions.")
+        mask = np.zeros(img_shape, dtype=np.uint8)
+        mask[y : y + height, x : x + width] = 1
+        self.mask = mask
+
+    def _start_live(self):
+        """Start the live mode for the camera."""
+        if self._live_mode_thread is not None:
+            logger.info("Live mode thread is already running.")
+            return
+        self._stop_live_mode_event.clear()
+        self._live_mode_thread = threading.Thread(
+            target=self._live_mode_loop, args=(self._stop_live_mode_event,)
+        )
+        self._live_mode_thread.start()
+
+    def _stop_live(self):
+        """Stop the live mode for the camera."""
+        if self._live_mode_thread is None:
+            logger.info("Live mode thread is not running.")
+            return
+        self._stop_live_mode_event.set()
+        self._live_mode_thread.join(timeout=5)
+        if self._live_mode_thread.is_alive():
+            logger.warning("Live mode thread did not stop gracefully.")
+        else:
+            self._live_mode_thread = None
+            logger.info("Live mode stopped.")
+
+    def _live_mode_loop(self, stop_event: threading.Event):
+        """Loop to capture images in live mode."""
+        self.cam.set_camera_rate_limiting(True)
+        while not stop_event.is_set():
+            try:
+                self.process_data(self.cam.get_image_data())
+            except Exception as e:
+                logger.error(f"Error in live mode loop: {e}")
+                break
+            self._live_frame_times.append(time.time())
+            stop_event.wait(self._live_mode_poll_interval_s)
+        self.cam.set_camera_rate_limiting(False)
+
+    def get_live_fps(self) -> float | None:
+        """Measured live-mode frame rate, from the timestamps of the last
+        pushed frames. Returns None if live mode hasn't pushed at least two
+        frames yet (e.g. not running, or just started).
+        """
+        times = self._live_frame_times
+        if len(times) < 2:
+            return None
+        return (len(times) - 1) / (times[-1] - times[0])
+
+    def process_data(self, image: np.ndarray | None):
+        """Process the image data before sending it to the preview signal."""
+        if image is None:
+            return
+        self.image.put(image)
+
+    def get_last_image(self) -> np.ndarray:
+        """Get the last captured image from the camera."""
+        image = self.image.get()
+        if image:
+            return image.data
+
+    ############## Exposure / Gain ##############
+
+    def get_exposure_time(self) -> float:
+        """Get the current exposure time in microseconds."""
+        return self.cam.exposure_time
+
+    def set_exposure_time(self, value: float) -> None:
+        """Set the exposure time in microseconds."""
+        self.cam.exposure_time = value
+
+    def get_gain(self) -> float:
+        """Get the current gain."""
+        return self.cam.gain
+
+    def set_gain(self, value: float) -> None:
+        """Set the gain."""
+        self.cam.gain = value
+
+    ############## User Interface Methods ##############
+
+    def on_connected(self):
+        """Connect to the camera."""
+        self.cam.on_connect()
+        self.live_mode_enabled.put(bool(self._inputs.get("live_mode", False)))
+        self.set_rect_roi(0, 0, self.cam.width, self.cam.height)
+
+    def on_destroy(self):
+        """Clean up resources when the device is destroyed."""
+        self.cam.on_disconnect()
+        super().on_destroy()
+
+    def on_trigger(self):
+        """Handle the trigger event."""
+        if not bool(self.live_mode_enabled.get()):
+            return
+        image = self.image.get()
+        if image is not None:
+            image: messages.DevicePreviewMessage
+            if self.mask.shape != image.data.shape:
+                logger.info(
+                    f"ROI shape does not match image shape, skipping ROI application for device {self.name}."
+                )
+                return
+            data = image.data * self.mask
+            self.roi_signal.put(np.sum(data) / np.sum(self.mask))
+
+
+if __name__ == "__main__":
+    # Example usage of the AlliedVisionAravisCamera class
+    camera = AlliedVisionAravisCamera(
+        name="TestCamera",
+        camera_id="Allied Vision Technologies-50-0536885929",
+        live_mode=False,
+    )
+    print(f"Camera {camera.name} initialized with ID {camera.cam.camera_id}.")

csaxs_bec/devices/allied_vision_cameras/base_integration/camera.py

@@ -0,0 +1,197 @@
+"""
+This module provides a Camera class for handling Allied Vision GigE cameras using Aravis,
+the open-source GenICam/GigE Vision library. Aravis implements the GigE Vision network
+protocol (GVCP/GVSP) itself, so it talks to the camera directly - no vendor SDK, account,
+or EPICS IOC/areaDetector driver is involved. Aravis is accessed here through its GObject
+Introspection Python bindings (PyGObject / `gi.repository.Aravis`).
+
+The Camera class is the high level interface, and only opens the Aravis camera connection
+when on_connect() is called. This allows for lazy initialization of the camera, so CI/CD
+pipelines can run without PyGObject or the Aravis GObject Introspection typelib installed
+on the system.
+
+Cameras are identified by their Aravis device ID string (e.g. as reported by `arv-tool-0.8`
+or `Aravis.get_device_id()`), which is resolved via GigE Vision discovery broadcast and
+does not depend on the camera's current IP address.
+"""
+
+from __future__ import annotations
+
+import atexit
+import time
+
+import numpy as np
+from bec_lib.logger import bec_logger
+
+logger = bec_logger.logger
+
+try:
+    import gi
+
+    gi.require_version("Aravis", "0.8")
+    from gi.repository import Aravis, GLib
+except (ImportError, ValueError) as exc:
+    logger.warning(f"The Aravis library is not properly installed : {exc}")
+    Aravis = None  # type: ignore[assignment]
+    GLib = None  # type: ignore[assignment]
+
+
+class Camera:
+    """High level camera wrapper for Allied Vision GigE cameras via Aravis.
+
+    Args:
+        camera_id (str): The Aravis device ID string (e.g. as reported by `arv-tool-0.8`
+            or `Aravis.get_device_id()`). Resolved via GigE Vision discovery broadcast,
+            independent of the camera's current IP address.
+    """
+
+    def __init__(self, camera_id: str, connect: bool = True):
+        self.Aravis = Aravis
+        self.GLib = GLib
+        self.camera_id = camera_id
+        self._cam = None
+        self._connected = False
+        self._pixel_format: str | None = None
+        self.width = 0
+        self.height = 0
+
+        atexit.register(self.on_disconnect)
+
+        self._enable_warning_rate_limit: bool = False
+        self._last_rate_limited_log: float = 0
+        self._warning_log_rate_limit_s: float = 10
+
+        if connect:
+            self.on_connect()
+
+    def on_connect(self):
+        """Connect to the camera and initialize it."""
+        if self.Aravis is None:
+            raise ImportError(
+                "The Aravis library is not installed, or its GObject Introspection typelib"
+                " is missing. Please check your Python environment (PyGObject) and Aravis"
+                " installation."
+            )
+        if self._connected:
+            logger.warning("Camera is already connected.")
+            return
+        try:
+            self._cam = self.Aravis.Camera.new(self.camera_id)
+        except self.GLib.Error:
+            self._cam = None
+            raise
+        self._configure_pixel_format()
+        self.width, self.height = self._cam.get_region()[2:]
+        self._connected = True
+
+    def on_disconnect(self, delay_after: float = 0.3):
+        """Disconnect from the camera and optionally wait a short time for driver cleanup."""
+        try:
+            self._cam = None  # drop the reference; Aravis closes the socket on finalization
+            self._connected = False
+            if delay_after > 0:
+                time.sleep(delay_after)
+                logger.debug(f"Waited {delay_after:.2f}s after camera disconnect for cleanup.")
+        except Exception as e:  # pylint: disable=broad-except
+            logger.info(f"Error during camera disconnection: {e}")
+
+    def _configure_pixel_format(self):
+        """Set the camera to a supported monochrome pixel format."""
+        for fmt in ("Mono8", "Mono12"):
+            try:
+                self._cam.set_pixel_format_from_string(fmt)
+            except self.GLib.Error:
+                continue
+            self._pixel_format = fmt
+            return
+        raise ValueError(
+            f"No supported monochrome pixel format available on camera {self.camera_id}."
+        )
+
+    @property
+    def exposure_time(self) -> float:
+        """Get the exposure time of the camera, in microseconds."""
+        return self._cam.get_exposure_time()
+
+    @exposure_time.setter
+    def exposure_time(self, value: float):
+        """Set the exposure time of the camera, in microseconds."""
+        self._cam.set_exposure_time(value)
+
+    @property
+    def gain(self) -> float:
+        """Get the gain of the camera."""
+        return self._cam.get_gain()
+
+    @gain.setter
+    def gain(self, value: float):
+        """Set the gain of the camera."""
+        self._cam.set_gain(value)
+
+    def set_roi(self, x: int, y: int, width: int, height: int):
+        """Set the region of interest (ROI) for the camera sensor readout."""
+        # Reset to the full sensor first so a growing width/height isn't rejected because
+        # the previous offset + new size would exceed the sensor bounds.
+        _, max_w = self._cam.get_width_bounds()
+        _, max_h = self._cam.get_height_bounds()
+        self._cam.set_region(0, 0, max_w, max_h)
+
+        width = self._align(width, self._cam.get_width_increment())
+        height = self._align(height, self._cam.get_height_increment())
+        x = self._align(x, self._cam.get_x_offset_increment())
+        y = self._align(y, self._cam.get_y_offset_increment())
+        self._cam.set_region(x, y, width, height)
+
+        self.width, self.height = self._cam.get_region()[2:]
+        logger.info(f"ROI set to: x={x}, y={y}, width={self.width}, height={self.height}")
+
+    @staticmethod
+    def _align(value: int, increment: int) -> int:
+        """Round value down to the nearest multiple of increment."""
+        if not increment:
+            return value
+        return (value // increment) * increment
+
+    def get_image_data(self) -> np.ndarray | None:
+        """Get a single image frame from the camera (blocking, synchronous)."""
+        if not self._connected:
+            self._rate_limited_warning_log(f"Camera with id {self.camera_id} is not connected.")
+            return None
+        try:
+            buffer = self._cam.acquisition(1_000_000)  # timeout in microseconds
+        except self.GLib.Error as exc:
+            self._rate_limited_warning_log(
+                f"Timeout/error acquiring frame from camera {self.camera_id}: {exc}"
+            )
+            return None
+        if buffer is None or buffer.get_status() != self.Aravis.BufferStatus.SUCCESS:
+            self._rate_limited_warning_log(
+                f"Invalid buffer from camera {self.camera_id}: "
+                f"{buffer.get_status() if buffer is not None else 'no buffer'}."
+            )
+            return None
+        dtype = np.uint16 if self._pixel_format == "Mono12" else np.uint8
+        data = np.frombuffer(buffer.get_data(), dtype=dtype)
+        return data.reshape(buffer.get_image_height(), buffer.get_image_width())
+
+    def set_camera_rate_limiting(self, enabled: bool, rate_limit_s: float | None = None):
+        if rate_limit_s is not None:
+            if rate_limit_s <= 0:
+                raise ValueError(f"Invalid rate limit: {rate_limit_s}, must be positive nonzero.")
+            self._warning_log_rate_limit_s = rate_limit_s
+        self._enable_warning_rate_limit = enabled
+
+    def _rate_limited_warning_log(self, msg: str):
+        if (
+            self._enable_warning_rate_limit
+            and time.monotonic() < self._last_rate_limited_log + self._warning_log_rate_limit_s
+        ):
+            return
+        self._last_rate_limited_log = time.monotonic()
+        logger.warning(msg)
+
+
+if __name__ == "__main__":
+    # Example usage
+    camera = Camera(camera_id="Allied Vision Technologies-50-0536885929")
+    camera.on_connect()

csaxs_bec/devices/sim/sim_cameras.py

@@ -1,8 +1,9 @@
 """
 Simulated flOMNI cameras.
 
-`SimWebcamViewer` and `SimIDSCamera` reuse the real device classes and replace only the
-frame source: the webcam viewer's MJPEG HTTP stream and the IDS camera's pyueye backend.
+`SimWebcamViewer`, `SimIDSCamera` and `SimAlliedVisionAravisCamera` reuse the real device
+classes and replace only the frame source: the webcam viewer's MJPEG HTTP stream, the IDS
+camera's pyueye backend, and the AlliedVision camera's Aravis backend, respectively.
 All signals (`preview` / `image`, `roi_signal`, `live_mode_enabled`), the USER_ACCESS
 methods (`start_live_mode`, `set_rect_roi`, `get_last_image`, ...) and the ROI/trigger
 logic therefore run through the unmodified code paths, so GUIs and the X-ray eye
@@ -24,6 +25,9 @@ from __future__ import annotations
 import numpy as np
 from bec_lib.logger import bec_logger
 
+from csaxs_bec.devices.allied_vision_cameras.allied_vision_aravis_camera import (
+    AlliedVisionAravisCamera,
+)
 from csaxs_bec.devices.ids_cameras.ids_camera import IDSCamera
 from csaxs_bec.devices.omny.webcam_viewer import WebcamViewer
 from csaxs_bec.devices.sim.sim_galil import SimGalilState
@@ -363,3 +367,77 @@ class SimIDSCamera(IDSCamera):
             axis_offset_seed=sim_axis_offset_seed,
         )
         self.cam.force_monochrome = self._force_monochrome
+
+
+class _SimAlliedVisionBackend:
+    """Drop-in replacement for `allied_vision_cameras.base_integration.camera.Camera`
+    serving synthetic frames."""
+
+    def __init__(self, width: int, height: int, noise_std: float = 0.0):
+        self.width = width
+        self.height = height
+        self._connected = False
+        self._noise_std = float(noise_std)
+        self._frame = make_test_pattern(height, width, rgb=False)
+        self._exposure_time = 10000.0  # us
+        self._gain = 0.0
+
+    def on_connect(self):
+        self._connected = True
+
+    def on_disconnect(self):
+        self._connected = False
+
+    def get_image_data(self) -> np.ndarray:
+        return add_frame_noise(self._frame, self._noise_std)
+
+    def set_camera_rate_limiting(self, enabled: bool):
+        pass
+
+    @property
+    def exposure_time(self) -> float:
+        return self._exposure_time
+
+    @exposure_time.setter
+    def exposure_time(self, value: float):
+        self._exposure_time = value
+
+    @property
+    def gain(self) -> float:
+        return self._gain
+
+    @gain.setter
+    def gain(self, value: float):
+        self._gain = value
+
+
+class SimAlliedVisionAravisCamera(AlliedVisionAravisCamera):
+    """AlliedVisionAravisCamera with the Aravis backend replaced by a synthetic frame source."""
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        camera_id: str = "sim",
+        prefix: str = "",
+        scan_info=None,
+        live_mode: bool = False,
+        num_rotation_90: int = 0,
+        transpose: bool = False,
+        sim_shape=(1024, 1280),
+        sim_noise_std=0.0,
+        **kwargs,
+    ):
+        super().__init__(
+            name=name,
+            camera_id=camera_id,
+            prefix=prefix,
+            scan_info=scan_info,
+            live_mode=live_mode,
+            num_rotation_90=num_rotation_90,
+            transpose=transpose,
+            **kwargs,
+        )
+        self.cam = _SimAlliedVisionBackend(
+            int(sim_shape[1]), int(sim_shape[0]), noise_std=sim_noise_std
+        )

pyproject.toml

@@ -22,6 +22,12 @@ dependencies = [
     "rich",
     "pyepics",
     "pyueye",                    # for the IDS uEye camera
+    "pygobject<3.52",             # for the AlliedVision Alvium camera (native Aravis/GenICam, no EPICS IOC)
+                                  # pinned below 3.52: that version switched to requiring
+                                  # girepository-2.0 (GLib >=2.80/gobject-introspection >=1.80),
+                                  # which RHEL9's system packages don't provide (only the older
+                                  # girepository-1.0 via gobject-introspection 1.68) -- unpin once
+                                  # deployment hosts are upgraded past that.
     "bec_widgets",
     "zmq",
     "opencv-python",

tests/tests_devices/test_allied_vision_aravis_camera.py

@@ -0,0 +1,138 @@
+"""Unit tests for the AlliedVisionAravisCamera device."""
+
+import threading
+from unittest import mock
+
+import numpy as np
+import pytest
+
+from csaxs_bec.devices.allied_vision_cameras.allied_vision_aravis_camera import (
+    AlliedVisionAravisCamera,
+)
+
+
+@pytest.fixture(scope="function")
+def allied_vision_camera():
+    """Fixture for creating an instance of the AlliedVisionAravisCamera."""
+    camera = AlliedVisionAravisCamera(
+        name="test_camera", camera_id="TEST-DEV", prefix="test:", scan_info=None, live_mode=False
+    )
+    # Mock camera connection and attributes
+    camera.cam = mock.Mock()
+    camera.cam._connected = True
+    camera.cam.width = 2
+    camera.cam.height = 2
+    yield camera
+
+
+def test_mask_setter_getter(allied_vision_camera):
+    """Test the mask setter and getter methods."""
+    mask = np.zeros((2, 2), dtype=np.uint8)
+    mask[0, 0] = 1
+    allied_vision_camera.mask = mask
+    assert np.array_equal(allied_vision_camera.mask, mask)
+
+
+def test_mask_setter_invalid_shape(allied_vision_camera):
+    """Test the mask setter with an invalid shape."""
+    with pytest.raises(ValueError):
+        allied_vision_camera.mask = np.zeros(
+            (3, 3), dtype=np.uint8
+        )  # Exceeds mocked camera dimensions
+
+
+def test_on_connected_sets_mask_and_live_mode(allied_vision_camera):
+    """Test the on_connected method to ensure it sets the mask and live mode."""
+    allied_vision_camera.cam.on_connect = mock.Mock()
+    allied_vision_camera.on_connected()
+    allied_vision_camera.cam.on_connect.assert_called_once()
+    expected_mask = np.ones((2, 2), dtype=np.uint8)
+    assert np.array_equal(allied_vision_camera.mask, expected_mask)
+
+
+def test_on_trigger_roi_signal(allied_vision_camera):
+    """Test the on_trigger method to ensure it processes the ROI signal correctly."""
+    allied_vision_camera.start_live_mode()
+    test_image = np.array([[2, 4], [6, 8]])
+    test_mask = np.array([[1, 0], [0, 1]], dtype=np.uint8)
+    allied_vision_camera.mask = test_mask
+    mock_image = mock.Mock()
+    mock_image.data = test_image
+    allied_vision_camera.image.get = mock.Mock(return_value=mock_image)
+    allied_vision_camera.roi_signal.put = mock.Mock(side_effect=allied_vision_camera.roi_signal.put)
+    allied_vision_camera.on_trigger()
+    expected_value = (2 * 1 + 4 * 0 + 6 * 0 + 8 * 1) / np.sum(test_mask)
+    result = allied_vision_camera.roi_signal.get()
+    assert np.isclose(
+        result.content["signals"][allied_vision_camera.roi_signal.name]["value"],
+        expected_value,
+        atol=1e-6,
+    )
+
+
+def test_get_last_image(allied_vision_camera):
+    """Test the get_last_image method to ensure it returns the last captured image."""
+    test_image = np.array([[1, 2], [3, 4]], dtype=np.uint8)
+    mock_image = mock.Mock()
+    mock_image.data = test_image
+    allied_vision_camera.image.get = mock.Mock(return_value=mock_image)
+
+    result = allied_vision_camera.get_last_image()
+    assert np.array_equal(result, test_image)
+
+
+def test_get_live_fps_none_before_frames(allied_vision_camera):
+    """No frames pushed yet -> get_live_fps() returns None."""
+    assert allied_vision_camera.get_live_fps() is None
+
+
+def test_get_live_fps_computes_rate(allied_vision_camera):
+    """get_live_fps() computes rate from the recorded push timestamps."""
+    allied_vision_camera._live_frame_times.extend([0.0, 0.2, 0.4, 0.6, 0.8])
+    assert allied_vision_camera.get_live_fps() == pytest.approx(5.0)
+
+
+def test_live_mode_loop_records_frame_times(allied_vision_camera):
+    """_live_mode_loop() appends one timestamp per pushed frame."""
+    allied_vision_camera.cam.get_image_data = mock.Mock(
+        return_value=np.zeros((2, 2), dtype=np.uint8)
+    )
+    stop_event = threading.Event()
+    calls = {"n": 0}
+
+    def fake_wait(timeout):
+        calls["n"] += 1
+        if calls["n"] >= 3:
+            stop_event.set()
+        return False
+
+    stop_event.wait = fake_wait
+    allied_vision_camera._live_mode_loop(stop_event)
+
+    assert len(allied_vision_camera._live_frame_times) == 3
+    assert allied_vision_camera.get_live_fps() is not None
+
+
+def test_on_destroy(allied_vision_camera):
+    """Test the on_destroy method to ensure it cleans up resources."""
+    allied_vision_camera.cam.on_disconnect = mock.Mock()
+    allied_vision_camera.on_destroy()
+    allied_vision_camera.cam.on_disconnect.assert_called_once()
+
+
+def test_get_set_exposure_time(allied_vision_camera):
+    """Test the exposure time getter/setter delegate to the low-level camera wrapper."""
+    allied_vision_camera.cam.exposure_time = 5000.0
+    assert allied_vision_camera.get_exposure_time() == 5000.0
+
+    allied_vision_camera.set_exposure_time(1234.5)
+    assert allied_vision_camera.cam.exposure_time == 1234.5
+
+
+def test_get_set_gain(allied_vision_camera):
+    """Test the gain getter/setter delegate to the low-level camera wrapper."""
+    allied_vision_camera.cam.gain = 3.0
+    assert allied_vision_camera.get_gain() == 3.0
+
+    allied_vision_camera.set_gain(7.5)
+    assert allied_vision_camera.cam.gain == 7.5