Add digital multiple waveform writer (#838)

* first draft

* multi-channel implementation and a few tests

* test___digital_multi_channel_writer___write_waveforms_mixed_lines___outputs_match_final_values

* test___digital_multi_channel_writer___write_waveforms_ports___outputs_match_final_values

* test___digital_multi_channel_writer___write_waveforms_port_and_lines___outputs_match_final_values

* test___digital_multi_channel_writer___write_waveforms_with_non_contiguous_data___outputs_match_final_values

* tests for channel count, sample count, and line count mismatches

* task tests

* cleanup comment

* brad's feedback

* better type annotations

Author: mikeprosserni

generated/nidaqmx/_base_interpreter.py

@@ -1938,4 +1938,13 @@ def write_digital_waveform(
         auto_start: bool,
         timeout: float,
     ) -> int:
-        raise NotImplementedError
+        raise NotImplementedError
+        
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveform: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        raise NotImplementedError

generated/nidaqmx/_grpc_interpreter.py

@@ -3682,6 +3682,15 @@ def write_digital_waveform(
     ) -> int:
         raise NotImplementedError
 
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveform: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        raise NotImplementedError
+
 def _assign_numpy_array(numpy_array, grpc_array):
     """
     Assigns grpc array to numpy array maintaining the original shape.

generated/nidaqmx/_library_interpreter.py

@@ -7108,6 +7108,50 @@ def _get_digital_write_array(self, waveform: DigitalWaveform[Any]) -> numpy.typi
             return data
         return data.copy(order="C")
 
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveforms: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        """Write digital waveforms."""
+        channel_count = len(waveforms)
+        assert channel_count > 0
+        sample_count = waveforms[0].sample_count
+
+        for waveform in waveforms:
+            if waveform.sample_count != sample_count:
+                raise DaqError(
+                    "The waveforms must all have the same sample count.",
+                    DAQmxErrors.UNKNOWN
+                )
+                
+        bytes_per_chan_array = numpy.array([wf.signal_count for wf in waveforms], dtype=numpy.uint32)
+
+        # build a temporary contiguous array to write the data from multiple channels into.
+        # write_array must be in the format (numChans x numSampsPerChan x maxDataWidth)
+        write_array = numpy.zeros(
+            (channel_count, sample_count, max(bytes_per_chan_array)),
+            dtype=numpy.uint8,
+        )
+        for i, waveform in enumerate(waveforms):
+            signal_count = waveform.signal_count
+            write_array[i, :, :signal_count] = waveform.data
+
+        error_code, samples_written = self._internal_write_digital_waveform(
+            task_handle,
+            sample_count,
+            auto_start,
+            timeout,
+            FillMode.GROUP_BY_CHANNEL.value,
+            write_array,
+            bytes_per_chan_array,
+        )
+
+        self.check_for_error(error_code, samps_per_chan_written=samples_written)
+        return samples_written
+
     def _internal_write_digital_waveform(
         self,
         task_handle: object,

generated/nidaqmx/stream_writers/_digital_multi_channel_writer.py

@@ -1,3 +1,10 @@
+from __future__ import annotations
+
+from typing import Any, Sequence
+
+from nitypes.waveform import DigitalWaveform
+
+from nidaqmx._feature_toggles import WAVEFORM_SUPPORT, requires_feature
 from nidaqmx.constants import FillMode
 from nidaqmx.stream_writers._channel_writer_base import (
     AUTO_START_UNSET,
@@ -9,7 +16,7 @@ class DigitalMultiChannelWriter(ChannelWriterBase):
     """Writes samples to one or more digital output channels in an NI-DAQmx task."""
 
     def write_many_sample_port_byte(self, data, timeout=10.0):
-        """Writes one or more 8-bit unsigned integer samples to one or more digital output channels in a task.
+        """Writes 8-bit unsigned integer samples to one or more digital output channels in a task.
 
         Use this method for devices with up to 8 lines per port.
 
@@ -47,7 +54,7 @@ def write_many_sample_port_byte(self, data, timeout=10.0):
 
             Specifies the actual number of samples this method
             successfully wrote to each channel in the task.
-        """  # noqa: W505 - doc line too long (110 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, True)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else False
@@ -57,7 +64,7 @@ def write_many_sample_port_byte(self, data, timeout=10.0):
         )
 
     def write_many_sample_port_uint16(self, data, timeout=10.0):
-        """Writes one or more 16-bit unsigned integer samples to one or more digital output channels in a task.
+        """Writes 16-bit unsigned integer samples to one or more digital output channels in a task.
 
         Use this method for devices with up to 16 lines per port.
 
@@ -95,7 +102,7 @@ def write_many_sample_port_uint16(self, data, timeout=10.0):
 
             Specifies the actual number of samples this method
             successfully wrote to each channel in the task.
