Python读取力传感器串口数据时绘图/sleep/存csv出现延时问题求解

核心问题定位 #

• 每次循环重复读取两次传感器数据：gsv_data.append(dev.read_value())和print(dev.read_value())各读一次，导致串口缓冲区数据堆积，读取到的均为滞后的历史数据
• 每次循环全量覆写CSV文件：随着gsv_data长度增长，IO耗时会线性升高，直接阻塞串口读取逻辑
• Matplotlib绘图逻辑效率极低：每次循环重绘全量数据、清空画布、重新生成绘图对象，随着数据量增加耗时会快速飙升
• 所有逻辑耦合在同一线程：高优先级的串口采集被低优先级的绘图、IO操作阻塞，缓冲区数据积压越来越严重，最终出现数十秒滞后

优化方案 #

1. 逻辑解耦 #

拆分串口采集线程和业务处理线程，用队列传递采集到的原始数据，保证采集逻辑不会被其他操作阻塞

2. 绘图优化 #

复用绘图对象，仅更新曲线数据而非重绘整张图，取消每次清空画布的操作

3. 存储优化 #

CSV改为低频写入，或仅在程序退出时一次性写入全量数据，避免高频IO阻塞

4. 移除冗余操作 #

删除重复的传感器读取逻辑、不必要的短sleep语句

修正后代码 #

import serial
import struct
import platform
import threading
import queue
import time
import csv
import matplotlib.pyplot as plt

class MeasurementConverter:
    def convertValue(self, bytes):
        pass

class ForceMeasurementConverterKG(MeasurementConverter):
    def __init__(self, F_n, S_n, u_e):
        self.F_n = F_n
        self.S_n = S_n
        self.u_e = u_e

    def convertValue(self, value):
        A = struct.unpack('>H', value)[0]
        return self.F_n / self.S_n * ((A - 0x8000) / 0x8000) * self.u_e * 2

class GSV3USB:
    def __init__(self, com_port, baudrate=38400):
        com_path = f'/dev/ttyUSB{com_port}' if platform.system() == 'Linux' else f'COM{com_port}'
        self.sensor = serial.Serial(com_path, baudrate, timeout=1)
        self.converter = ForceMeasurementConverterKG(10, 0.499552, 2)

    def read_value(self):
        self.sensor.read_until(b'\xA5')
        read_val = self.sensor.read(2)
        return self.converter.convertValue(read_val)

# 数据队列，用于采集线程和主线程传数据
data_queue = queue.Queue(maxsize=100)
# 退出标记
exit_flag = False

def collect_data():
    """采集线程，单独运行不被其他逻辑阻塞"""
    dev = GSV3USB(6)
    while not exit_flag:
        val = dev.read_value()
        t = time.time()
        try:
            data_queue.put_nowait((t, val))
        except queue.Full:
            # 队列满了就丢最老的数据，保证取到的都是最新值
            data_queue.get_nowait()
            data_queue.put_nowait((t, val))

def data_gsv():
    global exit_flag
    # 启动采集线程
    collect_thread = threading.Thread(target=collect_data, daemon=True)
    collect_thread.start()

    gsv_data = []
    temps = []
    t_0 = time.time()

    # 初始化绘图仅执行一次
    plt.ion()
    fig, ax = plt.subplots()
    line, = ax.plot([], [], 'b-')
    ax.set_ylim(-2, 10)

    try:
        while True:
            # 读取队列里的所有新数据
            while not data_queue.empty():
                t_raw, val = data_queue.get()
                t1 = t_raw - t_0
                gsv_data.append(val)
                temps.append(t1)
                print(val)  # 打印当前读到的最新值

            # 只更新曲线数据，不重绘整张图
            if len(temps) > 0:
                line.set_data(temps, gsv_data)
                # 只保留最近6秒的数据，避免数据量无限增长
                cutoff = temps[-1] - 6
                idx = 0
                while idx < len(temps) and temps[idx] < cutoff:
                    idx += 1
                if idx > 0:
                    temps = temps[idx:]
                    gsv_data = gsv_data[idx:]
                ax.set_xlim(temps[0], temps[-1] + 0.5)
                fig.canvas.draw_idle()
                fig.canvas.flush_events()

            # 每1秒存一次数据，大幅降低IO频率
            if int(time.time() - t_0) % 1 == 0:
                with open('student_gsv.csv', 'w', newline='') as f:
                    write = csv.writer(f)
                    write.writerow(["时间", "力值"])
                    write.writerows(zip(temps, gsv_data))

            time.sleep(0.016)  # 固定60帧刷新率，足够实验使用
    except KeyboardInterrupt:
        exit_flag = True
        collect_thread.join()
        # 退出前存最终全量数据
        with open('student_gsv.csv', 'w', newline='') as f:
            write = csv.writer(f)
            write.writerow(["时间", "力值"])
            write.writerows(zip(temps, gsv_data))
        print("Exiting")
        return

if __name__ == "__main__":
    data_gsv()

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