基于nidaqmx的NI PXI采集卡10小时长时监测数据落盘实现问题

你原来的代码核心问题是一次性把全时段的采样数拉到内存读取，完全没用上nidaqmx驱动层自带的环形缓冲能力，按你200kHz采样的参数算，哪怕单通道10小时的单精度数据都有2.88GB，多通道场景下直接就把Python进程内存吃满了。
连续采集模式下不需要把所有数据堆在内存里，只要改成分块轮询读、读一块写一块、及时回收内存的逻辑，哪怕连续跑几十小时，内存占用也能稳定在几十MB级别，根本不会出现内存溢出的问题。

核心配置逻辑 #

• 先给驱动层的硬件环形缓冲留够足够空间：按10秒的总采样量设置缓冲大小，公式为 硬件缓冲采样数 = 总通道数 * 200000 * 10，这个大小哪怕程序轮询偶尔卡顿，也不会出现缓冲溢出丢数的问题。
• 计算固定块的读取大小：按照你想要的块存储容量算对应的单通道采样数，比如想每20MB写一次盘，公式为 单块读取采样数 = (目标块大小MB * 1024 * 1024) / (总通道数 * 4)，这里单采样按float32的4字节算，比默认的float64省一半内存和磁盘空间。
• 循环分块读取：每次固定读指定块大小的采样数，读出来的数组直接写磁盘，写完立刻删除内存引用让Python回收，全程内存里不会堆超过单块大小的数据。

可直接复用的改造代码 #

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

# 基础参数，按实际场景调整
total_sample_time = 10 * 3600  # 总采集时长10小时，单位秒
s_freq = 200000
target_chunk_mb = 20  # 每20MB触发一次磁盘写入
num_channels = 6  # 两块卡各取3通道的总通道数，按实际通道数修改

# 换算单块需要读的单通道采样数
chunks_samples = int((target_chunk_mb * 1024 * 1024) / (num_channels * 4))
# 硬件环形缓冲大小：留10秒的采样余量
hw_buffer_size = num_channels * s_freq * 10
# 总循环读取的块数
total_chunks = int(np.ceil(total_sample_time * s_freq / chunks_samples))

# 打开二进制raw文件存储，写入速度远快于csv、xlsx格式
with open("long_term_vibration_data.raw", "wb") as f:
    # 先写文件头，存基础参数方便后续读取解析
    file_header = np.array([s_freq, num_channels, chunks_samples, total_chunks], dtype=np.float64)
    file_header.tofile(f)

    print("开始长时采集！")
    with nidaqmx.Task() as task:
        # 分别添加两块采集卡的加速度通道，按实际设备路径修改
        task.ai_channels.add_ai_accel_chan("PXI1Slot2/ai1:3",
                                           sensitivity=10000.0,
                                           max_val=1,
                                           min_val=-1)
        task.ai_channels.add_ai_accel_chan("PXI1Slot3/ai1:3",
                                           sensitivity=10000.0,
                                           max_val=1,
                                           min_val=-1)
        # 配置连续采样模式，指定硬件缓冲大小
        task.timing.cfg_samp_clk_timing(s_freq,
                                        sample_mode=AcquisitionType.CONTINUOUS,
                                        buffer_size=hw_buffer_size)
        # 启动采集任务
        task.start()

        for idx in range(total_chunks):
            # 分块读取数据，超时设置比单块采集时长久即可
            data_chunk = task.read(number_of_samples_per_channel=chunks_samples, timeout=10)
            # 转成float32格式存储，省一半磁盘和内存
            data_chunk = np.array(data_chunk, dtype=np.float32)
            # 直接写盘，不把数据堆在内存
            data_chunk.tofile(f)
            # 删除当前块的内存引用，触发垃圾回收
            del data_chunk
            # 每10块打印一次采集进度和硬件缓冲占用
            if idx % 10 == 0:
                elapsed_hour = (idx * chunks_samples / s_freq) / 3600
                buffer_usage = task.in_stream.available_samples_in_buffer / hw_buffer_size
                print(f"已采集 {elapsed_hour:.2f} 小时，硬件缓冲占用 {buffer_usage*100:.1f}%")

print("10小时采集任务完成！")

实操避坑要点 #

• 采集循环里别做实时画图、复杂信号计算这类耗时间的操作，这类操作一次就要占几百毫秒，很容易把硬件缓冲挤爆，所有可视化、分析逻辑等采集完读盘再做。
• 存储数据的硬盘别用系统盘，关掉硬盘自动休眠功能，避免采集过程中硬盘进入休眠状态，导致写入跟不上采集速度丢数。
• 正式跑长时采集前，先跑5分钟的短测试，观察硬件缓冲的占用率：如果占用率经常超过30%，就把单块读取的采样数调小，或者把硬件缓冲的预留空间从10秒提至30秒，避免缓冲溢出。
• Windows系统下可以适当把Python进程的优先级调高，避免后台更新、杀毒软件抢占CPU导致程序响应变慢。

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