Python data acquisition across DAQ vendors

NI, LabJack, MCC and Digilent all publish official Python bindings, and SCPI instruments only need a socket. The bindings differ in operating system support, naming and data layout. A good acquisition script has the same structure on any of them. The device’s clock paces the samples, the script reads in blocks, and anything slow stays out of the read loop.

Packages, vendors and operating systems

The nidaqmx package runs on Windows and Linux wherever the NI-DAQmx driver does, and NI ships the Linux driver for x86-64 only, so neither a Mac nor a Raspberry Pi can run it. Mac options are in using NI DAQ hardware with a Mac. All four vendor packages are ctypes wrappers around a C library, which affects threading (covered below).

A finite, hardware-timed read in each

Each snippet has the device’s clock take a fixed number of samples and returns them. Use this mode for anything faster than a few readings a second, for the reasons in hardware-timed vs software-timed sampling.

NI-DAQmx

import nidaqmx
from nidaqmx.constants import AcquisitionType, READ_ALL_AVAILABLE

with nidaqmx.Task() as task:
    task.ai_channels.add_ai_voltage_chan("Dev1/ai0:1")
    task.timing.cfg_samp_clk_timing(1000.0, sample_mode=AcquisitionType.FINITE,
                                    samps_per_chan=5000)
    data = task.read(READ_ALL_AVAILABLE)   # one list per channel

LabJack LJM

LabJack’s hardware-timed mode is stream mode. Ten reads of 500 scans gives 5,000 scans per channel:

from labjack import ljm

handle = ljm.openS("T7", "ANY", "ANY")
names = ["AIN0", "AIN1"]
addrs, _ = ljm.namesToAddresses(len(names), names)
rate = ljm.eStreamStart(handle, 500, len(names), addrs, 1000.0)
data = []
try:
    for _ in range(10):
        block, dev_backlog, ljm_backlog = ljm.eStreamRead(handle)
        data.extend(block)                 # interleaved AIN0, AIN1, ...
finally:
    ljm.eStreamStop(handle)
    ljm.close(handle)

MCC on Linux or macOS (uldaq)

from uldaq import (get_daq_device_inventory, DaqDevice, InterfaceType,
                   AiInputMode, Range, ScanOption, AInScanFlag,
                   WaitType, create_float_buffer)

dev = DaqDevice(get_daq_device_inventory(InterfaceType.USB)[0])
ai = dev.get_ai_device()
dev.connect()
buf = create_float_buffer(2, 5000)
rate = ai.a_in_scan(0, 1, AiInputMode.SINGLE_ENDED, Range.BIP10VOLTS, 5000,
                    1000.0, ScanOption.DEFAULTIO, AInScanFlag.DEFAULT, buf)
ai.scan_wait(WaitType.WAIT_UNTIL_DONE, 10)   # seconds
dev.disconnect()
dev.release()

Build and permission details for Linux are in running MCC DAQ devices on Linux and a Raspberry Pi.

MCC on Windows (mcculw)

mcculw uses the board numbers you assign in InstaCal. Without the BACKGROUND option, a_in_scan blocks until the scan completes:

from ctypes import cast, POINTER, c_double
from mcculw import ul
from mcculw.enums import ScanOptions, ULRange

board, low, high, per_chan = 0, 0, 1, 5000
total = (high - low + 1) * per_chan            # count is across all channels
memhandle = ul.scaled_win_buf_alloc(total)
try:
    rate = ul.a_in_scan(board, low, high, total, 1000, ULRange.BIP10VOLTS,
                        memhandle, ScanOptions.SCALEDATA)
    buf = cast(memhandle, POINTER(c_double))
    data = [buf[i] for i in range(total)]      # interleaved ch0, ch1, ...
finally:
    ul.win_buf_free(memhandle)

Digilent WaveForms SDK

The default analog-in mode fills the device buffer once at the rate you set. Digilent’s samples ship a dwfconstants.py with names for the state values. State 2 is Done.

import time
from ctypes import *

