Add AnalogSingleChannelWriter.write_waveform() (#828)

* 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

Author: mikeprosserni

generated/nidaqmx/_base_interpreter.py

@@ -4,7 +4,7 @@
 import abc
 import numpy
 from nitypes.waveform import AnalogWaveform, DigitalWaveform
-from typing import Sequence
+from typing import Any, Sequence
 from nidaqmx.constants import WaveformAttributeMode
 
 
@@ -1909,3 +1909,13 @@ def read_new_digital_waveforms(
         waveform_attribute_mode: WaveformAttributeMode,
     ) -> Sequence[DigitalWaveform[numpy.uint8]]:
         raise NotImplementedError
+
+    @abc.abstractmethod
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: AnalogWaveform[Any],
+        auto_start: bool,
+        timeout: float
+    ) -> int:
+        raise NotImplementedError

generated/nidaqmx/_feature_toggles.py

@@ -121,7 +121,8 @@ def is_enabled(self) -> bool:
             return self._is_enabled_override
         return self.readiness <= get_code_readiness_level()
 
-    def _raise_if_disabled(self) -> None:
+    def raise_if_disabled(self) -> None:
+        """Raises an error if the feature is disabled."""
         if self.is_enabled:
             return
 
@@ -143,7 +144,7 @@ def requires_feature(
     def decorator(func: Callable[_P, _T]) -> Callable[_P, _T]:
         @functools.wraps(func)
         def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
-            feature_toggle._raise_if_disabled()
+            feature_toggle.raise_if_disabled()
             return func(*args, **kwargs)
 
         return wrapper

generated/nidaqmx/_grpc_interpreter.py

@@ -3655,6 +3655,15 @@ def read_new_digital_waveforms(
     ) -> Sequence[DigitalWaveform[numpy.uint8]]:
         raise NotImplementedError
 
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: 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

@@ -11,7 +11,7 @@
 from datetime import timezone
 from hightime import datetime as ht_datetime
 from hightime import timedelta as ht_timedelta
-from typing import Callable, List, Sequence, Tuple, TYPE_CHECKING, Union
+from typing import Any, Callable, List, Sequence, Tuple, TYPE_CHECKING, Union
 
 from nidaqmx._base_interpreter import BaseEventHandler, BaseInterpreter
 from nidaqmx._lib import lib_importer, ctypes_byte_str, c_bool32, wrapped_ndpointer, TaskHandle
@@ -6967,6 +6967,28 @@ def read_raw(self, task, num_samps_per_chan, timeout, read_array):
 
         return read_array, samples_read.value, number_of_bytes_per_sample.value
 
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: AnalogWaveform[Any],
+        auto_start: bool,
+        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,
+        )
+
+
     def write_raw(
             self, task_handle, num_samps_per_chan, auto_start, timeout, numpy_array):
         samps_per_chan_written = ctypes.c_int()

generated/nidaqmx/stream_writers/_analog_single_channel_writer.py

@@ -1,4 +1,11 @@
+from __future__ import annotations
+
+import numpy
+from typing import Any
+
+from nidaqmx._feature_toggles import WAVEFORM_SUPPORT, requires_feature
 from nidaqmx.constants import FillMode
+from nitypes.waveform import AnalogWaveform
 
 from nidaqmx.stream_writers._channel_writer_base import ChannelWriterBase, AUTO_START_UNSET
 
@@ -80,3 +87,45 @@ def write_one_sample(self, data, timeout=10):
         return self._interpreter.write_analog_scalar_f64(
             self._handle, auto_start, timeout, data)
 
+    @requires_feature(WAVEFORM_SUPPORT)
+    def write_waveform(
+        self, 
+        waveform: AnalogWaveform[Any], 
+        timeout: float = 10.0
+    ) -> int:
+        """
+        Writes a waveform to a single analog output channel 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:
+            waveform (AnalogWaveform[Any]): Specifies the
+                waveform 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 this method
+            successfully wrote.
+        """
+        auto_start = (self._auto_start if self._auto_start is not 
+                      AUTO_START_UNSET else False)
+
+        return self._interpreter.write_analog_waveform(
+            self._handle, waveform, auto_start, timeout)
+    

generated/nidaqmx/task/_task.py

@@ -2,6 +2,7 @@
 
 import threading
 import warnings
+from collections.abc import Iterable
 from enum import Enum
 
 import numpy
@@ -1240,6 +1241,7 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
           sample for multiple channels.
         - List of lists/2D numpy.ndarray: Multiple samples for multiple
           channels.
+        - AnalogWaveform: Waveform data for a single analog output channel.
 
         The data type of the samples passed in must be appropriate for
         the channel type of the task.
@@ -1316,6 +1318,14 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
                 number_of_samples_per_channel = len(data)
                 element = data[0]
 