-        """  # noqa: W505 - doc line too long (111 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, True)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else False
@@ -105,7 +112,7 @@ def write_many_sample_port_uint16(self, data, timeout=10.0):
         )
 
     def write_many_sample_port_uint32(self, data, timeout=10.0):
-        """Writes one or more 32-bit unsigned integer samples to one or more digital output channels in a task.
+        """Writes 32-bit unsigned integer samples to one or more digital output channels in a task.
 
         Use this method for devices with up to 32 lines per port.
 
@@ -143,7 +150,7 @@ def write_many_sample_port_uint32(self, data, timeout=10.0):
 
             Specifies the actual number of samples this method
             successfully wrote to each channel in the task.
-        """  # noqa: W505 - doc line too long (111 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, True)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else False
@@ -219,7 +226,7 @@ def write_one_sample_one_line(self, data, timeout=10):
         )
 
     def write_one_sample_port_byte(self, data, timeout=10):
-        """Writes a single 8-bit unsigned integer sample to one or more digital output channels in a task.
+        """Writes a single 8-bit unsigned integer sample to one or more digital output channels.
 
         Use this method for devices with up to 8 lines per port.
 
@@ -242,7 +249,7 @@ def write_one_sample_port_byte(self, data, timeout=10):
                 once to write the submitted samples. If the method could
                 not write all the submitted samples, it returns an error
                 and the number of samples successfully written.
-        """  # noqa: W505 - doc line too long (106 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, False)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else True
@@ -252,7 +259,7 @@ def write_one_sample_port_byte(self, data, timeout=10):
         )
 
     def write_one_sample_port_uint16(self, data, timeout=10):
-        """Writes a single 16-bit unsigned integer sample to one or more digital output channels in a task.
+        """Writes a single 16-bit unsigned integer sample to one or more digital output channels.
 
         Use this method for devices with up to 16 lines per port.
 
@@ -275,7 +282,7 @@ def write_one_sample_port_uint16(self, data, timeout=10):
                 once to write the submitted samples. If the method could
                 not write all the submitted samples, it returns an error
                 and the number of samples successfully written.
-        """  # noqa: W505 - doc line too long (107 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, False)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else True
@@ -285,7 +292,7 @@ def write_one_sample_port_uint16(self, data, timeout=10):
         )
 
     def write_one_sample_port_uint32(self, data, timeout=10):
-        """Writes a single 32-bit unsigned integer sample to one or more digital output channels in a task.
+        """Writes a single 32-bit unsigned integer sample to one or more digital output channels.
 
         Use this method for devices with up to 32 lines per port.
 
@@ -308,11 +315,51 @@ def write_one_sample_port_uint32(self, data, timeout=10):
                 once to write the submitted samples. If the method could
                 not write all the submitted samples, it returns an error
                 and the number of samples successfully written.
-        """  # noqa: W505 - doc line too long (107 > 100 characters) (auto-generated noqa)
+        """
         self._verify_array(data, True, False)
 
         auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else True
 
         return self._interpreter.write_digital_u32(
             self._handle, 1, auto_start, timeout, FillMode.GROUP_BY_CHANNEL.value, data
         )
+
+    @requires_feature(WAVEFORM_SUPPORT)
+    def write_waveforms(
+        self, waveforms: Sequence[DigitalWaveform[Any]], timeout: float = 10.0
+    ) -> int:
+        """Writes waveforms to one or more digital output channels in a task.
+
+        If the task uses on-demand timing, this method returns only
+        after the device generates all samples. On-demand is the default
+        timing type if you do not use the timing property on the task to
+        configure a sample timing type. If the task uses any timing type
+        other than on-demand, this method returns immediately and does
+        not wait for the device to generate all samples. Your
+        application must determine if the task is done to ensure that
+        the device generated all samples.
+
+        Args:
+            waveforms (Sequence[DigitalWaveform[Any]]): Specifies the
+                waveforms to write to the task.
+            timeout (Optional[float]): Specifies the amount of time in
+                seconds to wait for the method to write all samples.
+                NI-DAQmx performs a timeout check only if the method
+                must wait before it writes data. This method returns an
+                error if the time elapses. The default timeout is 10
+                seconds. If you set timeout to
+                nidaqmx.constants.WAIT_INFINITELY, the method waits
+                indefinitely. If you set timeout to 0, the method tries
+                once to write the submitted samples. If the method could
+                not write all the submitted samples, it returns an error
+                and the number of samples successfully written.
+
+        Returns:
+            int: Specifies the actual number of samples per channel this method
+            successfully wrote.
+        """
+        auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else False
+
+        return self._interpreter.write_digital_waveforms(
+            self._handle, waveforms, auto_start, timeout
+        )

