Low-Latency single-point analog output from ESP32 to NI 9264

[InfiniteArray]

Hello NI Community,

I have an ESP32 microcontroller that is receiving a quasi-random signal via Bluetooth. This signal is written one sample at a time via UART to the PC which connects to both the ESP32 and the DAQ. The update rate fluctuates between roughly 1900 and 1600 samples per second. The signal itself is bandwidth limited to < 200Hz.

I would like to re-generate this signal with as little latency as possible using a cDAQ-9171 with a NI 9264 analog output module.

The basic idea is that I would read a sample from the USB port, and then immediately write it to the cDAQ which would then hold this value until the next piece of data is written.

Since I am familiar with Python I first tried naively implementing the example given by stan036. However, I found that it took 50 seconds to write out a single period of my test signal. The signal looks complete, no discontinuities, just takes forever to write out.

I then found this thread by glucenac but in their example samples are written out continuously at 100Hz and generated internally. My samples do not come in at a regular 100Hz so I cannot reliably call writeCallback every n samples.

I then tried a new approach: The serial port is read as fast as possible, and the resulting data then downsampled to fit the buffer of the cDAQ. I'll post the code here for posterity:

import nidaqmx
from nidaqmx.constants import AcquisitionType
import serial
import time

from threading import Thread, Event
import numpy as np

# Define parameters for the waveform
channel = "cDAQ2Mod1/ao0"  # Adjust as per your module and channel
sampling_rate = 1000  # Samples per second
num_samples = 100

# Waveform data
global write_data
global raw_data
global raw_data_index

write_data = np.zeros(num_samples)
raw_data = np.zeros(10*num_samples) # Some large array, must be larger than the number of samples that can be read between different calls to write.
raw_data_index = 0

end_program = Event()
task_started = False

s = serial.Serial('COM8', 115200, timeout=1)

def is_float(string):
    try:
        float(string)
        return True
    except ValueError:
        return False


def raw_read_data_from_serial():
    print("Serial Port read thread running")

    global raw_data_index
    global raw_data

    while True:
        # Acquiring waveform data from the serial port
        read_data = s.readline()
        read_data = read_data.strip()
        read_data = read_data.decode("utf-8")

        if is_float(read_data):
            raw_data[raw_data_index] = float(read_data)
            raw_data_index += 1

        # This makes sure the thread ends when the program does
        if end_program.is_set():
            break

# From https://stackoverflow.com/questions/20322079/downsample-a-1d-numpy-array
# with minor adjustments to make sure I don't downsample all of raw_data but only
# valid indices.
def ResampleLinear1D(original, original_len, targetLen):
    original = np.array(original, dtype=float)
    index_arr = np.linspace(0, original_len - 1, num=targetLen, dtype=float)
    index_floor = np.array(index_arr, dtype=int)  # Round down
    index_ceil = index_floor + 1
    index_rem = index_arr - index_floor  # Remain

    val1 = original[index_floor]
    val2 = original[index_ceil % original_len]
    interp = val1 * (1.0 - index_rem) + val2 * index_rem
    assert (len(interp) == targetLen)
    return interp


# Create a DAQmx task and configure the analog output channel
with nidaqmx.Task() as task:
    task.ao_channels.add_ao_voltage_chan(channel)
    task.timing.cfg_samp_clk_timing(
        rate=sampling_rate,
        sample_mode=AcquisitionType.CONTINUOUS,
        samps_per_chan=num_samples
    )

    task.out_stream.regen_mode = nidaqmx.constants.RegenerationMode.DONT_ALLOW_REGENERATION

    t_raw_serial = Thread(target=raw_read_data_from_serial, args=())
    t_raw_serial.start()

    try:
        while True:
            try:

                write_data = ResampleLinear1D(raw_data, raw_data_index, num_samples)
                raw_data_index = 0
                task.write(write_data)

            except nidaqmx.errors.DaqWriteError:
                print("Attempted DAC conversion before data was ready")
                end_program.set()
                break

            if not task_started:
                task.start()
                task_started = True

    except KeyboardInterrupt:
        pass

    end_program.set()
    t_raw_serial.join()

    print("All threads merged.")

    task.stop()
    print("Program stopped.")

However, while this appears to be somewhat real-time the output is very distorted for anything > 5Hz. Likely an artifact of applying an input signal with a varying sampling frequency onto an output signal with a fixed frequency.

I have tried using a second thread to sample the raw value coming from the serial port at a fixed frequency but turns out python isn't very good for precise loop timing.

So I am wondering if there is a better way. From the thread with glucenac it seems that I will not be able to write data to the cDAQ more than ~10 times per second.

Has anyone else tried forwarding a signal in real time to a cDAQ? I am hoping for < 100ms latencies.