Python写入文件时丢失ADC采样样本的解决方法咨询

你的代码存在几个核心问题导致样本丢失：

• 全局列表value_list未做线程同步，读写操作同时进行时会触发竞态条件，导致数据被覆盖或丢失
• 写入文件时直接重新赋值value_list=[]，而非清空原列表，会导致读取线程的列表引用失效
• 计时逻辑用轮询方式浪费资源，且无法保证精确的60秒周期
• 未控制采样频率，无法稳定维持100Hz的采样率

具体修复方案 #

1. 引入线程锁保护共享数据 #

用threading.Lock对value_list的读写操作加锁，避免多线程同时操作导致的数据异常。

2. 分离数据缓存与写入操作 #

每次写入文件前，先复制当前缓存的样本数据，再清空原缓存，这样读取线程可以继续写入新数据，不会被写入操作阻塞。

3. 优化计时逻辑 #

用time.sleep()替代轮询，更高效且能保证精确的时间间隔。

4. 精确控制采样率 #

通过计算每次采样的间隔时间，确保稳定的100Hz采样频率，避免因ADC读取速度波动导致的采样不均。

修改后的完整代码 #

import threading
import time
import csv
import ADS1256
import RPi.GPIO as GPIO

# 初始化ADC
ADC = ADS1256.ADS1256()
ADC.ADS1256_init()

# 共享数据与锁
value_list = []
data_lock = threading.Lock()
SAMPLE_RATE = 100  # 100Hz采样率
SAMPLE_INTERVAL = 1.0 / SAMPLE_RATE  # 采样间隔0.01秒

def adc_reading():
    global value_list
    while True:
        start_time = time.perf_counter()
        # 读取ADC数据
        adc_value = ADC.ADS1256_GetAll()
        timestamp = time.time()
        x = adc_value[1]
        y = adc_value[2]
        z = adc_value[3]
        
        # 加锁写入共享列表
        with data_lock:
            value_list.append([timestamp, x, y, z])
        
        # 控制采样间隔，保证100Hz
        elapsed = time.perf_counter() - start_time
        if elapsed < SAMPLE_INTERVAL:
            time.sleep(SAMPLE_INTERVAL - elapsed)

def cronometro():
    while True:
        # 等待60秒
        time.sleep(60)
        # 读取并清空缓存数据
        with data_lock:
            data_to_write = value_list.copy()
            value_list.clear()
        # 写入文件
        write_to_file(data_to_write)

def write_to_file(lista):
    if not lista:
        return  # 空数据不写入
    nombre_archivo = f"{int(time.time())}.finish"
    with open(nombre_archivo, 'w', newline='') as f:
        write = csv.writer(f)
        write.writerows(lista)

# 启动线程
escritor = threading.Thread(target=adc_reading, daemon=True)
temporizador = threading.Thread(target=cronometro, daemon=True)
escritor.start()
temporizador.start()

# 主线程保持运行
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    GPIO.cleanup()
    print("程序已终止")

关键修改说明 #

• 线程锁：所有对value_list的操作都放在with data_lock:块中，确保同一时间只有一个线程能操作列表
• 数据复制：写入文件前复制列表，原列表清空后读取线程可立即继续写入，避免阻塞采样
• 采样率控制：用time.perf_counter()计算读取耗时，不足采样间隔时休眠补全，稳定维持100Hz
• 守护线程：设置daemon=True，确保主线程终止时子线程自动退出
• 空数据判断：避免写入空文件，减少无效操作

内容的提问来源于stack exchange，提问作者Nahum Mendez Chazarra