"""Templates for creating camera device representations."""
import math
from functools import cache
from numbers import Real
from typing import ClassVar
from herosdevices.core.templates.acq_device import AcquisitionDeviceTemplate
from herosdevices.helper import FeatureNotSupportedError, log
from herosdevices.interfaces import atomiq
@cache
def _supported_features(cls: type["CameraTemplate"]) -> tuple[str, ...]:
"""Return the sorted supported feature names of a camera class, warning once per class on partial overrides."""
features = []
for feature, hooks in cls._FEATURE_HOOKS.items():
overridden = [getattr(cls, hook) is not getattr(CameraTemplate, hook) for hook in hooks]
if all(overridden):
features.append(feature)
elif any(overridden):
log.warning(f"{cls.__name__} overrides only one of {hooks} - feature '{feature}' is unsupported")
return tuple(sorted(features))
[docs]
class CameraTemplate(atomiq.Camera, AcquisitionDeviceTemplate):
"""
Template (base class) for cameras.
To make a functional camera device, the user needs to implement all abstract methods.
The camera helper hooks (``_get_*``/``_set_*`` for exposure time, ROI, binning, gain and frame rate) are optional.
A feature counts as supported if the subclass overrides both hooks of its pair, see
:meth:`supported_features`. Hooks that are not overridden raise :class:`FeatureNotSupportedError`.
In addition, this call does not cover the mechanism the actually retrieve the image buffers
from the device since it is typically special to each camera vendor/API. A general guideline
should be to start a separate thread for the acquisition which uses the _acquisition_lock
to prohibit concurrent exposures. For each received image, the event :meth:`acquisition_data`
should be called. In addition, :meth:`acquisition_stopped` should be emitted
after the acquisition.
"""
get_camera = AcquisitionDeviceTemplate.get_device
# feature name (public property name) -> (getter hook name, setter hook name)
_FEATURE_HOOKS: ClassVar[dict[str, tuple[str, str]]] = {
"exposure_time": ("_get_exposure_time", "_set_exposure_time"),
"roi_coordinates": ("_get_roi_coordinates", "_set_roi_coordinates"),
"binning": ("_get_binning", "_set_binning"),
"gain": ("_get_gain", "_set_gain"),
"frame_rate": ("_get_frame_rate", "_set_frame_rate"),
}
[docs]
def supported_features(self) -> list[str]:
"""Return the names of the camera helper features this class supports.
A feature is supported if the class overrides both its getter and its setter hook. If only one
of the two is overridden, a warning is logged once per class and the feature is treated as unsupported.
Returns:
sorted feature names, equal to the names of the corresponding properties
"""
return list(_supported_features(type(self)))
def _get_exposure_time(self) -> float | None:
"""Return the current exposure time in seconds from the active configuration.
Note:
Implementations must read this value from the stored configuration only,
never by querying the camera hardware directly.
Returns:
exposure time in seconds or `None` if not set
"""
raise FeatureNotSupportedError(type(self).__name__, "exposure_time")
def _set_exposure_time(self, exposure_time: float) -> dict: # noqa: ARG002
"""Return a config patch dict for the given exposure time.
Note:
Implementations must only compute and return the patch; they must not
write to the camera or mutate `self._config` directly.
Args:
exposure_time: exposure time in seconds
Returns:
dict to merge into the active configuration
"""
raise FeatureNotSupportedError(type(self).__name__, "exposure_time")
def _get_roi_coordinates(self) -> tuple[int, int, int, int] | None:
"""Return the current ROI from the active configuration.
Note:
Implementations must read this value from the stored configuration only,
never by querying the camera hardware directly.
Returns:
(x_offset, y_offset, width, height) in sensor pixels
"""
raise FeatureNotSupportedError(type(self).__name__, "roi_coordinates")
def _set_roi_coordinates(self, roi: tuple[int, int, int, int]) -> dict: # noqa: ARG002
"""Return a config patch dict for the given ROI.
Note:
Implementations must only compute and return the patch; they must not
write to the camera or mutate `self._config` directly.
Args:
roi: (x_offset, y_offset, width, height) in sensor pixels
Returns:
dict to merge into the active configuration
"""
raise FeatureNotSupportedError(type(self).__name__, "roi_coordinates")
def _get_binning(self) -> tuple[int, int] | None:
"""Return the current binning from the active configuration.
Note:
Implementations must read this value from the stored configuration only,
never by querying the camera hardware directly.
Returns:
(horizontal, vertical) binning factors
"""
raise FeatureNotSupportedError(type(self).__name__, "binning")
def _set_binning(self, binning: tuple[int, int]) -> dict: # noqa: ARG002
"""Return a config patch dict for the given binning factors.
Note:
Implementations must only compute and return the patch; they must not
write to the camera or mutate `self._config` directly.
Args:
binning: (horizontal, vertical) binning factors
Returns:
dict to merge into the active configuration
"""
raise FeatureNotSupportedError(type(self).__name__, "binning")
def _get_gain(self) -> float | None:
"""Return the current gain from the active configuration.
Note:
Implementations must read this value from the stored configuration only,
never by querying the camera hardware directly.
Returns:
gain in the native unit of the driver or `None` if not set
"""
raise FeatureNotSupportedError(type(self).__name__, "gain")
def _set_gain(self, gain: float) -> dict: # noqa: ARG002
"""Return a config patch dict for the given gain.
Note:
Implementations must only compute and return the patch; they must not
write to the camera or mutate `self._config` directly. A patch that enables a manual gain must put
the key that disables automatic gain before the gain key.
Args:
gain: gain in the native unit of the driver
Returns:
dict to merge into the active configuration
"""
raise FeatureNotSupportedError(type(self).__name__, "gain")
def _get_frame_rate(self) -> float | None:
"""Return the current frame rate limit from the active configuration.
Note:
Implementations must read this value from the stored configuration only,
never by querying the camera hardware directly.
Returns:
frame rate limit in Hz or `None` if the limit is removed or not set
"""
raise FeatureNotSupportedError(type(self).__name__, "frame_rate")
def _set_frame_rate(self, frame_rate: float | None) -> dict: # noqa: ARG002
"""Return a config patch dict for the given frame rate limit.
Note:
Implementations must only compute and return the patch; they must not
write to the camera or mutate `self._config` directly.
Args:
frame_rate: frame rate limit in Hz, `None` removes the limit
Returns:
dict to merge into the active configuration
"""
raise FeatureNotSupportedError(type(self).__name__, "frame_rate")
@property
def exposure_time(self) -> float | None:
"""Exposure time in seconds. `None` if not set or not supported, see :meth:`supported_features`."""
try:
return self._get_exposure_time()
except FeatureNotSupportedError:
return None
@exposure_time.setter
def exposure_time(self, exposure_time: float) -> None:
self._apply_config_patch(self._set_exposure_time(exposure_time), "exposure_time")
@property
def roi_coordinates(self) -> tuple[int, int, int, int] | None:
"""
ROI as (x_offset, y_offset, width, height) in sensor pixels.
`None` if not set or not supported, see :meth:`supported_features`.
"""
try:
return self._get_roi_coordinates()
except FeatureNotSupportedError:
return None
@roi_coordinates.setter
def roi_coordinates(self, roi: tuple[int, int, int, int]) -> None:
self._apply_config_patch(self._set_roi_coordinates(roi), "roi_coordinates")
@property
def binning(self) -> tuple[int, int] | None:
"""Binning as (horizontal, vertical). `None` if not set or not supported, see :meth:`supported_features`."""
try:
return self._get_binning()
except FeatureNotSupportedError:
return None
@binning.setter
def binning(self, binning: tuple[int, int]) -> None:
self._apply_config_patch(self._set_binning(binning), "binning")
@property
def gain(self) -> float | None:
"""
Gain in the native unit of the driver, see :ref:`camera-gain-frame-rate`.
The setter raises :class:`ValueError` for NaN, infinity, booleans and non-numbers. The valid range is driver
specific and checked by the driver. The getter returns `None` if not set or not supported, see
:meth:`supported_features`.
"""
try:
return self._get_gain()
except FeatureNotSupportedError:
return None
@gain.setter
def gain(self, gain: float) -> None:
if isinstance(gain, bool) or not isinstance(gain, Real) or not math.isfinite(gain):
msg = f"The gain must be a finite number, got {gain!r}"
raise ValueError(msg)
self._apply_config_patch(self._set_gain(gain), "gain")
@property
def frame_rate(self) -> float | None:
"""
Frame rate limit in Hz, see :ref:`camera-gain-frame-rate`.
Setting `None` removes the limit. Other values must be positive finite numbers, not booleans, otherwise the
setter raises :class:`ValueError`. The getter returns `None` if the limit is removed, if no frame rate is
configured and if the feature is not supported, see :meth:`supported_features`.
"""
try:
return self._get_frame_rate()
except FeatureNotSupportedError:
return None
@frame_rate.setter
def frame_rate(self, frame_rate: float | None) -> None:
if frame_rate is not None and (
isinstance(frame_rate, bool)
or not isinstance(frame_rate, Real)
or not (math.isfinite(frame_rate) and frame_rate > 0)
):
msg = f"The frame rate must be a positive finite number or None, got {frame_rate!r}"
raise ValueError(msg)
self._apply_config_patch(self._set_frame_rate(frame_rate), "frame_rate")