"""Template for general acquisition devices like cameras and oscilloscopes."""
import threading
from abc import abstractmethod
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Any
import numpy as np
from heros import event
from herosdevices.helper import log, merge_dicts
from herosdevices.interfaces import UnionInterfaceMeta
from herosdevices.interfaces.heros import ConfiguredDevice
[docs]
class AcquisitionDeviceTemplate(ConfiguredDevice, metaclass=UnionInterfaceMeta):
"""
Template (base class) for acquisition-like devices.
Template for devices which are based on the concept of configuring the device first and
then running an acquisition thread which collects data (like images, osci traces,...).
To make a functional device driver, the user needs to implement all abstract methods.
In addition, this call does not cover the mechanism the actually retrieve the data
from the device since it is typically special to each 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 data set, the event :meth:`acquisition_data`
should be called. In addition, :meth:`acquisition_stopped` should be emitted
after the acquisition.
The driver must also define the `default_config_dict` attribute, intended for configuration values which are applied
to all user configurations if the corresponding keys are not specified.
Two class attributes describe config keys the driver treats specially. ``_special_config_keys`` lists keys which
are handled by the driver itself instead of being passed on to the vendor SDK. They may be absent from a
configuration. ``_required_config_keys`` lists keys that must be present in every configuration, either set
explicitly or provided by ``default_config_dict``. :meth:`update_configuration` with ``merge=False`` enforces
the latter.
"""
_special_config_keys: list | tuple = ()
_required_config_keys: list | tuple = ()
# config keys of the patch that _apply_config_patch is applying, empty otherwise. Drivers read it in _set_config to
# tell keys written by a setter (a missing node raises FeatureNotSupportedError) from user keys
_patch_keys: frozenset[str] = frozenset()
acquisition_running: bool = False
payload_metadata: dict
default_config_dict: dict
def __init__(
self,
config_dict: dict,
default_config: str | None = None,
keep_device_open: bool = True,
payload_metadata: dict | None = None,
) -> None:
"""
Create a new device object.
Args:
config_dict: Configuration dictionary, see EXAMPLE_CONFIG
default_configuration: Default (starting) configuration
keep_device_open: Keep the device open after it is first opened
"""
self.keep_device_open = keep_device_open
self._device = None
self._acquisition_thread: threading.Thread | None = None
self._acquisition_lock = threading.Lock() # lock to ensure single exposure at a time
self._stop_acquisition_event = threading.Event() # event to signal the thread to stop
self._config_dict: dict = {}
self.update_configuration(config_dict)
self.default_config: str = default_config if default_config is not None else "default"
self._config: str = self.default_config
self.configure() # configure with default config
self.payload_metadata: dict = payload_metadata if payload_metadata is not None else {}
[docs]
@contextmanager
def get_device(self) -> Iterator[None]:
"""
Yield a device handle.
.. code-block:: python
# use the device in a with statement
with self.get_device() as camera:
camera.do_something()
"""
log.spam("Requesting device")
if self._device is None:
self.open()
yield self._device
if not self.keep_device_open:
self.teardown()
[docs]
def open(self) -> Any:
"""Open the device handler and assign it to `self._device`."""
log.info("Opening device handler")
self._device = self._open()
[docs]
def teardown(self) -> None:
"""Release the device handler and potentially de-initialize the API."""
log.info("Closing device")
self._teardown()
self._device = None
__del__ = teardown
[docs]
def start(self) -> bool:
"""
Fire a software trigger.
Returns:
True if successful
"""
log.info("Firing software trigger")
return self._start()
[docs]
def stop(self) -> bool:
"""
Abort the exposure and release queued buffers.
Returns:
True if successful
"""
log.error("Aborting any exposure")
return self._stop()
[docs]
def reset(self) -> None:
"""Reset the device by aborting any ongoing exposure, closing and re-opening the handler."""
log.error("Resetting device")
self.stop()
self.teardown()
try:
self.open()
self.configure()
except Exception as e: # noqa: BLE001
log.warn(f"Opening and configuration during reset failed: {e}")
[docs]
def get_status(self) -> dict:
"""
Get a dict with the current device status.
Returns:
A dict with the device status
"""
status = {
"acquisition_running": self.acquisition_running,
}
# add device specific status
status.update(self._get_status())
return status
[docs]
def get_available_options(self) -> dict:
"""
Get all configurable options/features of the device.
Returns:
Dict mapping option names to their properties (value, readable, writable, etc.).
"""
return self._get_available_options()
@property
def config_dict(self) -> dict:
"""
Get a copy of the configuration dict.
Returns:
Copy of the configuration dict
"""
return {**self._config_dict}
[docs]
def update_configuration(self, new_config_dict: dict, merge: bool = True) -> None:
"""
Update the configuration dict with new configuration.
Each dict key corresponds a (new) configuration name.
Each value is a dict with config property -> config value.
Args:
new_config_dict: A dict of configurations where the keys are the configuration names
merge: If ``True``, the new dict is recursively merged with the current set config dict. If ``False`` the
old configurations with the provided names (keys) are overwritten. In this case each new configuration,
merged over ``default_config_dict``, must contain all ``_required_config_keys``.
Raises:
ValueError: If ``merge`` is ``False`` and a required config key is missing. No configuration is changed.
"""
if merge:
self._config_dict = merge_dicts(self._config_dict, new_config_dict)
return
for config_name, config in new_config_dict.items():
for key in self._required_config_keys:
if key not in config and key not in self.default_config_dict:
msg = f"Required config key {key} not found in config {config_name}!"
raise ValueError(msg)
self._config_dict.update(new_config_dict)
[docs]
def get_configuration(self) -> dict:
"""
Get the currently active configuration.
Returns:
The currently active configuration.
"""
return merge_dicts(self.default_config_dict, self.config_dict[self._config])
set_configuration = configure # satisfy interface requirement
def _apply_config_patch(self, patch: dict, feature: str) -> None:
"""Merge a config patch into the active configuration and apply it to the device.
Patch keys are removed from the stored configuration of the active name and appended in patch order. Keys that
are also in ``default_config_dict`` keep their default position in :meth:`get_configuration`. If
:meth:`configure` returns `False` or raises, the stored configuration is restored and the device is
re-configured with the previous configuration. If that also fails, the failure is logged and the hardware may
be partially configured. While :meth:`configure` runs for the patch, ``_patch_keys`` holds the patch keys, so
a driver can raise :class:`~herosdevices.helper.FeatureNotSupportedError` if the device lacks the node of a
patch key.
Args:
patch: config patch, for example as returned by a camera helper hook
feature: name of the changed setting, used in error messages
Raises:
RuntimeError: if the acquisition is running (nothing is changed) or :meth:`configure` returned `False`
(the previous configuration is restored)
~herosdevices.helper.FeatureNotSupportedError: if the driver reports that the device lacks a patch key
(the previous configuration is restored)
ValueError: if the patched configuration lacks a required config key (nothing is changed)
"""
if self.acquisition_running:
msg = f"Could not set {feature}: acquisition running. Call .stop() and try again."
raise RuntimeError(msg)
name = self._config
previous = self._config_dict[name] # replaced below, not mutated, so this reference is a snapshot
merged = merge_dicts(previous, patch) # nested dicts are merged like update_configuration(merge=True)
self.update_configuration(
{name: {k: v for k, v in merged.items() if k not in patch} | {k: merged[k] for k in patch}}, merge=False
)
self._patch_keys = frozenset(patch)
try:
try:
success = self.configure(name)
finally:
self._patch_keys = frozenset() # cleared before the restore calls configure again
except BaseException:
self._restore_config(name, previous)
raise
if not success:
self._restore_config(name, previous)
msg = f"Could not set {feature}: _set_config failed. The previous configuration was restored."
raise RuntimeError(msg)
def _restore_config(self, name: str, previous: dict) -> None:
"""Restore a stored configuration and best-effort re-configure the device with it.
Args:
name: name of the active configuration
previous: stored configuration to restore
"""
# bypasses update_configuration: previous may have entered unvalidated (__init__, merge=True), and a failing
# validation here would mask the original error
self._config_dict[name] = previous
msg = f"Could not re-configure the device with the previous configuration '{name}'"
try:
if not self.configure(name):
log.error(msg)
except Exception: # noqa: BLE001 - best effort, the original error is raised by the caller
log.exception(msg)
[docs]
def arm(self, metadata: dict | None = None, kill_running: bool = False) -> bool:
"""
Arm the device with the currently active configuration.
Args:
metadata: Metadata that is merged into the current payload metadata dict which is send with every emitted
:code:`acquisition_data` event.
kill_running: If ``True`` any running acquisition will be stopped. If ``False`` and an acquisition is
already running, an error will be raised.
Returns:
True if arming was successful else False
"""
with self._acquisition_lock: # Acquire lock to prevent concurrent exposures
if self._acquisition_thread is not None:
if kill_running:
self.stop()
else:
log.error("Acquisition already in progress")
return False
if metadata is not None:
self.update_payload_metadata(metadata)
if self._arm():
# emit event
self.acquisition_started(self.get_configuration())
return True
return False
# events which emitted by each device
@event
def acquisition_data(self, frame: np.ndarray, metadata: dict | None = None) -> tuple:
"""
Event to emit new frames.
.. note::
The dtype of the frame is not changed here.
Args:
frame: The frame payload data (for example an image or an scope trace)
metadata: The metadata which is passed along the payload. This argument takes precedence over the
:code:`payload_metadata` attribute (for example set by the :code:`update_payload_metadata` method)
while merging the two dicts.
Returns:
A tuple of image and metadata(-dict)
"""
if metadata is None:
metadata = {}
log.info(f"Emitting frame without metadata with shape {frame.shape} and dtype {frame.dtype}")
else:
log.info(f"Emitting frame {metadata['frame']} with shape {frame.shape} and dtype {frame.dtype}")
metadata = self.payload_metadata | metadata
return frame, metadata
@event
def acquisition_started(self, metadata: dict | None = None) -> dict:
"""
Event emitted when the acquisition thread starts.
Returns:
A dict with acquisition metadata
"""
self.acquisition_running = True
if metadata is None:
return {}
return metadata
@event
def acquisition_stopped(self, metadata: dict | None = None) -> dict:
"""
Event emitted when the acquisition thread stops.
Returns:
A dict with acquisition metadata
"""
self.acquisition_running = False
if metadata is None:
return {}
return metadata
# the following methods need to be implement for specific devices
@abstractmethod
def _open(self) -> Any:
"""
Device specific code to open the device handler and return it.
.. code-block:: python
# open camera via vendor API
camera = vencor_api.get_handle()
camera.connect()
return camera
"""
@abstractmethod
def _teardown(self) -> None:
"""
Device specific code to release the device handler and potentially de-initialize the API.
.. code-block:: python
# close camera
with self.get_device() as camera:
camera.disconnect()
vendor_api.close()
"""
@abstractmethod
def _start(self) -> bool:
"""
Device specific code to fire a software trigger.
.. code-block:: python
# trigger camera via API
with self.get_device() as camera:
camera.trigger_now()
return True
Returns:
True if successful
"""
@abstractmethod
def _stop(self) -> bool:
"""
Device specific code to abort the exposure and release queued buffers.
.. attention::
Stop any running acquisition threads here. To signal a potention thread,
you can use :attr:`_stop_acquisition_event`.
.. code-block:: python
# abort exposure
with self.get_device() as camera:
camera.abort_exposure()
camera.buffers.release()
Returns:
True if successful
"""
@abstractmethod
def _get_status(self) -> dict:
"""
Device specific code to get a dict with the current device status.
.. code-block:: python
# get sensor temp
with self.get_device() as camera:
temp = camera.get_prop("sensor_temp")
return {"sensor_temp": temp}
Returns:
A dict with the device status
"""
@abstractmethod
def _set_config(self, config: dict) -> bool:
"""
Device specific code to configure device features.
.. code-block:: python
# apply config
for key, val in config:
try:
with self.get_device() as camera:
camera.apply_prop(name=key, value=val)
except Exception as e:
log.error(e)
return False
return True
Args:
config: A valid configuration dict passed from :meth:`set_config`
Returns:
True if configuration is possible
"""
@abstractmethod
def _arm(self) -> bool:
"""
Device specific code to arm the device with the currently active configuration.
.. code-block:: python
# arm exposure
try:
with self.get_device() as camera:
camera.arm(frame_count = 4)
self._start_acquisition_thread() # has to be implement for the specific device
return True
except Exception as e:
log.error(e)
return False
Returns:
True if arming was successful else False
"""
@abstractmethod
def _get_available_options(self) -> dict:
"""
Device specific code to enumerate all configurable options/features of the device.
.. code-block:: python
# enumerate camera features
options = {}
with self.get_device() as camera:
for feat in camera.get_all_features():
options[feat.get_name()] = {
"value": feat.get(),
"readable": feat.is_readable(),
"writable": feat.is_writable(),
}
return options
Note:
Devices that do not support enumerating options should raise a :class:`NotImplementedError`.
Returns:
Dict mapping option names to their properties (value, readable, writable, etc.).
Raises:
NotImplementedError: If the device does not support enumerating options.
"""