Add AnalogMultiChannelWriter.write_waveforms (#834)

* start with a stub and some tests

* Implement write_analog_waveform method in interpreters and update AnalogSingleChannelWriter to use it

* Add support for writing AnalogWaveform in Task class and implement related tests

* cleanup

* cleanup

* Refactor counter output data validation to check for Iterable type

* misc feedback

* Refactor write_analog_waveform method to use write_analog_f64 and add tests for scaling and noncontiguous arrays

* test improvements

* use AnalogWaveform[Any] for all write() type annotations

* write_analog_waveforms and _internal_write_analog_waveform_per_chan

* Add write_waveforms method to AnalogMultiChannelWriter and update task handling for AnalogWaveform lists

* tests

* style cleanup

* fix styling

* cleanup

* cleanup

* cleanup and write_waveforms()

* move logic to write_waveform()

* cleanup

* fix test name

* number_of_samples_per_channel = waveforms[0].sample_count

Author: mikeprosserni

generated/nidaqmx/_base_interpreter.py

@@ -1919,3 +1919,13 @@ def write_analog_waveform(
         timeout: float
     ) -> int:
         raise NotImplementedError
+
+    @abc.abstractmethod
+    def write_analog_waveforms(
+        self,
+        task_handle: object,
+        waveforms: Sequence[AnalogWaveform[Any]],
+        auto_start: bool,
+        timeout: float
+    ) -> int:
+        raise NotImplementedError

generated/nidaqmx/_grpc_interpreter.py

@@ -3664,6 +3664,15 @@ def write_analog_waveform(
     ) -> int:
         raise NotImplementedError
 
+    def write_analog_waveforms(
+        self,
+        task_handle: object,
+        waveforms: Sequence[AnalogWaveform[typing.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

@@ -6975,19 +6975,102 @@ def write_analog_waveform(
         timeout: float
     ) -> int:
         """Write an analog waveform."""
-        array = waveform.scaled_data
-        if not array.flags.c_contiguous:
-            array = array.copy(order="C")
-
         return self.write_analog_f64(
             task_handle,
             waveform.sample_count,
             auto_start,
             timeout,
             FillMode.GROUP_BY_CHANNEL.value,
-            array,
+            self._get_write_array(waveform),
         )
 
+    def write_analog_waveforms(
+        self,
+        task_handle: object,
+        waveforms: Sequence[AnalogWaveform[Any]],
+        auto_start: bool,
+        timeout: float
+    ) -> int:
+        """Write analog waveforms."""
+        assert len(waveforms) > 0
+        num_samps_per_chan = waveforms[0].sample_count
+
+        for waveform in waveforms:
+            if waveform.sample_count != num_samps_per_chan:
+                raise DaqError(
+                    "The waveforms must all have the same sample count.",
+                     DAQmxErrors.UNKNOWN
+                )
+
+        write_arrays = [self._get_write_array(waveform) for waveform in waveforms]
+
+        error_code, samples_written = self._internal_write_analog_waveform_per_chan(
+            task_handle,
+            num_samps_per_chan,
+            auto_start,
+            timeout,
+            write_arrays,
+        )
+
+        self.check_for_error(error_code, samps_per_chan_written=samples_written)
+        return samples_written
+
+    def _internal_write_analog_waveform_per_chan(
+        self,
+        task_handle: object,
+        num_samps_per_chan: int,
+        auto_start: bool,
+        timeout: float,
+        write_arrays: Sequence[numpy.typing.NDArray[numpy.float64]],
+    ) -> Tuple[
+        int, # error code
+        int, # The number of samples per channel that were written
+    ]:
+        assert isinstance(task_handle, TaskHandle)
+        samps_per_chan_written = ctypes.c_int()
+
+        channel_count = len(write_arrays)
+        assert channel_count > 0
+        assert all(write_array.size >= num_samps_per_chan for write_array in write_arrays)
+
+        cfunc = lib_importer.windll.DAQmxInternalWriteAnalogWaveformPerChan
+        if cfunc.argtypes is None:
+            with cfunc.arglock:
+                if cfunc.argtypes is None:
+                    cfunc.argtypes = [
+                        TaskHandle,
+                        ctypes.c_int,
+                        c_bool32,
+                        ctypes.c_double,
+                        ctypes.POINTER(ctypes.POINTER(ctypes.c_double)),
+                        ctypes.c_uint,
+                        ctypes.POINTER(ctypes.c_int),
+                        ctypes.POINTER(c_bool32),
+                    ]
+
+        write_array_ptrs = (ctypes.POINTER(ctypes.c_double) * channel_count)()
+        for i, write_array in enumerate(write_arrays):
+            write_array_ptrs[i] = write_array.ctypes.data_as(ctypes.POINTER(ctypes.c_double))
+
+        error_code = cfunc(
+            task_handle,
+            num_samps_per_chan,
+            auto_start,
+            timeout,
+            write_array_ptrs,
+            channel_count,
+            ctypes.byref(samps_per_chan_written),
+            None,
+        )
+
+        return error_code, samps_per_chan_written.value
+
+    def _get_write_array(self, waveform: AnalogWaveform[Any]) -> numpy.typing.NDArray[numpy.float64]:  
+        scaled_data = waveform.scaled_data
+        if scaled_data.flags.c_contiguous:
+            return scaled_data
+        return scaled_data.copy(order="C")
+
 
     def write_raw(
             self, task_handle, num_samps_per_chan, auto_start, timeout, numpy_array):

generated/nidaqmx/stream_writers/_analog_multi_channel_writer.py

@@ -1,3 +1,11 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from nitypes.waveform import AnalogWaveform
+
+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,
@@ -41,9 +49,7 @@ def write_many_sample(self, data, timeout=10.0):
                 and the number of samples successfully written.
 
         Returns:
-            int:
-
-            Specifies the actual number of samples this method
+            int: Specifies the actual number of samples this method
             successfully wrote to each channel in the task.
         """  # noqa: W505 - doc line too long (101 > 100 characters) (auto-generated noqa)
         self._verify_array(data, True, True)
@@ -84,3 +90,48 @@ def write_one_sample(self, data, timeout=10):
         return self._interpreter.write_analog_f64(
             self._handle, 1, auto_start, timeout, FillMode.GROUP_BY_CHANNEL.value, data
         )
+
+    @requires_feature(WAVEFORM_SUPPORT)
+    def write_waveforms(
+        self, waveforms: Sequence[AnalogWaveform[Any]], timeout: float = 10.0
+    ) -> int:
+        """Writes waveforms to one or more analog 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[AnalogWaveform[Any]]): Contains one or
+                more waveforms to write to the task. All waveforms must
+                have the same number of samples.
+
+                Each waveform corresponds to a channel in the task.
+                The order of the waveforms corresponds to
+                the order in which you add the channels 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 this method
+            successfully wrote to each channel in the task.
+        """
+        auto_start = self._auto_start if self._auto_start is not AUTO_START_UNSET else False
+
+        return self._interpreter.write_analog_waveforms(
+            self._handle, waveforms, auto_start, timeout
+        )