When using AI (analog input) and AO (analog output) channels simultaneously for real-time data acquisition, there will be a delay in the AI channels.

[sq2100]

I want to use the following code to feedback control my signal amplifier to act as a constant current source. When I test the AI (analog input) and AO (analog output) functionalities separately, their real-time capabilities seem to be guaranteed, or at least appear to be quite real-time. However, when I integrate them, I find that the AI has significant delays and cannot acquire data in real-time. What could be causing this issue? Currently, my solution is to separate AI and AO into two different threads and use global variables to share data

import nidaqmx
from nidaqmx import stream_readers, stream_writers
from nidaqmx.constants import AcquisitionType
import numpy as np
import time
from simple_pid import PID

# PID settings
pid = PID(0.1, 0.01, 0.05, setpoint=1)  # Setpoint set to 1 mA

with nidaqmx.Task() as task_voltage_output, nidaqmx.Task() as task_input:
    # Configure voltage output channel
    task_voltage_output.ao_channels.add_ao_voltage_chan("DAO/ao0")
    voltage_writer = stream_writers.AnalogSingleChannelWriter(task_voltage_output.out_stream)

    # Configure voltage and current reading channels
    task_input.ai_channels.add_ai_voltage_chan("DAI/ai0")  # Voltage channel
    task_input.ai_channels.add_ai_voltage_chan("DAI/ai1")  # Current channel
    task_input.timing.cfg_samp_clk_timing(rate=1000)  # Set sample rate to 1000 Hz

    reader = stream_readers.AnalogMultiChannelReader(task_input.in_stream)

    output_voltage = 0.0  # Initial voltage output
    data = np.zeros(2)  # Prepare data array for two channels

    try:
        while True:
            # Read current voltage and current simultaneously
            reader.read_one_sample(data)
            current_voltage, current_current = data

            # PID calculation for voltage adjustment
            correction = pid(current_current)

            # Update output voltage
            output_voltage += correction
            output_voltage = np.clip(output_voltage, 0, 2)  # Limit voltage output between 0 and 2V

            # Set output voltage
            voltage_writer.write_one_sample(output_voltage)

            # Print and store data
            print(f"Voltage: {current_voltage} V, Current: {current_current} mA, Control Voltage: {output_voltage} V")

            # time.sleep(0.1)  # Sleep for a while before next measurement and adjustment

    except KeyboardInterrupt:
        print("Stopped by user.")

[bkeryan]

Hi sq2100,

First, make sure your application isn't doing unnecessary task state transitions. See Task State Model for more details. To do this, add calls to task_input.start() and task_voltage_output.start(), above the while loop.

Second, your AI task is configured to perform a finite acquisition of 1000 samples at 1000 samples/second. Part of this is configured by optional parameters on the cfg_samp_clk_timing method:

def cfg_samp_clk_timing(
            self, rate, source="", active_edge=Edge.RISING,
            sample_mode=AcquisitionType.FINITE, samps_per_chan=1000):

I suspect that your call to reader.read_one_sample might be acquiring 1000 samples, returning the first one, and leaving the other 999 samples unread in the DAQmx task's circular buffer.

You can reduce the latency somewhat by passing AcquisitionType.CONTINUOUS instead of AcquisitionType.FINITE, setting input_task.in_stream.relative_to = ReadRelativeTo.MOST_RECENT_SAMPLE, input_task.in_stream.offset = -1, and input_task.in_stream.overwrite = OverwriteMode.OVERWRITE_UNREAD_SAMPLES and starting the task above the loop. This will keep the AI task continuously running, and each call to reader.read_one_sample will pull the most recent sample from the DAQmx task's circular buffer. (These in_stream settings are from memory, so they may be incomplete. There may be community-written LabVIEW examples of this on ni.com.)

For the lowest possible latency, you should use AcquisitionType.HW_TIMED_SINGLE_POINT and DAQmx Wait For Next Sample Clock for both the AI and AO tasks. This disables buffering, allows you to explicitly synchronize your PID loop to the sample clock, uses device-specific read/write code that is optimized for latency, and returns an error if your PID loop is late. However, this functionality is best used with an RTOS like NI Linux RT, and nidaqmx-python doesn't support the DAQmx Wait For Next Sample Clock function that is needed for this mode.

Also, you can control the delay between the AI and AO tasks by sharing a start trigger or sample clock. This isn't relevant for the MOST_RECENT_SAMPLE mode that discards less recent samples, but it is relevant when reading all of the AI samples.

Brad

[sq2100]

Dear Brad,

Thank you very much for your detailed suggestions on optimizing our data acquisition setup. Your advice on managing task transitions and setting up continuous acquisition modes has been extremely helpful and has significantly reduced the latency in our system.

However, we are still experiencing a latency of about 30ms, which I suspect might be due to hardware limitations. Based on your recommendations, I'm planning to experiment with LabVIEW, which might offer more robust capabilities for our requirements.

Thanks again for your invaluable assistance!