为何读取DigiUSB.print()的USB输出时Python与Rust代码结果不一致？

我编写了Rust代码通过USB读取DigiUSB.print()的输出，但结果出现字符丢失、错乱的情况。测试官方提供的Python代码时输出完全正常，无法排查出自家代码的问题。

我的Rust代码 #

use std::{io::Write, process::exit};

use rusb::{request_type, Device, DeviceList, Direction, GlobalContext, Recipient, RequestType};


fn main() -> rusb::Result<()> {
    // Find and open the USB device handle
    let device_desc = || -> Option<rusb::Device<rusb::GlobalContext>> {
        let mut temp: Option<Device<GlobalContext>> = None;
        for device in DeviceList::new().unwrap().iter() {
            let descriptor = device.device_descriptor().unwrap();
            if descriptor.vendor_id() == 0x16c0 && descriptor.product_id() == 0x05df {
                temp = Some(device)
            } else {
                continue;
            }
        }
        temp
    }();
    let device_desc = match device_desc {
        Some(a) => a,
        None => {
            println!("Device Not Found");
            exit(0);
        }
    };

    let request_type = request_type(Direction::In, RequestType::Class, Recipient::Device);
    let request = 0x01;
    let value = 0x00;
    let index = 0x00;
    loop {
        let handle = device_desc.open();
        match handle {
            Ok(hand) => {
                let buf: &mut [u8] = &mut [0];
                let response = hand.read_control(
                    request_type,
                    request,
                    value,
                    index,
                    buf,
                    std::time::Duration::from_secs(1),
                );
                match response {
                    Ok(_) => {
                        std::io::stdout()
                            .flush()
                            .expect("Error while flushing Buffer");
                        print!("{}",char::from(buf[0]));
                        std::io::stdout()
                            .flush()
                            .expect("Error while flushing Buffer");
                    }
                    Err(_) => std::thread::sleep(std::time::Duration::from_millis(500)),
                }
            }
            Err(_) => std::thread::sleep(std::time::Duration::from_secs(1)),
        }
    }
    Ok(());
}

Rust代码输出 #

Hello Wrl
Hell Word
Helo World
Hello orld
Helo Wold
HellWorldHelloWorld
Hello Wr

参考Python代码（主程序） #

#/usr/bin/python

#
# Written for PyUSB 1.0 (w/libusb 1.0.3)
#
# Includes functionality to retrieve string descriptors
#
# Author: follower@rancidbacon.com
#
# Version: 20091021
#

#
# Assumes 'UsbStreamDemo1.pde' is loaded on Arduino and 
# LEDs are present on pins 11, 12 and 13.
#

import usb # 1.0 not 0.4


import sys
sys.path.append("..")

from arduino.usbdevice import ArduinoUsbDevice


if __name__ == "__main__":
    try:
        theDevice = ArduinoUsbDevice(idVendor=0x16c0, idProduct=0x05df)

        print "Found: 0x%04x 0x%04x %s %s" % (theDevice.idVendor, 
                                              theDevice.idProduct,
                                              theDevice.productName,
                                              theDevice.manufacturer)
    except:
        pass



    import sys
    import time

    while 1 == 1:
        try:
            theDevice = ArduinoUsbDevice(idVendor=0x16c0, idProduct=0x05df)
            try:
                sys.stdout.write(chr(theDevice.read()))
                sys.stdout.flush()
            except:
                # TODO: Check for exception properly
                time.sleep(0.5)
                
        except:
            # TODO: Check for exception properly
            time.sleep(1)

arduino.usbdevice模块代码 #

#/usr/bin/python

#
# Written for PyUSB 1.0 (w/libusb 1.0.3)
#
# Includes functionality to retrieve string descriptors
#
# Author: follower@rancidbacon.com
#
# Version: 20091022
#

import usb # 1.0 not 0.4


def getStringDescriptor(device, index):
    """
    """
    response = device.ctrl_transfer(usb.util.ENDPOINT_IN,
                                    usb.legacy.REQ_GET_DESCRIPTOR,
                                    (usb.util.DESC_TYPE_STRING << 8) | index,
                                    0, # language id
                                    255) # length

    # TODO: Refer to 'libusb_get_string_descriptor_ascii' for error handling
    
    return response[2:].tostring().decode('utf-16')


