Raspberry Pi 3B搭配高精度AD/DA板 10KPS以上CSV写入方案咨询

你当前遇到的写入速度远低于采样速度的问题，和ADS1256硬件性能无关，完全是代码逻辑的开销导致：

• 逐采样点触发磁盘IO：树莓派SD卡的随机写入延迟极高，单次写入操作的开销可达10-30ms，每采一个点写一次的话，理论上限也就30-100点/秒，和你实测的27点/秒完全吻合
• 手动字符串拼接效率极低：Python字符串是不可变对象，连续用+拼接9个字符串会生成8个临时字符串对象，高频调用下CPU开销非常高
• 采样和写入同线程同步执行：磁盘IO阻塞时，采样流程完全卡住，根本跑不到标称的20kSPS采样率

按优先级做以下调整，可稳定跑到10kSPS以上的CSV写入速率：

1. 彻底取消逐行IO，改用批量缓冲写入 #

这是提升性能最核心的手段。SD卡的连续写入性能是随机写入的50-100倍，不要来一个点写一次，攒够100-5000个点（对应10kSPS下10ms-500ms的数据量，内存占用不足1MB）再一次性写入，文件打开时设置1MB以上的写入缓冲，不要手动频繁调用flush()。

2. 替换手动字符串拼接，用原生C实现的CSV模块 #

Python标准库的csv模块底层是C实现的，批量写入接口writerows()的转码、拼接效率比你手动逐行用+拼接高100倍以上。采样过程中不要提前转字符串，直接把原始数值存到内存，写入时统一交给csv模块处理。

3. 采样、写入线程分离 #

用队列做生产者-消费者解耦：采样线程只负责读ADC数据、打时间戳，把数据丢进内存队列，全程不碰任何IO操作；单独开一个写入线程从队列批量取数据，攒批后写入CSV。这样磁盘IO的延迟完全不会阻塞采样流程。

import time
import csv
from queue import Queue
import threading
# 此处导入你已有的ADS1256驱动模块

# 可调参数
BATCH_SIZE = 1000  # 每攒1000个点写一次，10kSPS下对应100ms延迟，可根据需求在100-5000区间调整
FILE_BUFFER = 1024 * 1024  # 1MB文件写入缓冲
data_queue = Queue(maxsize=BATCH_SIZE*2)
stop_flag = False

def csv_writer_worker(save_path):
    write_buffer = []
    # 开文件时配置大缓冲，newline=''是csv模块官方要求的参数，避免多余空行
    with open(save_path, 'w', newline='', buffering=FILE_BUFFER) as f:
        writer = csv.writer(f)
        # 写入CSV表头
        writer.writerow(['ch0', 'ch1', 'ch2', 'ch3', 'ch4', 'ch5', 'ch6', 'ch7', 'timestamp'])
        while not stop_flag or not data_queue.empty():
            try:
                # 从队列取采样数据，1秒超时防止线程卡死
                sample = data_queue.get(timeout=1)
                write_buffer.append(sample)
                # 攒够批次一次性写入
                if len(write_buffer) >= BATCH_SIZE:
                    writer.writerows(write_buffer)
                    write_buffer.clear()
            except Exception:
                # 队列空闲时把残留的缓冲数据写入
                if write_buffer:
                    writer.writerows(write_buffer)
                    write_buffer.clear()
        # 程序退出前写完所有剩余数据
        if write_buffer:
            writer.writerows(write_buffer)

if __name__ == "__main__":
    # 启动写入线程
    writer_thread = threading.Thread(target=csv_writer_worker, args=('./ads1256_sample.csv',))
    writer_thread.start()

    # 原有采样逻辑
    try:
        # 此处替换成你原有的ADS1256初始化代码
        # ads = ADS1256()
        while True:
            # 读8通道ADC值
            adc_values = ads1256.ReadAllChannels()
            # 直接把原始数值+时间戳丢队列，不要在这里转字符串
            data_queue.put( (*adc_values, time.time()) )
    except KeyboardInterrupt:
        stop_flag = True
        writer_thread.join()

• 如果担心意外掉电丢太多数据，可以把BATCH_SIZE调到200，对应10kSPS下20ms写入一次，性能也比逐行写高200倍，完全满足10kSPS需求
• 不要在采样循环里加任何打印、sleep逻辑，会直接拉低采样率
• 如果需要进一步降低CPU开销，可以不用每个采样点都调用time.time()，每采10个点打一次时间戳，中间的点时间按采样间隔偏移计算即可，精度损失可以忽略

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