无原生支持下NI-DAQmx Python实现重触发功能的技术问询

由于USB-6002不支持可重触发启动触发，我们可以通过计数器生成同步触发脉冲的方式，实现每次PFI0上升沿触发时采集指定数量的AI样本，彻底解决短间隔触发下的信号丢失问题。

实现思路 #

1. 计数器任务：将板载计数器配置为脉冲序列输出模式，每个PFI0的上升沿触发计数器输出一个短脉冲，作为AI采集的同步触发信号。
2. AI采集任务：设置AI任务为连续采样模式，使用内部高速时钟保证采样速率，同时绑定计数器的输出作为参考触发，每次触发自动采集2-5个样本。
3. 回调函数：通过注册回调函数自动处理样本读取，消除手动循环的延迟开销，适配1ms级的触发间隔。

完整代码实现 #

import nidaqmx
from nidaqmx.constants import (
    Edge, AcquisitionType, Level, TriggerType
)

# 核心配置参数
SAMPLES_PER_TRIGGER = 5       # 每次触发采集的样本数（可设置为2-5）
AI_SAMPLE_RATE = 50000.0      # AI通道最大支持采样速率
TRIGGER_INPUT = "PFI0"        # 平台触发信号输入接口
COUNTER_DEVICE = "Dev1/ctr0"  # 板载计数器编号
AI_CHANNEL = "Dev1/ai0"       # 模拟输入通道

# 存储所有触发采集到的数据
collected_data = []

def on_samples_acquired(task_handle, event_type, num_samples, callback_data):
    """回调函数：每次完成触发采集后自动读取样本"""
    samples = task_handle.read(number_of_samples_per_channel=SAMPLES_PER_TRIGGER)
    collected_data.append(samples)
    return 0  # 返回0表示继续监听触发事件

with nidaqmx.Task() as ai_task, nidaqmx.Task() as counter_task:
    # --------------------------
    # 配置AI采集任务
    # --------------------------
    ai_task.ai_channels.add_ai_voltage_chan(AI_CHANNEL)
    
    # 配置连续采样模式，预分配足够缓冲区避免数据丢失
    ai_task.timing.cfg_samp_clk_timing(
        rate=AI_SAMPLE_RATE,
        sample_mode=AcquisitionType.CONTINUOUS,
        samps_per_chan=SAMPLES_PER_TRIGGER * 1000
    )
    
    # --------------------------
    # 配置计数器触发任务
    # --------------------------
    # 配置计数器输出短脉冲，低/高电平时间可根据硬件调整
    counter_task.co_channels.add_co_pulse_chan_time(
        COUNTER_DEVICE,
        low_time=0.0001,
        high_time=0.0001,
        idle_state=Level.LOW
    )
    
    # 设置计数器为连续触发模式，由PFI0上升沿启动脉冲输出
    counter_task.timing.cfg_implicit_timing(sample_mode=AcquisitionType.CONTINUOUS)
    counter_task.triggers.start_trigger.cfg_dig_edge_start_trig(
        trigger_source=TRIGGER_INPUT,
        trigger_edge=Edge.RISING
    )
    
    # --------------------------
    # 关联AI任务与计数器触发
    # --------------------------
    # 将AI任务的参考触发绑定到计数器内部输出，每次脉冲上升沿触发采集
    ai_task.triggers.reference_trigger.cfg_dig_edge_ref_trig(
        trigger_source=f"/Dev1/ctr0InternalOutput",
        trigger_edge=Edge.RISING,
        num_samples_per_trigger=SAMPLES_PER_TRIGGER
    )
    
    # 注册回调函数，自动处理每次触发后的样本读取
    ai_task.register_every_n_samples_acquired_into_buffer_event(
        SAMPLES_PER_TRIGGER,
        on_samples_acquired
    )
    
    # --------------------------
    # 启动任务并执行采集
    # --------------------------
    ai_task.start()
    counter_task.start()
    
    # 等待用户输入停止采集（实际场景可替换为平台移动完成的判断逻辑）
    input("Press Enter to stop the acquisition...\n")

# 输出采集结果
print("Collected data from each trigger:")
for idx, sample_group in enumerate(collected_data):
    print(f"Trigger {idx+1}: {sample_group}")

关键细节说明 #

• 计数器脉冲配置：通过短脉冲保证触发信号的精准性，避免与AI采样时钟冲突。
• 连续采样模式：消除手动循环的延迟，确保DAQmx内核自动处理高速触发与采集。
• 回调机制：样本读取操作在DAQmx内核回调中执行，完全适配1ms级的触发间隔要求。

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