Add Unit test for cfg_time_start_trig (#488)

Author: DeborahOoi96

generated/nidaqmx/_grpc_interpreter.py

@@ -8,6 +8,7 @@
 from typing import Callable, Generic, Optional, TypeVar
 
 import google.protobuf.message
+from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
 import grpc
 import numpy
 

generated/nidaqmx/_grpc_time.py

@@ -14,9 +14,11 @@
 _YS_PER_FS = 10**9
 
 
-def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTimestamp) -> None:
+def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: Optional[GrpcTimestamp] = None) -> GrpcTimestamp:
     utc_dt = dt.astimezone(tz=timezone.utc)
     seconds = int(utc_dt.timestamp())
+    if ts is None:
+        ts = GrpcTimestamp()
 
     if isinstance(dt, ht_datetime):
         total_yoctoseconds = dt.yoctosecond
@@ -30,6 +32,7 @@ def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTime
         nanos = utc_dt.microsecond * _NS_PER_US
 
     ts.FromNanoseconds(seconds * _NS_PER_S + nanos)
+    return ts
 
 
 def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: Optional[timezone] = None) -> ht_datetime:
@@ -47,4 +50,4 @@ def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: Optional[timezone] = No
     # Add in precision
     dt = dt.replace(microsecond=microsecond, femtosecond=femtosecond, yoctosecond=yoctosecond)
     # Then convert to requested timezone
-    return dt.astimezone(tz=tzinfo)
+    return dt.astimezone(tz=tzinfo)

generated/nidaqmx/_lib_time.py

@@ -96,4 +96,4 @@ def __lt__(self, other) -> bool:
         if self.msb == other.msb:
             return self.lsb < other.lsb
         else:
-            return self.msb < other.msb
+            return self.msb < other.msb

generated/nidaqmx/_library_interpreter.py

@@ -3684,7 +3684,7 @@ def get_trig_attribute_timestamp(self, task, attribute):
         error_code = cfunc(
             task, attribute, ctypes.byref(value))
         self.check_for_error(error_code)
-        return value
+        return value.to_datetime()
 
     def get_trig_attribute_uint32(self, task, attribute):
         value = ctypes.c_uint32()

src/codegen/templates/_grpc_interpreter.py.mako

@@ -24,6 +24,7 @@ import warnings
 from typing import Callable, Generic, Optional, TypeVar
 
 import google.protobuf.message
+from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
 import grpc
 import numpy
 

src/codegen/utilities/interpreter_helpers.py

@@ -447,8 +447,10 @@ def get_return_values(func):
                 return_values.append(param.parameter_name)
             else:
                 return_values.append(f"{param.parameter_name}.tolist()")
-        elif param.type == "TaskHandle" or param.type == "CVIAbsoluteTime":
+        elif param.type == "TaskHandle":
             return_values.append(param.parameter_name)
+        elif param.type == "CVIAbsoluteTime":
+            return_values.append(f"{param.parameter_name}.to_datetime()")
         else:
             return_values.append(f"{param.parameter_name}.value")
     if func.is_init_method:

src/handwritten/_grpc_time.py

@@ -14,9 +14,11 @@
 _YS_PER_FS = 10**9
 
 
-def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTimestamp) -> None:
+def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: Optional[GrpcTimestamp] = None) -> GrpcTimestamp:
     utc_dt = dt.astimezone(tz=timezone.utc)
     seconds = int(utc_dt.timestamp())
+    if ts is None:
+        ts = GrpcTimestamp()
 
     if isinstance(dt, ht_datetime):
         total_yoctoseconds = dt.yoctosecond
@@ -30,6 +32,7 @@ def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTime
         nanos = utc_dt.microsecond * _NS_PER_US
 
     ts.FromNanoseconds(seconds * _NS_PER_S + nanos)
+    return ts
 
 
 def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: Optional[timezone] = None) -> ht_datetime:
@@ -47,4 +50,4 @@ def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: Optional[timezone] = No
     # Add in precision
     dt = dt.replace(microsecond=microsecond, femtosecond=femtosecond, yoctosecond=yoctosecond)
     # Then convert to requested timezone
