如何将NI DAQ实时测温数据写入Pandas DataFrame并导出CSV

问题分析 #

你遇到的两个报错其实指向两个核心问题：

1. 第一个报错是因为你在循环中频繁将列表转为DataFrame后，尝试直接append字典，但pandas的DataFrame.append()方法需要将字典转为Series（指定name）或设置ignore_index=True；
2. 第二个报错则是因为pandas 2.0及以上版本已经移除了DataFrame.append()方法，所以这个方法不再支持任何关键字参数。

另外，你的代码还有个效率问题：每次循环都重新创建NI DAQ Task，并且反复将列表转为DataFrame，这会拖慢程序运行速度，也容易引发数据不一致。

解决方案 #

推荐两种优化思路，优先选第一种（效率更高）：

思路1：先收集所有数据到列表，循环结束后再生成DataFrame

这种方式避免了循环中频繁操作DataFrame，性能更好，也不会触发append相关的报错：

import nidaqmx
import matplotlib.pyplot as plt
import pandas as pd

plt.ion()
i = 0
data_list = []  # 初始化空列表存储所有数据字典

# 将DAQ Task放在循环外，避免重复创建
with nidaqmx.Task() as task:
    # 添加热电偶通道
    FreezerT = task.ai_channels.add_ai_thrmcpl_chan("cDAQ1Mod1/ai0")
    RightSideT = task.ai_channels.add_ai_thrmcpl_chan("cDAQ1Mod1/ai3")
    task.timing.samp_clk_rate = 1
    
    # 建议添加终止条件，比如运行100次或捕获键盘中断
    try:
        while i >= 0:
            data = task.read()
            # 实时绘图
            plt.scatter(i, data[0], c='r')
            plt.scatter(i, data[1], c='b')
            plt.xlabel('Time (s)')
            plt.ylabel('Temperature (C)')
            plt.pause(0.05)
            
            # 将当前数据存入列表
            data_list.append({
                'Time (s)': i,
                'Freezer Temperature': data[0],
                'Right Side': data[1]
            })
            
            i += 1
            # 每10次循环打印一次最新数据，避免刷屏
            if i % 10 == 0:
                temp_df = pd.DataFrame(data_list)
                print(temp_df.tail())
    except KeyboardInterrupt:
        print("测量已终止")

# 循环结束后生成完整DataFrame并导出CSV
df = pd.DataFrame(data_list, columns=['Time (s)', 'Freezer Temperature', 'Right Side'])
print("\n完整测量数据：")
print(df)
df.to_csv('temperature_data.csv', index=False)
print("数据已导出到temperature_data.csv")

思路2：循环中逐行更新DataFrame（适合需要实时查看完整DataFrame的场景）

如果需要在循环中实时维护DataFrame，用pd.concat()替代已弃用的append()：

import nidaqmx
import matplotlib.pyplot as plt
import pandas as pd

plt.ion()
i = 0
# 初始化空DataFrame
df = pd.DataFrame(columns=['Time (s)', 'Freezer Temperature', 'Right Side'])

with nidaqmx.Task() as task:
    FreezerT = task.ai_channels.add_ai_thrmcpl_chan("cDAQ1Mod1/ai0")
    RightSideT = task.ai_channels.add_ai_thrmcpl_chan("cDAQ1Mod1/ai3")
    task.timing.samp_clk_rate = 1
    
    try:
        while i >= 0:
            data = task.read()
            plt.scatter(i, data[0], c='r')
            plt.scatter(i, data[1], c='b')
            plt.xlabel('Time (s)')
            plt.ylabel('Temperature (C)')
            plt.pause(0.05)
            
            # 将当前数据转为单行DataFrame，然后用concat合并
            new_row = pd.DataFrame({
                'Time (s)': [i],
                'Freezer Temperature': [data[0]],
                'Right Side': [data[1]]
            })
            df = pd.concat([df, new_row], ignore_index=True)
            
            print("\n最新测量数据：")
            print(df.tail())  # 打印最新几行，避免刷屏
            i += 1
    except KeyboardInterrupt:
        print("测量已终止")

# 导出CSV
df.to_csv('temperature_data.csv', index=False)
print("数据已导出到temperature_data.csv")

额外建议 #

• 给循环加终止条件（比如捕获KeyboardInterrupt按Ctrl+C终止），否则程序会无限运行；
• 实时绘图时可以通过plt.xlim()限制x轴范围，避免图表因数据过多变得拥挤；
• NI DAQ Task放在循环外可以减少资源消耗，提升测量稳定性。

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