哈哈哈哈 发表于 2026-1-24 11:25:29

【开源分享】STC8H 触摸传感器实时监控系统(Python 可视化 + 串口通信)

前因:修改“STC8H4K64Txx-触摸按键校验检测工程”至 STC8H1K08T-33I未果
索性自己问AI写了一个可视化监控程序。
程序功能:

[*]支持 3 路触摸通道的实时波形显示(采样值 + 基线值)
[*]单独展示所有通道的差值波形
[*]串口自动扫描、一键连接 / 断开,状态可视化提示
[*]实时显示通道最新数值,差值实时更新
[*]纵轴自适应数据范围,X 轴固定 200 个数据点

[*]效果图片:



单片机代码:
while(1) {
    touch_scan();
    // 发送:基线0 基线1 基线2 当前0 当前1 当前2
    printf("%u %u %u %u %u %u %x\r\n",
         TK_zero, TK_zero, TK_zero,
         TK_cnt,TK_cnt,TK_cnt,);
    delay_ms(50); // ~20Hz 发送频率
}

上位机代码:
import serial
import serial.tools.list_ports
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import numpy as np
import threading
import time
import tkinter as tk
from tkinter import ttk, messagebox
from collections import deque
import queue

# # 中文字体设置(放在导入 matplotlib 之后)
# plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS', 'DejaVu Sans']
# # 优先尝试常见的中文字体,如果都没有会自动回退
# plt.rcParams['axes.unicode_minus'] = False# 解决负号显示问题

# ================== 配置与全局变量 ==================
# 优先使用Windows原生中文等宽字体,兜底兼容
plt.rcParams['font.sans-serif'] = [
    'Microsoft YaHei Mono',# 核心:Win10+ 中文等宽(优先)
    'SimSun-ExtB',         # 兜底:Win全版本中文等宽
    'Courier New'            # 备用:跨平台等宽(防止前两者缺失)
]
plt.rcParams['axes.unicode_minus'] = False# 解决负号显示问题
plt.rcParams['font.family'] = 'monospace'   # 强制启用等宽模式(关键)
plt.rcParams['font.size'] = 10            # 可选:统一字体大小

DATA_POINTS = 200
BAUDRATE = 9600
ANIMATE_INTERVAL = 50# 动画刷新间隔(ms)
SERIAL_READ_INTERVAL = 0.01# 串口读取间隔(s)
Y_AXIS_MARGIN = 50# Y轴边距(缩小边距,适配更精准)


