Raise an error if irregular timing is encountered during conversion. (#31)

* Raise an error if irregular timing is encountered during conversion.

Signed-off-by: Michael Johansen <[email protected]>

* Fix typo.

Signed-off-by: Michael Johansen <[email protected]>

* Rename method based on PR feedback.

Signed-off-by: Michael Johansen <[email protected]>

---------

Signed-off-by: Michael Johansen <[email protected]>
Co-authored-by: Michael Johansen <[email protected]>

Author: mjohanse-emr

packages/ni.protobuf.types/src/ni/protobuf/types/waveform_conversion.py

@@ -18,6 +18,7 @@
     ExtendedPropertyValue,
     LinearScaleMode,
     NoneScaleMode,
+    SampleIntervalMode,
     Spectrum,
     Timing,
 )
@@ -42,6 +43,7 @@ def float64_analog_waveform_to_protobuf(
     value: AnalogWaveform[np.float64], /
 ) -> DoubleAnalogWaveform:
     """Convert the Python AnalogWaveform to a protobuf DoubleAnalogWaveform."""
+    _validate_timing(value)
     t0 = _t0_from_waveform(value)
     time_interval = _time_interval_from_waveform(value)
     attributes = _extended_properties_to_attributes(value.extended_properties)
@@ -74,6 +76,7 @@ def float64_complex_waveform_to_protobuf(
     value: ComplexWaveform[np.complex128], /
 ) -> DoubleComplexWaveform:
     """Convert the Python ComplexWaveform to a protobuf DoubleComplexWaveform."""
+    _validate_timing(value)
     t0 = _t0_from_waveform(value)
     time_interval = _time_interval_from_waveform(value)
     attributes = _extended_properties_to_attributes(value.extended_properties)
@@ -111,6 +114,7 @@ def int16_complex_waveform_to_protobuf(
     value: ComplexWaveform[ComplexInt32Base], /
 ) -> I16ComplexWaveform:
     """Convert the Python ComplexWaveform to a protobuf DoubleComplexWaveform."""
+    _validate_timing(value)
     t0 = _t0_from_waveform(value)
     time_interval = _time_interval_from_waveform(value)
     attributes = _extended_properties_to_attributes(value.extended_properties)
@@ -205,6 +209,13 @@ def _value_to_attribute(value: ExtendedPropertyValue) -> WaveformAttributeValue:
     return attr_value
 
 
+def _validate_timing(
+    waveform: AnalogWaveform[Any] | ComplexWaveform[Any],
+) -> None:
+    if waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR:
+        raise ValueError("Cannot convert irregular sample interval to protobuf.")
+
+
 def _t0_from_waveform(
     waveform: AnalogWaveform[Any] | ComplexWaveform[Any],
 ) -> PrecisionTimestamp | None:

packages/ni.protobuf.types/tests/unit/test_waveform_conversion.py

@@ -1,6 +1,7 @@
 import datetime as dt
 
 import numpy as np
+import pytest
 from nitypes.bintime import DateTime
 from nitypes.complex import ComplexInt32DType
 from nitypes.waveform import (
@@ -94,6 +95,19 @@ def test___analog_waveform_with_standard_timing___convert___valid_protobuf() ->
     assert dbl_analog_waveform.t0 == converted_t0
 
 
+def test___analog_waveform_with_irregular_timing___convert___raises_value_error() -> None:
+    analog_waveform = AnalogWaveform.from_array_1d(np.array([1.0, 2.0, 3.0]))
+    t0_dt = dt.datetime(2000, 12, 1, tzinfo=dt.timezone.utc)
+    analog_waveform.timing = Timing.create_with_irregular_interval(
+        [t0_dt, t0_dt + dt.timedelta(milliseconds=1000), t0_dt + dt.timedelta(milliseconds=3000)]
+    )
+
+    with pytest.raises(ValueError) as exc:
+        _ = float64_analog_waveform_to_protobuf(analog_waveform)
+
+    assert exc.value.args[0].startswith("Cannot convert irregular sample interval to protobuf.")
+
+
 # ========================================================
 # DoubleAnalogWaveform to AnalogWaveform
 # ========================================================
@@ -221,6 +235,19 @@ def test___float64_complex_waveform_with_standard_timing___convert___valid_proto
     assert dbl_complex_waveform.t0 == converted_t0
 
 
+def test___float64_complex_waveform_with_irregular_timing___convert___raises_value_error() -> None:
+    complex_waveform = ComplexWaveform.from_array_1d([1.5 + 2.5j, 3.5 + 4.5j], np.complex128)
+    t0_dt = dt.datetime(2000, 12, 1, tzinfo=dt.timezone.utc)
+    complex_waveform.timing = Timing.create_with_irregular_interval(
+        [t0_dt, t0_dt + dt.timedelta(milliseconds=1000)]
+    )
+
+    with pytest.raises(ValueError) as exc:
+        _ = float64_complex_waveform_to_protobuf(complex_waveform)
+
+    assert exc.value.args[0].startswith("Cannot convert irregular sample interval to protobuf.")
+
+
 # ========================================================
 # DoubleComplexWaveform to ComplexWaveform
 # ========================================================
@@ -349,6 +376,19 @@ def test___int16_complex_waveform_with_standard_timing___convert___valid_protobu
     assert i16_complex_waveform.t0 == converted_t0
 
 
+def test___int16_complex_waveform_with_irregular_timing___convert___raises_value_error() -> None:
+    complex_waveform = ComplexWaveform.from_array_1d([(1, 2), (3, 4)], ComplexInt32DType)
+    t0_dt = dt.datetime(2000, 12, 1, tzinfo=dt.timezone.utc)
+    complex_waveform.timing = Timing.create_with_irregular_interval(
+        [t0_dt, t0_dt + dt.timedelta(milliseconds=1000)]
+    )
+
+    with pytest.raises(ValueError) as exc:
+        _ = int16_complex_waveform_to_protobuf(complex_waveform)
+
+    assert exc.value.args[0].startswith("Cannot convert irregular sample interval to protobuf.")
+
+
 def test___int16_complex_waveform_with_scaling___convert___valid_protobuf() -> None:
     scale_mode = LinearScaleMode(2.0, 3.0)
     complex_waveform = ComplexWaveform.from_array_1d(