+            elif isinstance(data, AnalogWaveform):
+                WAVEFORM_SUPPORT.raise_if_disabled()
+                if number_of_channels != 1:
+                    self._raise_invalid_write_num_chans_error(
+                        number_of_channels, 1)
+                number_of_samples_per_channel = data.sample_count
+                element = data.raw_data[0]
+
             else:
                 number_of_samples_per_channel = 1
                 element = data
@@ -1356,10 +1366,14 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
                 auto_start = True
 
         if write_chan_type == ChannelType.ANALOG_OUTPUT:
-            data = numpy.asarray(data, dtype=numpy.float64)
-            return self._interpreter.write_analog_f64(
-                self._handle, number_of_samples_per_channel, auto_start,
-                timeout, FillMode.GROUP_BY_CHANNEL.value, data)
+            if isinstance(data, AnalogWaveform):
+                return self._interpreter.write_analog_waveform(
+                    self._handle, data, auto_start, timeout)
+            else:
+                data = numpy.asarray(data, dtype=numpy.float64)
+                return self._interpreter.write_analog_f64(
+                    self._handle, number_of_samples_per_channel, auto_start,
+                    timeout, FillMode.GROUP_BY_CHANNEL.value, data)
 
         elif write_chan_type == ChannelType.DIGITAL_OUTPUT:
             if self.out_stream.do_num_booleans_per_chan == 1:
@@ -1396,6 +1410,11 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
 
             if number_of_samples_per_channel == 1:
                 data = [data]
+            elif not isinstance(data, Iterable):
+                raise DaqError(
+                    'Write failed, because the provided data type is not supported '
+                    'for counter output channels.',
+                    DAQmxErrors.UNKNOWN, task_name=self.name)
 
             if output_type == UsageTypeCO.PULSE_FREQUENCY:
                 if not all(isinstance(sample, CtrFreq) for sample in data):

src/codegen/templates/_base_interpreter.py.mako

@@ -15,7 +15,7 @@ from __future__ import annotations
 import abc
 import numpy
 from nitypes.waveform import AnalogWaveform, DigitalWaveform
-from typing import Sequence
+from typing import Any, Sequence
 from nidaqmx.constants import WaveformAttributeMode
 
 
@@ -118,3 +118,13 @@ class BaseInterpreter(abc.ABC):
         waveform_attribute_mode: WaveformAttributeMode,
     ) -> Sequence[DigitalWaveform[numpy.uint8]]:
         raise NotImplementedError
+
+    @abc.abstractmethod
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: AnalogWaveform[Any],
+        auto_start: bool,
+        timeout: float
+    ) -> int:
+        raise NotImplementedError

src/codegen/templates/_grpc_interpreter.py.mako

@@ -300,6 +300,15 @@ class GrpcStubInterpreter(BaseInterpreter):
     ) -> Sequence[DigitalWaveform[numpy.uint8]]:
         raise NotImplementedError
 
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: 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.

src/codegen/templates/_library_interpreter.py.mako

@@ -28,7 +28,7 @@ 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, Sequence, Tuple, TYPE_CHECKING, Union
+from typing import Any, Callable, List, Sequence, Tuple, TYPE_CHECKING, Union
 
 from nidaqmx._base_interpreter import BaseEventHandler, BaseInterpreter
 from nidaqmx._lib import lib_importer, ctypes_byte_str, c_bool32, wrapped_ndpointer, TaskHandle
@@ -793,6 +793,29 @@ class LibraryInterpreter(BaseInterpreter):
 
         return read_array, samples_read.value, number_of_bytes_per_sample.value
 
+    ## write_analog_waveform has special handling
+    def write_analog_waveform(
+        self,
+        task_handle: object,
+        waveform: AnalogWaveform[Any],
+        auto_start: bool,
+        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,
+        )
+
+
     ## The datatype of 'write_array' is incorrect in daqmxAPISharp.json file.
     def write_raw(
             self, task_handle, num_samps_per_chan, auto_start, timeout, numpy_array):

src/handwritten/_feature_toggles.py

@@ -121,7 +121,8 @@ def is_enabled(self) -> bool:
             return self._is_enabled_override
         return self.readiness <= get_code_readiness_level()
 
-    def _raise_if_disabled(self) -> None:
+    def raise_if_disabled(self) -> None:
+        """Raises an error if the feature is disabled."""
         if self.is_enabled:
             return
 
@@ -143,7 +144,7 @@ def requires_feature(
     def decorator(func: Callable[_P, _T]) -> Callable[_P, _T]:
         @functools.wraps(func)
         def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
-            feature_toggle._raise_if_disabled()
+            feature_toggle.raise_if_disabled()
             return func(*args, **kwargs)
 
         return wrapper