Add nidaqmx.stream_readers.AnalogSingleChannelReader.read_waveform() (#798)

* first draft - based on the PoC with just enough changes to run the example without crashing

* fix waveform.timing

* clean up pyproject.toml

* clean up formatting

* more cleanup

* fix extended_properties

* generate stream_readers.py properly

* generate _library_interpreter.py properly

* generate types.py properly

* generate constants.py properly

* test___analog_single_channel_reader___read_waveform___returns_valid_waveform

* BaseInterpreter.internal_read_analog_waveform_ex()

* cleanup

* cleanup

* assert channel_index == 0

* updated changelog

* add documentation

* add documentation (generated)

* move set_wfm_attr_callback to LibraryInterpreter

* refactor

* revert types.py

* whitespace

* initialize arrays to zero instead of inf or min

* update pyproject.toml

* Add nitypes to the intersphinx mapping

* refactor to avoid unecessary copies and allow reading into an existing waveform

* move all the waveform stuff into the interpreter

* "for reading samples into."

* _INT64_WFM_SEC_PER_TICK and _T0_EPOCH should be module-level constants

* fix mypy issue on linux?

* make _internal_read_analog_waveform_ex internal to LIbraryInterpreter

* better handling for waveform buffer size issues

* use hightime

* update tests

* update tests

* f"{sim_6363_device.name}/ai0"

---------

Co-authored-by: Mike Prosser <[email protected]>

Author: mikeprosserni

CHANGELOG.md

@@ -32,7 +32,7 @@ All notable changes to this project will be documented in this file.
     * ...
 
 * ### Major Changes
-    * ...
+    * (IN PROGRESS) Added support for reading and writing Waveform data. 
 
 * ### Known Issues
     * ...

docs/conf.py

@@ -47,6 +47,7 @@
 
 intersphinx_mapping = {
     "grpc": ("https://grpc.github.io/grpc/python/", None),
+    "nitypes": ("https://nitypes.readthedocs.io/en/latest/", None),
     "numpy": ("https://numpy.org/doc/stable/", None),
     "protobuf": ("https://googleapis.dev/python/protobuf/latest/", None),
     "python": ("https://docs.python.org/3", None),

examples/analog_in/voltage_acq_int_clk_wfm.py

@@ -0,0 +1,21 @@
+"""Example of analog input voltage waveform acquisition.
+
+This example demonstrates how to acquire a finite amount
+of data using the DAQ device's internal clock.
+"""
+
+import nidaqmx
+from nidaqmx.constants import AcquisitionType, READ_ALL_AVAILABLE
+from nidaqmx.stream_readers import AnalogSingleChannelReader
+
+with nidaqmx.Task() as task:
+    task.ai_channels.add_ai_voltage_chan("Dev1/ai0")
+    task.timing.cfg_samp_clk_timing(1000.0, sample_mode=AcquisitionType.FINITE, samps_per_chan=50)
+
+    reader = AnalogSingleChannelReader(task.in_stream)
+    waveform = reader.read_waveform(READ_ALL_AVAILABLE)
+    print(f"Acquired data: {waveform.scaled_data}")
+    print(f"Channel name: {waveform.channel_name}")
+    print(f"Unit description: {waveform.unit_description}")
+    print(f"t0: {waveform.timing.start_time}")
+    print(f"dt: {waveform.timing.sample_interval}")

generated/nidaqmx/_base_interpreter.py

@@ -1,6 +1,7 @@
 # Do not edit this file; it was automatically generated.
 import abc
-from typing import Optional
+import numpy
+from nitypes.waveform import AnalogWaveform
 
 
 class BaseEventHandler(abc.ABC):
@@ -1845,4 +1846,14 @@ def write_to_teds_from_file(
 
     @abc.abstractmethod
     def hash_task_handle(self, task_handle):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def read_analog_waveform(
+        self,
+        task_handle: object,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        waveform: AnalogWaveform[numpy.float64]
+    ) -> None:
         raise NotImplementedError

generated/nidaqmx/_grpc_interpreter.py

@@ -5,6 +5,7 @@
 import threading
 import typing
 import warnings
+from nitypes.waveform import AnalogWaveform
 from typing import Callable, Generic, TypeVar
 
 import google.protobuf.message
@@ -3600,6 +3601,14 @@ def set_runtime_environment(
     def internal_get_last_created_chan(self):
         raise NotImplementedError
 
+    def read_analog_waveform(
+        self,
+        task_handle: object,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        waveform: AnalogWaveform[numpy.float64]
+    ) -> None:
+        raise NotImplementedError
 
 def _assign_numpy_array(numpy_array, grpc_array):
     """

generated/nidaqmx/_library_interpreter.py

@@ -1,24 +1,57 @@
 # Do not edit this file; it was automatically generated.
 
+from __future__ import annotations
 import ctypes
 import logging
+import numpy
 import platform
 import warnings
-from typing import Optional
-
-import numpy
-from typing import List
+import sys
+from enum import Enum
+from datetime import timezone
+from hightime import datetime as ht_datetime
+from hightime import timedelta as ht_timedelta
+from typing import Callable, List, Optional, Sequence, Tuple, TYPE_CHECKING
 
 from nidaqmx._base_interpreter import BaseEventHandler, BaseInterpreter
-from nidaqmx._lib import lib_importer, ctypes_byte_str, c_bool32, wrapped_ndpointer
+from nidaqmx._lib import lib_importer, ctypes_byte_str, c_bool32, wrapped_ndpointer, TaskHandle
+from nidaqmx.constants import FillMode
 from nidaqmx.error_codes import DAQmxErrors, DAQmxWarnings
 from nidaqmx.errors import DaqError, DaqFunctionNotSupportedError, DaqReadError, DaqWarning, DaqWriteError
 from nidaqmx._lib_time import AbsoluteTime
+from nitypes.waveform.typing import ExtendedPropertyValue
+from nitypes.waveform import AnalogWaveform, SampleIntervalMode, Timing, ExtendedPropertyDictionary
 
+if TYPE_CHECKING:
+    if sys.version_info >= (3, 10):
+        from typing import TypeAlias
+    else:
+        from typing_extensions import TypeAlias
 
 _logger = logging.getLogger(__name__)
 _was_runtime_environment_set = None
 
+_INT64_WFM_SEC_PER_TICK = 100e-9
+_T0_EPOCH = ht_datetime(1, 1, 1, tzinfo=timezone.utc)
+
+# typedef int32 (CVICALLBACK *DAQmxSetWfmAttrCallbackPtr)(uInt32 channelIndex, const char attributeName[], int32 attributeType, const void* value, uInt32 valueSizeInBytes, void *callbackData);  # noqa: W505 - doc line too long
+CSetWfmAttrCallbackPtr = ctypes.CFUNCTYPE(
+    ctypes.c_int32,  # return value (error code)
+    ctypes.c_uint32,  # channel_index
+    ctypes.c_char_p,  # attribute_name
+    ctypes.c_int32,  # attribute_type
+    ctypes.c_void_p,  # value
+    ctypes.c_uint32,  # value_size_in_bytes
+    ctypes.c_void_p,  # callback_data
+)
+
+class WfmAttrType(Enum):
+    BOOL32 = 1
+    FLOAT64 = 2
+    INT32 = 3
+    STRING = 4
+
+SetWfmAttrCallback: TypeAlias = Callable[[int, str, WfmAttrType, ExtendedPropertyValue, object], int]
 
 class LibraryEventHandler(BaseEventHandler):
     """Manage the lifetime of a ctypes callback method pointer.
@@ -6338,6 +6371,152 @@ def get_extended_error_info(self):
             return 'Failed to retrieve error description.'
         return error_buffer.value.decode(lib_importer.encoding)
 
+    def read_analog_waveform(
+        self,
+        task_handle: object,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        waveform: AnalogWaveform[numpy.float64]
+    ) -> None:
+        """Read an analog waveform with timing and attributes."""
+        error_code, samples_read, timestamps, sample_intervals = self._internal_read_analog_waveform_ex(
+            task_handle,
+            1, # single channel
+            number_of_samples_per_channel,
+            timeout,
+            FillMode.GROUP_BY_CHANNEL.value,
+            waveform.raw_data,
+            [waveform.extended_properties]
+        )
+
+        waveform.timing = Timing(
+            sample_interval_mode=SampleIntervalMode.REGULAR,
+            timestamp=timestamps[0],
+            sample_interval=sample_intervals[0],
+        )
+
+        # TODO: AB#3228924 - if the read was short, set waveform.sample_count before throwing the exception
+        self.check_for_error(error_code, samps_per_chan_read=samples_read)
+
+    def _internal_read_analog_waveform_ex(
+        self,
+        task_handle: object,
+        channel_count: int,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        fill_mode: int,
+        read_array: numpy.typing.NDArray[numpy.float64],
+        properties: Sequence[ExtendedPropertyDictionary]
+    ) -> Tuple[
+        int, # error code
+        int, # The number of samples per channel that were read
+        Sequence[ht_datetime], # The timestamps for each sample, indexed by channel
+        Sequence[ht_timedelta], # The sample intervals, indexed by channel
+    ]:
+        assert isinstance(task_handle, TaskHandle)
+        samps_per_chan_read = ctypes.c_int()
+
+        cfunc = lib_importer.windll.DAQmxInternalReadAnalogWaveformEx
+        if cfunc.argtypes is None:
+            with cfunc.arglock:
+                if cfunc.argtypes is None:
+                    cfunc.argtypes = [
+                        TaskHandle,
+                        ctypes.c_int,
+                        ctypes.c_double,
+                        ctypes.c_int,
+                        wrapped_ndpointer(dtype=numpy.int64, flags=("C", "W")),
+                        wrapped_ndpointer(dtype=numpy.int64, flags=("C", "W")),
+                        ctypes.c_uint,
+                        CSetWfmAttrCallbackPtr,
+                        ctypes.c_void_p,
+                        wrapped_ndpointer(dtype=numpy.float64, flags=("C", "W")),
+                        ctypes.c_uint,
+                        ctypes.POINTER(ctypes.c_int),
+                        ctypes.POINTER(c_bool32),
+                    ]
+
+        t0_array = numpy.zeros(channel_count, dtype=numpy.int64)
+        dt_array = numpy.zeros(channel_count, dtype=numpy.int64)
+
+        def set_wfm_attr_callback(
+            channel_index: int,
+            attribute_name: str,
+            attribute_type: WfmAttrType,
+            value: ExtendedPropertyValue,
+            callback_data: object,
+        ) -> int:
+            properties[channel_index][attribute_name] = value
+            return 0
+
+        error_code = cfunc(
+            task_handle,
+            number_of_samples_per_channel,
+            timeout,
+            fill_mode,
+            t0_array,
+            dt_array,
+            0 if t0_array is None else t0_array.size,
+            self._get_wfm_attr_callback_ptr(set_wfm_attr_callback),
+            None,
+            read_array,
+            read_array.size,
+            ctypes.byref(samps_per_chan_read),
+            None,
+        )
+
+        timestamps = [_T0_EPOCH + ht_timedelta(seconds=t0 * _INT64_WFM_SEC_PER_TICK) for t0 in t0_array]
+        sample_intervals = [ht_timedelta(seconds=dt * _INT64_WFM_SEC_PER_TICK) for dt in dt_array]
+
+        return error_code, samps_per_chan_read.value, timestamps, sample_intervals
+
+    def _get_wfm_attr_value(
+        self, attribute_type: int, value: ctypes.c_void_p, value_size_in_bytes: int
+    ) -> ExtendedPropertyValue:
+        if attribute_type == WfmAttrType.BOOL32.value:
+            assert value_size_in_bytes == 4
+            return ctypes.cast(value, ctypes.POINTER(ctypes.c_int32))[0] != 0
+        elif attribute_type == WfmAttrType.FLOAT64.value:
+            assert value_size_in_bytes == 8
+            return float(ctypes.cast(value, ctypes.POINTER(ctypes.c_double))[0])
+        elif attribute_type == WfmAttrType.INT32.value:
+            assert value_size_in_bytes == 4
+            return int(ctypes.cast(value, ctypes.POINTER(ctypes.c_int32))[0])
+        elif attribute_type == WfmAttrType.STRING.value:
+            value_c_bytes = ctypes.cast(value, ctypes.POINTER(ctypes.c_byte))
+            assert value_c_bytes[value_size_in_bytes - 1] == 0
+            return bytes(value_c_bytes[0 : value_size_in_bytes - 1]).decode(lib_importer.encoding)
+        else:
+            raise ValueError(f"Unsupported attribute type {attribute_type}")
+
+    def _get_wfm_attr_callback_ptr(
+        self, set_wfm_attr_callback: Optional[SetWfmAttrCallback]
+    ) -> ctypes._FuncPointer:
+        if set_wfm_attr_callback is None:
+            return CSetWfmAttrCallbackPtr()
+
+        def _invoke_callback(
+            channel_index: int,
+            attribute_name: bytes,
+            attribute_type: int,
+            value: ctypes.c_void_p,
+            value_size_in_bytes: int,
+            callback_data: object,
+        ) -> int:
+            try:
+                return set_wfm_attr_callback(
+                    channel_index,
+                    attribute_name.decode(lib_importer.encoding),
+                    WfmAttrType(attribute_type),
+                    self._get_wfm_attr_value(attribute_type, value, value_size_in_bytes),
+                    callback_data,
+                )
+            except Exception:
+                _logger.exception("Unhandled exception in set_wfm_attr_callback")
+                return -1
+
+        return CSetWfmAttrCallbackPtr(_invoke_callback)
+
     def read_id_pin_memory(self, device_name, id_pin_name):
         data_length_read = ctypes.c_uint()
         format_code = ctypes.c_uint()