USB DAQ 协议与上位机：分帧、Python 与 WinUSB

学习教程更新于 2026-10-02编辑本文

本页定义一个用于学习的 USB DAQ 协议，并给出 Python 编解码、PyUSB 访问和 C++ WinUSB 接入方法。它承接 STM32 USB DAQ 设计。

这是重新明确过的 VitaLogos DAQ 示例协议 v1，并不是某个商业采集卡协议，也不声称兼容旧草案中不完整的 64 字节命令包。固件和主机必须同时实现本页的字段约定才能互通。下面的离线测试不需要 USB 硬件；真实设备枚举、传输和采样仍须上板验证。

1. 为什么要有应用帧

USB 保证的是相应传输机制，不会替应用定义“这 4 字节是采样率”。一次读操作也不能当成一个完整业务帧。应用协议至少应约定：

消息种类、协议版本、长度和字节序。
命令与响应的关联方法。
数据流中的通道、样本数量、顺序和采集会话。
非法参数、未实现功能、超时和溢出如何报告。

原例把控制应答和采样数据都发到同一个 IN 端点，却没有统一的消息类型。主机可能把样本首字节当作“启动成功”。本协议用类型和序号分流；同一 IN 端点在应用层只应由一个读取者负责，再分发给不同业务。

2. 帧头：固定 16 字节，小端序

所有多字节整数使用 little-endian。字段宽度是协议的一部分，不依赖 C/C++ 编译器的结构体填充。

偏移	字节数	字段	含义
0	2	magic	ASCII `DQ`，即 `44 51`
2	1	version	本页为 1
3	1	kind	1 命令，2 响应，3 样本
4	4	sequence	命令流水号；响应回显；样本在会话内独立递增
8	2	code	命令/响应的命令码；样本固定为 0
10	2	payload_length	不含帧头的长度，0～496
12	4	session	采集会话号，未开始采集的普通查询为 0

一帧最多 512 字节，是本示例的应用层选择。它可以通过 FS 的多个 USB 包传输，并不要求只能用 HS。接收器在读到完整帧头后检查长度，再等待剩余字节；一次读取收到多帧时逐个解析。

新开始一次采集必须分配新的非零 session。主机不把前一次采集的残留样本混入新结果；当前连接内不能在仍可能存在旧数据时复用会话号。USB 重连后重新查询并建立新会话。

3. 命令、状态与载荷

命令码	名称	命令载荷	行为
`0x01`	GET_INFO	空	查询已实现的硬件能力
`0x10`	AI_CONFIG	8 字节配置	校验并应用采样配置
`0x11`	AI_START	空	启动；成功响应携带新会话号
`0x12`	AI_STOP	空	停止当前会话，按固件约定处理尾部数据
`0x13`	AI_READ	由扩展定义	本页连续推送方式不实现，返回 unsupported
`0x20`	AO_WRITE	由扩展定义	没有模拟输出实现则返回 unsupported
`0x30`～`0x32`	DIO_CONFIG/READ/WRITE	由扩展定义	没有数字I/O实现则返回 unsupported

响应帧的 code 与 sequence 回显对应命令。载荷第一个字节为状态：0 成功、1 参数或长度错误、2 状态不允许、3 功能未实现、4 暂时忙、5 硬件故障。失败不能伪装成空的成功响应。

3.1 采样配置

按 <BBIH 编码，共 8 字节：

字段	类型	约定
channels	uint8	有效通道数，示例允许 1～16；实际还受设备能力限制
range_code	uint8	设备定义的量程编号，先查 GET_INFO/设备量程表
sample_rate	uint32	每通道目标采样率，Hz，必须大于0
samples_per_channel	uint16	有限采集点数；0表示连续采集

例如 1 通道、量程编号0、10000 Hz、1000点的载荷为：

01 00 10 27 00 00 E8 03

旧 Python 例中的 <BBIHH 会产生10字节，而旧 C 结构只有8字节。本协议没有额外的隐式 padding。

3.2 设备信息

