用 Python 调用 DAQmx：把真实信号采下来，而不是跟驱动较劲

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)

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

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

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()

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()

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()

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.")

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()