【开源分享】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[0], TK_zero[1], TK_zero[2],
           TK_cnt[0],  TK_cnt[1],  TK_cnt[2],);
    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 = [deque([0] * DATA_POINTS, maxlen=DATA_POINTS) for _ in range(3)]
        self.zero_data = [deque([0] * DATA_POINTS, maxlen=DATA_POINTS) for _ in range(3)]
        self.current_delta = [0, 0, 0]

        # 预计算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': [1, 1, 1, 1.2]},
            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 = [self.ax1, self.ax2, self.ax3]

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

            # 基线线条
            baseline, = ax.plot(self.x_data, list(self.zero_data[i]),
                                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, [0] * 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 = [p.device for p in ports]
        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[0]), int(parts[1]), int(parts[2])]
                                        cnts = [int(parts[3]), int(parts[4]), int(parts[5])]

                                        # 使用队列传递数据，避免直接操作共享变量
                                        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[i].append(zeros[i])
                        self.cnt_data[i].append(cnts[i])
                        self.current_delta[i] = zeros[i] - cnts[i]

            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 = [np.array(list(d)) for d in self.cnt_data]
            zero_arrays = [np.array(list(d)) for d in self.zero_data]

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

                # 【关键修复】强制更新Y轴范围，确保自适应
                ax = axes[i]
                all_vals = np.concatenate([cnt_arrays[i], zero_arrays[i]])
                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[i][-1]  # 获取deque最后一个元素（最新值）
                ax.set_title(f'Channel {i} ({latest_value:06d})')

            # 更新差值图 + 纵轴自适应
            delta_arrays = []
            for i in range(3):
                delta = zero_arrays[i] - cnt_arrays[i]
                delta_arrays.append(delta)
                self.delta_line_objects[i].set_ydata(delta)
                # self.delta_line_objects[i].set_label(f'CH{i}Δ:({self.current_delta[i]:06d})')
            self.ax4.set_title(f'CH0Δ:({self.current_delta[0]:06d}) CH1Δ:({self.current_delta[1]:06d}) CH2Δ:({self.current_delta[2]: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()
复制代码