dwf = cdll.LoadLibrary("/Library/Frameworks/dwf.framework/dwf")  # macOS
hdwf, sts, n = c_int(), c_ubyte(), 8192
dwf.FDwfDeviceOpen(c_int(-1), byref(hdwf))
dwf.FDwfAnalogInFrequencySet(hdwf, c_double(100e3))
dwf.FDwfAnalogInBufferSizeSet(hdwf, c_int(n))
dwf.FDwfAnalogInChannelEnableSet(hdwf, c_int(0), c_int(1))
dwf.FDwfAnalogInChannelRangeSet(hdwf, c_int(0), c_double(5.0))
dwf.FDwfAnalogInConfigure(hdwf, c_int(1), c_int(1))     # apply and start
while dwf.FDwfAnalogInStatus(hdwf, c_int(1), byref(sts)) and sts.value != 2:
    time.sleep(0.01)
samples = (c_double * n)()
dwf.FDwfAnalogInStatusData(hdwf, c_int(0), samples, c_int(n))
dwf.FDwfDeviceClose(hdwf)

On Windows load cdll.dwf and on Linux libdwf.so. For captures longer than the buffer, see the record-mode example in data logging with an Analog Discovery 3.

SCPI over a raw socket

Benchtop DMMs and data acquisition mainframes time their own readings. With the pure-Python pyvisa-py backend you don’t need a vendor VISA install. This sequence is for 34401A-style DMMs, and other instruments use different commands:

import pyvisa

rm = pyvisa.ResourceManager("@py")
dmm = rm.open_resource("TCPIP0::192.168.1.50::5025::SOCKET",
                       read_termination="\n", write_termination="\n",
                       timeout=20000)
print(dmm.query("*IDN?"))
dmm.write("CONF:VOLT:DC 10")
dmm.write("TRIG:SOUR IMM")
dmm.write("SAMP:COUN 100")
values = [float(v) for v in dmm.query("READ?").split(",")]

The instrument paces those 100 readings itself, so its integration time and trigger delay set their spacing, not your script. The spacing is regular, but you don’t set it the way you set a sample clock.

Buffering and threads

Every hardware-timed device fills a buffer the host must drain in time. On NI hardware, falling behind ends the task with error -200279 because unread samples were overwritten. LJM reports a growing device and LJM backlog on each eStreamRead. The WaveForms SDK counts lost samples, and the daqhats library sets buffer_overrun. Size each read to roughly 50 to 200 ms of data. Smaller reads spend more time in per-call overhead, and larger ones add latency.

Python releases the global interpreter lock during each ctypes call, so a thread blocked in eStreamRead or task.read doesn’t stop your other threads. Give each device one reader thread that only reads and pushes blocks onto a queue.Queue. For NI hardware, nidaqmx.stream_readers can read directly into a preallocated NumPy array, which avoids building Python lists at high rates.

Writing to disk without blocking

Disk writes stall when a USB drive flushes its cache, antivirus scans a growing file or a network share hangs. Put file writing in its own thread that consumes the queue, so a slow write delays the file and not the acquisition. Write binary, with ndarray.tofile on an open file or appended chunks in an HDF5 dataset via h5py. Formatting floats as CSV text in the read loop is the most common reason a script that worked at 1 kS/s falls over at 50 kS/s. Convert to CSV afterward if someone needs it. Choosing a file format for test data compares the options.

Time alignment across devices

Each device timestamps its samples with its own oscillator, and the host only knows roughly when each one started. When each task starts, record time.time() for wall-clock context and time.monotonic() for interval math, and store the rate each API returns rather than the one you requested. Starts then line up to within the host’s scheduling latency, but the clocks still drift apart over a long run. Synchronizing multiple DAQ devices covers that drift and how to correct it with a shared marker signal.

Live plotting at high sample rates

Two channels at 10 kS/s over a 60 s window is 1.2 million points, and redrawing them in matplotlib several times a second takes more CPU than the acquisition. A plot has only as many columns as it has pixels across, so reduce each channel to a min/max pair per pixel column before drawing. Spikes stay visible that way. Refresh 10 to 30 times a second from the main thread, reading a copy of the latest data, never from inside the reader thread. pyqtgraph is designed for fast-updating plots. Leave matplotlib for analysis after the run.

DAQAtlas, which is in development, records on its own thread and runs NI, LabJack, MCC, Digilent and SCPI hardware in one run. The beta list is open.