Implement gRPC Waveform Support (#878)

* remove all "not implemented in GRPC" test markers

* implement methods in _grpc_interpreter.mako

* update changelog

* update nidaqmx.proto and stub_generator.py

* fix lint and some mypy issues

* generate .pyi for timestamp and waveform

* _fix_ni_protobuf_imports()

* lint

* Sort proto files for deterministic generation order

* lint

* normalize line endings

* revert sorting fix

* lint

* fix some line endings

* fix stub_generator line endings

* write_bytes in stub_generator

* stub_generator cleanup

* force crlf endings

* line endings

* cleanup

* cleanup

* fix line endings

* cleanup

* cleanup

* _get_num_samps_per_chan helper

* cleanup

* improve changelog

* _waveform_utils.py

* add ni-protobuf-types dependency and revert changes in stub_generator

* grpc interpreter cleanup

* use waveform_conversion methods

* cleanup

* target_waveform.timing = temp_waveform.timing

* simplify read_analog_waveforms and read_digital_waveforms

* simplify write_analog_waveforms and write_digital_waveforms

* cleanup

* revert _handle_rpc_error

* cleanup

---------

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

Author: mikeprosserni

CHANGELOG.md

@@ -34,6 +34,7 @@ All notable changes to this project will be documented in this file.
 
 * ### Major Changes
     * Removed support for Python 3.9.
+    * (IN PROGRESS behind "WAVEFORM_SUPPORT" feature toggle) Added support for reading and writing Waveform data through gRPC using [NI gRPC Device Server](https://github.com/ni/grpc-device).
 
 * ### Known Issues
     * ...
@@ -47,7 +48,7 @@ All notable changes to this project will be documented in this file.
     * [639: first_samp_timestamp_val property does not work because LibraryInterpreter is missing a method](https://github.com/ni/nidaqmx-python/issues/639)
 
 * ### Major Changes
-    * (IN PROGRESS behind "WAVEFORM_SUPPORT" feature toggle) Added support for reading and writing Waveform data. 
+    * (IN PROGRESS behind "WAVEFORM_SUPPORT" feature toggle) Added support for reading and writing Waveform data.
     * Simplify `numpy` and `click` version constraints.
         * `nidaqmx` supports a wider range of Python versions than `numpy` and `click`, so testing
           `nidaqmx` against its oldest supported Python version (currently 3.9) requires downgrading

generated/nidaqmx/_base_interpreter.py

@@ -1879,7 +1879,7 @@ def read_analog_waveforms(
         waveform_attribute_mode: WaveformAttributeMode
     ) -> int:
         raise NotImplementedError
-        
+
     @abc.abstractmethod
     def read_digital_waveform(
         self,
@@ -1935,7 +1935,7 @@ def write_analog_waveforms(
         timeout: float
     ) -> int:
         raise NotImplementedError
-        
+
     @abc.abstractmethod
     def write_digital_waveform(
         self,
@@ -1945,7 +1945,7 @@ def write_digital_waveform(
         timeout: float,
     ) -> int:
         raise NotImplementedError
-        
+
     def write_digital_waveforms(
         self,
         task_handle: object,

generated/nidaqmx/_grpc_interpreter.py

@@ -18,9 +18,17 @@
 from nidaqmx._base_interpreter import BaseEventHandler, BaseInterpreter
 from nidaqmx._stubs import nidaqmx_pb2 as grpc_types
 from nidaqmx._stubs import nidaqmx_pb2_grpc as nidaqmx_grpc
+from ni.protobuf.types.waveform_conversion import (
+    digital_waveform_from_protobuf,
+    digital_waveform_to_protobuf,
+    float64_analog_waveform_from_protobuf,
+    float64_analog_waveform_to_protobuf
+)
 from nidaqmx.constants import WaveformAttributeMode
 from nidaqmx.error_codes import DAQmxErrors
 from nidaqmx._grpc_time import convert_time_to_timestamp, convert_timestamp_to_time
+from nidaqmx._waveform_utils import get_num_samps_per_chan
+from session_pb2 import Session
 
 _logger = logging.getLogger(__name__)
 
@@ -3616,7 +3624,13 @@ def read_analog_waveform(
         waveform: AnalogWaveform[numpy.float64],
         waveform_attribute_mode: WaveformAttributeMode
     ) -> int:
-        raise NotImplementedError
+        return self.read_analog_waveforms(
+            task_handle,
+            number_of_samples_per_channel,
+            timeout,
+            [waveform],
+            waveform_attribute_mode
+        )
 
     def read_analog_waveforms(
         self,
@@ -3626,7 +3640,30 @@ def read_analog_waveforms(
         waveforms: Sequence[AnalogWaveform[numpy.float64]],
         waveform_attribute_mode: WaveformAttributeMode
     ) -> int:
-        raise NotImplementedError
+        assert isinstance(task_handle, Session)
+        response = self._invoke(
+            self._client.ReadAnalogWaveforms,
+            grpc_types.ReadAnalogWaveformsRequest(
+                task=task_handle,
+                num_samps_per_chan=number_of_samples_per_channel,
+                timeout=timeout,
+                waveform_attribute_mode_raw=waveform_attribute_mode.value
+            ))
+
+        if len(response.waveforms) != len(waveforms):
+            raise ValueError(f"Expected {len(waveforms)} waveforms but received {len(response.waveforms)} from server")
+
+        for i, grpc_waveform in enumerate(response.waveforms):
+            temp_waveform = float64_analog_waveform_from_protobuf(grpc_waveform)
+            waveforms[i].load_data(temp_waveform.scaled_data)
+
+            waveforms[i].scale_mode = temp_waveform.scale_mode
+            waveforms[i].timing = temp_waveform.timing
+            waveforms[i].extended_properties.clear()
+            waveforms[i].extended_properties.update(temp_waveform.extended_properties)
+
+        self._check_for_error_from_response(response.status, samps_per_chan_read=response.samps_per_chan_read)
+        return response.samps_per_chan_read
 
     def read_digital_waveform(
         self,
@@ -3636,7 +3673,15 @@ def read_digital_waveform(
         waveform: DigitalWaveform[Any],
         waveform_attribute_mode: WaveformAttributeMode
     ) -> int:
-        raise NotImplementedError
+        return self.read_digital_waveforms(
+            task_handle,
+            1,  # channel_count
+            number_of_samples_per_channel,
+            waveform.signal_count,  # number_of_signals_per_sample
+            timeout,
+            [waveform],
+            waveform_attribute_mode
+        )
 
     def read_digital_waveforms(
         self,
@@ -3648,7 +3693,32 @@ def read_digital_waveforms(
         waveforms: Sequence[DigitalWaveform[Any]],
         waveform_attribute_mode: WaveformAttributeMode,
     ) -> int:
-        raise NotImplementedError
+        assert isinstance(task_handle, Session)
+        response = self._invoke(
+            self._client.ReadDigitalWaveforms,
+            grpc_types.ReadDigitalWaveformsRequest(
+                task=task_handle,
+                num_samps_per_chan=number_of_samples_per_channel,
+                timeout=timeout,
+                waveform_attribute_mode_raw=waveform_attribute_mode.value
+            ))
+
+        if len(response.waveforms) != len(waveforms):
+            raise ValueError(f"Expected {len(waveforms)} waveforms but received {len(response.waveforms)} from server")
+
+        for i, grpc_waveform in enumerate(response.waveforms):
+            temp_waveform = digital_waveform_from_protobuf(grpc_waveform)
+            data = temp_waveform.data
+            if data.dtype != waveforms[i].dtype:
+                data = data.view(waveforms[i].dtype)
+            waveforms[i].load_data(data)
+
+            waveforms[i].timing = temp_waveform.timing
+            waveforms[i].extended_properties.clear()
+            waveforms[i].extended_properties.update(temp_waveform.extended_properties)
+
+        self._check_for_error_from_response(response.status, samps_per_chan_read=response.samps_per_chan_read)
+        return response.samps_per_chan_read
 
     def read_new_digital_waveforms(
         self,
@@ -3659,7 +3729,20 @@ def read_new_digital_waveforms(
         timeout: float,
         waveform_attribute_mode: WaveformAttributeMode,
     ) -> Sequence[DigitalWaveform[numpy.uint8]]:
-        raise NotImplementedError
+        assert isinstance(task_handle, Session)
+        response = self._invoke(
+            self._client.ReadDigitalWaveforms,
+            grpc_types.ReadDigitalWaveformsRequest(
+                task=task_handle,
+                num_samps_per_chan=number_of_samples_per_channel,
+                timeout=timeout,
+                waveform_attribute_mode_raw=waveform_attribute_mode.value
+            ))
+
+        waveforms = [digital_waveform_from_protobuf(grpc_waveform) for grpc_waveform in response.waveforms]
+
+        self._check_for_error_from_response(response.status, samps_per_chan_read=response.samps_per_chan_read)
+        return waveforms
 
     def write_analog_waveform(
         self,
@@ -3668,7 +3751,7 @@ def write_analog_waveform(
         auto_start: bool,
         timeout: float
     ) -> int:
-        raise NotImplementedError
+        return self.write_analog_waveforms(task_handle, [waveform], auto_start, timeout)
 
     def write_analog_waveforms(
         self,
@@ -3677,7 +3760,22 @@ def write_analog_waveforms(
         auto_start: bool,
         timeout: float
     ) -> int:
-        raise NotImplementedError
+        assert isinstance(task_handle, Session)
+        num_samps_per_chan = get_num_samps_per_chan(waveforms)
+
+        grpc_waveforms = [float64_analog_waveform_to_protobuf(waveform) for waveform in waveforms]
+
+        response = self._invoke(
+            self._client.WriteAnalogWaveforms,
+            grpc_types.WriteAnalogWaveformsRequest(
+                task=task_handle,
+                auto_start=auto_start,
+                timeout=timeout,
+                waveforms=grpc_waveforms
+            ))
+
+        self._check_for_error_from_response(response.status, samps_per_chan_written=response.samps_per_chan_written)
+        return response.samps_per_chan_written
 
     def write_digital_waveform(
         self,
@@ -3686,16 +3784,31 @@ def write_digital_waveform(
         auto_start: bool,
         timeout: float,
     ) -> int:
-        raise NotImplementedError
+        return self.write_digital_waveforms(task_handle, [waveform], auto_start, timeout)
 
     def write_digital_waveforms(
         self,
         task_handle: object,
-        waveform: Sequence[DigitalWaveform[Any]],
+        waveforms: Sequence[DigitalWaveform[Any]],
         auto_start: bool,
         timeout: float,
     ) -> int:
-        raise NotImplementedError
+        assert isinstance(task_handle, Session)
+        num_samps_per_chan = get_num_samps_per_chan(waveforms)
+
+        grpc_waveforms = [digital_waveform_to_protobuf(waveform) for waveform in waveforms]
+
+        response = self._invoke(
+            self._client.WriteDigitalWaveforms,
+            grpc_types.WriteDigitalWaveformsRequest(
+                task=task_handle,
+                auto_start=auto_start,
+                timeout=timeout,
+                waveforms=grpc_waveforms
+            ))
+
+        self._check_for_error_from_response(response.status, samps_per_chan_written=response.samps_per_chan_written)
+        return response.samps_per_chan_written
 
 def _assign_numpy_array(numpy_array, grpc_array):
     """

generated/nidaqmx/_library_interpreter.py

@@ -20,6 +20,7 @@
 from nidaqmx.error_codes import DAQmxErrors, DAQmxWarnings
 from nidaqmx.errors import DaqError, DaqFunctionNotSupportedError, DaqReadError, DaqWarning, DaqWriteError
 from nidaqmx._lib_time import AbsoluteTime
+from nidaqmx._waveform_utils import get_num_samps_per_chan
 from nitypes.waveform.typing import ExtendedPropertyValue
 from nitypes.waveform import AnalogWaveform, DigitalWaveform, SampleIntervalMode, Timing, ExtendedPropertyDictionary
 
@@ -7011,15 +7012,7 @@ def write_analog_waveforms(
         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
-                )
+        num_samps_per_chan = get_num_samps_per_chan(waveforms)
 
         write_arrays = [self._get_analog_write_array(waveform) for waveform in waveforms]
 
@@ -7130,15 +7123,7 @@ def write_digital_waveforms(
     ) -> 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
-                )
+        sample_count = get_num_samps_per_chan(waveforms)
 
         bytes_per_chan_array = numpy.array([wf.signal_count for wf in waveforms], dtype=numpy.uint32)