如何合并蓝牙Socket数据采集与Kivy绘图代码实现实时可视化？

问题根源 #

蓝牙Socket的recv()是阻塞式操作，直接在Kivy主线程执行会彻底卡住UI事件循环，导致程序无响应；同时Matplotlib绘图操作必须在主线程执行，子线程不能直接修改UI相关元素。

解决方案核心思路 #

• 用独立线程处理蓝牙数据接收和CSV写入，与UI线程完全解耦
• 通过线程安全的队列传递采集到的数据
• 利用Kivy的Clock调度器，在主线程定期读取队列数据并更新绘图

完整实现代码 #

import threading
import queue
import csv
import time
from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.clock import Clock
from kivy.graphics.texture import Texture
from kivy.uix.image import Image
import matplotlib.pyplot as plt
from matplotlib.backends.backend_agg import FigureCanvasAgg
import bluetooth

# 线程安全的数据队列
data_queue = queue.Queue()
csv_file_path = "bluetooth_data.csv"

class BluetoothReceiver(threading.Thread):
    def __init__(self, target_addr, port):
        super().__init__(daemon=True)  # 守护线程，随主线程退出
        self.target_addr = target_addr
        self.port = port
        self.running = True
        # 初始化CSV表头
        with open(csv_file_path, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(["Timestamp", "Value1", "Value2"])

    def run(self):
        try:
            sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
            sock.connect((self.target_addr, self.port))
            print("蓝牙连接成功")
            
            while self.running:
                data = sock.recv(1024).decode('utf-8').strip()
                if not data:
                    break
                # 解析数据（格式示例："value1,value2"）
                values = data.split(',')
                if len(values) == 2:
                    timestamp = time.time()
                    # 追加写入CSV
                    with open(csv_file_path, 'a', newline='') as f:
                        writer = csv.writer(f)
                        writer.writerow([timestamp, values[0], values[1]])
                    # 放入队列供绘图使用
                    data_queue.put((timestamp, float(values[0]), float(values[1])))
            sock.close()
        except Exception as e:
            print(f"蓝牙操作错误: {str(e)}")

class PlotWidget(BoxLayout):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 添加显示绘图的Image控件
        self.add_widget(Image(id='plot_canvas'))
        # 初始化Matplotlib图表
        self.figure, self.ax = plt.subplots(figsize=(8, 4))
        self.line1, = self.ax.plot([], [], label='Value1')
        self.line2, = self.ax.plot([], [], label='Value2')
        self.ax.set_xlabel('时间')
        self.ax.set_ylabel('数值')
        self.ax.legend()
        self.ax.grid(True)
        # 存储历史数据
        self.timestamps = []
        self.values1 = []
        self.values2 = []
        # 每100ms更新一次绘图
        Clock.schedule_interval(self.update_plot, 0.1)

    def update_plot(self, dt):
        # 读取队列中所有待处理数据
        while not data_queue.empty():
            ts, v1, v2 = data_queue.get()
            self.timestamps.append(ts)
            self.values1.append(v1)
            self.values2.append(v2)
        
        # 限制历史数据量，避免卡顿
        if len(self.timestamps) > 50:
            self.timestamps = self.timestamps[-50:]
            self.values1 = self.values1[-50:]
            self.values2 = self.values2[-50:]
        
        # 更新图表数据
        self.line1.set_data(self.timestamps, self.values1)
        self.line2.set_data(self.timestamps, self.values2)
        self.ax.relim()
        self.ax.autoscale_view()
        
        # 转换Matplotlib画布为Kivy纹理
        canvas = FigureCanvasAgg(self.figure)
        canvas.draw()
        buf = canvas.buffer_rgba()
        texture = Texture.create(size=(buf.shape[1], buf.shape[0]), colorfmt='rgba')
        texture.blit_buffer(buf, colorfmt='rgba', bufferfmt='ubyte')
        texture.flip_vertical()  # 匹配Kivy的Y轴方向
        
        # 更新UI控件
        self.ids.plot_canvas.texture = texture

class BluetoothPlotApp(App):
    def build(self):
        # 替换为你的蓝牙设备MAC地址和端口
        bluetooth_thread = BluetoothReceiver(target_addr="00:1A:7D:DA:71:13", port=1)
        bluetooth_thread.start()
        return PlotWidget()

if __name__ == "__main__":
    BluetoothPlotApp().run()

关键细节说明 #

• 线程设计：蓝牙线程设为守护线程，主线程退出时自动终止；通过self.running标志控制循环退出
• 数据传递：用queue.Queue实现线程间安全通信，避免竞态条件
• 绘图优化：限制历史数据条数（示例保留50条），防止数据过多导致渲染卡顿
• UI同步：所有绘图更新通过Kivy的Clock调度到主线程执行，严格遵循UI线程安全规则
• CSV写入：初始化时写入表头，每次接收有效数据后追加写入，保证文件格式正确

注意事项 #

1. 替换代码中的蓝牙设备MAC地址和端口为实际值
2. 确保蓝牙设备发送的数据格式与代码解析逻辑匹配（示例为value1,value2）
3. 运行前安装依赖：pip install kivy matplotlib pybluez

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