"""Driver for cameras based on the Allied Vision VimbaX library."""
import os
import sys
from typing import Any
from heros.helper import log
from herosdevices.core.templates.camera import CameraTemplate
from herosdevices.helper import mark_driver
try:
import vmbpy
DEFAULT_FEATURE_VISIBLIY = vmbpy.FeatureVisibility.Expert
except ModuleNotFoundError:
log.exception("Could not import the 'vmbpy' python module, required for using Allied Vision cameras")
DEFAULT_FEATURE_VISIBLIY = 2
DEFAULT_CONFIG = {}
[docs]
@mark_driver(
name="VimbaX Camera",
info="Camera compatible with the Allied Vision VimbaX driver",
state="beta",
requires={"vmbpy": "vmbpy"},
)
class VimbaCompatibleCamera(CameraTemplate):
"""
A class to interface with Allied Vision VimbaX cameras.
The class provides functionality to control and capture images from Allied Vision VimbaX cameras.
It manages camera configuration, acquisition, and data streaming.
Important:
The vendor underlying library must be obtained from the
`official website <https://www.alliedvision.com/de/support/software-downloads/vimba-x-sdk/vimba-x>`_.
Download the `Allied Vision VimbaX SDK`, unpack it and install it following the supplied ``README`` file.
Note:
For use with :ref:`BOSS in docker <boss-additional-deps>`, you can adapt the VimbaX install script
``cti/Install_GenTL_Path.sh``
To find the ``cam_id`` of your camera, you can either use the VimbaXViewer
(or start the driver with a wrong id, it will print the available ids).
The ids look something like ``DEV_000A4717239E``.
"""
_connection_calibrated: bool = False
_expected_frames: int = 0
_last_frame_id: int = 0
"""Valid special keys are:
- ``pixel_format``: Sets the pixel format of the camera, for example "Mono8"
:meta public:
"""
_special_config_keys: tuple = ("pixel_format",)
def __init__(
self,
cam_id: str,
config_dict: dict,
default_config: str | None = None,
reset_to_continuous: bool = False,
lib_path: str | None = None,
) -> None:
"""Create a class to interface with Allied Vision VimbaX cameras.
Args:
cam_id: Serial number of the cam. See note above to learn how to obtain the correct serial.
config_dict: Dict of configuration values like shown in the json example above. The keys starting with a
capital letter in the example are features of the camera. For available features, see the manual
of your camera model.
Valid special keys are:
- ``pixel_format``: Sets the pixel format of the camera, for example "Mono8"
Example: { "default":
{"ExposureTime": 1000, "TriggerSelector": "ExposureStart", "TriggerMode": "On", "TriggerSource":
"Software","AcquisitionMode": "MultiFrame", "AcquisitionFrameCount": 5}}
default_config: Default key in :code:`config_dict` to use. Example: "default"
reset_to_continuous: If True, the camera will be set to continuous acquisition mode on teardown.
lib_path: Path to vendor transport library. Example: "/opt/vimba/cti/"
"""
self.cam_id = cam_id
self.default_config_dict = DEFAULT_CONFIG
self.reset_to_continuous = reset_to_continuous
if lib_path is not None:
os.environ["GENICAM_GENTL32_PATH"] = lib_path
os.environ["GENICAM_GENTL64_PATH"] = lib_path
super().__init__(config_dict, default_config)
[docs]
def read_feature(self, feature_name: str) -> Any:
"""Read the current value of a VimbaX camera feature.
Args:
feature_name: Name of the feature to read (e.g., "ExposureTime", "Gain")
Returns:
The current value of the feature, or False if the feature cannot be read.
For enum features, returns the string representation of the enum value.
Example:
Read exposure time::
exposure = camera.read_feature("ExposureTime")
print(f"Current exposure: {exposure} µs")
"""
with self.get_camera() as cam:
try:
value = getattr(cam, feature_name).get()
if isinstance(value, vmbpy.feature.EnumEntry):
value = str(value)
except (AttributeError, vmbpy.VmbFeatureError):
log.exception("Failed reading VimbaX feature %s ", feature_name)
return False
else:
return value
[docs]
def set_feature(self, feature_name: str, value: Any) -> bool:
"""Set the value of a VimbaX camera feature.
Warning:
This method is discouraged for production use. Instead, use the configuration
system via :meth:`set_configuration` which provides better configuration management.
Args:
feature_name: Name of the feature to set (e.g., "ExposureTime", "Gain")
value: Value to set for the feature
Returns:
True if the feature was set successfully, False otherwise.
Example:
Set exposure time::
success = camera.set_feature("ExposureTime", 1000)
if not success:
print("Failed to set exposure time")
"""
with self.get_camera() as cam:
try:
getattr(cam, feature_name).set(value)
except (AttributeError, vmbpy.VmbFeatureError):
log.exception("Failed setting VimbaX feature %s to value %s, ignoring...", feature_name, value)
return False
return True
[docs]
def run_feature(self, feature_name: str) -> Any:
"""Execute a command-type VimbaX camera feature.
Command features are used for actions like triggering software triggers,
saving settings, or other one-time operations.
Args:
feature_name: Name of the command feature to execute (e.g., "TriggerSoftware")
Returns:
True if the feature was ran successfully, False otherwise.
Example:
Trigger a software trigger::
result = camera.run_feature("TriggerSoftware")
if result:
print("Trigger executed successfully")
"""
with self.get_camera() as cam:
try:
getattr(cam, feature_name).run()
except (AttributeError, vmbpy.VmbFeatureError):
log.exception("Failed running VimbaX feature %s", feature_name)
return False
else:
return True
[docs]
def get_config_features(
self, max_visibility_level: "vmbpy.FeatureVisibility | int" = DEFAULT_FEATURE_VISIBLIY
) -> dict:
"""Get all features from the camera in form of a dict including information about if they can be set/read.
Args:
max_visibility_level: Maximum visibility level for features to include (1=Beginner, 2=Expert, etc.)
"""
feature_dict = {}
with self.get_camera() as cam:
if cam is not None:
for feat in cam.get_all_features():
if feat.get_visibility() <= max_visibility_level:
try:
value = feat.get()
if isinstance(value, vmbpy.feature.EnumEntry):
value = str(value)
except (AttributeError, vmbpy.VmbFeatureError):
value = None
readable, writable = feat.get_access_mode()
feature_props = {
"tooltip": feat.get_tooltip(),
"desc": feat.get_description(),
"value": value,
"readable": readable,
"writable": writable,
}
feature_dict[feat.get_name()] = feature_props
return feature_dict
def _open(self) -> "vmbpy.Camera":
"""Open the connection to the device. Don't call directly.
:meta private:
"""
self._vmb_instance = vmbpy.VmbSystem.get_instance().__enter__()
try:
cam = self._vmb_instance.get_camera_by_id(self.cam_id).__enter__()
if not self._connection_calibrated:
# calibrate network package size
stream = cam.get_streams()[0]
stream.GVSPAdjustPacketSize.run()
log.info("Running VimbaX network package calibration...")
while not stream.GVSPAdjustPacketSize.is_done():
pass
log.info("VimbaX network package calibration successful")
self._connection_calibrated = True
except vmbpy.VmbCameraError:
log.error(
"Available cameras are %s",
[str(cam._Camera__info.cameraName) for cam in self._vmb_instance.get_all_cameras()],
)
raise
else:
return cam
def _set_config(self, config: dict) -> bool:
success = True
for key, value in config.items():
if key not in self._special_config_keys:
success &= self.set_feature(key, value) and success
elif key == "pixel_format":
with self.get_camera() as cam:
try:
cam.set_pixel_format(vmbpy.frame.PixelFormat[value])
except (ValueError, KeyError):
msg = "Failed to set unknown pixel format %s. Available formats are %s"
avail_formats = [en.name for en in cam.get_pixel_formats()]
log.error(msg, value, avail_formats)
return success
def _get_expected_frames(self) -> int:
current_config = self.get_configuration()
if current_config["AcquisitionMode"] == "Continuous":
return -1
if current_config["AcquisitionMode"] == "SingleFrame":
return 1
# MultiFrame
return current_config["AcquisitionFrameCount"]
def _frame_handler(self, cam: "vmbpy.Camera", _stream: "vmbpy.Stream", frame: "vmbpy.Frame") -> None:
frame_id = frame.get_id() - 1
self._last_frame_id = max(frame_id or 0, self._last_frame_id)
meta_data = {"frame": frame_id, "acquisition_time": frame.get_timestamp()}
self.acquisition_data(frame.as_numpy_ndarray().squeeze(), meta_data)
cam.queue_frame(frame)
def _arm(self) -> bool:
self._expected_frames = self._get_expected_frames()
with self.get_camera() as cam:
if cam is not None:
log.debug("Starting acquisition...")
cam.start_streaming(handler=self._frame_handler)
return True
return False
def _start(self) -> bool:
with self.get_camera() as cam:
if cam is not None:
cam.TriggerSoftware.run()
return True
return False
def _stop(self) -> bool:
if self._expected_frames > 0:
if self._expected_frames > 0 and self._last_frame_id != self._expected_frames:
log.error(
"Incorrect number of received VimbaX frames: %s instead of %s!",
self._last_frame_id,
self._expected_frames,
)
self._expected_frames = 0
self._last_frame_id = 0
self.acquisition_running = False
with self.get_camera() as cam:
if cam is not None:
if cam.is_streaming():
cam.stop_streaming()
return True
return False
def _get_status(self) -> dict:
return {
"acquisition_running": self.acquisition_running,
}
def _get_exposure_time(self) -> float | None:
"""Return the current exposure time in seconds from the active configuration.
Returns:
exposure time in seconds, or None if not set
"""
val = self.get_configuration().get("ExposureTime")
return val / 1e6 if val is not None else None
def _get_roi_coordinates(self) -> tuple[int, int, int, int] | None:
"""Return the current ROI from the active configuration.
Returns:
(x_offset, y_offset, width, height) in sensor pixels, or None if not set
"""
c = self.get_configuration()
x, y, w, h = c.get("OffsetX"), c.get("OffsetY"), c.get("Width"), c.get("Height")
if None in (x, y, w, h):
return None
return (x, y, w, h)
def _get_binning(self) -> tuple[int, int] | None:
"""Return the current binning from the active configuration.
Returns:
(horizontal, vertical) binning factors, or None if not set
"""
c = self.get_configuration()
h, v = c.get("BinningHorizontal"), c.get("BinningVertical")
if None in (h, v):
return None
return (h, v)
def _set_exposure_time(self, exposure_time: float) -> dict:
"""Return IDS Peak config patch for the given exposure time.
Args:
exposure_time: exposure time in seconds
Returns:
config patch dict
"""
return {"ExposureTime": int(exposure_time * 1e6)}
def _set_roi_coordinates(self, roi: tuple[int, int, int, int]) -> dict:
"""Return IDS Peak config patch for the given ROI.
Args:
roi: (x_offset, y_offset, width, height) in sensor pixels
Returns:
config patch dict
"""
x_offset, y_offset, width, height = roi
return {"OffsetX": x_offset, "OffsetY": y_offset, "Width": width, "Height": height}
def _set_binning(self, binning: tuple[int, int]) -> dict:
"""Return IDS Peak config patch for the given binning factors.
Args:
binning: (horizontal, vertical) binning factors
Returns:
config patch dict
"""
h_bin, v_bin = binning
valid = (1, 2, 4)
if h_bin not in valid or v_bin not in valid:
msg = f"Binning factors must be one of {valid}, got ({h_bin}, {v_bin})"
raise ValueError(msg)
return {"BinningSelector": "Sensor", "BinningHorizontal": int(h_bin), "BinningVertical": int(v_bin)}
def _teardown(self, *_args) -> None:
if self._device is not None or self._vmb_instance is not None:
# only close if not already closed. This avoids that repeated calls are not opening the cam again.
if self.reset_to_continuous:
self._set_config({"AcquisitionMode": "Continuous"})
log.debug(f"closing down connection to Allied Vision camera {self.cam_id}")
self.stop()
if self._device is not None:
self._device.__exit__(*sys.exc_info())
self._device = None
if self._vmb_instance is not None:
self._vmb_instance.__exit__(*sys.exc_info())
self._vmb_instance = None
def __del__(self) -> None:
"""Call teardown method and close VimbaX library on delete."""
self.teardown()