DAQmx from Python: acquiring real signals instead of fighting the driver

Installing the driver and the package, the four calls every task is made of, why the sample rate is an aggregate rather than a per-channel figure, finite versus continuous versus software-timed, analog and counter output, the error codes you will actually hit, and the requests the hardware will refuse.

Before writing a line of Python, accept one thing: Python is not a real-time system. The instant a sample is taken is decided by the DAQmx driver and the clock on the card. Python's job is to configure the task and then collect data that the hardware has already acquired. Once that is straight, the usual first-week puzzles — why my loop is not fast enough, why the timestamps are wrong, why it works until I add a print — all have the same answer.

1. Two layers, and the order matters

There are two pieces to install, and they are not interchangeable. The lower one is the NI-DAQmx driver, which is the part that actually talks to the hardware; it installs separately and is not a Python package. The upper one is nidaqmx, which wraps the driver's C API in Python objects. The PyPI name is nidaqmx — nidaqmx-python is only the GitHub repository name, and installing it will not work.

import nidaqmx

print(nidaqmx.__version__)          # e.g. 1.6.0

system = nidaqmx.system.System.local()
for device in system.devices:
    print(device.name, device.product_type)
    for channel in device.ai_physical_chans:
        print("   ", channel.name)

The name column — Dev1 and so on — is the prefix for every channel string that follows. If a device name is wrong, the failure happens when a channel is added, not at import, and the message will name the device rather than the line of your code.

2. Every task is the same four calls

The API has a very fixed shape. Every acquisition task, from a single thermocouple to thirty-two channels at 250 kS/s, is these four steps with different arguments:

1. Create the channel — tell the driver which physical channel and what range.
2. Configure the timing — who supplies the sample clock, how fast, how many samples.
3. Start, and read or write.
4. Close the task and release the device.

import nidaqmx
from nidaqmx.constants import AcquisitionType

RATE = 10_000.0     # samples per second, shared by every enabled channel
COUNT = 5_000       # samples per channel

with nidaqmx.Task() as task:
    task.ai_channels.add_ai_voltage_chan(
        "Dev1/ai0",
        min_val=-10.0,
        max_val=10.0,
    )
    task.timing.cfg_samp_clk_timing(
        rate=RATE,
        sample_mode=AcquisitionType.FINITE,
        samps_per_chan=COUNT,
    )
    data = task.read(number_of_samples_per_channel=COUNT)

print(len(data))                    # 5000
print(sum(data) / len(data))        # mean, in volts

Step four is the one people skip, and the device does not release itself. A task left open from a previous run means the next one fails with "the specified resource is reserved". The with statement clears the task on the way out and is the reason to use it even in a throwaway script. As of nidaqmx 1.6.0, close() became an alias for clear_task(), so either name works today.

min_val and max_val are not "how big is my signal" — they set the range, and the range sets the resolution. On a 16-bit converter, ±10 V costs 305 µV per code; ±0.2 V costs 6.1 µV. Set the range wider than the signal and you spend resolution you did not have to spend; set it narrower and a signal that overshoots is clipped flat.

3. The rate is the whole device's, not one channel's

This is the single most common misunderstanding in a first DAQmx program. The rate in cfg_samp_clk_timing is the aggregate sample rate for the device, shared among every channel you enabled. Four channels at 50 kS/s each needs rate=200_000; writing rate=50_000 gets you 12.5 kS/s per channel and a waveform that is wrong by a factor of four. The "maximum sample rate" on a datasheet is the same aggregate figure.

The architecture behind it is multiplexing: the card has one converter, and a scan switches it round the channels in turn. (The USB-4431 and USB-4432 are the exception — they carry a converter per channel.) Two consequences show up directly in the code.

import numpy as np
import nidaqmx
from nidaqmx.constants import AcquisitionType, TerminalConfiguration

RATE = 50_000.0     # aggregate across all four channels -> 12.5 kS/s each
COUNT = 20_000      # samples per channel

with nidaqmx.Task() as task:
    task.ai_channels.add_ai_voltage_chan(
        "Dev1/ai0:3",
        min_val=-1.0,
        max_val=1.0,
        terminal_config=TerminalConfiguration.DIFF,
    )
    task.timing.cfg_samp_clk_timing(
        rate=RATE,
        sample_mode=AcquisitionType.FINITE,
        samps_per_chan=COUNT,
    )
    data = task.read(number_of_samples_per_channel=COUNT)

block = np.array(data, dtype=np.float64)    # shape (channels, samples)
print(block.shape)                          # (4, 20000)

t = np.arange(COUNT) / RATE                 # one time axis, shared by all four