REQUEST_TYPE_SEND = usb.util.build_request_type(usb.util.CTRL_OUT,
                                                usb.util.CTRL_TYPE_CLASS,
                                                usb.util.CTRL_RECIPIENT_DEVICE)

REQUEST_TYPE_RECEIVE = usb.util.build_request_type(usb.util.CTRL_IN,
                                                usb.util.CTRL_TYPE_CLASS,
                                                usb.util.CTRL_RECIPIENT_DEVICE)

USBRQ_HID_GET_REPORT = 0x01
USBRQ_HID_SET_REPORT = 0x09
USB_HID_REPORT_TYPE_FEATURE = 0x03


class ArduinoUsbDevice(object):
    """
    """
    
    def __init__(self, idVendor, idProduct):
        """
        """
        self.idVendor = idVendor
        self.idProduct = idProduct

        # TODO: Make more compliant by checking serial number also.
        self.device = usb.core.find(idVendor=self.idVendor,
                                    idProduct=self.idProduct)

        if not self.device:
            raise Exception("Device not found")


    def write(self, byte):
        """
        """
        # TODO: Return bytes written?
        #print "Write:"+str(byte)
        self._transfer(REQUEST_TYPE_SEND, USBRQ_HID_SET_REPORT,
                       byte,
                       []) # ignored


    def read(self):
        """
        """
        response = self._transfer(REQUEST_TYPE_RECEIVE, USBRQ_HID_GET_REPORT,
                              0, # ignored
                              1) # length

        if not response:
            raise Exception("No Data")
        
        return response[0]
    

    def _transfer(self, request_type, request, index, value):
        """
        """
        return self.device.ctrl_transfer(request_type, request,
                                        (USB_HID_REPORT_TYPE_FEATURE << 8) | 0,
                                         index,
                                         value)


    @property
    def productName(self):
        """
        """
        return getStringDescriptor(self.device, self.device.iProduct)

    
    @property
    def manufacturer(self):
        """
        """
        return getStringDescriptor(self.device, self.device.iManufacturer)

Python代码输出 #

Hello World
Hello World
Hello World
Hello World
Hello World
Hello World

问题分析与修复 #

对比两段代码的USB控制传输参数，发现核心问题：
Rust代码中read_control的value参数设置为0x00，但Python代码的_transfer方法里，该参数是(USB_HID_REPORT_TYPE_FEATURE << 8) | 0，即0x0300。DigiUSB基于HID特征报告传输数据，必须指定这个正确的报告类型标识，设备才能返回完整的数据流。

另外，Rust代码每次循环都重新打开设备句柄，虽然不直接导致错误，但会增加资源开销，可优化为只打开一次。

修复后的Rust代码 #

use std::{io::Write, process::exit, thread::sleep};

use rusb::{request_type, Device, DeviceList, Direction, GlobalContext, Recipient, RequestType};

fn main() -> rusb::Result<()> {
    // 查找目标USB设备
    let device = DeviceList::new()?
        .iter()
        .find(|dev| {
            let desc = dev.device_descriptor().unwrap();
            desc.vendor_id() == 0x16c0 && desc.product_id() == 0x05df
        })
        .unwrap_or_else(|| {
            println!("设备未找到");
            exit(0);
        });

    // 仅打开一次设备句柄，避免重复操作
    let mut handle = device.open()?;

    let request_type = request_type(Direction::In, RequestType::Class, Recipient::Device);
    const REQUEST: u8 = 0x01;
    // 正确的value参数：HID特征报告类型标识
    const VALUE: u16 = 0x0300;
    const INDEX: u16 = 0x00;
    let mut buf = [0u8; 1];

    loop {
        match handle.read_control(
            request_type,
            REQUEST,
            VALUE,
            INDEX,
            &mut buf,
            std::time::Duration::from_secs(1),
        ) {
            Ok(_) => {
                print!("{}", char::from(buf[0]));
                std::io::stdout().flush().expect("刷新输出缓冲区失败");
            }
            Err(_) => {
                sleep(std::time::Duration::from_millis(500));
            }
        }
    }
}

修复说明 #

1. 修正核心参数：将value改为0x0300，匹配Python代码中HID特征报告的类型要求，确保设备能正确返回数据。
2. 优化设备句柄：把设备打开操作移到循环外，减少重复连接的开销和潜在问题。
3. 简化代码逻辑：使用find方法简化设备查找过程，提升可读性。

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