"""Dummy camera devices for use in testing. Do not use in production."""
import threading
import time
import numpy as np
from numpy.typing import NDArray
from herosdevices.core.templates import CameraTemplate
from herosdevices.helper import log
[docs]
class ImageGeneratorDummy:
"""Act like a real camera, no one will notice."""
def __init__(self) -> None:
self.sensor_width: int = 800
self.sensor_height: int = 600
self.exposure_time: float = 1.0
self.gain: float = 1.0
self.roi: tuple[int, int, int, int] | None = None
self.h_binning: int = 1
self.v_binning: int = 1
self.frame_count = 1
self._is_armed = False
self._image_buffer = []
[docs]
@staticmethod
def generate_gaussian_image(
w: int, h: int, amplitude: float = 65535, noise_level: float = 0.05
) -> NDArray[np.uint16]:
"""Generate a 2D Gaussian image with added random noise.
The Gaussian is centered in the image with a fixed standard deviation,
scaled to the specified amplitude. Additive Gaussian noise is applied
and the result is clipped to the valid `uint16` range.
Args:
w: Width of the image.
h: Height of the image.
amplitude: Peak value of the Gaussian. Defaults to 65535.
noise_level: Standard deviation of noise relative to
the amplitude (e.g., 0.05 means ±5% noise). Defaults to 0.05.
Returns:
np.ndarray: A (h, w) image array of dtype `np.uint16`.
"""
x = np.linspace(-1, 1, w)
y = np.linspace(-1, 1, h)
xv, yv = np.meshgrid(x, y)
sigma = 0.3
gaussian = np.exp(-(xv**2 + yv**2) / (2 * sigma**2))
gaussian *= amplitude
noise = np.random.default_rng().normal(loc=0, scale=noise_level * amplitude, size=(h, w))
image = gaussian + noise
image = np.clip(image, 0, 65535)
return image.astype(np.uint16)
[docs]
def arm(self) -> None:
"""Arm the dummy device."""
self._is_armed = True
[docs]
def trigger(self) -> None:
"""Append an image to the buffer."""
if self._is_armed:
if len(self._image_buffer) == self.frame_count:
raise RuntimeError("Camera was armed and triggered while buffer was full!")
amplitude = min(65535 * self.exposure_time * self.gain, 65535)
image = self.generate_gaussian_image(self.sensor_width, self.sensor_height, amplitude=amplitude)
if self.roi is not None:
x, y, w, h = self.roi
image = image[y : y + h, x : x + w]
if self.h_binning > 1 or self.v_binning > 1:
h_out = image.shape[0] // self.v_binning
w_out = image.shape[1] // self.h_binning
image = (
image[: h_out * self.v_binning, : w_out * self.h_binning]
.reshape(h_out, self.v_binning, w_out, self.h_binning)
.mean(axis=(1, 3))
.astype(np.uint16)
)
self._image_buffer.append(image)
if len(self._image_buffer) == self.frame_count:
self._is_armed = False
[docs]
def get_image(self) -> np.ndarray:
"""Get the last image from the buffer."""
if len(self._image_buffer) > 0:
return self._image_buffer.pop(0)
raise RuntimeError("Image buffer empty!")
[docs]
def clear_buffer(self) -> None:
"""Clear the image buffer."""
self._image_buffer = []
[docs]
def abort(self) -> None:
"""Abort the acquisition."""
self._is_armed = False
self.clear_buffer()
[docs]
class CameraDummy(CameraTemplate):
"""A dummy camera.
``gain`` is a factor >= 0 that scales the generated amplitude (default 1). ``frame_rate`` throttles the
acquisition thread, ``None`` (default) removes the throttle.
"""
_auto_trigger: bool = True
default_config_dict: dict = {"exposure_time": 1.0}
def _open(self) -> ImageGeneratorDummy:
"""Device specific code to open the camera handler and return it."""
return ImageGeneratorDummy()
def _teardown(self) -> None:
"""Device specific code to release the camera handler and potentially de-initialize the API."""
def _start(self) -> bool:
"""
Device specific code to fire a software trigger via DCAM.
Returns:
True if successful
"""
with self.get_camera() as camera:
camera.trigger()
return True
def _stop(self) -> bool:
"""
Device specific code to abort the exposure and release queued buffers.
Returns:
True if successful
"""
self._stop_acquisition_thread()
with self.get_camera() as camera:
camera.abort()
camera.clear_buffer()
return True
def _get_status(self) -> dict:
"""
Device specific code to get a dict with the current device status.
Returns:
A dict with the device status
"""
return {"foo": "bar"}
def _set_config(self, config: dict) -> bool:
"""
Device specific code to configure camera features.
Args:
config: A valid configuration dict passed from :meth:`set_config`
Returns:
True if configuration is possible
"""
with self.get_camera() as camera:
camera.exposure_time = config.get("exposure_time", 1.0)
camera.gain = config.get("gain", 1.0) # ty: ignore[unresolved-attribute]
w, h = config.get("width"), config.get("height")
camera.roi = (config.get("x_offset", 0), config.get("y_offset", 0), w, h) if None not in (w, h) else None
camera.h_binning = config.get("h_binning", 1)
camera.v_binning = config.get("v_binning", 1)
if "auto_trigger" in config:
self._auto_trigger = config["auto_trigger"]
return True
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
"""
return self.get_configuration().get("exposure_time")
def _set_exposure_time(self, exposure_time: float) -> dict:
"""Return config patch with the exposure time.
Args:
exposure_time: exposure time in seconds
Returns:
config patch dict
"""
return {"exposure_time": exposure_time}
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()
w, h = c.get("width"), c.get("height")
if None in (w, h):
return None
return (c.get("x_offset", 0), c.get("y_offset", 0), w, h)
def _set_roi_coordinates(self, roi: tuple[int, int, int, int]) -> dict:
"""Return config patch with ROI fields.
Args:
roi: (x_offset, y_offset, width, height)
Returns:
config patch dict
"""
x_offset, y_offset, width, height = roi
return {"x_offset": x_offset, "y_offset": y_offset, "width": width, "height": height}
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("h_binning"), c.get("v_binning")
if None in (h, v):
return None
return (h, v)
def _set_binning(self, binning: tuple[int, int]) -> dict:
"""Return config patch with binning factors.
Args:
binning: (horizontal, vertical) binning factors
Returns:
config patch dict
"""
h_bin, v_bin = binning
return {"h_binning": h_bin, "v_binning": v_bin}
def _get_available_options(self) -> dict:
"""Return the configurable options of the dummy camera.
Returns:
Dict mapping option names to their current values and metadata.
"""
config = self.get_configuration()
return {
"exposure_time": {"value": config.get("exposure_time"), "writable": True},
"width": {"value": config.get("width"), "writable": True},
"height": {"value": config.get("height"), "writable": True},
"x_offset": {"value": config.get("x_offset", 0), "writable": True},
"y_offset": {"value": config.get("y_offset", 0), "writable": True},
"h_binning": {"value": config.get("h_binning", 1), "writable": True},
"v_binning": {"value": config.get("v_binning", 1), "writable": True},
"gain": {"value": config.get("gain"), "writable": True},
"frame_rate": {"value": config.get("frame_rate"), "writable": True},
"frame_count": {"value": config.get("frame_count"), "writable": True},
"auto_trigger": {"value": self._auto_trigger, "writable": True},
}
def _get_gain(self) -> float | None:
"""Return the current gain from the active configuration.
Returns:
factor that scales the generated amplitude, or None if not set
"""
return self.get_configuration().get("gain")
def _set_gain(self, gain: float) -> dict:
"""Return config patch with the gain.
Args:
gain: factor >= 0 that scales the generated amplitude, 1 leaves it unchanged, 0 gives black images
Returns:
config patch dict
Raises:
ValueError: if the gain is negative
"""
if gain < 0:
msg = f"The gain must not be negative, got {gain!r}"
raise ValueError(msg)
return {"gain": gain}
def _get_frame_rate(self) -> float | None:
"""Return the current frame rate limit from the active configuration.
Returns:
frame rate in Hz, or None if unthrottled
"""
return self.get_configuration().get("frame_rate")
def _set_frame_rate(self, frame_rate: float | None) -> dict:
"""Return config patch with the frame rate limit.
Args:
frame_rate: frame rate in Hz, None removes the limit
Returns:
config patch dict
"""
return {"frame_rate": frame_rate}
def _arm(self) -> bool:
"""
Device specific code to arm the camera with the currently active configuration.
Returns:
True if arming was successful else False
"""
try:
with self.get_camera() as camera:
camera.frame_count = self.get_configuration()["frame_count"]
camera.arm()
self._start_acquisition_thread() # has to be implement for the specific device
except Exception as e: # noqa: BLE001
log.error(e)
return False
return True
def _start_acquisition_thread(self) -> None:
"""Start the acquisition thread."""
log.debug("Starting acquisition thread")
self._stop_acquisition_event.clear()
self._acquisition_thread = threading.Thread(target=self._acquisition_loop)
self._acquisition_thread.daemon = False # daemon thread?
self._acquisition_thread.start()
def _stop_acquisition_thread(self) -> None:
"""Stop the acquisition thread and wait for it to terminate."""
thread = self._acquisition_thread
if thread is not None:
if threading.current_thread().ident != thread.ident:
# set the stop event and mark acquisition as not running
self._stop_acquisition_event.set()
self.acquisition_running = False
# join the thread to wait for its termination
thread.join(timeout=1)
# check if the thread is still alive
if thread.is_alive():
log.warn("Acquisition thread did not terminate gracefully")
else:
log.debug("Acquisition thread stopped successfully")
self._acquisition_thread = None
self.acquisition_running = False
def _acquisition_loop(self) -> None:
"""Grab all images from queued buffers and release buffers."""
images = []
frame_count = self.get_configuration()["frame_count"]
frame_id = 0
frame_rate = self.get_configuration().get("frame_rate")
next_frame_time = time.monotonic()
with self.get_camera() as camera:
while not self._stop_acquisition_event.is_set() and (frame_id < frame_count or frame_count < 0):
log.debug(f"Waiting for frame {frame_id} / {frame_count}")
if self._auto_trigger:
camera.trigger()
try:
image = camera.get_image()
except RuntimeError:
continue
images.append(image)
# emit image via event
self.acquisition_data(image, {"frame": frame_id})
frame_id += 1
if frame_rate is not None:
# wait on the stop event so that stop() is not delayed by the throttle
# fixed schedule without drift, after a delay at most one frame follows without a wait
next_frame_time = max(next_frame_time + 1 / frame_rate, time.monotonic())
self._stop_acquisition_event.wait(max(0.0, next_frame_time - time.monotonic()))
log.debug("Stopping exposure")
camera.clear_buffer()
# cleanup
self.acquisition_stopped({"frames": len(images), "frame_count": frame_count})
if len(images) != frame_count and frame_count >= 0:
log.error(f"Incorrect number of received frames: {len(images)} instead of {frame_count}!")
self.stop()
if self._acquisition_thread is threading.current_thread():
self._acquisition_thread = None
self.acquisition_running = False