• All four channels share one time axis, but they are not sampled at the same instant. A scan takes them one after another, separated by the channel interval — channel count divided by aggregate rate, which here is 4 / 50 000 = 80 µs. Any phase comparison between channels has to allow for that. Genuine simultaneous sampling needs a card with a converter per channel.
• Dev1/ai0:3 with TerminalConfiguration.DIFF consumes four pairs of pins, not four pins. A module specified as 16SE/8DI has eight usable channels in differential mode, which is the mode you want whenever the signal is small or the ground between the sensor and the card is not the same ground.

4. Finite, continuous, and the mode that has no sample rate

There are three ways to acquire, and only two of them are timed by hardware.

Swipe the table sideways to see all columns

The third one has to be labelled clearly, because it does not have a sample rate. Samples happen when your program asks for them, so the interval depends on the operating system scheduler, on the interpreter, and on what else your code is doing that millisecond. Jitter is measured in milliseconds. It cannot measure anything where time matters — no frequency, no phase, no waveform. Hardware timing is the default for the other two modes, and it is the only thing that produces a trustworthy time axis.

import functools
import numpy as np
import nidaqmx
from nidaqmx import constants, stream_readers

RATE = 20_000.0        # aggregate across all enabled channels
CHUNK = 2_000          # samples per channel handed over on each callback
CHANNELS = 4

def on_samples(reader, store, task_handle, event_type, num_samples, callback_data):
    reader.read_many_sample(store["buffer"], num_samples)
    store["blocks"].append(store["buffer"].copy())
    return 0        # a non-zero return tells the driver to stop the task

def main():
    store = {
        "buffer": np.zeros((CHANNELS, CHUNK), dtype=np.float64),
        "blocks": [],
    }

    with nidaqmx.Task() as task:
        task.ai_channels.add_ai_voltage_chan(
            "Dev1/ai0:3", min_val=-10.0, max_val=10.0
        )
        task.timing.cfg_samp_clk_timing(
            rate=RATE,
            sample_mode=constants.AcquisitionType.CONTINUOUS,
            samps_per_chan=CHUNK * 10,      # buffer size: ten chunks
        )
        reader = stream_readers.AnalogMultiChannelReader(task.in_stream)
        task.register_every_n_samples_acquired_into_buffer_event(
            CHUNK, functools.partial(on_samples, reader, store)
        )
        task.start()
        input("Acquiring. Press Enter to stop.")
        task.stop()

if __name__ == "__main__":
    main()

Three rules are specific to this package, and every one of them causes a silent failure if you do not know it:

• The callback must return 0. The driver reads the return value, and anything non-zero is taken as a request to stop the task.
• The buffer is preallocated and reused. read_many_sample fills the array you hand it rather than returning a new one, so the copy inside the callback is not optional — without it, every stored block ends up pointing at the same memory and your whole recording is the last chunk, repeated.
• Extra state cannot be passed through callback_data. The Python binding does not deliver it — the C layer passes None — so context has to come in through functools.partial or a closure, as above.

The callback runs on the driver's own thread, and its only job is to move data out of the way. Plotting, writing files or doing arithmetic in there will stall acquisition and cost you samples. If you would rather not use a callback, polling is shorter and easier to reason about, at the cost of a loop that has to keep up.

import numpy as np
import nidaqmx
from nidaqmx.constants import AcquisitionType

RATE = 20_000.0
CHUNK = 2_000       # samples per channel per read

with nidaqmx.Task() as task:
    task.ai_channels.add_ai_voltage_chan("Dev1/ai0:3", min_val=-10.0, max_val=10.0)
    task.timing.cfg_samp_clk_timing(
        rate=RATE,
        sample_mode=AcquisitionType.CONTINUOUS,
        samps_per_chan=CHUNK * 10,
    )
    task.start()
    try:
        while True:
            block = np.array(task.read(number_of_samples_per_channel=CHUNK))
            print(block.shape, block.mean(axis=1))
    except KeyboardInterrupt:
        task.stop()

In continuous mode samps_per_chan is the buffer size, not a target count. The example leaves ten chunks of headroom: as long as the loop comes back for data before ten chunks have been filled, nothing is lost. Fall behind and the driver overwrites the oldest samples, which is the error in the next section but one.

5. The output side: waveforms and counters

Output is the same shape as input with the direction reversed. There is one detail worth knowing when several tasks have to start together: write() starts the task for you by default, so a shared start needs auto_start=False followed by an explicit start().

import numpy as np
import nidaqmx
from nidaqmx.constants import AcquisitionType

