You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python实现带启停/暂停功能的带延迟串行命令发送咨询

Python实现带延迟的命令串行发送与暂停/停止控制

针对你的需求,我会用Tkinter(Python自带GUI库)结合多线程和事件控制来实现,核心思路是用后台线程处理命令发送逻辑,避免阻塞GUI,同时通过threading.Event实现暂停、停止的精准控制,还能记录执行断点实现续传。

完整实现代码

import tkinter as tk
from tkinter import ttk
import threading
import time

class CommandSenderApp:
    def __init__(self, root):
        self.root = root
        self.root.title("命令发送控制器")
        
        # 初始化控制事件
        self.stop_event = threading.Event()
        self.pause_event = threading.Event()
        self.pause_event.set()  # 默认不暂停
        self.is_running = False
        self.current_index = 0  # 记录当前执行的命令索引
        self.command_list = []  # 存储从TreeView提取的命令
        
        # 创建TreeView
        self.tree = ttk.Treeview(root, columns=("行号", "cmd", "delay"), show="headings")
        self.tree.heading("行号", text="行号")
        self.tree.heading("cmd", text="命令")
        self.tree.heading("delay", text="延迟(秒)")
        self.tree.column("行号", width=60)
        self.tree.column("cmd", width=100)
        self.tree.column("delay", width=80)
        
        # 添加示例数据
        self.tree.insert("", "end", values=(1, "50", 2))
        self.tree.insert("", "end", values=(2, "60", 3))
        self.tree.insert("", "end", values=(3, "70", 1))
        self.tree.pack(pady=10, padx=10, fill=tk.BOTH, expand=True)
        
        # 控制按钮框架
        btn_frame = ttk.Frame(root)
        btn_frame.pack(pady=5)
        
        self.start_btn = ttk.Button(btn_frame, text="启动", command=self.start_sending)
        self.start_btn.grid(row=0, column=0, padx=5)
        
        self.pause_btn = ttk.Button(btn_frame, text="暂停", command=self.toggle_pause)
        self.pause_btn.grid(row=0, column=1, padx=5)
        
        self.stop_btn = ttk.Button(btn_frame, text="停止", command=self.stop_sending)
        self.stop_btn.grid(row=0, column=2, padx=5)
        
    def extract_commands(self):
        """从TreeView提取命令列表,格式为[(cmd, delay), ...]"""
        self.command_list = []
        for item in self.tree.get_children():
            values = self.tree.item(item)["values"]
            cmd = values[1]
            delay = int(values[2])
            self.command_list.append((cmd, delay))
    
    def send_command_thread(self):
        """后台线程:处理命令发送、延迟、暂停/停止逻辑"""
        self.is_running = True
        total_commands = len(self.command_list)
        
        while self.current_index < total_commands:
            # 检查是否需要停止
            if self.stop_event.is_set():
                break
            
            # 检查是否需要暂停
            self.pause_event.wait()
            
            # 取出当前命令
            cmd, delay = self.command_list[self.current_index]
            
            # 模拟发送命令(这里替换成你的实际发送逻辑)
            print(f"发送命令: `{cmd}`")
            
            # 等待指定延迟,期间每隔0.1秒检查一次停止信号
            start_time = time.time()
            while time.time() - start_time < delay:
                if self.stop_event.is_set():
                    break
                time.sleep(0.1)
            
            # 如果没被停止,才推进索引
            if not self.stop_event.is_set():
                self.current_index += 1
        
        # 执行结束,重置状态
        self.is_running = False
        self.stop_event.clear()
        self.pause_event.set()
        self.root.after(0, lambda: self.start_btn.config(state=tk.NORMAL))
        print("命令发送结束/已停止")
    
    def start_sending(self):
        """启动按钮回调"""
        if not self.is_running:
            self.start_btn.config(state=tk.DISABLED)
            self.extract_commands()
            self.current_index = 0  # 从头开始,如需断点续传可保留当前索引
            # 创建并启动线程
            threading.Thread(target=self.send_command_thread, daemon=True).start()
    
    def toggle_pause(self):
        """暂停/恢复按钮回调"""
        if self.is_running:
            if self.pause_event.is_set():
                self.pause_event.clear()
                self.pause_btn.config(text="恢复")
                print("已暂停")
            else:
                self.pause_event.set()
                self.pause_btn.config(text="暂停")
                print("已恢复")
    
    def stop_sending(self):
        """停止按钮回调"""
        if self.is_running:
            self.stop_event.set()
            self.pause_event.set()  # 暂停状态下也能停止
            self.start_btn.config(state=tk.NORMAL)
            self.pause_btn.config(text="暂停")
            print("已停止")

if __name__ == "__main__":
    root = tk.Tk()
    app = CommandSenderApp(root)
    root.mainloop()

关键逻辑说明

1. 事件控制机制

  • stop_event:一旦触发(set()),线程会立即终止当前流程,不管是在等待延迟还是准备发送命令
  • pause_event:默认是激活状态(set()),调用clear()会让线程进入等待状态,直到再次set()恢复执行

2. 断点续传实现

  • 用self.current_index记录当前执行到第几条命令,暂停后恢复时,线程会从这个索引继续执行
  • 如果需要每次启动都从头开始,就在start_sending里重置self.current_index = 0;如果要保留断点,去掉这行即可

3. 延迟等待的优化

  • 等待延迟时没有用time.sleep(delay),而是每隔0.1秒检查一次stop_event,这样在等待过程中点击停止也能立即响应,不会卡住

4. GUI线程安全

  • 所有修改GUI组件的操作(比如按钮状态)都用root.after(0, ...)放到主线程执行,避免多线程直接操作GUI导致的异常

替换实际发送逻辑

把代码中print(f"发送命令: {cmd}")这一行替换成你的实际命令发送代码,比如串口发送、网络请求等。

内容的提问来源于stack exchange,提问作者mScientist

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.13 08:58:56