Add support for WaitForValidTimestamp to nidaqmx-python

generated/nidaqmx/_base_interpreter.py

@@ -1661,6 +1661,10 @@ def unregister_signal_event(self, task, signal_id):
     def unreserve_network_device(self, device_name):
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def wait_for_valid_timestamp(self, task, timestamp_event, timeout):
+        raise NotImplementedError
+
     @abc.abstractmethod
     def wait_until_task_done(self, task, time_to_wait):
         raise NotImplementedError

generated/nidaqmx/_grpc_interpreter.py

@@ -3254,6 +3254,14 @@ def unreserve_network_device(self, device_name):
             self._client.UnreserveNetworkDevice,
             grpc_types.UnreserveNetworkDeviceRequest(device_name=device_name))
 
+    def wait_for_valid_timestamp(self, task, timestamp_event, timeout):
+        response = self._invoke(
+            self._client.WaitForValidTimestamp,
+            grpc_types.WaitForValidTimestampRequest(
+                task=task, timestamp_event_raw=timestamp_event,
+                timeout=timeout))
+        return convert_timestamp_to_time(response.timestamp)
+
     def wait_until_task_done(self, task, time_to_wait):
         response = self._invoke(
             self._client.WaitUntilTaskDone,

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()
@@ -5636,6 +5636,22 @@ def unreserve_network_device(self, device_name):
             device_name)
         self.check_for_error(error_code)
 
+    def wait_for_valid_timestamp(self, task, timestamp_event, timeout):
+        timestamp = AbsoluteTime()
+
+        cfunc = lib_importer.windll.DAQmxWaitForValidTimestamp
+        if cfunc.argtypes is None:
+            with cfunc.arglock:
+                if cfunc.argtypes is None:
+                    cfunc.argtypes = [
+                        lib_importer.task_handle, ctypes.c_int32,
+                        ctypes.c_double, ctypes.POINTER(AbsoluteTime)]
+
+        error_code = cfunc(
+            task, timestamp_event, timeout, ctypes.byref(timestamp))
+        self.check_for_error(error_code)
+        return timestamp.to_datetime()
+
     def wait_until_task_done(self, task, time_to_wait):
         cfunc = lib_importer.windll.DAQmxWaitUntilTaskDone
         if cfunc.argtypes is None:

generated/nidaqmx/task.py

@@ -1019,6 +1019,22 @@ def stop(self):
         """
         self._interpreter.stop_task(self._handle)
 
+    def wait_for_valid_timestamp(self, timestamp_event, timeout=10.0):
+        """
+        Wait until the specified timestamp has a value.
+
+        Use this method to ensure the timestamp has a valid value to prevent an error when querying a timestamp value.
+
+        Args:
+            timestamp_event(nidaqmx.constants.TimestampEvent): Specifies the timestamp type to wait on. 
+            timeout (float): Specifies the maximum amount of time in
+                seconds to wait for a valid timestamp.
+                This method returns an error if the time elapses. The
+                default is 10. If you set timeout (sec) to
+                nidaqmx.WAIT_INFINITELY, the method waits indefinitely.
+        """
+        self._interpreter.wait_for_valid_timestamp(self._handle, timestamp_event.value, timeout)
+
     def wait_until_done(self, timeout=10.0):
         """
         Waits for the measurement or generation to complete.

src/codegen/metadata/functions.py

@@ -23606,6 +23606,11 @@
     },
     'WaitForValidTimestamp': {
         'calling_convention': 'StdCall',
+        'handle_parameter': {
+            'ctypes_data_type': 'lib_importer.task_handle',
+            'cvi_name': 'taskHandle',
+            'python_accessor': 'self._handle'
+        },
         'parameters': [
             {
                 'ctypes_data_type': 'ctypes.TaskHandle',
@@ -23644,13 +23649,14 @@
                 'direction': 'out',
                 'is_optional_in_python': False,
                 'name': 'timestamp',
-                'python_data_type': 'DateTime',
+                'python_data_type': 'datetime',
                 'python_description': 'Specifies the timestamp type to wait on.',
-                'python_type_annotation': 'nidaqmx.constants.DateTime',
+                'python_type_annotation': 'datetime',
                 'type': 'CVIAbsoluteTime'
             }
         ],
-        'python_codegen_method': 'no',
+        'python_class_name': 'Task',
+        'python_codegen_method': 'CustomCode',
         'python_description': 'DAQmx Wait for Valid Timestamp',
         'returns': 'int32'
     },

src/codegen/utilities/function_helpers.py

@@ -197,12 +197,17 @@ def to_param_argtype(parameter):
         if parameter.ctypes_data_type == "ctypes.TaskHandle":
             return "lib_importer.task_handle"
         elif parameter.python_data_type == "datetime":
-            return "AbsoluteTime"
+            if parameter.direction == "in":
+                return "AbsoluteTime"
+            else:
+                return f"ctypes.POINTER(AbsoluteTime)"
         elif parameter.direction == "in":
             # If is string input parameter, use separate custom
             # argtype to convert from unicode to bytes.
             if parameter.ctypes_data_type == "ctypes.c_char_p":
                 return "ctypes_byte_str"
+            elif parameter.python_data_type == "timestampEvent":
+                return "ctypes.c_int32"
             else:
                 return parameter.ctypes_data_type or parameter.python_data_type
         else:

src/codegen/utilities/interpreter_helpers.py

@@ -70,7 +70,6 @@
     "SetSyncPulseTimeWhen",
     "SetTimingAttributeExTimestamp",
     "SetTimingAttributeTimestamp",
-    "WaitForValidTimestamp",
     # Deprecated, not working
     "GetAnalogPowerUpStates",
 ]
@@ -447,8 +446,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/task.py

@@ -1019,6 +1019,22 @@ def stop(self):
         """
         self._interpreter.stop_task(self._handle)
 
+    def wait_for_valid_timestamp(self, timestamp_event, timeout=10.0):
+        """
+        Wait until the specified timestamp has a value.
+
+        Use this method to ensure the timestamp has a valid value to prevent an error when querying a timestamp value.
+
+        Args:
+            timestamp_event(nidaqmx.constants.TimestampEvent): Specifies the timestamp type to wait on. 
+            timeout (float): Specifies the maximum amount of time in
+                seconds to wait for a valid timestamp.
+                This method returns an error if the time elapses. The
+                default is 10. If you set timeout (sec) to
+                nidaqmx.WAIT_INFINITELY, the method waits indefinitely.
+        """
+        self._interpreter.wait_for_valid_timestamp(self._handle, timestamp_event.value, timeout)
+
     def wait_until_done(self, timeout=10.0):
         """
         Waits for the measurement or generation to complete.