RATE = 100_000.0
N = 1_000

with nidaqmx.Task() as task:
    task.ao_channels.add_ao_voltage_chan("Dev1/ao0", min_val=-10.0, max_val=10.0)
    task.timing.cfg_samp_clk_timing(
        rate=RATE,
        sample_mode=AcquisitionType.FINITE,
        samps_per_chan=N,
    )
    wave = 5.0 * np.sin(2.0 * np.pi * 100.0 * np.arange(N) / RATE)
    task.write(wave, auto_start=False)
    task.start()
    task.wait_until_done()

The counter is the most overlooked and most useful part of this layer. It runs entirely in hardware: no CPU, no software jitter. Measuring a frequency, counting encoder pulses and generating an exact pulse train all live here. The multifunction models carry two 32-bit counters on an 80 MHz timebase.

import nidaqmx
from nidaqmx.constants import AcquisitionType, CountDirection, Edge

# Counter input: count rising edges arriving on the counter source terminal.
with nidaqmx.Task() as task:
    task.ci_channels.add_ci_count_edges_chan(
        "Dev1/ctr0",
        edge=Edge.RISING,
        initial_count=0,
        count_direction=CountDirection.COUNT_UP,
    )
    task.start()
    print(task.read())          # edges counted since the task started

# Counter output: a 1 kHz square wave, made by the hardware alone.
with nidaqmx.Task() as task:
    task.co_channels.add_co_pulse_chan_freq(
        "Dev1/ctr1",
        freq=1_000.0,
        duty_cycle=0.5,
    )
    task.timing.cfg_implicit_timing(sample_mode=AcquisitionType.CONTINUOUS)
    task.start()
    input("Generating 1 kHz. Press Enter to stop.")

Note that the counter output has no sample rate. The sample_mode in cfg_implicit_timing only says "keep going"; the frequency is divided down from the 80 MHz on-board timebase and has nothing to do with how fast your program runs.

6. The error message usually contains the answer

DAQmx errors are more explicit than most drivers'. Catch DaqmxError, read error_code for the number, and print the exception itself — the driver writes a full sentence, and that sentence is the documentation for the case you are in.

import nidaqmx
from nidaqmx.errors import DaqmxError

with nidaqmx.Task() as task:
    # An entry-level module is fixed at +/-10 V and cannot run at 500 kS/s.
    task.ai_channels.add_ai_voltage_chan("Dev1/ai0", min_val=-0.2, max_val=0.2)
    task.timing.cfg_samp_clk_timing(rate=500_000.0)
    try:
        task.start()
        data = task.read(number_of_samples_per_channel=1_000)
    except DaqmxError as err:
        print("error code:", err.error_code)
        print(err)              # the driver's own explanation, in full
        task.stop()

Swipe the table sideways to see all columns

7. What you ask for is not always what the hardware can do

Python will not stop you. rate=500_000 and min_val=-0.2 are valid arguments to a valid call; the refusal comes later, when the driver tries to configure the hardware. It is worth knowing which tier you are buying before you write the script, not after.

Swipe the table sideways to see all columns

8. The same handful of mistakes, every time

• Not closing the task. The device stays reserved and the next run fails with -50103.
• Treating rate as per-channel. Enable four channels and each one gets a quarter of what you asked for.
• Doing real work inside the callback. Move the data out and process it on another thread.
• Using time.sleep as a clock. Python's timing is millisecond-class; the counter timebase is 80 MHz.
• Ignoring the buffer size. In continuous mode samps_per_chan is the buffer, and too small a buffer overflows by construction.
• Running the acquisition loop on a GUI main thread. Every redraw is a chance to miss a read.
• Not recording version numbers. The package has changed shape across major versions — 1.6.0 turned close() into an alias for clear_task() — and a driver/package mismatch is the hardest kind of problem to diagnose. Pin both in anything you ship.

9. Where this leaves you

Everything above is plain DAQmx: the same calls, the same constants, the same channel strings. Our modules run a DAQmx-compatible driver, so the code does not change — you swap the device name for ours and the rest stays as it is. That is the concrete meaning of the claim that an existing program can move across, which the product note at "Industrial, not laboratory" sets out in more detail.

Runnable examples for Python, LabVIEW and .NET are on the downloads page. If a script of yours fails and the message is not obvious, send us the full error text together with the channel configuration and the module you are using — between the code, the error code and the wiring, there is usually only one thing it can be.

Every figure in this note is either arithmetic from published resolution and sample-rate specifications, or a value stated by the standard it cites. Where a number depends on a particular module, confirm it against that module datasheet before design freeze.