使用cfg_anlg_edge_ref_trig()触发后nidaqmx丢失实时数据的问题

问题 #

使用Python的nidaqmx包读取NI 6356采集卡模拟通道数据时，常规无触发配置下可正常实时获取电压读数，但调用cfg_anlg_edge_ref_trig()设置触发后，nidaqmx会等待整个测量完成才返回数据，无法实现实时数据流，不符合预期。切换代码中trigger参数的True/False可复现问题，最小复现代码如下：

import nidaqmx as daq
import numpy as np
import matplotlib.pyplot as plt

from datetime import datetime
from nidaqmx.stream_readers import AnalogMultiChannelReader

def set_ai_measurement(number_samples, ai_channels, clock_frequency, trigger=False, trigger_channel='Dev1/ai0', pretrigger_samples=0, trigger_threshold=1):
    """ Configures the counter for a finite measurement.

    Args:
        ai_voltage_range (float): expected max voltages value encountered during the measurement.
        clock_frequency (int): rate of the measurement's clock.
        pretrigger_samples (int, optional): number of samples recorded before the trigger.
        trigger_threshold (int, optional): threshold for the detection of the trigger, in volts.
    """
    current_time = datetime.now().strftime("%m%d%Y_%H%M%S")
    task = daq.Task(new_task_name='task_{}'.format(current_time))
    reader = AnalogMultiChannelReader(task.in_stream)

    for chan in ai_channels:
        task.ai_channels.add_ai_voltage_chan(
            chan,
            max_val=10,
            min_val=-10)

    if trigger:
        if pretrigger_samples < 0:
            pretrigger_samples = 0
        if trigger_threshold < 0:
            trigger_threshold = 1
        if pretrigger_samples == 0:
            pretrigger_samples = 2 # Minimum value allowed
        task.triggers.reference_trigger.cfg_anlg_edge_ref_trig(
            trigger_channel,
            pretrigger_samples,
            trigger_slope=daq.constants.Slope.RISING,
            trigger_level=trigger_threshold)
        
    task.timing.cfg_samp_clk_timing(
        clock_frequency,
        sample_mode=daq.constants.AcquisitionType.FINITE,
        samps_per_chan=number_samples)
    task.start()

    return task, reader

def read_daq(number_samples, number_ai_channels, reader):
        """ Reads the current values of the channels.

        Args:
            number_samples (int): number of samples to read.

        Returns:
            np.array: array containing the datas.
        """
        _stop_event = False
        number_measurement = 0
        raw_data = np.empty((number_ai_channels, number_samples))
        while not _stop_event:
            number_measurement += 1
            current_time = datetime.now()
            if number_measurement >= number_samples and number_samples > 0:
                break
            channels_data = np.empty((number_ai_channels, 1))
            channels_data.fill(np.nan)
            try:
                reader.read_many_sample(
                    channels_data,
                    number_of_samples_per_channel=1)
                print(channels_data)
            except:
                pass
            for i in range(number_ai_channels):
                raw_data[i, 0] = channels_data[i]
            raw_data = np.roll(raw_data, -1, axis=1)

        plt.figure()
        for i in range(number_ai_channels):
            plt.plot(raw_data[i])
        plt.show()
        return 1

number_samples = 100
ai_channels = ['Dev1/ai0', 'Dev1/ai1']
clock_frequency = 20
trigger = True
trigger_channel = 'Dev1/ai0'
pretrigger_samples = 0
trigger_threshold = 4

task, reader = set_ai_measurement(number_samples, ai_channels, clock_frequency,
                   trigger, trigger_channel, pretrigger_samples, trigger_threshold)
read_daq(number_samples, len(ai_channels), reader)

分析与解决建议 #

你的实现存在两处关键疏漏，导致触发模式下无法实时读取数据：

1. 有限采样模式的触发逻辑限制 #

当配置参考触发+有限采样时，NI采集卡会先缓存预触发样本，触发事件发生后继续采集剩余样本，直到完成number_samples总数才会将全部数据一次性提交给主机。这是硬件触发在有限采样模式下的默认行为，并非代码bug，但与实时读取需求不匹配。

2. 读取逻辑的低效与错误 #

• read_daq中每次尝试读取1个样本，但触发模式下，采集卡未完成全部采样前不会返回单样本数据，导致循环阻塞直到整个采集完成。
• raw_data的滚动逻辑存在错误，每次只更新第一列后滚动，最终仅能保留最后几次样本，无法正确记录完整数据流。

针对硬件触发+实时读取的需求，可按以下方案调整：

方案1：改用连续采样模式配合触发

将采样模式改为AcquisitionType.CONTINUOUS，触发后采集卡会持续输出数据，支持实时读取：

• 修改cfg_samp_clk_timing中的sample_mode为daq.constants.AcquisitionType.CONTINUOUS，并设置合适的samps_per_chan（建议设为时钟频率的1-2倍，作为缓存区大小）。
• 在read_daq中，循环读取固定数量的样本（比如每次读取10个），直到达到目标总样本数后停止任务。

方案2：配置触发后的实时读取回调

利用nidaqmx的回调机制，当采集卡有新数据可用时自动触发读取函数，无需主动循环等待：

def callback(task_handle, every_n_samples_event_type, number_of_samples, callback_data):
    reader = callback_data
    data = np.empty((len(ai_channels), number_of_samples))
    reader.read_many_sample(data, number_of_samples_per_channel=number_of_samples)
    print(data)
    return 0

# 在set_ai_measurement函数中，task.start()之前添加回调注册
task.register_every_n_samples_acquired_into_buffer_event(10, callback, reader)

方案3：优化有限采样模式下的读取逻辑

若必须使用有限采样，可将总样本数拆分为预触发样本+触发后样本，配置触发后，先读取预触发缓存，触发事件发生后再实时读取后续样本，但这种方式需精确控制缓存区，实现复杂度较高。

不建议自行实现软件峰值检测，NI硬件触发的精度和可靠性远高于软件实现，优先调整硬件触发的配置逻辑即可满足需求。

内容的提问来源于stack exchange，提问作者Florentin Fabre