GET_INFO 成功状态字节之后，可使用下面的固定 52 字节信息块：名称32字节（UTF-8，剩余填0）、固件版本4字节、AI/AO/DI/DO通道数各uint16、最高总采样率uint32、功能位图uint32。若设备实际只有 AI，就把其他通道数/功能位写0。

名称必须按32字节有界解析，不应直接假设来自设备的数组总有字符串终止符。量程表、校准系数和扩展信息应通过后续明确版本的扩展查询获取，不能依靠主机猜测。

3.3 样本帧

样本载荷先放12字节元信息，格式为 <BBHQ：

字段	类型	约定
channel	uint8	从0开始的通道编号
format	uint8	本页1表示小端无符号16bit样本
count	uint16	后续样本数量，1～242
first_sample_index	uint64	该通道本次采集中的首样本序号

其后严格为 count × 2 字节。容量推导是：

N_{\max}=\left\lfloor\frac{512-16-12}{2}\right\rfloor=242.

帧序号用于观察消息流连续性；每通道样本序号用于发现通道内缺样。发现缺口时保留原序号并报告，不能把前后两段悄悄拼接成等间隔连续波形。位宽、格式和量程转换参见固件设计中的 ADC 转换。

4. Python 编解码：可离线运行

保存为 daq_protocol.py。代码只依赖 Python 标准库，执行文件会验证固定字节样例、分片、多帧及非法长度。

from dataclasses import dataclass
import struct

HEADER = struct.Struct('<2sBBIHHI')
CONFIG = struct.Struct('<BBIH')
SAMPLE_META = struct.Struct('<BBHQ')
MAX_PAYLOAD = 496
MAX_SAMPLES = 242
COMMAND, RESPONSE, SAMPLES = 1, 2, 3


@dataclass(frozen=True)
class Frame:
    kind: int
    sequence: int
    code: int
    session: int
    payload: bytes


def encode(frame):
    if frame.kind not in (COMMAND, RESPONSE, SAMPLES):
        raise ValueError('unknown message kind')
    if len(frame.payload) > MAX_PAYLOAD:
        raise ValueError('payload exceeds 496 bytes')
    return HEADER.pack(b'DQ', 1, frame.kind, frame.sequence,
                       frame.code, len(frame.payload), frame.session) + frame.payload


class Decoder:
    def __init__(self):
        self.buffer = bytearray()

    def feed(self, chunk):
        # A bounded USB read is expected; reject unreasonable accumulation.
        if len(self.buffer) + len(chunk) > 65536:
            raise ValueError('receive buffer limit exceeded')
        self.buffer.extend(chunk)
        result = []
        while len(self.buffer) >= HEADER.size:
            magic, version, kind, seq, code, size, session = HEADER.unpack_from(self.buffer)
            if magic != b'DQ' or version != 1:
                raise ValueError('invalid magic or protocol version')
            if kind not in (COMMAND, RESPONSE, SAMPLES) or size > MAX_PAYLOAD:
                raise ValueError('invalid kind or payload length')
            end = HEADER.size + size
            if len(self.buffer) < end:
                break
            payload = bytes(self.buffer[HEADER.size:end])
            del self.buffer[:end]
            result.append(Frame(kind, seq, code, session, payload))
        return result

    def finish(self):
        if self.buffer:
            raise ValueError('stream ended inside a frame')


def encode_config(channels, range_code, sample_rate, samples_per_channel=0):
    if not 1 <= channels <= 16:
        raise ValueError('channel count must be 1..16')
    if not 0 <= range_code <= 255 or not 1 <= sample_rate <= 0xffffffff:
        raise ValueError('invalid range code or sample rate')
    if not 0 <= samples_per_channel <= 65535:
        raise ValueError('sample count does not fit uint16')
    return CONFIG.pack(channels, range_code, sample_rate, samples_per_channel)


def sample_frame(sequence, session, channel, first_index, values):
    values = tuple(values)
    if not 1 <= len(values) <= MAX_SAMPLES or not 0 <= channel <= 255:
        raise ValueError('invalid sample count or channel')
    if session == 0:
        raise ValueError('samples require an acquisition session')
    payload = SAMPLE_META.pack(channel, 1, len(values), first_index)
    payload += struct.pack('<' + 'H' * len(values), *values)
    return Frame(SAMPLES, sequence, 0, session, payload)