-    return dt.astimezone(tz=tzinfo)
+    return dt.astimezone(tz=tzinfo)

src/handwritten/_lib_time.py

@@ -96,4 +96,4 @@ def __lt__(self, other) -> bool:
         if self.msb == other.msb:
             return self.lsb < other.lsb
         else:
-            return self.msb < other.msb
+            return self.msb < other.msb

tests/component/_task_modules/test_triggers.py

@@ -0,0 +1,53 @@
+from datetime import timedelta
+
+import pytest
+from hightime import datetime as ht_datetime
+
+from nidaqmx.constants import Timescale
+from nidaqmx.task import Task
+
+
+@pytest.fixture()
+def ai_voltage_field_daq_task(task, sim_field_daq_device):
+    """Gets AI voltage task."""
+    task.ai_channels.add_ai_voltage_chan(sim_field_daq_device.ai_physical_chans[0].name)
+    yield task
+
+
+def test___default_arguments___cfg_time_start_trig___no_errors(
+    ai_voltage_field_daq_task: Task,
+):
+    ai_voltage_field_daq_task.timing.cfg_samp_clk_timing(1000)
+    trigger_time = ht_datetime.now() + timedelta(seconds=10)
+
+    ai_voltage_field_daq_task.triggers.start_trigger.cfg_time_start_trig(trigger_time)
+
+    when_value = ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+    timescale_value = ai_voltage_field_daq_task.triggers.start_trigger.timestamp_timescale
+    assert timescale_value == Timescale.USE_HOST
+    assert when_value.year == trigger_time.year
+    assert when_value.month == trigger_time.month
+    assert when_value.day == trigger_time.day
+    assert when_value.hour == trigger_time.hour
+    assert when_value.minute == trigger_time.minute
+    assert when_value.second == trigger_time.second
+
+
+def test___arguments_provided___cfg_time_start_trig___no_errors(
+    ai_voltage_field_daq_task: Task,
+):
+    ai_voltage_field_daq_task.timing.cfg_samp_clk_timing(1000)
+    trigger_time = ht_datetime.now() + timedelta(seconds=10)
+    # simulated devices don't support setting timescale to USE_IO_DEVICE
+    timescale = Timescale.USE_HOST
+
+    ai_voltage_field_daq_task.triggers.start_trigger.cfg_time_start_trig(trigger_time, timescale)
+
+    when_value = ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+    assert when_value.year == trigger_time.year
+    assert when_value.month == trigger_time.month
+    assert when_value.day == trigger_time.day
+    assert when_value.hour == trigger_time.hour
+    assert when_value.minute == trigger_time.minute
+    assert when_value.second == trigger_time.second
+    assert ai_voltage_field_daq_task.triggers.start_trigger.timestamp_timescale == timescale

tests/component/_task_modules/test_triggers_properties.py

@@ -1,9 +1,13 @@
+from datetime import timezone
+
 import pytest
+from hightime import datetime as ht_datetime
 
 from nidaqmx.constants import TaskMode, TriggerType
 from nidaqmx.error_codes import DAQmxErrors
 from nidaqmx.errors import DaqError
 from nidaqmx.task import Task
+from tests.unit._time_utils import JAN_01_1904_HIGHTIME, JAN_01_2002_HIGHTIME
 
 
 @pytest.fixture()
@@ -13,6 +17,20 @@ def ai_voltage_task(task, sim_6363_device):
     yield task
 
 
+@pytest.fixture()
+def ai_voltage_field_daq_task(task, sim_field_daq_device):
+    """Gets AI voltage task."""
+    task.ai_channels.add_ai_voltage_chan(sim_field_daq_device.ai_physical_chans[0].name)
+    yield task
+
+
+def convert_to_local_timezone(expected_time_utc):
+    current_time_utc = ht_datetime.now(timezone.utc)
+    local_timezone_offset = current_time_utc.astimezone().utcoffset()
+    local_expected_time = expected_time_utc + local_timezone_offset
+    return local_expected_time
+
+
 def test___ai_task___get_float_property___returns_default_value(ai_voltage_task: Task):
     assert ai_voltage_task.triggers.start_trigger.dig_edge_dig_fltr_timebase_rate == 0.0
 