class SerialApp:
    def __init__(self, root):
      self.root = root
      self.root.title("STC8H 触摸监控系统")

      # 串口对象及状态
      self.ser = serial.Serial()
      self.is_running = False
      self.read_thread = None
      self.data_lock = threading.Lock()# 线程安全锁
      self.data_queue = queue.Queue(maxsize=100)# 数据缓冲队列

      # 数据存储 - 使用deque替代numpy数组,优化数据更新
      self.cnt_data = * DATA_POINTS, maxlen=DATA_POINTS) for _ in range(3)]
      self.zero_data = * DATA_POINTS, maxlen=DATA_POINTS) for _ in range(3)]
      self.current_delta =

      # 预计算X轴数据
      self.x_data = np.arange(DATA_POINTS)

      # 绘图对象缓存
      self.line_objects = []
      self.baseline_objects = []
      self.delta_line_objects = []

      # 动画控制标志
      self.animation_running = False
      self.ani = None# 延迟初始化动画

      self.setup_ui()
      self.setup_plot()

    def setup_ui(self):
      """创建控制面板"""
      control_frame = ttk.Frame(self.root)
      control_frame.pack(side=tk.TOP, fill=tk.X, padx=10, pady=5)

      # 1. 串口选择
      ttk.Label(control_frame, text="串口选择:").grid(row=0, column=0, padx=5)
      self.port_combo = ttk.Combobox(control_frame, width=15)
      self.port_combo.grid(row=0, column=1, padx=5)
      self.refresh_ports()# 初始化扫描

      # 2. 刷新串口列表按钮
      ttk.Button(control_frame, text="刷新串口", command=self.refresh_ports).grid(row=0, column=2, padx=5)

      # 3. 开始/断开按钮
      self.btn_start = ttk.Button(control_frame, text="开始连接", command=self.toggle_serial)
      self.btn_start.grid(row=0, column=3, padx=10)

      self.status_label = ttk.Label(control_frame, text="状态: 未连接", foreground="red")
      self.status_label.grid(row=0, column=4, padx=10)

    def setup_plot(self):
      """创建嵌入式 Matplotlib 图表 - 优化版本"""
      # 创建图表时减少不必要的参数,优化布局
      self.fig, (self.ax1, self.ax2, self.ax3, self.ax4) = plt.subplots(
            4, 1, figsize=(10, 8), gridspec_kw={'height_ratios': },
            tight_layout=True# 自动优化布局,替代tight_layout调用
      )
      self.fig.suptitle('Touch Channel Monitor', y=0.98)

      # 将图表嵌入 Tkinter 窗口
      self.canvas = FigureCanvasTkAgg(self.fig, master=self.root)
      self.canvas.get_tk_widget().pack(side=tk.TOP, fill=tk.BOTH, expand=1)

      # 初始化绘图对象(只创建一次,后续更新数据)
      colors = ['blue', 'green', 'red']
      axes =

      # 初始化通道波形图
      for i, (ax, color) in enumerate(zip(axes, colors)):
            # 主数据线条
            line, = ax.plot(self.x_data, list(self.cnt_data),
                            label=f'CH{i} Count', color=color, linewidth=1)
            self.line_objects.append(line)

            # 基线线条
            baseline, = ax.plot(self.x_data, list(self.zero_data),
                              color='black', linestyle='--', alpha=0.6,
                              label='Baseline', linewidth=1)
            self.baseline_objects.append(baseline)

            # 初始化坐标轴设置(只做一次)
            ax.set_title(f'Channel {i} (Delta: 0)')
            ax.set_ylim(-Y_AXIS_MARGIN, Y_AXIS_MARGIN)
            ax.grid(True, alpha=0.3)
            ax.legend(loc='upper right', fontsize='small')
            # 关闭x轴自动缩放,避免布局抖动
            ax.set_xlim(0, DATA_POINTS - 1)

      # 初始化差值图
      for i, color in enumerate(colors):
            delta_line, = self.ax4.plot(self.x_data, * DATA_POINTS,
                                        label=f'CH{i} Δ (0)', color=color, linewidth=1)
            self.delta_line_objects.append(delta_line)

      self.ax4.set_title('Real-time Channel Differences (Baseline - Count)')
      self.ax4.set_ylim(-50, 50)# 初始值,后续动态更新
      self.ax4.legend(loc='upper right', ncol=3)
      self.ax4.grid(True, alpha=0.3)
      self.ax4.set_xlim(0, DATA_POINTS - 1)# 固定x轴范围

    def start_animation(self):
      """启动动画(仅在连接时)"""
      if not self.animation_running:
            self.ani = FuncAnimation(
                self.fig,
                self.animate,
                interval=ANIMATE_INTERVAL,
                cache_frame_data=False,
                blit=True,# 开启blit优化
                repeat=True
            )
            self.animation_running = True

    def stop_animation(self):
      """停止动画(断开连接时)"""
      if self.animation_running and self.ani:
            self.ani.event_source.stop()
            self.animation_running = False

    def refresh_ports(self):
      """扫描当前可用串口 - 优化:减少不必要的刷新"""
      ports = serial.tools.list_ports.comports()
      port_list =
      self.port_combo['values'] = port_list
      if port_list and not self.port_combo.get():
            self.port_combo.current(0)

    def toggle_serial(self):
      """切换开始/断开状态"""
      if not self.is_running:
            self.start_serial()
      else:
            self.stop_serial()

    def start_serial(self):
      """打开串口并启动读取线程"""
      port = self.port_combo.get()
      if not port:
            messagebox.showwarning("警告", "请先选择一个串口!")
            return

      try:
            self.ser.port = port
            self.ser.baudrate = BAUDRATE
            self.ser.timeout = 0.5# 优化超时时间
            self.ser.open()

            self.is_running = True
            self.btn_start.config(text="断开连接")
            self.status_label.config(text=f"状态: 已连接 {port}", foreground="green")
            self.port_combo.config(state="disabled")

            # 启动动画
            self.start_animation()

            # 启动线程
            self.read_thread = threading.Thread(target=self.read_serial_data, daemon=True)
            self.read_thread.start()
      except Exception as e:
            messagebox.showerror("错误", f"无法打开串口: {e}")

    def stop_serial(self):
      """停止读取并关闭串口 - 优化:更安全的线程退出"""
      self.is_running = False
      # 停止动画,减少资源占用
      self.stop_animation()

      # 清空数据队列
      while not self.data_queue.empty():
            try:
                self.data_queue.get_nowait()
            except queue.Empty:
                break

      # 等待线程退出
      if self.read_thread and self.read_thread.is_alive():
            self.read_thread.join(timeout=0.5)

      if self.ser.is_open:
            self.ser.close()

      self.btn_start.config(text="开始连接")
      self.status_label.config(text="状态: 未连接", foreground="red")
      self.port_combo.config(state="normal")

    def read_serial_data(self):
      """后台读取线程 - 大幅优化"""
      buffer = ""# 串口数据缓冲
      while self.is_running:
            if self.ser.is_open:
                try:
                  # 读取可用数据(而非整行),减少阻塞
                  if self.ser.in_waiting > 0:
                        data = self.ser.read(self.ser.in_waiting).decode('utf-8', errors='ignore')
                        buffer += data

                        # 按行分割处理
                        while '\n' in buffer:
                            line, buffer = buffer.split('\n', 1)
                            line = line.strip()
                            if line:
                              parts = line.split()
                              if len(parts) >= 6:
                                    try:
                                        zeros = ), int(parts), int(parts)]
                                        cnts = ), int(parts), int(parts)]

                                        # 使用队列传递数据,避免直接操作共享变量
                                        if not self.data_queue.full():
                                          self.data_queue.put((zeros, cnts))
                                    except ValueError:
                                        # 数据格式错误时忽略,不中断线程
                                        continue

                  # 优化休眠时间,减少CPU占用
                  time.sleep(SERIAL_READ_INTERVAL)

                except Exception as e:
                  print(f"读取异常: {e}")
                  time.sleep(0.1)
            else:
                time.sleep(0.1)

    def update_data(self):
      """更新数据缓存 - 线程安全"""
      if self.data_queue.empty():
            return False

      try:
            # 批量处理队列中的数据(最多处理10条,避免阻塞)
            for _ in range(min(10, self.data_queue.qsize())):
                zeros, cnts = self.data_queue.get_nowait()

                with self.data_lock:
                  # 更新数据 - 使用deque的append自动维护长度,无需roll
                  for i in range(3):
                        self.zero_data.append(zeros)
                        self.cnt_data.append(cnts)
                        self.current_delta = zeros - cnts

            return True
      except queue.Empty:
            return False

    def animate(self, i):
      """图表刷新逻辑 - 核心优化(解决卡顿+纵轴自适应)"""
      # 只在有数据更新时才处理绘图逻辑
      data_updated = self.update_data()
      if not data_updated:
            return []

      with self.data_lock:
            # 转换为numpy数组用于绘图
            cnt_arrays =
            zero_arrays =

            # 更新通道波形 + 纵轴自适应
            axes =
            for i in range(3):
                # 更新线条数据(不重建,只更新)
                self.line_objects.set_ydata(cnt_arrays)
                self.baseline_objects.set_ydata(zero_arrays)

                # 【关键修复】强制更新Y轴范围,确保自适应
                ax = axes
                all_vals = np.concatenate(, zero_arrays])
                val_min, val_max = all_vals.min(), all_vals.max()

                # 计算新的Y轴范围(加边距)
                new_ylim = (val_min - Y_AXIS_MARGIN, val_max + Y_AXIS_MARGIN)
                ax.set_ylim(new_ylim)

                # 更新标题(只更新文本,不重建)- 显示当前通道最新值(6位补零)
                latest_value = self.cnt_data[-1]# 获取deque最后一个元素(最新值)
                ax.set_title(f'Channel {i} ({latest_value:06d})')

            # 更新差值图 + 纵轴自适应
            delta_arrays = []
            for i in range(3):
                delta = zero_arrays - cnt_arrays
                delta_arrays.append(delta)
                self.delta_line_objects.set_ydata(delta)
                # self.delta_line_objects.set_label(f'CH{i}Δ:({self.current_delta:06d})')
            self.ax4.set_title(f'CH0Δ:({self.current_delta:06d}) CH1Δ:({self.current_delta:06d}) CH2Δ:({self.current_delta:06d})')
            # 差值图纵轴自适应
            all_delta = np.concatenate(delta_arrays)
            delta_min, delta_max = all_delta.min(), all_delta.max()
            self.ax4.set_ylim(delta_min - 20, delta_max + 20)
            self.ax4.legend(loc='upper right', ncol=3)

      # 返回需要更新的对象(blit优化)
      return self.line_objects + self.baseline_objects + self.delta_line_objects


if __name__ == "__main__":
    root = tk.Tk()
    # 优化Tkinter渲染优先级
    root.attributes('-topmost', False)
    root.update_idletasks()

    # 减少Tkinter事件循环负载
    root.after(100, lambda: root.update())

    app = SerialApp(root)
    root.mainloop()

页: [1]
查看完整版本: 【开源分享】STC8H 触摸传感器实时监控系统(Python 可视化 + 串口通信)