利用Python调试串口的示例代码

目录

• 概述：为什么需要专业的串口调试工具
• 项目架构设计
• 1.1 技术栈选型
• 1.2 关键类说明
• 1.3 线程模型
• 环境配置指南
• 2.1 基础环境
• 2.2 硬件准备
• 2.3 兼容性说明
• 核心功能实现
• 3.1 串口通信核心
• 3.2 多线程数据处理
• 3.3 自动发送机制
• UI界面详解
• 4.1 三栏式布局
• 4.2 主题切换实现
• 4.3 控件亮点功能
• 运行效果展示
• 5.1 主题切换演示
• 5.2 数据收发演示
• 5.3 统计功能展示
• 源码下载
• 总结与扩展
• 7.1 项目总结
• 7.2 扩展方向

在嵌入式开发、物联网设备调试过程中，串口通信是最基础的调试手段。但系统自带的串口工具功能简陋，商业软件又价格昂贵。本文将带你用Python+ttkbootstrap打造一款高颜值、多功能的串口调试助手，具备以下亮点功能：

核心功能亮点：

• 现代化UI界面 – 基于ttkbootstrap的多主题切换
• 实时数据统计 – 发送/接收字节计数
• 自动发送功能 – 可配置间隔时间
• 发送历史记录 – 支持上下箭头导航
• 数据持久化 – 接收内容保存为文件
• 自动端口检测 – 实时监控串口热插拔

import serial  # 串口通信核心库
import serial.tools.list_ports  # 串口设备枚举
import threading  # 多线程处理
import queue  # 线程安全队列
import ttkbootstrap as ttk  # 现代化UI框架
from tkinter import filedialog  # 文件对话框

SerialTool：主控制类，采用MVC设计模式

数据层：serial_port管理物理连接

视图层：create_widgets()构建界面

控制层：事件处理方法群

定时轮询异步推送定时触发主线程接收队列接收子线程自动发送任务

# 必需库安装
pip install pyserial ttkbootstrap

任意USB转串口设备（如CH340、CP2102等）

开发板或目标设备

支持Windows/macOS/Linux

测试Python版本：3.8+

def open_serial(self):
    # 参数映射转换
    parity_map = {
        'None': serial.PARITY_NONE,
        'Even': serial.PARITY_EVEN,
        # ...其他校验位映射
    }
    
    self.serial_port = serial.Serial(
        port=self.port_cb.get(),
        baudrate=int(self.baudrate_cb.get()),
        parity=parity_map[self.parity_cb.get()],
        timeout=0.1  # 非阻塞读取
    )

def receive_worker(self):
    while not self.receive_thread_event.is_set():
        try:
            # 非阻塞读取
            if self.serial_port.in_waiting > 0:
                data = self.serial_port.read(self.serial_port.in_waiting)
                self.receive_queue.put(data)
        except serial.SerialException:
            break

def auto_send_task(self):
    if self.auto_send_flag:
        try:
            interval = int(self.interval_entry.get())
            self.send_data()  # 执行发送
            self.master.after(interval, self.auto_send_task)  # 定时循环
        except ValueError:
            self.auto_var.set(False)

main_frame = ttk.Frame(self.master)
left_frame = ttk.Labelframe(main_frame, text="串口配置")  # 左侧配置区
send_frame = ttk.Labelframe(right_frame, text="数据发送")  # 右上发送区
recv_frame = ttk.Labelframe(right_frame, text="数据接收")  # 右下接收区

def change_theme(self):
    selected_theme = self.theme_cb.get()
    self.style.theme_use(selected_theme)  # 动态切换主题

历史记录导航：通过<Up>/<Down>键遍历

智能滚动文本框：自动滚动到最新内容

状态栏提示：实时显示连接状态

[2023-08-20 14:25:36] [Send] AT+GMR
[2023-08-20 14:25:36] AT version:2.1.0.0-dev

发送: 2456 字节 | 接收: 18923 字节

import serial
import serial.tools.list_ports
import threading
import queue
import os
import time
import ttkbootstrap as ttk
from ttkbootstrap.constants import *
from ttkbootstrap.dialogs import Messagebox
from ttkbootstrap.scrolled import ScrolledText
from tkinter import BooleanVar, StringVar, IntVar
import platform
from tkinter import filedialog
import json