@@ -99,3 +117,57 @@ def test___ai_task___reset_uint32_property___returns_default_value(ai_voltage_ta
     del ai_voltage_task.triggers.reference_trigger.pretrig_samples
 
     assert ai_voltage_task.triggers.reference_trigger.pretrig_samples == 2
+
+
+@pytest.mark.xfail(reason="Timestamp conversion doesn't work on dates before 1970", raises=OSError)
+def test___ai_voltage_field_daq_task___get_timestamp_property___returns_default_value(
+    ai_voltage_field_daq_task: Task,
+):
+    ai_voltage_field_daq_task.timing.cfg_samp_clk_timing(1000)
+
+    when_value = ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+
+    localized_default_value = convert_to_local_timezone(JAN_01_1904_HIGHTIME)
+    assert when_value.year == localized_default_value.year
+    assert when_value.month == localized_default_value.month
+    assert when_value.day == localized_default_value.day
+    assert when_value.hour == localized_default_value.hour
+    assert when_value.minute == localized_default_value.minute
+    assert when_value.second == localized_default_value.second
+
+
+def test___ai_voltage_field_daq_task___set_timestamp_property___returns_assigned_value(
+    ai_voltage_field_daq_task: Task,
+):
+    value_to_test = JAN_01_2002_HIGHTIME
+    ai_voltage_field_daq_task.timing.cfg_samp_clk_timing(1000)
+
+    ai_voltage_field_daq_task.triggers.start_trigger.trig_when = value_to_test
+
+    when_value = ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+    localized_value_to_test = convert_to_local_timezone(value_to_test)
+    assert when_value.year == localized_value_to_test.year
+    assert when_value.month == localized_value_to_test.month
+    assert when_value.day == localized_value_to_test.day
+    assert when_value.hour == localized_value_to_test.hour
+    assert when_value.minute == localized_value_to_test.minute
+    assert when_value.second == localized_value_to_test.second
+
+
+@pytest.mark.xfail(reason="Timestamp conversion doesn't work on dates before 1970", raises=OSError)
+def test___ai_voltage_field_daq_task___reset_timestamp_property___returns_default_value(
+    ai_voltage_field_daq_task: Task,
+):
+    ai_voltage_field_daq_task.timing.cfg_samp_clk_timing(1000)
+    ai_voltage_field_daq_task.triggers.start_trigger.trig_when = JAN_01_2002_HIGHTIME
+
+    del ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+
+    when_value = ai_voltage_field_daq_task.triggers.start_trigger.trig_when
+    localized_default_value = convert_to_local_timezone(JAN_01_1904_HIGHTIME)
+    assert when_value.year == localized_default_value.year
+    assert when_value.month == localized_default_value.month
+    assert when_value.day == localized_default_value.day
+    assert when_value.hour == localized_default_value.hour
+    assert when_value.minute == localized_default_value.minute
+    assert when_value.second == localized_default_value.second

tests/conftest.py

@@ -123,6 +123,7 @@ def _device_by_product_type(
             device_type == DeviceType.ANY
             or (device_type == DeviceType.REAL and not device.is_simulated)
             or (device_type == DeviceType.SIMULATED and device.is_simulated)
+            and len(device.ai_physical_chans) >= 1
         )
         if device_type_match and device.product_type == product_type:
             return device
@@ -147,6 +148,12 @@ def sim_6363_device(system: nidaqmx.system.System) -> nidaqmx.system.Device:
     return _device_by_product_type("PCIe-6363", DeviceType.SIMULATED, system)
 
 
+@pytest.fixture(scope="function")
+def sim_field_daq_device(system):
+    """Gets simulated Field DAQ device information."""
+    return _device_by_product_type("FD-11601", DeviceType.SIMULATED, system)
+
+
 @pytest.fixture(scope="function")
 def sim_ts_chassis(system: nidaqmx.system.System) -> nidaqmx.system.Device:
     """Gets simulated TestScale chassis information."""

tests/max_config/nidaqmxMaxConfig.ini

@@ -257,3 +257,27 @@ DevIsSimulated = 1
 CompactDAQ.ChassisDevName = cdaqChassisTester
 CompactDAQ.SlotNum = 1
 
+[DAQmxFieldDAQ FieldDAQ1]
+ProductType = FD-11601
+DevSerialNum = 0x0
+DevIsSimulated = 1
+BusType = TCP/IP
+TCPIP.Hostname = 
+TCPIP.EthernetIP = 0.0.0.0
+TCPIP.EthernetMAC = 00:00:00:00:00:00
+TCPIP.EthernetMDNSServiceInstance = 
+TCPIP.DevIsReserved = 0
+
+[DAQmxFieldDAQBank FieldDAQ1-Bank1]
+ProductType = FD-11601
+DevSerialNum = 0x0
+DevIsSimulated = 1
+FieldDAQ.BankNum = 1
+FieldDAQ.DevName = FieldDAQ1
+
+[DAQmxFieldDAQBank FieldDAQ1-Bank2]
+ProductType = FD-11601
+DevSerialNum = 0x0
+DevIsSimulated = 1
+FieldDAQ.BankNum = 2
+FieldDAQ.DevName = FieldDAQ1

tests/unit/_time_utils.py

@@ -10,3 +10,5 @@
 
 JAN_01_2002_DATETIME = std_datetime(2002, 1, 1, tzinfo=timezone.utc)
 JAN_01_2002_HIGHTIME = ht_datetime(2002, 1, 1, tzinfo=timezone.utc)
+JAN_01_1904_DATETIME = std_datetime(1904, 1, 1, tzinfo=timezone.utc)
+JAN_01_1904_HIGHTIME = ht_datetime(1904, 1, 1, tzinfo=timezone.utc)

tests/unit/test_grpc_time.py

@@ -1,17 +1,17 @@
-from datetime import datetime as std_datetime
-from datetime import timedelta, timezone
+from datetime import datetime as std_datetime, timedelta, timezone
 
 import pytest
 from hightime import datetime as ht_datetime
 
 from tests.unit._time_utils import (
-    JAN_01_2002_TIMESTAMP_1970_EPOCH,
     JAN_01_2002_DATETIME,
     JAN_01_2002_HIGHTIME,
+    JAN_01_2002_TIMESTAMP_1970_EPOCH,
 )
 
 try:
     from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
+
     import nidaqmx._grpc_time as grpc_time
     import nidaqmx._stubs.nidaqmx_pb2 as nidaqmx_pb2
 except ImportError:
@@ -22,8 +22,7 @@
 
 @pytest.mark.parametrize("from_dt", [(JAN_01_2002_DATETIME), (JAN_01_2002_HIGHTIME)])
 def test___utc_datetime___convert_to_timestamp___is_reversible(from_dt):
-    to_ts = GrpcTimestamp()
-    grpc_time.convert_time_to_timestamp(from_dt, to_ts)
+    to_ts = grpc_time.convert_time_to_timestamp(from_dt)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
     total_nanoseconds = to_ts.ToNanoseconds()
@@ -73,8 +72,7 @@ def test___tz_datetime___convert_to_timestamp___is_reversible(
 ):
     from_dt = datetime_cls(2002, 1, 1, tzinfo=tzinfo)
 
-    to_ts = GrpcTimestamp()
-    grpc_time.convert_time_to_timestamp(from_dt, to_ts)
+    to_ts = grpc_time.convert_time_to_timestamp(from_dt)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=tzinfo)
 
     assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH + expected_offset
@@ -104,8 +102,7 @@ def test___datetime_with_microseconds___convert_to_timestamp___is_reversible(
 ):
     from_dt = base_dt.replace(microsecond=microsecond)
 
-    to_ts = GrpcTimestamp()
-    grpc_time.convert_time_to_timestamp(from_dt, to_ts)
+    to_ts = grpc_time.convert_time_to_timestamp(from_dt)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
     assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH
@@ -135,8 +132,7 @@ def test___datetime_with_femtoseconds___convert_to_timestamp___is_reversible(
 ):
     from_dt = base_dt.replace(femtosecond=femtosecond)
 
-    to_ts = GrpcTimestamp()
-    grpc_time.convert_time_to_timestamp(from_dt, to_ts)
+    to_ts = grpc_time.convert_time_to_timestamp(from_dt)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
     assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH