当前位置: 首页 > backend >正文

python实现usb热插拔检测(windows)

文章目录

    • 1 概述
    • 2 实现代码
    • 3 演示效果
    • 4 源码地址


更多精彩内容
👉内容导航 👈
👉Qt开发 👈
👉python开发 👈

1 概述

这段代码实现了一个Windows平台的USB设备实时监控类,主要功能如下:

核心功能

  1. 设备事件监听
    • 基于WMI技术实时捕获USB设备的插入(creation)/移除(deletion)事件
    • 通过Win32_PnPEntity类监听硬件变更事件
    • 使用双事件过滤器实现双向监控
  2. 异步处理机制
    • 独立监控线程避免阻塞主程序(通过threading.Thread实现)
    • 守护线程模式确保主程序退出时自动终止
    • 非阻塞式事件检测(0.5秒超时机制)
  3. 回调机制
    • 可自定义设备插入/移除回调函数
    • 内置安全校验确保回调可调用性

适用场景

  • USB设备热插拔检测
  • 外设连接状态监控
  • 自动化设备管理工具开发

WMI基础概念

  • WMI (Windows Management Instrumentation) 是微软的Windows系统管理技术
  • 通过wmi模块可以获取/管理Windows系统的:
    • 硬件信息(CPU/内存/磁盘)
    • 系统配置(服务/进程/网络)
    • 性能指标(CPU负载/内存使用率)

关键方法说明:

  • __init__: 初始化监控实例,可以传入设备插入和移除时的回调函数。
  • get_all_devices: 获取当前已连接的所有 USB 设备的快照。
  • device_added_callback: 内部方法,调用传入的设备插入回调函数。
  • device_removed_callback: 内部方法,调用传入的设备移除回调函数。
  • start_monitoring: 监控线程的主循环,监听设备的插入和移除事件。
  • start: 启动监控服务。
  • stop: 安全停止监控服务。

环境说明

环境版本
pythonV3.13.2
系统windows10
pycharmpycharm-community-2024.3.5.exe
依赖wmi

依赖安装

pip install wmi

2 实现代码

import wmi
import threading
import time
import pythoncom  # 用于处理COM库的线程安全初始化class USBMonitor:"""USB设备实时监控类(Windows平台专用)功能特性:- 基于WMI实现设备热插拔事件监听- 支持设备插入/移除的双向回调机制- 使用独立监控线程避免阻塞主程序- 自动处理COM库线程初始化问题"""def __init__(self, on_device_added=None, on_device_removed=None):"""初始化监控实例:param on_device_added:   设备插入回调函数,格式 func(device_obj):param on_device_removed: 设备移除回调函数,格式 func(device_obj)"""self.on_device_added = on_device_addedself.on_device_removed = on_device_removedself.monitor_thread = threading.Thread(target=self.start_monitoring)self.monitor_thread.daemon = True  # 设为守护线程确保主程序退出时自动终止self.thread_running = False  # 线程运行状态标志def get_all_devices(self):"""获取当前已连接的USB设备列表(快照)"""c = wmi.WMI(moniker="root/cimv2")return c.Win32_USBControllerDevice()  # 返回Win32_USBControllerDevice对象集合def device_added_callback(self, new_device):"""内部设备插入事件处理器"""if callable(self.on_device_added):  # 安全检查回调函数self.on_device_added(new_device)def device_removed_callback(self, removed_device):"""内部设备移除事件处理器"""if callable(self.on_device_removed):self.on_device_removed(removed_device)def start_monitoring(self):"""监控线程主循环(核心逻辑)"""pythoncom.CoInitialize()  # 每个线程必须独立初始化COM库try:c = wmi.WMI(moniker="root/cimv2")# 配置插入事件监听器(1秒轮询间隔)arr_filter = c.Win32_PnPEntity.watch_for(notification_type="creation",delay_secs=1)# 配置移除事件监听器rem_filter = c.Win32_PnPEntity.watch_for(notification_type="deletion",delay_secs=1)self.thread_running = Trueprint("[监控系统] USB设备监控服务启动")while self.thread_running:# 非阻塞式检测插入事件(0.5秒超时)try:new_device = arr_filter(0.5)self.device_added_callback(new_device)except wmi.x_wmi_timed_out:pass  # 正常超时,继续循环# 非阻塞式检测移除事件try:removed_device = rem_filter(0.5)self.device_removed_callback(removed_device)except wmi.x_wmi_timed_out:passexcept KeyboardInterrupt:print("\n[监控系统] 用户主动终止监控")finally:pythoncom.CoUninitialize()  # 必须释放COM资源print("[监控系统] 监控服务已安全关闭")def start(self):"""启动监控服务"""if not self.monitor_thread.is_alive():self.monitor_thread.start()def stop(self):"""安全停止监控服务"""self.thread_running = Falseif self.monitor_thread.is_alive():self.monitor_thread.join(timeout=2)  # 等待线程结束# 示例回调实现
def on_device_added(device):"""设备插入事件处理示例"""print(f"[事件] 新设备接入: {device.Description}")def on_device_removed(device):"""设备移除事件处理示例"""print(f"[事件] 设备移除: {device.Description}")# 使用示例
if __name__ == "__main__":# 初始化带回调的监控实例monitor = USBMonitor(on_device_added, on_device_removed)try:monitor.start()print("主程序运行中,按Ctrl+C停止监控...")while True:  # 主线程保持活动time.sleep(1)except KeyboardInterrupt:monitor.stop()print("程序正常退出")

3 演示效果

在这里插入图片描述

4 源码地址

  • gitee
  • github


http://www.xdnf.cn/news/5778.html

相关文章:

  • Android Framework
  • LWIP传输层协议笔记
  • Git 用法总结
  • 微信小程序原生swiper高度自适应图片,不同屏幕适配,正方形1:1等比例图片轮播
  • E+H流量计与Profibus DP主站转Modbus RTU/TCP网关通讯
  • DeepSeek新玩法: RAG Chatbot 3.0测试人的新大脑
  • 深入探讨dubbo组件的实践
  • Dapp开发-如何开发一个dapp
  • Vue 2 项目中配置 Tailwind CSS 和 Font Awesome 的最佳实践
  • MYSQL之表的约束
  • rbac模型详解
  • PHP编写图书信息爬虫程序
  • 力扣451:根据字符频率排序(桶排序)
  • 快解析为TPDDNS用户提供免费替换服务
  • 小白学习Java第18天(上):mybatis
  • 994. 腐烂的橘子
  • MYSQL时间函数、group by 和partition by的区别、组内编号leetcode学习
  • GitHub 趋势日报 (2025年05月11日)
  • LeetCode热题100——链表
  • docker-compose的yml文件配置deploy参数失效use the ‘deploy‘ key, which will be ignored.
  • MIMO 检测(2)--噪声白化
  • 雷池WAF的身份认证 - 钉钉配置教程
  • hi3516cv610的VPSS_ONLINE支持在vpss做图片放大的操作吗
  • IT团队如何通过ManageEngine卓豪Endpoint Central有效管理远程终端
  • 解决echartsV5+ restore后echarts显示空白
  • 防火墙来回路径不一致导致的业务异常
  • 当用户在浏览器输入一个 URL 并访问服务器时, 这个请求是如何到达对应的 Servlet 的?
  • 基于大模型预测的吉兰 - 巴雷综合征综合诊疗方案研究报告大纲
  • 5.11 - 5.12 JDBC+Mybatis+StringBoot项目配置文件
  • 【NextPilot日志移植】日志写入流程