我有一个以通用批量模式运行的USB设备 . 最后查看我的原型代码 . 该设备是一个数据采集单元，可以10kHz 9通道（每个2字节）流式传输数据，使其达到每秒180 000字节 . 一旦发出命令start，它就会启动流数据 . 我需要读取该数据而不是跳过任何字节 .
现在我看到缓冲区溢出，因为我在流中松开了注册，我再也无法解析它了 .
这里有几个问题：1）如何在Linux端控制缓冲区的大小？对于串行（pyserial 3.4）通信，我可以将缓冲区的大小更改为我想要的任何内容 . 通过这种方式，我可以确保我拥有以后需要的所有数据 . 2）或者我如何确保PyUSB库有足够的CPU汁来执行此任务？如果我可以将足够的CPU注意力集中在这个过程中，我可以解决这个问题 .
现在，我使用 dev = usb.core.find() 函数来获取有关设备的信息：
<DEVICE ID 0683:4108 on Bus 002 Address 016>
我设置了内部缓冲区大小（至少我认为这样做）
epi.wMaxPacketSize = 7200000 # ~7.2 MB
后来我用read函数读取缓冲区的1/10（大小= ~0.72 MB）
dev.read(endpoint, size, timeout)
我的理解是，有一种内部过程（缓冲区）不断从USB设备读取数据 . 缓冲区大小为7.2 M字节，填充它需要3秒钟 . 但是，它有时会发生得更快 .
网上没有太多的描述 . 如果有人可以帮助解决这个问题，我会很高兴 .
import psutil, os
p = psutil.Process(os.getpid())
p.nice(0)
from pdb import pm
import usb.core
import usb.util
from time import sleep,time,clock
# find our device
#dev = usb.core.find(idVendor=0x0403, idProduct=0xFAF0, address = k) #ThorLabs
dev = usb.core.find(idVendor=0x0683, idProduct=0x4108)
# was it found?
if dev is None:
raise ValueError('Device not found')
dev.reset()
# set the active configuration. With no arguments, the first
# configuration will be the active one
dev.set_configuration()
# get an endpoint instance
cfg = dev.get_active_configuration()
intf = cfg[(0,0)]
epo = usb.util.find_descriptor(
intf,
# match the first OUT endpoint
custom_match = \
lambda e: \
usb.util.endpoint_direction(e.bEndpointAddress) == \
usb.util.ENDPOINT_OUT)
epi = usb.util.find_descriptor(
intf,
# match the first IN endpoint
custom_match = \
lambda e: \
usb.util.endpoint_direction(e.bEndpointAddress) == \
usb.util.ENDPOINT_IN)
assert epo is not None
assert epi is not None
epi.wMaxPacketSize = 7200000
epo.wMaxPacketSize = 7200000
def write(command):
try:
dev.write(epo,command)
except Exception as err:
print(err)
def inquiry(command,buffer):
write(command)
res = read(buffer)
return res
def read(buffer, timeout = 1000):
string = ""
try:
data = dev.read(epi,buffer,timeout)
except Exception as err:
data = ''
#print(len(data))
if len(data) != 0:
for i in data:
string += chr(i)
return string
def run_once(buffer):
from time import time
global string
global running
write('start \r')
tstart = time()
while running:
t = time()
string = read(buffer)
print('%r, %r, %r, %r' % (string[-2:],len(string),round(time()-t,3),round(time() - tstart,2)))
if 'stop' in string:
running = False
#print(time()-t)
print('done')
print('%r' % inquiry('stop \r',buffer))
def run():
from time import clock, time
from thread import start_new_thread
global running
global string
string = ''
running = True
buffer = epi.wMaxPacketSize/100
start_new_thread(run_once,(buffer,))
print('Device serial number: %r' % dev.serial_number)
print('Product description: %r' % dev.product)
print('dev.write(epo,"led 1")')
print('dev.write(epo,"stop\r")')
print('dev.write(epo,"info 0")')
print('dev.read(epi,1,1000)')
print('usb.util.dispose_resources(dev)')
print('dev.write(epo,"start 0 "); res = dev.read(epi,1000000,1000); len(res)')
print("dev.write(epo,'string = 'srate ' + str(60000000/60000000) +'\r'")
buffer = epi.wMaxPacketSize/100
print('%r' % inquiry('stop \r',buffer))
print('%r' % inquiry('endo 127 \r',buffer))
print('%r' % inquiry('dout 100 \r',buffer))
print('%r' % inquiry('slist 0 256 \r',buffer))
print('%r' % inquiry('slist 1 257 \r',buffer))
print('%r' % inquiry('slist 2 1282 \r',buffer))
print('%r' % inquiry('slist 3 1283 \r',buffer))
print('%r' % inquiry('slist 4 1284 \r',buffer))
print('%r' % inquiry('slist 5 1285 \r',buffer))
print('%r' % inquiry('slist 6 262 \r',buffer))
print('%r' % inquiry('slist 7 263 \r',buffer))
print('%r' % inquiry('slist 8 8 \r',buffer))
print('%r' % inquiry('info 0 \r',buffer))
print('%r' % inquiry('srate '+str(60000000/10000)+ ' \r',buffer))
run()