class SerialTool:
    def __init__(self, master):
        self.master = master
        self.master.title("串口调试助手")
        self.master.geometry("900x520")
        self.master.resizable(False, False)  # 禁止调整窗口大小
        self.master.update()  # 强制应用尺寸限制

        # 初始化样式
        self.style = ttk.Style(theme='cosmo')
        # 配置边框线为纯黑色的样式
        self.style.configure('BlackBorder.TLabelframe', bordercolor='#D3D3D3', relief='solid', borderwidth=1)

        # 串口参数
        self.serial_port = None
        self.receive_queue = queue.Queue()
        self.auto_send_flag = False
        self.send_count = 0
        self.receive_count = 0
        self.receive_thread = None
        self.receive_thread_event = threading.Event()  # 用于控制接收线程的事件

        # 发送历史记录
        self.send_history = []
        self.history_index = -1

        # 自动检测串口变化
        self.last_port_count = 0

        # 创建界面
        self.create_widgets()
        self.refresh_ports()
        self.master.after(100, self.process_queue)
        self.check_ports_change()  # 开始检测串口变化

    def create_widgets(self):
        """创建三栏式布局"""
        main_frame = ttk.Frame(self.master)
        main_frame.pack(fill=BOTH, expand=True, padx=10, pady=10)

        # 主题切换控件
        theme_frame = ttk.Frame(self.master)
        theme_frame.pack(fill=X, padx=10, pady=(0, 5))
        
        ttk.Label(theme_frame, text="主题:").pack(side=LEFT, padx=5)
        self.theme_cb = ttk.Combobox(
            theme_frame, 
            values=sorted(ttk.Style().theme_names()),
            state='readonly'
        )
        self.theme_cb.pack(side=LEFT, padx=5)
        self.theme_cb.set('cosmo')
        self.theme_cb.bind('<<ComboboxSelected>>', self.change_theme)
        
        # 左侧串口配置区
        left_frame = ttk.Labelframe(main_frame, text="串口配置", padding=15, style='BlackBorder.TLabelframe')
        left_frame.grid(row=0, column=0, sticky=NSEW, padx=5, pady=5)

        # 右侧上下分区
        right_frame = ttk.Frame(main_frame)
        right_frame.grid(row=0, column=1, sticky=NSEW, padx=5, pady=5)

        # 发送区（右上）
        send_frame = ttk.Labelframe(right_frame, text="数据发送", padding=15, style='BlackBorder.TLabelframe')
        send_frame.pack(fill=BOTH, expand=True, side=TOP)

        # 接收区（右下）
        recv_frame = ttk.Labelframe(right_frame, text="数据接收", padding=15, style='BlackBorder.TLabelframe')
        recv_frame.pack(fill=BOTH, expand=True, side=TOP)

        # 配置网格权重
        main_frame.columnconfigure(1, weight=1)
        main_frame.rowconfigure(0, weight=1)
        right_frame.rowconfigure(1, weight=1)

        # 创建各区域组件
        self.create_serial_controls(left_frame)
        self.create_send_controls(send_frame)
        self.create_recv_controls(recv_frame)

        # 状态栏
        self.status_var = StringVar(value="就绪")
        ttk.Label(self.master, textvariable=self.status_var,
                  bootstyle=(SECONDARY, INVERSE)).pack(fill=X, side=BOTTOM)

    def change_theme(self, event=None):
        """切换主题"""
        selected_theme = self.theme_cb.get()
        self.style.theme_use(selected_theme)
        
    def create_serial_controls(self, parent):
        """串口参数控件"""
        param_frame = ttk.Frame(parent)
        param_frame.pack(fill=X)

        # 串口号
        ttk.Label(param_frame, text="COM端口:").grid(row=0, column=0, padx=5, pady=5, sticky=W)
        self.port_cb = ttk.Combobox(param_frame, width=15)
        self.port_cb.grid(row=0, column=1, padx=5, pady=5)

        # 波特率
        ttk.Label(param_frame, text="波特率:").grid(row=1, column=0, padx=5, pady=5, sticky=W)
        self.baudrate_cb = ttk.Combobox(param_frame, values=[
            '9600', '115200', '57600', '38400',
            '19200', '14400', '4800', '2400', '1200'
        ], width=15)
        self.baudrate_cb.set('9600')
        self.baudrate_cb.grid(row=1, column=1, padx=5, pady=5)

        # 校验位
        ttk.Label(param_frame, text="校验位:").grid(row=2, column=0, padx=5, pady=5, sticky=W)
        self.parity_cb = ttk.Combobox(param_frame, values=[
            'None', 'Even', 'Odd', 'Mark', 'Space'
        ], width=15)
        self.parity_cb.set('None')
        self.parity_cb.grid(row=2, column=1, padx=5, pady=5)

        # 数据位
        ttk.Label(param_frame, text="数据位:").grid(row=3, column=0, padx=5, pady=5, sticky=W)
        self.databits_cb = ttk.Combobox(param_frame, values=['8', '7', '6', '5'], width=15)
        self.databits_cb.set('8')
        self.databits_cb.grid(row=3, column=1, padx=5, pady=5)

        # 停止位
        ttk.Label(param_frame, text="停止位:").grid(row=4, column=0, padx=5, pady=5, sticky=W)
        self.stopbits_cb = ttk.Combobox(param_frame, values=['1', '1.5', '2'], width=15)
        self.stopbits_cb.set('1')
        self.stopbits_cb.grid(row=4, column=1, padx=5, pady=5)

        # 操作按钮
        # 按钮容器
        btn_frame = ttk.Frame(parent)
        btn_frame.pack(pady=10, fill=X)

        # 配置网格列权重实现自动伸缩
        btn_frame.columnconfigure((0, 1, 2), weight=1, uniform='btns')  # uniform 确保列宽一致

        # 刷新按钮
        ttk.Button(
            btn_frame,
            text="刷新端口",
            command=self.refresh_ports,
            bootstyle=OUTLINE
        ).grid(row=0, column=0, padx=5, sticky="ew")

        # 连接按钮
        self.conn_btn = ttk.Button(
            btn_frame,
            text="打开串口",
            command=self.toggle_connection,
            bootstyle=OUTLINE + SUCCESS
        )
        self.conn_btn.grid(row=0, column=1, padx=5, sticky="ew")

        # 手动发送按钮（移动到此处）
        ttk.Button(
            btn_frame,
            text="手动发送",
            command=self.send_data,
            bootstyle=OUTLINE + PRIMARY
        ).grid(row=0, column=2, padx=5, sticky="ew")

    def create_send_controls(self, parent):
        """发送区控件"""
        # 自动发送设置
        auto_frame = ttk.Frame(parent)
        auto_frame.pack(fill=X, pady=5)

        self.auto_var = BooleanVar()
        ttk.Checkbutton(auto_frame, text="自动发送", variable=self.auto_var,
                        command=self.toggle_auto_send).pack(side=LEFT)
        ttk.Label(auto_frame, text="间隔(ms):").pack(side=LEFT, padx=5)
        self.interval_entry = ttk.Entry(auto_frame, width=8)
        self.interval_entry.insert(0, "1000")
        self.interval_entry.pack(side=LEFT)

        # 发送内容
        self.send_text = ScrolledText(parent, height=4, autohide=True)
        self.send_text.pack(fill=BOTH, expand=True)
        
        # 绑定上下箭头键用于历史记录导航
        self.send_text.bind("<Up>", self.prev_history)
        self.send_text.bind("<Down>", self.next_history)

    def create_recv_controls(self, parent):
        """接收区控件"""
        # 接收显示
        self.recv_text = ScrolledText(parent, height=5, autohide=True)
        self.recv_text.pack(fill=BOTH, expand=True)

        # 统计栏
        stat_frame = ttk.Frame(parent)
        stat_frame.pack(fill=X, pady=5)
        ttk.Label(stat_frame, text="发送:").pack(side=LEFT, padx=5)
        self.send_label = ttk.Label(stat_frame, text="0")
        self.send_label.pack(side=LEFT)
        ttk.Label(stat_frame, text="接收:").pack(side=LEFT, padx=10)
        self.recv_label = ttk.Label(stat_frame, text="0")
        self.recv_label.pack(side=LEFT)
        
        # 添加保存接收按钮
        ttk.Button(stat_frame, text="保存接收", command=self.save_received,
                  bootstyle=OUTLINE + INFO).pack(side=RIGHT, padx=5)
        
        ttk.Button(stat_frame, text="清空", command=self.clear_received,
                  bootstyle=OUTLINE + WARNING).pack(side=RIGHT)

    def refresh_ports(self):
        """刷新端口列表"""
        try:
            ports = [p.device for p in serial.tools.list_ports.comports()]
            self.port_cb['values'] = ports
            self.status_var.set(f"自动检测到主板有{len(ports)} 个串口可用，请注意选择正确的。")
            self.last_port_count = len(ports)
        except Exception as e:
            print(f"Error refreshing ports: {e}")
            self.status_var.set(f"刷新端口时出错: {e}")

    def check_ports_change(self):
        """检查串口变化"""
        current_count = len(list(serial.tools.list_ports.comports()))
        if current_count != self.last_port_count:
            self.refresh_ports()
        self.master.after(1000, self.check_ports_change)  # 每秒检查一次

    def toggle_connection(self):
        """切换连接状态"""
        if self.serial_port and self.serial_port.is_open:
            self.close_serial()
        else:
            self.open_serial()

    def open_serial(self):
        """打开串口"""
        try:
            port = self.port_cb.get()
            if not port:
                raise ValueError("请选择串口")

            parity_map = {
                'None': serial.PARITY_NONE,
                'Even': serial.PARITY_EVEN,
                'Odd': serial.PARITY_ODD,
                'Mark': serial.PARITY_MARK,
                'Space': serial.PARITY_SPACE
            }

            self.serial_port = serial.Serial(
                port=port,
                baudrate=int(self.baudrate_cb.get()),
                parity=parity_map[self.parity_cb.get()],
                bytesize=int(self.databits_cb.get()),
                stopbits=float(self.stopbits_cb.get()),
                timeout=0.1
            )

            self.conn_btn.configure(text="关闭串口", bootstyle=OUTLINE + SUCCESS)
            self.status_var.set(f"已连接 {port}")
            self.receive_thread_event.clear()  # 清除事件标志
            self.receive_thread = threading.Thread(target=self.receive_worker, daemon=True)
            self.receive_thread.start()

        except Exception as e:
            Messagebox.show_error(f"主板上没有这个串口或你选的被测端口跟主板端口不对应，请在设备管理器中确认正确的端口: {str(e)}", "错误")
            self.status_var.set("连接失败")

    def close_serial(self):
        """关闭串口"""
        self.receive_thread_event.set()  # 设置事件标志，通知接收线程停止
        if self.receive_thread and self.receive_thread.is_alive():
            self.receive_thread.join()  # 等待接收线程结束

        if self.serial_port:
            try:
                self.serial_port.close()
            except Exception as e:
                print(f"关闭串口时出错: {e}")

        self.conn_btn.configure(text="打开串口", bootstyle=DANGER)
        self.status_var.set("已断开连接")

    def receive_worker(self):
        """接收线程工作函数"""
        while not self.receive_thread_event.is_set() and self.serial_port and self.serial_port.is_open:
            try:
                if self.serial_port.in_waiting > 0:
                    data = self.serial_port.read(self.serial_port.in_waiting)
                    self.receive_queue.put(data)
            except Exception as e:
                print(f"接收错误: {e}")
                break

    def process_queue(self):
        """处理接收队列"""
        while not self.receive_queue.empty():
            data = self.receive_queue.get()
            self.display_received(data)
            self.receive_count += len(data)
            self.recv_label.configure(text=str(self.receive_count))
        self.master.after(100, self.process_queue)

    def display_received(self, data):
        """显示接收数据（带时间戳）"""
        timestamp = time.strftime("[%Y-%m-%d %H:%M:%S] ", time.localtime())
        try:
            text = data.decode('utf-8')
            self.recv_text.insert(END, timestamp + text + 'n')
            self.recv_text.see(END)
        except UnicodeDecodeError:
            self.recv_text.insert(END, timestamp + data.hex(' ') + 'n')
            self.recv_text.see(END)

    def toggle_auto_send(self):
        """切换自动发送"""
        self.auto_send_flag = self.auto_var.get()
        if self.auto_send_flag:
            self.auto_send_task()

    def auto_send_task(self):
        """自动发送任务"""
        if self.auto_send_flag and self.serial_port and self.serial_port.is_open:
            try:
                interval = int(self.interval_entry.get())
                self.send_data()
                self.master.after(interval, self.auto_send_task)
            except ValueError:
                self.auto_var.set(False)
                Messagebox.show_error("无效的间隔时间", "错误")

    def send_data(self):
        """发送数据"""
        if not self.serial_port or not self.serial_port.is_open:
            Messagebox.show_warning("请先打开串口", "警告")
            return

        data = self.send_text.get(1.0, END).strip()
        if not data:
            return

        try:
            # 添加到历史记录
            if data and (not self.send_history or data != self.send_history[0]):
                self.send_history.insert(0, data)
                if len(self.send_history) > 20:  # 限制历史记录数量
                    self.send_history.pop()
                self.history_index = -1  # 重置历史索引

            self.serial_port.write(data.encode('utf-8'))
            self.send_count += len(data)
            self.send_label.configure(text=str(self.send_count))
            
            # 显示发送的数据（带时间戳）
            timestamp = time.strftime("[%Y-%m-%d %H:%M:%S] ", time.localtime())
            self.recv_text.insert(END, f"{timestamp}[Send] {data}n")
            self.recv_text.see(END)
        except Exception as e:
            Messagebox.show_error(f"发送失败: {str(e)}", "错误")

    def prev_history(self, event):
        """上一条历史记录"""
        if self.send_history:
            if self.history_index < len(self.send_history) - 1:
                self.history_index += 1
                self.send_text.delete(1.0, END)
                self.send_text.insert(END, self.send_history[self.history_index])
        return "break"

    def next_history(self, event):
        """下一条历史记录"""
        if self.history_index > 0:
            self.history_index -= 1
            self.send_text.delete(1.0, END)
            self.send_text.insert(END, self.send_history[self.history_index])
        elif self.history_index == 0:
            self.history_index = -1
            self.send_text.delete(1.0, END)
        return "break"

    def save_received(self):
        """保存接收内容到文件"""
        filename = filedialog.asksaveasfilename(
            defaultextension=".txt",
            filetypes=[("Text Files", "*.txt"), ("All Files", "*.*")]
        )
        if filename:
            try:
                with open(filename, 'w', encoding='utf-8') as f:
                    f.write(self.recv_text.get(1.0, END))
                self.status_var.set(f"接收内容已保存到 {filename}")
            except Exception as e:
                Messagebox.show_error(f"保存文件失败: {str(e)}", "错误")

    def clear_received(self):
        """清空接收区"""
        self.recv_text.delete(1.0, END)
        self.receive_count = 0
        self.recv_label.configure(text="0")
        self.send_text.delete(1.0, END)
        self.send_count = 0
        self.send_label.configure(text="0")

    def on_closing(self):
        """安全关闭程序"""
        # 停止自动发送循环
        self.auto_send_flag = False

        # 关闭串口连接
        self.close_serial()

        # 确保完全退出
        self.master.quit()  # 终止mainloop
        self.master.destroy()  # 销毁所有Tkinter对象
        self.master.after(500, self.force_exit)  # 500ms后强制退出

    def force_exit(self):
        """最终退出保障"""
        import os
        os._exit(0)  # 强制终止进程


if __name__ == "__main__":
    root = ttk.Window()
    app = SerialTool(root)
    root.protocol("WM_DELETE_WINDOW", app.on_closing)
    root.mainloop()

• 采用生产者-消费者模式处理串口数据
• 通过队列实现线程间安全通信
• 现代化UI提升使用体验

• 增加协议解析功能（Modbus/AT指令等）
• 实现数据图表可视化
• 添加插件系统支持

以上就是利用Python调试串口的示例代码的详细内容，更多关于Python调试串口的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• java使用Rxtx实现串口通信调试工具
• C#实现串口调试工具
• 使用python+Pyqt5实现串口调试助手
• 基于Python编写一个串口调试工具