def decode_samples(frame):
    if frame.kind != SAMPLES or frame.code != 0 or frame.session == 0:
        raise ValueError('not a valid sample frame')
    if len(frame.payload) < SAMPLE_META.size:
        raise ValueError('sample metadata truncated')
    channel, fmt, count, first = SAMPLE_META.unpack_from(frame.payload)
    if fmt != 1 or not 1 <= count <= MAX_SAMPLES:
        raise ValueError('unsupported sample format or count')
    if len(frame.payload) != SAMPLE_META.size + count * 2:
        raise ValueError('sample count/length mismatch')
    values = struct.unpack_from('<' + 'H' * count, frame.payload, SAMPLE_META.size)
    return channel, first, values


def require_success(frame, sequence, code):
    if frame.kind != RESPONSE or frame.sequence != sequence or frame.code != code:
        raise ValueError('response does not match the request')
    if not frame.payload:
        raise ValueError('response has no status byte')
    if frame.payload[0] != 0:
        raise RuntimeError(f'device error {frame.payload[0]}')
    return frame.payload[1:]


def self_test():
    cfg = encode_config(1, 0, 10000, 1000)
    assert cfg.hex() == '010010270000e803'
    command = Frame(COMMAND, 7, 0x10, 0, cfg)
    wire = encode(command)
    assert wire.hex() == '44510101070000001000080000000000010010270000e803'
    data = sample_frame(0, 42, 0, 100, (0, 32768, 65535))
    stream = wire + encode(data)
    decoder = Decoder()
    frames = []
    for byte in stream:  # A frame split at every possible byte boundary.
        frames.extend(decoder.feed(bytes((byte,))))
    decoder.finish()
    assert frames == [command, data]
    assert decode_samples(data) == (0, 100, (0, 32768, 65535))
    assert Decoder().feed(stream) == [command, data]
    assert len(encode(sample_frame(1, 42, 0, 103, range(242)))) == 512

    def rejects(action):
        try:
            action()
        except (ValueError, struct.error):
            return
        raise AssertionError('invalid input was accepted')

    rejects(lambda: encode_config(0, 0, 1000))
    rejects(lambda: sample_frame(0, 1, 0, 0, range(243)))
    rejects(lambda: Decoder().feed(HEADER.pack(b'DQ', 1, 1, 0, 0, 497, 0)))
    rejects(lambda: Decoder().feed(b'XX' + wire[2:]))
    rejects(lambda: decode_samples(Frame(SAMPLES, 0, 0, 1, data.payload[:-1])))
    partial = Decoder()
    partial.feed(wire[:-1])
    rejects(partial.finish)
    print('DAQ protocol tests passed')


if __name__ == '__main__':
    self_test()

这里把错误当成连接/协议故障交给调用者处理，没有在未知字节流中猜测恢复位置。实际应用应在重新建立会话时清空解析器，记录错误并重新同步。

5. PyUSB 访问：枚举、端点、超时与释放

PyUSB 是 Python API；它还需要可用的 USB 后端。Windows 上设备接口的驱动绑定、libusb 后端与 Python 包是不同层次。参考 PyUSB 官方教程。

下面保存为 usb_transport.py，与前面的文件放在一起。它只针对一个已配置的自定义接口；VID/PID、接口号、备用设置必须来自自己的设备，不能照搬他人标识。

import time
from collections import deque
from daq_protocol import Decoder, Frame, COMMAND, RESPONSE, SAMPLES, encode, require_success


class UsbTransport:
    def __init__(self, vid, pid, interface=0, alternate=0):
        import usb.core
        import usb.util
        self.core, self.util = usb.core, usb.util
        self.interface = interface
        self.dev = usb.core.find(idVendor=vid, idProduct=pid)
        if self.dev is None:
            raise RuntimeError('USB device not found')
        self.claimed = False
        self.decoder = Decoder()
        self.frames = deque()
        try:
            # This example selects the device default configuration.
            self.dev.set_configuration()
            usb.util.claim_interface(self.dev, interface)
            self.claimed = True
            self.dev.set_interface_altsetting(interface=interface, alternate_setting=alternate)
            itf = self.dev.get_active_configuration()[(interface, alternate)]
            bulk = [ep for ep in itf if usb.util.endpoint_type(ep.bmAttributes) == usb.util.ENDPOINT_TYPE_BULK]
            incoming = [ep for ep in bulk if usb.util.endpoint_direction(ep.bEndpointAddress) == usb.util.ENDPOINT_IN]
            outgoing = [ep for ep in bulk if usb.util.endpoint_direction(ep.bEndpointAddress) == usb.util.ENDPOINT_OUT]
            if len(incoming) != 1 or len(outgoing) != 1:
                raise RuntimeError('expected exactly one bulk IN and one bulk OUT')
            self.ep_in, self.ep_out = incoming[0], outgoing[0]
        except Exception:
            self.close()
            raise

    def send(self, frame, timeout_ms=1000):
        packet = encode(frame)
        written = self.ep_out.write(packet, timeout=timeout_ms)
        if written != len(packet):
            raise IOError('short USB write; restart the protocol session')

    def receive(self, deadline):
        while not self.frames:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError('application deadline exceeded')
            timeout_ms = max(1, min(100, int(remaining * 1000)))
            try:
                chunk = self.ep_in.read(4096, timeout=timeout_ms)
            except self.core.USBTimeoutError:
                continue
            self.frames.extend(self.decoder.feed(bytes(chunk)))
        return self.frames.popleft()

    def exchange(self, sequence, code, payload=b'', session=0, on_samples=None, timeout_s=2):
        self.send(Frame(COMMAND, sequence, code, session, payload))
        deadline = time.monotonic() + timeout_s
        while True:
            frame = self.receive(deadline)
            if frame.kind == SAMPLES:
                if on_samples is None:
                    raise RuntimeError('unexpected sample stream during command exchange')
                on_samples(frame)
                continue
            require_success(frame, sequence, code)
            return frame

    def close(self):
        if self.dev is not None:
            try:
                if self.claimed:
                    self.util.release_interface(self.dev, self.interface)
            finally:
                self.util.dispose_resources(self.dev)
                self.dev = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, traceback):
        self.close()

time.monotonic() 适合截止时间计算，因为它不会因系统时钟校正而倒退；其单位是秒。USB API 的 timeout 参数使用毫秒，代码中显式转换。Python：monotonic 时钟

这个例子没有自动从系统驱动夺取接口。若接口被内核驱动占用，应先确认选中的接口就是自定义 DAQ 功能，再按平台进行适当的驱动/权限配置。配置失败不应简单打印后继续通信。

5.1 有限与连续采集

业务流程是 GET_INFO → AI_CONFIG → AI_START → 读取样本 → AI_STOP → 释放资源。每次命令采用新的序号；启动响应中的会话号用于过滤样本。连续采集把 samples_per_channel 设为0，并由应用明确停止。

主机应维护每个通道的 expected_sample_index。收到样本后先验证会话、通道、格式和序号，再追加数据；若 first_sample_index 不等于预期，记录缺样并决定终止还是保存带间隙结果。内存队列必须有上限，长时间采集应流式写入文件。

停止也可能与尾部样本交错，因此给 exchange() 提供同一个样本处理函数。不要在另一个线程同时直接读取相同 IN 端点，否则响应和数据会被不同消费者随机取走。

得到有效数组后，可以计算样本均值与总体标准差：

\bar{x}=\frac1N\sum_{i=0}^{N-1}x_i,\qquad \sigma=\sqrt{\frac1N\sum_{i=0}^{N-1}(x_i-\bar{x})^2}.

需要估计总体方差的无偏估计时应使用 $N-1$ 分母，并要求 $N>1$ 。这两种统计口径不能混用。绘图横轴应由样本序号和实际采样率构造；保存 CSV 时保留 session、channel、sample_index、原始码值及校准版本，不能只保存一个失去上下文的 Voltage 列。

6. C++：显式编码与 WinUSB 资源管理

C++ 也按同一字段顺序编码。不要通过 *(uint32_t*)&buffer[2] 写入采样率：这依赖对齐、别名和宿主字节序。下面的独立 C++17 程序生成与 Python 相同的配置帧，可先在没有设备时编译运行。

#include <cassert>
#include <cstdint>
#include <iomanip>
#include <iostream>
#include <stdexcept>
#include <vector>

void appendLE(std::vector<std::uint8_t>& out, std::uint64_t n, unsigned bytes) {
    for (unsigned i = 0; i < bytes; ++i) {
        out.push_back(static_cast<std::uint8_t>((n >> (8U * i)) & 0xffU));
    }
}

std::vector<std::uint8_t> configCommand(std::uint32_t sequence,
    std::uint8_t channels, std::uint8_t range, std::uint32_t rate, std::uint16_t count) {
    if (channels == 0 || channels > 16 || rate == 0) {
        throw std::invalid_argument("invalid acquisition configuration");
    }
    std::vector<std::uint8_t> out{'D', 'Q', 1, 1};
    appendLE(out, sequence, 4);
    appendLE(out, 0x10, 2);
    appendLE(out, 8, 2);
    appendLE(out, 0, 4);
    out.push_back(channels);
    out.push_back(range);
    appendLE(out, rate, 4);
    appendLE(out, count, 2);
    return out;
}

int main() {
    const auto packet = configCommand(7, 1, 0, 10000, 1000);
    const std::vector<std::uint8_t> expected{
        0x44,0x51,1,1,7,0,0,0,0x10,0,8,0,0,0,0,0,
        1,0,0x10,0x27,0,0,0xe8,3};
    assert(packet == expected);
    for (auto byte : packet) {
        std::cout << std::hex << std::setfill('0') << std::setw(2) << unsigned(byte);
    }
    std::cout << '\n';
}

6.1 Windows 接入顺序

步骤	API / 资源	处理要点
枚举应用接口	SetupDiGetClassDevs / SetupDiEnumDeviceInterfaces	使用设备注册的接口 GUID，不是 USBDevice 安装类 GUID
获取设备路径	SetupDiGetDeviceInterfaceDetail	先查询所需长度，再分配缓冲；区分无设备与API失败
打开文件	CreateFileW	使用获取的路径，按WinUSB要求以重叠I/O标志打开
初始化接口	WinUsb_Initialize	失败时仍需释放已打开的文件句柄
查询端点	WinUsb_QueryInterfaceSettings / WinUsb_QueryPipe	检查方向和Bulk类型，不把0x01/0x81当所有设备固定地址
收发	WinUsb_WritePipe / WinUsb_ReadPipe	检查返回值和实际字节数，协议解码保留未完整部分
释放	WinUsb_Free / CloseHandle	用RAII保证异常和提前返回路径也释放

Microsoft 的示例说明了设备路径、接口句柄、端点查询与读写的关系。Microsoft：使用 WinUSB 访问设备

下面的传输类接受已取得的设备路径、端点和接口；只负责一次读写，不冒充完整的业务 SDK。上层仍需实现与 Python 相同的流式解析和消息分发。

#include <windows.h>
#include <winusb.h>
#include <cstdint>
#include <limits>
#include <stdexcept>
#include <string>
#include <system_error>
#include <vector>

class UsbPipe {
    HANDLE file_ = INVALID_HANDLE_VALUE;
    WINUSB_INTERFACE_HANDLE usb_ = nullptr;
    UCHAR in_ = 0, out_ = 0;

