fix: include content changes for vmbpy->Aravis swap
CI for csaxs_bec / test (push) Failing after 2m21s
CI for csaxs_bec / test (push) Failing after 2m21s
The prior commit only captured the file renames (a botched `git add` dropped the actual edits). This commit carries the real diff: Aravis GObject-introspection backend in base_integration/camera.py, renamed class/config references, and the rewritten README. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -34,10 +34,10 @@ eiger_1_5:
|
||||
# softwareTrigger: True
|
||||
|
||||
# allied_vision_cam:
|
||||
# description: AlliedVision Alvium G1-507m camera for live image acquisition (native vmbpy/GigE Vision, no EPICS IOC)
|
||||
# deviceClass: csaxs_bec.devices.allied_vision_cameras.AlliedVisionVmbCamera
|
||||
# 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: "DEV_1AB22C000000" # GigE ID/serial string; resolved via discovery, independent of current IP
|
||||
# camera_id: "Allied Vision Technologies-50-0536885929" # Aravis device ID; resolved via discovery, independent of current IP
|
||||
# live_mode: True
|
||||
# onFailure: raise
|
||||
# enabled: true
|
||||
|
||||
@@ -1,141 +1,156 @@
|
||||
# Allied Vision camera (native vmbpy integration)
|
||||
# Allied Vision camera (native Aravis integration)
|
||||
|
||||
`AlliedVisionVmbCamera` is the native-Python replacement for the old EPICS/Vimba-IOC
|
||||
`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 **vmbpy**, the vendor's Python bindings for the Vimba X SDK — no
|
||||
EPICS IOC or areaDetector driver involved.
|
||||
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 AlliedVisionVmbCamera
|
||||
├── allied_vision_vmb_camera.py # ophyd device (PSIDeviceBase)
|
||||
├── __init__.py # exports AlliedVisionAravisCamera
|
||||
├── allied_vision_aravis_camera.py # ophyd device (PSIDeviceBase)
|
||||
└── base_integration/
|
||||
└── camera.py # low-level vmbpy wrapper (class Camera)
|
||||
└── camera.py # low-level Aravis wrapper (class Camera)
|
||||
```
|
||||
|
||||
Simulation counterpart: `SimAlliedVisionVmbCamera` in `csaxs_bec/devices/sim/sim_cameras.py`
|
||||
(same device, `self.cam` swapped for a synthetic-frame backend — no hardware or vmbpy
|
||||
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_vmb_camera.py`): nothing
|
||||
extra. `Camera.__init__` guards the `import vmbpy` and the tests mock `camera.cam`
|
||||
directly, so `pytest tests/tests_devices/test_allied_vision_vmb_camera.py` runs on a
|
||||
plain dev environment, same as the IDS camera tests run without `pyueye`.
|
||||
- **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 `AlliedVisionVmbCamera`/`Camera` directly):
|
||||
1. The `vmbpy` Python package.
|
||||
2. The **Vimba X SDK runtime** (GenTL producer / transport layer `vmbpy` links against at
|
||||
runtime). The pip package alone does not include this — same class of local
|
||||
dependency as the DLLs `pyueye` needs for the IDS cameras.
|
||||
server (or wherever you instantiate `AlliedVisionAravisCamera`/`Camera` directly):
|
||||
1. **PyGObject** (`pygobject`, the `gi` module) — already listed in `pyproject.toml`.
|
||||
It's the GObject Introspection bridge, not Aravis itself.
|
||||
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 vmbpy + the Vimba X runtime (Linux)
|
||||
### Installing PyGObject + Aravis (Linux)
|
||||
|
||||
`vmbpy` is already listed in `pyproject.toml`, but `pip install vmbpy` from PyPI alone
|
||||
pulls in a package that has **no camera drivers/transport layers** — it can be imported
|
||||
but `VmbSystem.get_instance()` won't find any cameras until the SDK runtime below is also
|
||||
installed. There are two ways to get a working install:
|
||||
`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. **Install the Vimba X SDK first, then use the `.whl` it ships with** (recommended —
|
||||
this `.whl` bundles the VmbC libraries so you don't need a separate runtime step for
|
||||
the Python bindings):
|
||||
- Download the Vimba X SDK for Linux from Allied Vision
|
||||
(alliedvision.com → Support → Software → *Vimba X SDK*; free account required).
|
||||
- Extract it somewhere you have write access, e.g.
|
||||
`tar -xf VimbaX_Setup-<version>_Linux_x86_64.tar.gz -C /opt/vimbax`.
|
||||
- Register the GenTL transport layer path so cameras can be discovered:
|
||||
```bash
|
||||
cd /opt/vimbax/VimbaX_*/cti
|
||||
sudo ./Install_GenTL_Path.sh # writes GENICAM_GENTL64_PATH, may need a re-login
|
||||
```
|
||||
- Install the matching Python wheel from the SDK's `api/python` directory into this
|
||||
repo's venv:
|
||||
```bash
|
||||
pip install /opt/vimbax/VimbaX_*/api/python/vmbpy-*-py3-none-*.whl
|
||||
```
|
||||
This satisfies the `vmbpy` dependency already declared in `pyproject.toml` — pip
|
||||
won't reinstall over it unless the version constraint mismatches.
|
||||
2. **Or install `vmbpy` from PyPI** (`pip install vmbpy`, or just `pip install -e .` in
|
||||
this repo since it's already a dependency) **and separately install the Vimba X SDK**
|
||||
for the GenTL producer only, following the same extract + `Install_GenTL_Path.sh` steps
|
||||
above but skipping its wheel.
|
||||
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).
|
||||
|
||||
Either way, the GigE transport layer is what's needed here (this is a pure Ethernet
|
||||
camera); the USB3 transport layer in the SDK isn't required.
|
||||
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 vmbpy; print(vmbpy.VmbSystem.get_instance())"
|
||||
python3 -c "import vmbpy
|
||||
with vmbpy.VmbSystem.get_instance() as vmb:
|
||||
print(vmb.get_all_cameras())"
|
||||
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 the transport layer loaded
|
||||
without error).
|
||||
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: [VmbPy README (GitHub)](https://github.com/alliedvision/VmbPy),
|
||||
[Vimba X Python API Manual](https://docs.alliedvision.com/Vimba_X/Vimba_X_DeveloperGuide/pythonAPIManual.html),
|
||||
[vmbpy on PyPI](https://pypi.org/project/vmbpy/).
|
||||
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 GigE Vision device ID/serial
|
||||
string, e.g. `"DEV_1AB22C000000"` — not an IP address. `VmbSystem.get_camera_by_id()`
|
||||
resolves it via GigE Vision discovery broadcast, so it works regardless of the camera's
|
||||
current DHCP-assigned IP.
|
||||
`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 vmbpy + the Vimba X runtime are installed on a machine on the same network segment as
|
||||
the camera, list connected cameras and their IDs with:
|
||||
|
||||
```python
|
||||
import vmbpy
|
||||
with vmbpy.VmbSystem.get_instance() as vmb:
|
||||
for cam in vmb.get_all_cameras():
|
||||
print(cam.get_id(), cam.get_name(), cam.get_serial())
|
||||
```
|
||||
|
||||
Or from the shell, if the Vimba X SDK install included its CLI tools:
|
||||
Once PyGObject + Aravis are installed on a machine on the same network segment as the
|
||||
camera, list connected cameras and their IDs with:
|
||||
|
||||
```bash
|
||||
VmbCPPExamples/ListCameras # or wherever the SDK installed its example binaries
|
||||
arv-tool-0.8 --list
|
||||
```
|
||||
|
||||
Allied Vision's **Vimba X Viewer** GUI (bundled with the SDK installer) also 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.
|
||||
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_vmb_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 vmbpy.
|
||||
- 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
|
||||
`SimAlliedVisionVmbCamera` (`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`/
|
||||
`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 vmbpy + Vimba X SDK runtime as above.
|
||||
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 AlliedVisionVmbCamera
|
||||
cam = AlliedVisionVmbCamera(name="allied_vision_cam", camera_id="DEV_1AB22C000000")
|
||||
cam.on_connected() # opens the vmbpy VmbSystem/Camera context
|
||||
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() # closes the vmbpy contexts
|
||||
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
|
||||
@@ -143,11 +158,15 @@ the IP does, which is why it's the right identifier here given the camera has no
|
||||
|
||||
## Notes / open items
|
||||
|
||||
- Exception names used in `Camera.get_image_data()` (`vmbpy.VmbTimeout`,
|
||||
`vmbpy.VmbFeatureError`, `vmbpy.VmbCameraError`) and the feature-access API
|
||||
(`get_feature_by_name`) should be double-checked against whatever vmbpy version actually
|
||||
gets installed — they were written from the vmbpy API as documented, not verified against
|
||||
a running install.
|
||||
- `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.
|
||||
|
||||
@@ -1 +1 @@
|
||||
from .allied_vision_vmb_camera import AlliedVisionVmbCamera
|
||||
from .allied_vision_aravis_camera import AlliedVisionAravisCamera
|
||||
|
||||
@@ -20,12 +20,12 @@ if TYPE_CHECKING:
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class AlliedVisionVmbCamera(PSIDeviceBase):
|
||||
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 the vmbpy library (Vimba X SDK), directly
|
||||
over GigE Vision - no EPICS IOC is involved.
|
||||
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(
|
||||
@@ -80,7 +80,8 @@ class AlliedVisionVmbCamera(PSIDeviceBase):
|
||||
|
||||
Args:
|
||||
name (str): Name of the device.
|
||||
camera_id (str): The GigE Vision ID/serial string of the camera 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.
|
||||
@@ -243,6 +244,10 @@ class AlliedVisionVmbCamera(PSIDeviceBase):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage of the AlliedVisionVmbCamera class
|
||||
camera = AlliedVisionVmbCamera(name="TestCamera", camera_id="DEV_1AB22C000000", live_mode=False)
|
||||
# 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}.")
|
||||
|
||||
@@ -1,16 +1,18 @@
|
||||
"""
|
||||
This module provides a Camera class for handling Allied Vision GigE cameras using the
|
||||
vmbpy library, which links to the vendor's Vimba X C SDK. vmbpy talks to the camera
|
||||
directly over GigE Vision / GenICam - no EPICS IOC or areaDetector driver is involved.
|
||||
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 vmbpy VmbSystem/Camera
|
||||
context when on_connect() is called. This allows for lazy initialization of the camera,
|
||||
so CI/CD pipelines can run without vmbpy or the Vimba X runtime/GenTL producer installed
|
||||
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 GigE Vision ID/serial string (VmbSystem.get_camera_by_id),
|
||||
which is resolved via GigE discovery broadcast and does not depend on the camera's current
|
||||
IP address.
|
||||
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
|
||||
@@ -24,32 +26,38 @@ from bec_lib.logger import bec_logger
|
||||
logger = bec_logger.logger
|
||||
|
||||
try:
|
||||
import vmbpy
|
||||
except ImportError as exc:
|
||||
logger.warning(f"The vmbpy library is not properly installed : {exc}")
|
||||
vmbpy = None # type: ignore[assignment]
|
||||
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 vmbpy.
|
||||
"""High level camera wrapper for Allied Vision GigE cameras via Aravis.
|
||||
|
||||
Args:
|
||||
camera_id (str): The vmbpy camera ID/serial string (e.g. "DEV_1AB22C000000").
|
||||
Resolved via GigE Vision discovery broadcast, independent of the camera's
|
||||
current IP address.
|
||||
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):
|
||||
if vmbpy is None:
|
||||
if Aravis is None:
|
||||
raise ImportError(
|
||||
"The vmbpy library is not installed or the Vimba X runtime/GenTL producer"
|
||||
" is missing. Please check your Python environment and Vimba X installation."
|
||||
"The Aravis library is not installed, or its GObject Introspection typelib"
|
||||
" is missing. Please check your Python environment (PyGObject) and Aravis"
|
||||
" installation."
|
||||
)
|
||||
self.vmbpy = vmbpy
|
||||
self.Aravis = Aravis
|
||||
self.GLib = GLib
|
||||
self.camera_id = camera_id
|
||||
self._vmb = None
|
||||
self._cam = None
|
||||
self._connected = False
|
||||
self._pixel_format: str | None = None
|
||||
self.width = 0
|
||||
self.height = 0
|
||||
|
||||
@@ -67,36 +75,19 @@ class Camera:
|
||||
if self._connected:
|
||||
logger.warning("Camera is already connected.")
|
||||
return
|
||||
self._vmb = self.vmbpy.VmbSystem.get_instance()
|
||||
self._vmb.__enter__()
|
||||
try:
|
||||
self._cam = self._vmb.get_camera_by_id(self.camera_id)
|
||||
self._cam.__enter__()
|
||||
except Exception:
|
||||
self._vmb.__exit__(None, None, None)
|
||||
self._vmb = None
|
||||
self._cam = self.Aravis.Camera.new(self.camera_id)
|
||||
except self.GLib.Error:
|
||||
self._cam = None
|
||||
raise
|
||||
self._configure_pixel_format()
|
||||
self.width = self._feature("Width").get()
|
||||
self.height = self._feature("Height").get()
|
||||
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:
|
||||
if self._cam is not None:
|
||||
try:
|
||||
self._cam.__exit__(None, None, None)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.info(f"Error while closing camera context: {e}")
|
||||
if self._vmb is not None:
|
||||
try:
|
||||
self._vmb.__exit__(None, None, None)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.info(f"Error while closing VmbSystem context: {e}")
|
||||
self._cam = None
|
||||
self._vmb = None
|
||||
self._cam = None # drop the reference; Aravis closes the socket on finalization
|
||||
self._connected = False
|
||||
if delay_after > 0:
|
||||
time.sleep(delay_after)
|
||||
@@ -104,59 +95,54 @@ class Camera:
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.info(f"Error during camera disconnection: {e}")
|
||||
|
||||
def _feature(self, name: str):
|
||||
"""Return the GenICam feature object for the given name."""
|
||||
return self._cam.get_feature_by_name(name)
|
||||
|
||||
def _configure_pixel_format(self):
|
||||
"""Set the camera to a supported monochrome pixel format."""
|
||||
supported = self._cam.get_pixel_formats()
|
||||
if self.vmbpy.PixelFormat.Mono8 in supported:
|
||||
self._cam.set_pixel_format(self.vmbpy.PixelFormat.Mono8)
|
||||
elif self.vmbpy.PixelFormat.Mono12 in supported:
|
||||
self._cam.set_pixel_format(self.vmbpy.PixelFormat.Mono12)
|
||||
else:
|
||||
raise ValueError(f"No supported monochrome pixel format available: {supported}")
|
||||
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._feature("ExposureTime").get()
|
||||
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._feature("ExposureTime").set(value)
|
||||
self._cam.set_exposure_time(value)
|
||||
|
||||
@property
|
||||
def gain(self) -> float:
|
||||
"""Get the gain of the camera."""
|
||||
return self._feature("Gain").get()
|
||||
return self._cam.get_gain()
|
||||
|
||||
@gain.setter
|
||||
def gain(self, value: float):
|
||||
"""Set the gain of the camera."""
|
||||
self._feature("Gain").set(value)
|
||||
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 offsets first so a growing width/height isn't rejected because the
|
||||
# previous offset + new size would exceed the sensor bounds.
|
||||
self._feature("OffsetX").set(0)
|
||||
self._feature("OffsetY").set(0)
|
||||
# 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._feature("Width").get_increment())
|
||||
height = self._align(height, self._feature("Height").get_increment())
|
||||
self._feature("Width").set(width)
|
||||
self._feature("Height").set(height)
|
||||
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)
|
||||
|
||||
x = self._align(x, self._feature("OffsetX").get_increment())
|
||||
y = self._align(y, self._feature("OffsetY").get_increment())
|
||||
self._feature("OffsetX").set(x)
|
||||
self._feature("OffsetY").set(y)
|
||||
|
||||
self.width = self._feature("Width").get()
|
||||
self.height = self._feature("Height").get()
|
||||
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
|
||||
@@ -172,19 +158,21 @@ class Camera:
|
||||
self._rate_limited_warning_log(f"Camera with id {self.camera_id} is not connected.")
|
||||
return None
|
||||
try:
|
||||
frame = self._cam.get_frame(timeout_ms=1000)
|
||||
except self.vmbpy.VmbTimeout:
|
||||
buffer = self._cam.acquisition(1_000_000) # timeout in microseconds
|
||||
except self.GLib.Error as exc:
|
||||
self._rate_limited_warning_log(
|
||||
f"Timeout waiting for frame from camera {self.camera_id}."
|
||||
f"Timeout/error acquiring frame from camera {self.camera_id}: {exc}"
|
||||
)
|
||||
return None
|
||||
except (self.vmbpy.VmbFeatureError, self.vmbpy.VmbCameraError) as exc:
|
||||
logger.error(f"Error while acquiring frame from camera {self.camera_id}: {exc}")
|
||||
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
|
||||
img = frame.as_numpy_ndarray()
|
||||
if img.ndim == 3 and img.shape[2] == 1:
|
||||
img = img[:, :, 0]
|
||||
return img
|
||||
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:
|
||||
@@ -205,5 +193,5 @@ class Camera:
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
camera = Camera(camera_id="DEV_1AB22C000000")
|
||||
camera = Camera(camera_id="Allied Vision Technologies-50-0536885929")
|
||||
camera.on_connect()
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
"""
|
||||
Simulated flOMNI cameras.
|
||||
|
||||
`SimWebcamViewer`, `SimIDSCamera` and `SimAlliedVisionVmbCamera` reuse the real device
|
||||
`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 vmbpy backend, respectively.
|
||||
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
|
||||
@@ -20,7 +20,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_vmb_camera import AlliedVisionVmbCamera
|
||||
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
|
||||
|
||||
@@ -203,8 +205,8 @@ class _SimAlliedVisionBackend:
|
||||
self._gain = value
|
||||
|
||||
|
||||
class SimAlliedVisionVmbCamera(AlliedVisionVmbCamera):
|
||||
"""AlliedVisionVmbCamera with the vmbpy backend replaced by a synthetic frame source."""
|
||||
class SimAlliedVisionAravisCamera(AlliedVisionAravisCamera):
|
||||
"""AlliedVisionAravisCamera with the Aravis backend replaced by a synthetic frame source."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
||||
+1
-1
@@ -22,7 +22,7 @@ dependencies = [
|
||||
"rich",
|
||||
"pyepics",
|
||||
"pyueye", # for the IDS uEye camera
|
||||
"vmbpy", # for the AlliedVision Alvium camera (native SDK, no EPICS IOC)
|
||||
"pygobject", # for the AlliedVision Alvium camera (native Aravis/GenICam, no EPICS IOC)
|
||||
"bec_widgets",
|
||||
"zmq",
|
||||
"opencv-python",
|
||||
|
||||
@@ -1,20 +1,20 @@
|
||||
"""Unit tests for the AlliedVisionVmbCamera device."""
|
||||
"""Unit tests for the AlliedVisionAravisCamera device."""
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from csaxs_bec.devices.allied_vision_cameras.allied_vision_vmb_camera import (
|
||||
AlliedVisionVmbCamera,
|
||||
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 AlliedVisionVmbCamera."""
|
||||
camera = AlliedVisionVmbCamera(
|
||||
name="test_camera", camera_id="DEV_TEST", prefix="test:", scan_info=None, live_mode=False
|
||||
"""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()
|
||||
|
||||
Reference in New Issue
Block a user