generated/nidaqmx/task/_task.py

@@ -1392,9 +1392,7 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
             Specifies the actual number of samples this method
             successfully wrote.
         """
-        if isinstance(data, (AnalogWaveform, DigitalWaveform)) or (
-            isinstance(data, list) and data and all(isinstance(wf, AnalogWaveform) for wf in data)
-        ):
+        if self._is_waveform_data(data):
             return self.write_waveform(data, auto_start, timeout)
 
         channels_to_write = self.channels
@@ -1578,10 +1576,27 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
         else:
             self._raise_no_output_channels_error()
 
+    def _is_waveform_data(self, data):
+        """Check if data is waveform data (single waveform or list of waveforms)."""
+        if isinstance(data, (AnalogWaveform, DigitalWaveform)):
+            return True
+
+        if not isinstance(data, list) or not data:
+            return False
+
+        return all(isinstance(wf, AnalogWaveform) for wf in data) or all(
+            isinstance(wf, DigitalWaveform) for wf in data
+        )
+
     @requires_feature(WAVEFORM_SUPPORT)
     def write_waveform(
         self,
-        waveforms: AnalogWaveform[Any] | DigitalWaveform[Any] | Sequence[AnalogWaveform[Any]],
+        waveforms: (
+            AnalogWaveform[Any]
+            | DigitalWaveform[Any]
+            | Sequence[AnalogWaveform[Any]]
+            | Sequence[DigitalWaveform[Any]]
+        ),
         auto_start=AUTO_START_UNSET,
         timeout: float = 10.0,
     ) -> int:
@@ -1671,6 +1686,12 @@ def write_waveform(
                 return self._interpreter.write_digital_waveform(
                     self._handle, waveforms, auto_start, timeout
                 )
+            elif isinstance(waveforms, list) and isinstance(waveforms[0], DigitalWaveform):
+                if number_of_channels != len(waveforms):
+                    self._raise_invalid_write_num_chans_error(number_of_channels, len(waveforms))
+                return self._interpreter.write_digital_waveforms(
+                    self._handle, waveforms, auto_start, timeout
+                )
             else:
                 self._raise_unsupported_output_type_error(type(waveforms))
 

src/codegen/templates/_base_interpreter.py.mako

@@ -147,4 +147,13 @@ class BaseInterpreter(abc.ABC):
         auto_start: bool,
         timeout: float,
     ) -> int:
-        raise NotImplementedError
+        raise NotImplementedError
+        
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveform: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        raise NotImplementedError

src/codegen/templates/_grpc_interpreter.py.mako

@@ -327,6 +327,15 @@ class GrpcStubInterpreter(BaseInterpreter):
     ) -> int:
         raise NotImplementedError
 
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveform: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        raise NotImplementedError
+
 def _assign_numpy_array(numpy_array, grpc_array):
     """
     Assigns grpc array to numpy array maintaining the original shape.

src/codegen/templates/_library_interpreter.py.mako

@@ -936,6 +936,50 @@ class LibraryInterpreter(BaseInterpreter):
             return data
         return data.copy(order="C")
 
+    def write_digital_waveforms(
+        self,
+        task_handle: object,
+        waveforms: Sequence[DigitalWaveform[Any]],
+        auto_start: bool,
+        timeout: float,
+    ) -> int:
+        """Write digital waveforms."""
+        channel_count = len(waveforms)
+        assert channel_count > 0
+        sample_count = waveforms[0].sample_count
+
+        for waveform in waveforms:
+            if waveform.sample_count != sample_count:
+                raise DaqError(
+                    "The waveforms must all have the same sample count.",
+                    DAQmxErrors.UNKNOWN
+                )
+                
+        bytes_per_chan_array = numpy.array([wf.signal_count for wf in waveforms], dtype=numpy.uint32)
+
+        # build a temporary contiguous array to write the data from multiple channels into.
+        # write_array must be in the format (numChans x numSampsPerChan x maxDataWidth)
+        write_array = numpy.zeros(
+            (channel_count, sample_count, max(bytes_per_chan_array)),
+            dtype=numpy.uint8,
+        )
+        for i, waveform in enumerate(waveforms):
+            signal_count = waveform.signal_count
+            write_array[i, :, :signal_count] = waveform.data
+
+        error_code, samples_written = self._internal_write_digital_waveform(
+            task_handle,
+            sample_count,
+            auto_start,
+            timeout,
+            FillMode.GROUP_BY_CHANNEL.value,
+            write_array,
+            bytes_per_chan_array,
+        )
+
+        self.check_for_error(error_code, samps_per_chan_written=samples_written)
+        return samples_written
+
     def _internal_write_digital_waveform(
         self,
         task_handle: object,