pyusb调用POST请求后USB扫码设备反复断开重连问题求助

问题描述 #

已完成libusb过滤器配置，使用pyusb读取二维码扫描数据后触发POST请求，单次执行逻辑正确，但执行POST请求后电脑会出现libusb设备断开并重新连接的提示音；若注释掉POST请求代码，程序可正常持续运行。

附相关代码：

import usb.core
import usb.util
import requests

# USB QR code reader vendor and product IDs
VENDOR_ID = XXXX  # Vendor ID in hex
PRODUCT_ID = XXXX  # Product ID in hex

# Create a mapping dictionary
mapping = {
    4: 'a', 5: 'b', 6: 'c', 7: 'd', 8: 'e', 9: 'f', 10: 'g', 11: 'h', 12: 'i', 13: 'j',
    14: 'k', 15: 'l', 16: 'm', 17: 'n', 18: 'o', 19: 'p', 20: 'q', 21: 'r', 22: 's', 23: 't',
    24: 'u', 25: 'v', 26: 'w', 27: 'x', 28: 'y', 29: 'z', 30: '1', 31: '2', 32: '3', 33: '4',
    34: '5', 35: '6', 36: '7', 37: '8', 38: '9', 39: '0', 45: '-'
}

# Find the USB device
dev = usb.core.find(idVendor=VENDOR_ID, idProduct=PRODUCT_ID)

if dev is None:
    raise ValueError("QR Code Scanner not found")

# Set the active configuration
dev.set_configuration()

# Claim the interface
cfg = dev.get_active_configuration()
intf = cfg[(0, 0)]
usb.util.claim_interface(dev, intf)

# Endpoint to read from
endpoint = intf[0]

# API URL
api_url = "https://my-website/api/membership-manager/checkins"

# To store the concatenated code values
code_values = []


def post_qr_code(qr_code):
    headers = {
        'Provider': 'UUID',
        'Content-Type': 'application/json'
    }

    payload = {
        'userId': qr_code
    }

    try:
        response = requests.post(api_url, json=payload, headers=headers)
        response.raise_for_status()  # Raises HTTPError for bad responses (4xx and 5xx)

        data = response.json()
        print('RESULT', data)

    except requests.exceptions.RequestException as error:
        print('There was an error!', error)


def read_qr_code():
    global code_values

    try:
        # Read data from the USB endpoint
        data = dev.read(endpoint.bEndpointAddress, endpoint.wMaxPacketSize, timeout=5000)

        if len(data) > 2:
            code_value = data[2]

            # Skip processing if data[2] is 0
            if code_value == 0:
                return

            # Convert code_value to character using the dictionary
            qr_code_char = mapping.get(code_value, '?')

            if qr_code_char != '?':
                code_values.append(qr_code_char)

            # Check if the concatenated string length is 36
            if len(code_values) == 36:
                qr_code = ''.join(code_values)
                print(f"QR Code Scanned: {qr_code}")
                post_qr_code(qr_code)
                code_values = []  # Clear the list after processing

        else:
            print("Data length is less than expected.")

    except usb.core.USBError as e:
        # Handle USB errors and avoid re-attempting in a loop
        if e.errno != None:
            print(f"USB Error: {str(e)}")


if __name__ == "__main__":
    try:
        print("Waiting for QR code scan...")
        while True:
            read_qr_code()
    except KeyboardInterrupt:
        # Release the interface when the program ends
        usb.util.release_interface(dev, intf)
        usb.util.dispose_resources(dev)
        print("Program terminated and USB interface released.")
    except Exception as e:
        # Handle any other exceptions that occur
        print(f"An unexpected error occurred: {e}")
        usb.util.release_interface(dev, intf)
        usb.util.dispose_resources(dev)

问题原因 #

核心问题是同步POST请求阻塞了USB读取的主线程：

1. requests.post是同步阻塞操作，执行时会占用主线程，导致dev.read()长时间无法执行，USB设备因超时判定连接异常，触发自动重连。
2. 过长的USB读取超时设置（5000ms），进一步放大了主线程阻塞的影响，设备得不到响应的时间窗口变长。

解决方案 #

1. 用异步线程执行POST请求 #

将POST操作放到独立线程中，避免阻塞USB读取的主线程：

# 新增导入threading
import threading

# 修改read_qr_code中的POST调用逻辑
if len(code_values) == 36:
    qr_code = ''.join(code_values)
    print(f"QR Code Scanned: {qr_code}")
    # 用守护线程执行POST，不阻塞主线程
    threading.Thread(target=post_qr_code, args=(qr_code,), daemon=True).start()
    code_values = []  # Clear the list after processing

2. 优化USB读取的超时与错误处理 #

缩短读取超时时间，避免设备长时间等待响应；同时过滤超时错误，让循环持续运行：

# 修改read_qr_code中的读取超时
data = dev.read(endpoint.bEndpointAddress, endpoint.wMaxPacketSize, timeout=1000)  # 缩短至1秒

# 修改USBError处理逻辑
except usb.core.USBError as e:
    # 忽略超时错误，继续循环
    if e.errno == 110:  # 对应超时错误码
        return
    print(f"USB Error: {str(e)}")

3. 确保全局资源清理 #

添加退出钩子，保证程序在任何异常情况下都能正确释放USB资源：

# 新增导入atexit
import atexit

def cleanup_usb():
    try:
        usb.util.release_interface(dev, intf)
        usb.util.dispose_resources(dev)
        print("USB resources released.")
    except Exception as e:
        print(f"Cleanup error: {str(e)}")

# 注册退出钩子
atexit.register(cleanup_usb)

4. 调整系统USB电源设置（Windows） #

在设备管理器中找到扫码器设备，进入「属性」-「电源管理」，取消勾选「允许计算机关闭此设备以节约电源」，避免系统因空闲断开设备。

内容的提问来源于stack exchange，提问作者vkatsou