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

如何在Flask/Python中停止USB监控进程?

USB监控程序无法停止且无法触发页面渲染问题

我在项目中运行一个USB监控程序,但它会干扰其他功能,因此需要按需启动/停止它。目前启动功能正常,但停止功能无法生效,无法终止监控。触发停止逻辑时,监控未终止,页面也未渲染。之前HTML中的JavaScript会造成干扰,改用纯HTML页面后启动功能有所改善,但停止功能仍无法正常工作。

相关代码

# 大量导入代码省略

# USB监控进程(用于管理生命周期的全局变量,为解决问题尝试了很多变量)
usb_process = None
running = False
usb_monitor_thread = None
usb_monitor_started = False

def device_event_monitor(action, device):
    if device.subsystem == 'usb':
        if action == 'add' and device.device_type == 'usb_device':
            # 我的代码会处理设备,获取连接设备的信息
            # 尤其是生成哈希值插入SQL数据库
            device_hash = hash_object.hexdigest()
            if device_hash in hash_list:
                cursor.execute(
                    "UPDATE devices SET connection_time = ? WHERE hash = ?",
                    (start_time, device_hash),
                )
            else:
                # 这部分无法正常工作:我想停止监控并通过网页启动设备注册流程,但看不到日志,监控没停,页面也没渲染
                stop_usb_monitor()
                print(f"STOP MONITOR!!")
                with app.app_context():
                    return render_template("new_device.html")

def monitor_usb():
    global running
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by('usb')
    monitor.enable_receiving()
    observer = pyudev.MonitorObserver(monitor, device_event_monitor)
    observer.start()
    try:
        while running:  
            time.sleep(0.1)  
    except KeyboardInterrupt:
        print("Monitor manually stopped.")
    finally:
        observer.stop()
        print("Monitor USB stopped.")

def start_usb_monitor():
    global running, usb_monitor_thread
    if running:
        print("Already running!")
        return
    running = True
    usb_monitor_thread = threading.Thread(target=monitor_usb, daemon=True)
    usb_monitor_thread.start()
    print("Monitor started!")

def stop_usb_monitor():
    global running, usb_monitor_thread
    if not running:
        print("Monitor not running!")
        return
    running = False  
    if usb_monitor_thread:
        usb_monitor_thread.join()  # 等待线程终止
        print("Monitor USB stopped!")

@app.route("/")
def index():
    return render_template("index.html")

@app.route('/start', methods=['POST'])
def start():
    start_usb_monitor()
    return render_template("table.html")

@app.route("/new_device", methods=["POST"])
def new_device():
    device_name = request.form.get("owner_name")
    
if __name__ == "__main__":
    # 启动Flask应用
    app.run(debug=True)

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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.06.15 23:52:39