"""Dummy camera devices for use in testing. Do not use in production."""
import threading
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.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, 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."""
_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)
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 _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()
self.acquisition_running = True
self.acquisition_started(self.get_configuration())
def _stop_acquisition_thread(self) -> None:
"""Stop the acquisition thread and wait for it to terminate."""
if self._acquisition_thread is not None:
if threading.current_thread().ident != self._acquisition_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
self._acquisition_thread.join(timeout=1)
# check if the thread is still alive
if self._acquisition_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
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
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()
self.acquisition_running = False