Custom Acquisitions

General Guidelines

It is possible to create custom acquisition classes that can be used with the LDAQ library. This is done by subclassing the LDAQ.acquisition_base.BaseAcquisition class and implementing the abstract methods. The following is a list of guidelines that should be followed when creating custom acquisition classes:

class LDAQ.acquisition_base.BaseAcquisition

Parent acquisition class that should be used when creating new child acquisition source class. Child class should override methods the following methods:

• self.__init__()

• self.set_data_source()

• self.terminate_data_source()

• self.read_data()

• self.clear_buffer() (optional)

• self.get_sample_rate() (optional)

For further information on how to override these methods, see the listed methods docstrings.

Additionally, the __init__() or set_data_source() methods should override or be able to set the following attributes:

• self._channel_names_init - list of original data channels names from source

• self._channel_names_video_init - list of original video channels names from source

• self._channel_shapes_video_init - list of original video channels shapes from source

• self.sample_rate = 0 - sample rate of acquisition source

clear_buffer() → None

EDIT in child class (Optional).

The source buffer should be cleared with this method. It can either clear the buffer, or just read the data with self.read_data() and does not add/save data anywhere. By default, this method will read the data from the source and not add/save data anywhere.

Returns None.

get_sample_rate() → float

EDIT in child class (Optional).

Returns sample rate of acquisition class.

read_data() → ndarray

EDIT in child class.

This method only reads data from the source and transforms data into standard format used by other methods. It is called within self.acquire() method which properly handles acquiring data and saves it into pyTrigger ring buffer.

Must ALWAYS return a 2D numpy array of shape (n_samples, n_columns).

IMPORTANT: If some of the channels are videos (2D array - so shape is (n_samples, n_pixels_width, n_pixels_height)), then data has to be reshaped to shape (n_samples, n_pixels_width*n_pixels_height). Then data from multiple sources have to be concatenated into one array of shape (n_samples, n_cols), where cols is the combined number of pixel of all video sources and number of channels in data sources.

For an example where data source has 2 video sources with resolution 300x200 and 2 data channels, the final shape returned by this methods should be (n_samples, 300*200*2+2).

For video sources, the shape of the video is automatically stored in self.channel_shapes_video_init when self.set_data_source() is called. When data is retrieved from the source, it is reshaped to (n_samples, n_pixels_width, n_pixels_height).

Returns:

2D numpy array of shape (n_samples, n_columns)

Return type:

data (np.ndarray)

terminate_data_source() → None

EDIT in child class.

Properly closes/disconnects acquisition source after the measurement. The method should be able to handle mutliple calls in a row.

Example

In the example below, the source code of National Instrument acquisition class is shown.

Source code of National Instrument acquisition class

from __future__ import annotations

import numpy as np

from ..acquisition_base import BaseAcquisition

_NIDAQWRAPPER_AVAILABLE = False
try:
    from nidaqwrapper import AITask, get_task_by_name
    _NIDAQWRAPPER_AVAILABLE = True
except ImportError:
    pass


class NIAcquisition(BaseAcquisition):
    """Acquisition class for National Instruments devices via nidaqwrapper.

    A thin wrapper around ``nidaqwrapper.AITask`` that satisfies the
    ``BaseAcquisition`` contract. Supports both programmatic tasks
    (``AITask`` objects) and tasks defined in NI MAX (task name strings).

    Parameters
    ----------
    task : nidaqwrapper.AITask or str
        Either a fully configured ``AITask`` instance or the name of a task
        saved in NI MAX.  When a string is supplied the task is loaded via
        ``nidaqwrapper.get_task_by_name()``.
    acquisition_name : str or None, optional
        Human-readable name for this acquisition source.  Defaults to the
        underlying NI task name when ``None``.

    Raises
    ------
    ImportError
        If the ``nidaqwrapper`` package is not installed.
    TypeError
        If ``task`` is not an ``AITask`` instance or a string.

    Examples
    --------
    Wrap a programmatically created task:

    >>> ai = AITask("my_task", sample_rate=10_000)
    >>> ai.add_channel(...)
    >>> acq = NIAcquisition(ai)

    Load a task saved in NI MAX:

    >>> acq = NIAcquisition("MyNIMaxTask")
    """

    def __init__(
        self,
        task: AITask | str,
        acquisition_name: str | None = None,
    ) -> None:
        """
        Initialize NIAcquisition.

        Parameters
        ----------
        task : nidaqwrapper.AITask or str
            Source task: either an ``AITask`` object or an NI MAX task name.
        acquisition_name : str or None, optional
            Name for this acquisition source.  Uses the task name when None.

        Raises
        ------
        ImportError
            If ``nidaqwrapper`` is not installed.
        TypeError
            If ``task`` is neither an ``AITask`` nor a string.
        """
        if not _NIDAQWRAPPER_AVAILABLE:
            raise ImportError(
                "nidaqwrapper is not installed. "
                "Install it before using NIAcquisition."
            )

        super().__init__()

        if isinstance(task, AITask):
            self._ai_task: AITask = task
        elif isinstance(task, str):
            ni_task = get_task_by_name(task)
            if ni_task is None:
                raise ValueError(f"NI MAX task '{task}' not found.")
            self._ai_task = AITask.from_task(ni_task)
        else:
            raise TypeError(
                f"task must be an AITask instance or a string (NI MAX task name), "
                f"got {type(task).__name__!r}."
            )

        self.acquisition_name = (
            acquisition_name if acquisition_name is not None else self._ai_task.task_name
        )
        self.sample_rate = self._ai_task.sample_rate
        self._channel_names_init = list(self._ai_task.channel_list)
        self._task_active = False

        self._set_all_channels()  # populate channel_names_all before set_trigger
        self.set_trigger(1e20, 0, duration=1.0)

    def set_data_source(self) -> None:
        """Start the underlying AITask in preparation for acquisition.

        Calls ``super().set_data_source()`` at the end to satisfy the
        ``BaseAcquisition`` contract.
        """
        if self._task_active:
            return

        self._ai_task.start()
        self._task_active = True
        super().set_data_source()

    def terminate_data_source(self) -> None:
        """Stop the NIDAQmx task.

        Safe to call multiple times; subsequent calls when the task is
        already stopped are silently ignored.
        """
        if not self._task_active:
            return

        self._ai_task.task.stop()
        self._task_active = False

    def read_data(self) -> np.ndarray:
        """Read all currently available samples from the device buffer.

        Returns
        -------
        np.ndarray
            2-D array of shape ``(n_samples, n_channels)``.  Returns an
            empty array of shape ``(0, n_channels)`` when no data is
            available.
        """
        n_channels = len(self._channel_names_init)

        raw = self._ai_task.acquire(n_samples=None)

        if raw is None or raw.size == 0:
            return np.empty((0, n_channels))

        return raw

    def clear_buffer(self) -> None:
        """Read and discard all data currently in the device buffer."""
        self._ai_task.acquire(n_samples=None)