    [[noreturn]] static void fail(const char* operation) {
        throw std::system_error(static_cast<int>(GetLastError()),
                                std::system_category(), operation);
    }
public:
    explicit UsbPipe(const std::wstring& devicePath) {
        file_ = CreateFileW(devicePath.c_str(), GENERIC_READ | GENERIC_WRITE,
            FILE_SHARE_READ | FILE_SHARE_WRITE, nullptr, OPEN_EXISTING,
            FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, nullptr);
        if (file_ == INVALID_HANDLE_VALUE) fail("CreateFileW");
        try {
            if (!WinUsb_Initialize(file_, &usb_)) fail("WinUsb_Initialize");
            USB_INTERFACE_DESCRIPTOR descriptor{};
            if (!WinUsb_QueryInterfaceSettings(usb_, 0, &descriptor)) fail("QueryInterface");
            unsigned inCount = 0, outCount = 0;
            for (UCHAR i = 0; i < descriptor.bNumEndpoints; ++i) {
                WINUSB_PIPE_INFORMATION pipe{};
                if (!WinUsb_QueryPipe(usb_, 0, i, &pipe)) fail("QueryPipe");
                if (pipe.PipeType != UsbdPipeTypeBulk) continue;
                if ((pipe.PipeId & 0x80U) != 0) { in_ = pipe.PipeId; ++inCount; }
                else { out_ = pipe.PipeId; ++outCount; }
            }
            if (inCount != 1 || outCount != 1) {
                throw std::runtime_error("expected one bulk endpoint in each direction");
            }
            ULONG timeout = 1000;
            if (!WinUsb_SetPipePolicy(usb_, in_, PIPE_TRANSFER_TIMEOUT,
                                     sizeof(timeout), &timeout)) fail("IN timeout");
            if (!WinUsb_SetPipePolicy(usb_, out_, PIPE_TRANSFER_TIMEOUT,
                                     sizeof(timeout), &timeout)) fail("OUT timeout");
        } catch (...) {
            if (usb_) WinUsb_Free(usb_);
            CloseHandle(file_);
            throw;
        }
    }
    UsbPipe(const UsbPipe&) = delete;
    UsbPipe& operator=(const UsbPipe&) = delete;
    ~UsbPipe() {
        if (usb_) WinUsb_Free(usb_);
        if (file_ != INVALID_HANDLE_VALUE) CloseHandle(file_);
    }
    void write(const std::vector<std::uint8_t>& data) {
        if (data.size() > (std::numeric_limits<ULONG>::max)()) {
            throw std::length_error("USB request too large");
        }
        ULONG written = 0;
        if (!WinUsb_WritePipe(usb_, out_, const_cast<PUCHAR>(data.data()),
            static_cast<ULONG>(data.size()), &written, nullptr)) fail("WritePipe");
        if (written != data.size()) throw std::runtime_error("short USB write");
    }
    std::vector<std::uint8_t> read() {
        std::vector<std::uint8_t> data(4096);
        ULONG received = 0;
        if (!WinUsb_ReadPipe(usb_, in_, data.data(), static_cast<ULONG>(data.size()),
                             &received, nullptr)) fail("ReadPipe");
        data.resize(received);
        return data;
    }
};

使用 MSVC 时链接 winusb.lib；使用支持 Windows SDK 接口的其他工具链时，链接对应 WinUSB 库。枚举设备还需 SetupAPI。这个类使用同步等待的读写调用，应该放在工作线程，并由应用统一管理截止时间和取消；不要在GUI线程里阻塞一秒。多设备、复合设备关联接口、热插拔与异步吞吐优化需要在此基础上补全。

7. 驱动安装与验证范围

开发环境可以使用针对选定接口的 WinUSB 绑定工具；产品交付应按目标系统选择 Microsoft OS 描述符或自定义 INF/签名目录。系统安装类 GUID、应用接口 GUID、VID/PID 解决的是不同问题，必须分别配置。Microsoft：WinUSB 安装

本页验证分为三层：标准库 Python 编解码与异常输入、C++ 固定字节输出、Windows 传输类的编译检查。它们不能证明某块板已经枚举成功或ADC数据准确。上板报告还应记录具体硬件、固件提交、驱动、USB速度、传输持续时间、丢样和校准结果。

原始来源：USB DAQ 草案。旧草案中的命令、采样配置、Python测试、C++句柄管理、统计/CSV和驱动接入均在对应章节保留其用途，并修订字段边界和未验证的运行承诺。