Add support for reading digital waveforms (#809)

* add read_digital_waveform() to Interpreters and DigitalSingleChannelReader

* read_digital_waveforms

* cleanup

* add digital support to Task.read_waveform()

* handle LineGrouping.CHAN_FOR_ALL_LINES

* cleanup

* test for ports and better asserts

* test cleanup

* cleanup and add test___digital_multi_channel_multi_line_reader___reuse_waveform_in_place___overwrites_data_timing_and_attributes

* from __future__ import annotations

* api cleanup

* minor test cleanups

* fix test assertion

* bytes_per_chan_array and interpreter.read_new_digital_waveforms()

* assert signal names, and test lines and port together

* tests with lines and port together in a single channel

* use constructor instead of DigitalWaveform.from_lines()

* cleanup

---------

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

Author: mikeprosserni

examples/digital_in/acq_dig_port_int_clk_wfm.py

@@ -0,0 +1,29 @@
+"""Example for reading digital signals.
+
+This example demonstrates how to input a finite digital pattern
+using the DAQ device's internal clock.
+"""
+
+import os
+
+os.environ["NIDAQMX_ENABLE_WAVEFORM_SUPPORT"] = "1"
+
+import nidaqmx  # noqa: E402 # Must import after setting environment variable
+from nidaqmx.constants import (  # noqa: E402
+    READ_ALL_AVAILABLE,
+    AcquisitionType,
+    LineGrouping,
+)
+
+with nidaqmx.Task() as task:
+    task.di_channels.add_di_chan(
+        "cdaqTesterMod4/port0", line_grouping=LineGrouping.CHAN_FOR_ALL_LINES
+    )
+    task.timing.cfg_samp_clk_timing(1000.0, sample_mode=AcquisitionType.FINITE, samps_per_chan=50)
+
+    waveform = task.read_waveform(READ_ALL_AVAILABLE)
+    print("Acquired data:")
+    print(waveform.data)
+    print(f"Channel name: {waveform.channel_name}")
+    print(f"t0: {waveform.timing.start_time}")
+    print(f"dt: {waveform.timing.sample_interval}")

generated/nidaqmx/_base_interpreter.py

@@ -1,7 +1,9 @@
 # Do not edit this file; it was automatically generated.
+from __future__ import annotations
+
 import abc
 import numpy
-from nitypes.waveform import AnalogWaveform
+from nitypes.waveform import AnalogWaveform, DigitalWaveform
 from typing import Sequence
 from nidaqmx.constants import WaveformAttributeMode
 
@@ -1858,7 +1860,7 @@ def read_analog_waveform(
         timeout: float,
         waveform: AnalogWaveform[numpy.float64],
         waveform_attribute_mode: WaveformAttributeMode
-    ) -> None:
+    ) -> int:
         raise NotImplementedError
 
     @abc.abstractmethod
@@ -1869,5 +1871,41 @@ def read_analog_waveforms(
         timeout: float,
         waveforms: Sequence[AnalogWaveform[numpy.float64]],
         waveform_attribute_mode: WaveformAttributeMode
-    ) -> None:
-        raise NotImplementedError
+    ) -> int:
+        raise NotImplementedError
+        
+    @abc.abstractmethod
+    def read_digital_waveform(
+        self,
+        task_handle: object,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        waveform: DigitalWaveform[numpy.uint8],
+        waveform_attribute_mode: WaveformAttributeMode
+    ) -> int:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def read_digital_waveforms(
+        self,
+        task_handle: object,
+        channel_count: int,
+        number_of_samples_per_channel: int,
+        number_of_signals_per_sample: int,
+        timeout: float,
+        waveforms: Sequence[DigitalWaveform[numpy.uint8]],
+        waveform_attribute_mode: WaveformAttributeMode,
+    ) -> int:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def read_new_digital_waveforms(
+        self,
+        task_handle: object,
+        channel_count: int,
+        number_of_samples_per_channel: int,
+        number_of_signals_per_sample: int,
+        timeout: float,
+        waveform_attribute_mode: WaveformAttributeMode,
+    ) -> Sequence[DigitalWaveform[numpy.uint8]]:
+        raise NotImplementedError

generated/nidaqmx/_grpc_interpreter.py

@@ -5,7 +5,7 @@
 import threading
 import typing
 import warnings
-from nitypes.waveform import AnalogWaveform
+from nitypes.waveform import AnalogWaveform, DigitalWaveform
 from typing import Callable, Generic, Sequence, TypeVar
 
 import google.protobuf.message
@@ -3609,7 +3609,7 @@ def read_analog_waveform(
         timeout: float,
         waveform: AnalogWaveform[numpy.float64],
         waveform_attribute_mode: WaveformAttributeMode
-    ) -> None:
+    ) -> int:
         raise NotImplementedError
 
     def read_analog_waveforms(
@@ -3619,7 +3619,40 @@ def read_analog_waveforms(
         timeout: float,
         waveforms: Sequence[AnalogWaveform[numpy.float64]],
         waveform_attribute_mode: WaveformAttributeMode
-    ) -> None:
+    ) -> int:
+        raise NotImplementedError
+
+    def read_digital_waveform(
+        self,
+        task_handle: object,
+        number_of_samples_per_channel: int,
+        timeout: float,
+        waveform: DigitalWaveform[numpy.uint8],
+        waveform_attribute_mode: WaveformAttributeMode
+    ) -> int:
+        raise NotImplementedError
+
+    def read_digital_waveforms(
+        self,
+        task_handle: object,
+        channel_count: int,
+        number_of_samples_per_channel: int,
+        number_of_signals_per_sample: int,
+        timeout: float,
+        waveforms: Sequence[DigitalWaveform[numpy.uint8]],
+        waveform_attribute_mode: WaveformAttributeMode,
+    ) -> int:
+        raise NotImplementedError
+
+    def read_new_digital_waveforms(
+        self,
+        task_handle: object,
+        channel_count: int,
+        number_of_samples_per_channel: int,
+        number_of_signals_per_sample: int,
+        timeout: float,
+        waveform_attribute_mode: WaveformAttributeMode,
+    ) -> Sequence[DigitalWaveform[numpy.uint8]]:
         raise NotImplementedError
 
 def _assign_numpy_array(numpy_array, grpc_array):