Python实现USB热插拔检测

在Python中，我们可以使用多种方法来检测USB设备的插拔事件。以下是几种常见的实现方式：

方法一：使用pyudev库（Linux）

import pyudev
import threading

def monitor_usb():
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by(subsystem='usb')

    for device in iter(monitor.poll, None):
        if device.action == 'add':
            print(f"USB设备已插入: {device}")
        elif device.action == 'remove':
            print(f"USB设备已拔出: {device}")

# 启动监控线程
usb_thread = threading.Thread(target=monitor_usb, daemon=True)
usb_thread.start()

# 主线程继续执行其他任务
print("USB热插拔监控已启动...")
while True:
    pass

方法二：使用pywin32库（Windows）

import win32api
import win32con
import win32gui
import threading

def usb_notification(hwnd, msg, wparam, lparam):
    if msg == win32con.WM_DEVICECHANGE:
        if wparam == win32con.DBT_DEVICEARRIVAL:
            print("USB设备已插入")
        elif wparam == win32con.DBT_DEVICEREMOVECOMPLETE:
            print("USB设备已拔出")
    return True

def message_loop():
    win32gui.PumpMessages()

# 创建隐藏窗口接收消息
hwnd = win32gui.CreateWindow("STATIC", "USB Monitor", 0, 0, 0, 0, 0, 0, 0, 0, None)
win32gui.SetWindowLong(hwnd, win32con.GWL_WNDPROC, usb_notification)

# 启动消息循环线程
thread = threading.Thread(target=message_loop, daemon=True)
thread.start()

print("USB热插拔监控已启动...")
while True:
    pass

方法三：跨平台方案（使用pyudev或wmi）

import platform
import threading

def monitor_usb():
    system = platform.system()

    if system == 'Linux':
        import pyudev
        context = pyudev.Context()
        monitor = pyudev.Monitor.from_netlink(context)
        monitor.filter_by(subsystem='usb')

        for device in iter(monitor.poll, None):
            if device.action == 'add':
                print(f"USB设备已插入: {device}")
            elif device.action == 'remove':
                print(f"USB设备已拔出: {device}")

    elif system == 'Windows':
        import wmi
        c = wmi.WMI()
        watcher = c.Win32_DeviceChangeEvent.watch_for(
            notification_type="Creation",
            delay_secs=1
        )
        while True:
            watcher()
            print("USB设备状态变化")

    else:
        print(f"不支持的操作系统: {system}")

# 启动监控线程
usb_thread = threading.Thread(target=monitor_usb, daemon=True)
usb_thread.start()

print("USB热插拔监控已启动...")
while True:
    pass

方法四：使用psutil检测USB设备变化

import psutil
import time

def get_usb_devices():
    return [d for d in psutil.disk_partitions() if 'removable' in d.opts]

def monitor_usb():
    current_devices = set(get_usb_devices())
    while True:
        new_devices = set(get_usb_devices())
        added = new_devices - current_devices
        removed = current_devices - new_devices

        for device in added:
            print(f"USB设备已插入: {device.device}")

        for device in removed:
            print(f"USB设备已拔出: {device.device}")

        current_devices = new_devices
        time.sleep(1)

# 启动监控
print("USB热插拔监控已启动...")
monitor_usb()

注意事项

1. Linux系统需要安装pyudev库：pip install pyudev
2. Windows系统需要安装pywin32或wmi库：pip install pywin32 或 pip install wmi
3. 跨平台方案需要根据运行环境选择适当的实现
4. 某些方法可能需要管理员权限
5. 对于生产环境，建议添加异常处理和日志记录

以上方法可以根据你的具体需求和运行环境进行选择和调整。