Implement gRPC Waveform Support

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

@@ -5,10 +5,11 @@
 import threading
 import typing
 import warnings
-from nitypes.waveform import AnalogWaveform, DigitalWaveform
-from typing import Any, Generic, TypeVar
-
-from collections.abc import Callable, Sequence
+from datetime import timezone
+from hightime import datetime as ht_datetime, timedelta as ht_timedelta
+from nitypes import bintime
+from nitypes.waveform import AnalogWaveform, DigitalWaveform, SampleIntervalMode, Timing
+from typing import Any, Callable, Generic, Optional, Sequence, TypeVar
 
 import google.protobuf.message
 import grpc
@@ -18,9 +19,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 import waveform_pb2
+from ni.protobuf.types.waveform_conversion import (
+    digital_waveform_from_protobuf,
+    float64_analog_waveform_from_protobuf
+)
 from nidaqmx.constants import WaveformAttributeMode
+from nidaqmx.errors import DaqError
 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__)
 
@@ -128,7 +137,11 @@ def _handle_rpc_error(self, rpc_error):
                 'This operation is not supported by the NI gRPC Device Server being used. Upgrade NI gRPC Device Server.'
             )
         if error_code is None:
-            raise errors.RpcError(grpc_error, error_message) from None
+            # Convert certain gRPC validation errors to DaqError for consistency with library version
+            if grpc_error == grpc.StatusCode.INVALID_ARGUMENT:
+                raise DaqError(error_message, DAQmxErrors.UNKNOWN) from None
+            else:
+                raise errors.RpcError(grpc_error, error_message) from None
         else:
             self._raise_error(error_code, error_message, samps_per_chan_written, samps_per_chan_read)
 
@@ -3616,7 +3629,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 +3645,24 @@ 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):
+            _copy_protobuf_waveform_to_analog_waveform(grpc_waveform, waveforms[i])
+
+        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 +3672,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 +3692,26 @@ 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):
+            if waveforms[i].data.shape[1] != grpc_waveform.signal_count:
+                raise ValueError(f"waveforms[{i}].data has {waveforms[i].data.shape[1]} signals, but expected {grpc_waveform.signal_count}")
+            _copy_protobuf_waveform_to_digital_waveform(grpc_waveform, waveforms[i])
+
+        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 +3722,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 +3744,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 +3753,26 @@ 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 = []
+        for waveform in waveforms:
+            grpc_waveform = waveform_pb2.DoubleAnalogWaveform()
+            _copy_analog_waveform_to_protobuf_waveform(waveform, grpc_waveform)
+            grpc_waveforms.append(grpc_waveform)
+
+        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 +3781,99 @@ 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 = []
+        for waveform in waveforms:
+            grpc_waveform = waveform_pb2.DigitalWaveform()
+            _copy_digital_waveform_to_protobuf_waveform(waveform, grpc_waveform)
+            grpc_waveforms.append(grpc_waveform)
+
+        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 _copy_timing_and_properties_from_temp_waveform(temp_waveform, target_waveform):
+    try:
+        timestamp_raw = temp_waveform.timing.timestamp
+        if isinstance(timestamp_raw, bintime.DateTime):
+            timestamp = ht_datetime(
+                timestamp_raw.year, timestamp_raw.month, timestamp_raw.day,
+                timestamp_raw.hour, timestamp_raw.minute, timestamp_raw.second,
+                timestamp_raw.microsecond, timestamp_raw.femtosecond,
+                tzinfo=timestamp_raw.tzinfo
+            )
+        else:
+            timestamp = timestamp_raw
+    except RuntimeError:
+        timestamp = None
+
+    try:
+        sample_interval = temp_waveform.timing.sample_interval
+    except RuntimeError:
+        sample_interval = None
+
+    if timestamp is not None or sample_interval is not None:
+        target_waveform.timing = Timing(
+            temp_waveform.timing.sample_interval_mode,
+            timestamp=timestamp,
+            sample_interval=sample_interval
+        )
+    else:
+        target_waveform.timing = temp_waveform.timing
+
+    target_waveform.extended_properties.clear()
+    target_waveform.extended_properties.update(temp_waveform.extended_properties)
+
+
+def _copy_protobuf_waveform_to_analog_waveform(grpc_waveform, target_waveform):
+    temp_waveform = float64_analog_waveform_from_protobuf(grpc_waveform)
+    target_waveform.load_data(temp_waveform.scaled_data)
+    _copy_timing_and_properties_from_temp_waveform(temp_waveform, target_waveform)
+
+def _copy_protobuf_waveform_to_digital_waveform(grpc_waveform, target_waveform):
+    temp_waveform = digital_waveform_from_protobuf(grpc_waveform)
+
+    data = temp_waveform.data
+    if data.dtype != target_waveform.dtype:
+        data = data.view(target_waveform.dtype)
+    target_waveform.load_data(data)
+
+    _copy_timing_and_properties_from_temp_waveform(temp_waveform, target_waveform)
+
+def _copy_analog_waveform_to_protobuf_waveform(waveform, grpc_waveform):
+    scaled_data = waveform.scaled_data
+    if not scaled_data.flags.c_contiguous:
+        scaled_data = scaled_data.copy(order="C")
+    grpc_waveform.y_data[:] = scaled_data
+
+def _copy_digital_waveform_to_protobuf_waveform(waveform, grpc_waveform):
+    grpc_waveform.signal_count = waveform.signal_count
+    data = waveform.data
+    if data.dtype != numpy.uint8:
+        data = data.view(numpy.uint8)
+    if not data.flags.c_contiguous:
+        data = data.copy(order="C")
+    grpc_waveform.y_data = data.tobytes()
 
 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)