0
# Synchronous USB Transfers
1
2
Synchronous transfers provide blocking I/O operations for control, bulk, and interrupt endpoints. These methods are simpler to use than asynchronous transfers but block the calling thread until completion or timeout. They're ideal for simple request-response patterns and low-throughput applications.
3
4
## Capabilities
5
6
### Control Transfers
7
8
Control transfers handle device configuration, status queries, and vendor-specific commands using the standard USB control transfer format with setup packet and optional data phase.
9
10
```python { .api }
11
class USBDeviceHandle:
12
def controlRead(self, request_type, request, value, index, length, timeout=0):
13
"""
14
Perform synchronous control read transfer.
15
16
Args:
17
request_type (int): Request type bitmask (TYPE_* | RECIPIENT_* | ENDPOINT_IN)
18
request (int): Request ID (vendor-specific or standard)
19
value (int): Request-specific value parameter
20
index (int): Request-specific index parameter (often interface/endpoint)
21
length (int): Maximum number of bytes to read
22
timeout (int): Timeout in milliseconds (0 = no timeout)
23
24
Returns:
25
bytes: Received data (may be shorter than length)
26
27
Raises:
28
USBErrorTimeout: Transfer timed out
29
USBError: Other transfer errors
30
"""
31
32
def controlWrite(self, request_type, request, value, index, data, timeout=0):
33
"""
34
Perform synchronous control write transfer.
35
36
Args:
37
request_type (int): Request type bitmask (TYPE_* | RECIPIENT_* | ENDPOINT_OUT)
38
request (int): Request ID (vendor-specific or standard)
39
value (int): Request-specific value parameter
40
index (int): Request-specific index parameter (often interface/endpoint)
41
data (bytes): Data to send (use writable buffer like bytearray to avoid copies)
42
timeout (int): Timeout in milliseconds (0 = no timeout)
43
44
Returns:
45
int: Number of bytes actually sent
46
47
Raises:
48
USBErrorTimeout: Transfer timed out
49
USBError: Other transfer errors
50
"""
51
```
52
53
### Bulk Transfers
54
55
Bulk transfers provide reliable, large-volume data transfer for endpoints that can tolerate variable latency but require error-free delivery.
56
57
```python { .api }
58
class USBDeviceHandle:
59
def bulkRead(self, endpoint, length, timeout=0):
60
"""
61
Perform synchronous bulk read transfer.
62
63
Args:
64
endpoint (int): Bulk IN endpoint address (with ENDPOINT_IN bit set)
65
length (int): Maximum number of bytes to read
66
timeout (int): Timeout in milliseconds (0 = no timeout)
67
68
Returns:
69
bytes: Received data (may be shorter than length)
70
71
Raises:
72
USBErrorTimeout: Transfer timed out (exception has .received property)
73
USBError: Other transfer errors
74
"""
75
76
def bulkWrite(self, endpoint, data, timeout=0):
77
"""
78
Perform synchronous bulk write transfer.
79
80
Args:
81
endpoint (int): Bulk OUT endpoint address (with ENDPOINT_OUT bit set)
82
data (bytes): Data to send (use writable buffer like bytearray to avoid copies)
83
timeout (int): Timeout in milliseconds (0 = no timeout)
84
85
Returns:
86
int: Number of bytes actually sent
87
88
Raises:
89
USBErrorTimeout: Transfer timed out (exception has .transferred property)
90
USBError: Other transfer errors
91
"""
92
```
93
94
### Interrupt Transfers
95
96
Interrupt transfers provide low-latency, periodic data transfer with guaranteed maximum latency for time-sensitive applications like input devices.
97
98
```python { .api }
99
class USBDeviceHandle:
100
def interruptRead(self, endpoint, length, timeout=0):
101
"""
102
Perform synchronous interrupt read transfer.
103
104
Args:
105
endpoint (int): Interrupt IN endpoint address (with ENDPOINT_IN bit set)
106
length (int): Maximum number of bytes to read
107
timeout (int): Timeout in milliseconds (0 = no timeout)
108
109
Returns:
110
bytes: Received data (may be shorter than length)
111
112
Raises:
113
USBErrorTimeout: Transfer timed out (exception has .received property)
114
USBError: Other transfer errors
115
"""
116
117
def interruptWrite(self, endpoint, data, timeout=0):
118
"""
119
Perform synchronous interrupt write transfer.
120
121
Args:
122
endpoint (int): Interrupt OUT endpoint address (with ENDPOINT_OUT bit set)
123
data (bytes): Data to send (use writable buffer like bytearray to avoid copies)
124
timeout (int): Timeout in milliseconds (0 = no timeout)
125
126
Returns:
127
int: Number of bytes actually sent
128
129
Raises:
130
USBErrorTimeout: Transfer timed out (exception has .transferred property)
131
USBError: Other transfer errors
132
"""
133
```
134
135
## Usage Examples
136
137
### Standard Control Requests
138
139
```python
140
import usb1
141
142
with usb1.USBContext() as context:
143
device = context.getByVendorIDAndProductID(0x1234, 0x5678)
144
if device:
145
with device.open() as handle:
146
# Get device status (standard request)
147
try:
148
status = handle.controlRead(
149
request_type=usb1.TYPE_STANDARD | usb1.RECIPIENT_DEVICE | usb1.ENDPOINT_IN,
150
request=0x00, # GET_STATUS
151
value=0,
152
index=0,
153
length=2,
154
timeout=1000
155
)
156
print(f"Device status: {int.from_bytes(status, 'little')}")
157
158
except usb1.USBErrorTimeout:
159
print("Control request timed out")
160
except usb1.USBError as e:
161
print(f"Control request failed: {e}")
162
```
163
164
### Vendor-Specific Control Commands
165
166
```python
167
import usb1
168
169
def send_vendor_command(handle, command, data=None):
170
"""Send vendor-specific control command."""
171
try:
172
if data is None:
173
# Vendor command with no data (status/query)
174
response = handle.controlRead(
175
request_type=usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE | usb1.ENDPOINT_IN,
176
request=command,
177
value=0,
178
index=0,
179
length=64, # Maximum expected response
180
timeout=5000
181
)
182
return response
183
else:
184
# Vendor command with data
185
sent = handle.controlWrite(
186
request_type=usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE | usb1.ENDPOINT_OUT,
187
request=command,
188
value=0,
189
index=0,
190
data=data,
191
timeout=5000
192
)
193
return sent
194
195
except usb1.USBErrorTimeout:
196
print(f"Vendor command {command} timed out")
197
return None
198
except usb1.USBError as e:
199
print(f"Vendor command {command} failed: {e}")
200
return None
201
202
with usb1.USBContext() as context:
203
device = context.getByVendorIDAndProductID(0x1234, 0x5678)
204
if device:
205
with device.open() as handle:
206
# Query device version
207
version = send_vendor_command(handle, 0x01)
208
if version:
209
print(f"Device version: {version.hex()}")
210
211
# Send configuration data
212
config_data = b'\x42\x00\x01\x02'
213
result = send_vendor_command(handle, 0x02, config_data)
214
if result is not None:
215
print(f"Configuration sent, {result} bytes")
216
```
217
218
### Bulk Data Transfer
219
220
```python
221
import usb1
222
import time
223
224
def bulk_transfer_example(handle):
225
"""Demonstrate bulk transfer with error handling."""
226
bulk_out_ep = 0x02 # OUT endpoint
227
bulk_in_ep = 0x82 # IN endpoint (0x02 | ENDPOINT_IN)
228
229
try:
230
# Send data via bulk transfer
231
send_data = b"Hello, USB device!" * 10 # Some test data
232
sent = handle.bulkWrite(bulk_out_ep, send_data, timeout=5000)
233
print(f"Sent {sent} of {len(send_data)} bytes")
234
235
# Read response via bulk transfer
236
response = handle.bulkRead(bulk_in_ep, 1024, timeout=5000)
237
print(f"Received {len(response)} bytes: {response[:50]}...")
238
239
return True
240
241
except usb1.USBErrorTimeout as e:
242
if hasattr(e, 'transferred'):
243
print(f"Bulk write timed out after {e.transferred} bytes")
244
elif hasattr(e, 'received'):
245
print(f"Bulk read timed out with {len(e.received)} bytes received")
246
else:
247
print("Bulk transfer timed out")
248
return False
249
250
except usb1.USBErrorPipe:
251
print("Bulk endpoint stalled - clearing halt")
252
handle.clearHalt(bulk_out_ep)
253
handle.clearHalt(bulk_in_ep)
254
return False
255
256
except usb1.USBError as e:
257
print(f"Bulk transfer error: {e}")
258
return False
259
260
with usb1.USBContext() as context:
261
device = context.getByVendorIDAndProductID(0x1234, 0x5678)
262
if device:
263
with device.open() as handle:
264
with handle.claimInterface(0):
265
success = bulk_transfer_example(handle)
266
print(f"Bulk transfer {'succeeded' if success else 'failed'}")
267
```
268
269
### Interrupt Transfer for Input Device
270
271
```python
272
import usb1
273
import struct
274
275
def read_interrupt_data(handle, endpoint, packet_size, duration=10):
276
"""
277
Read interrupt data continuously for specified duration.
278
Typical for HID devices like mice, keyboards.
279
"""
280
print(f"Reading interrupt data for {duration} seconds...")
281
start_time = time.time()
282
packet_count = 0
283
284
while time.time() - start_time < duration:
285
try:
286
# Read interrupt data (non-blocking with short timeout)
287
data = handle.interruptRead(endpoint, packet_size, timeout=100)
288
packet_count += 1
289
290
# Parse HID data (example for mouse)
291
if len(data) >= 3:
292
buttons, dx, dy = struct.unpack('Bbb', data[:3])
293
if buttons or dx or dy: # Only print if there's activity
294
print(f"Packet {packet_count}: buttons=0x{buttons:02x}, dx={dx}, dy={dy}")
295
296
except usb1.USBErrorTimeout:
297
# Timeout is normal for interrupt endpoints with no data
298
continue
299
except usb1.USBError as e:
300
print(f"Interrupt read error: {e}")
301
break
302
303
print(f"Read {packet_count} interrupt packets")
304
305
with usb1.USBContext() as context:
306
# Find HID device (mouse example)
307
for device in context.getDeviceIterator(skip_on_error=True):
308
if device.getDeviceClass() == 3: # HID class
309
print(f"Found HID device: {device.getVendorID():04x}:{device.getProductID():04x}")
310
311
try:
312
with device.open() as handle:
313
# Detach kernel driver if needed (Linux)
314
if handle.kernelDriverActive(0):
315
handle.detachKernelDriver(0)
316
317
with handle.claimInterface(0):
318
# Read from interrupt IN endpoint (typically 0x81)
319
read_interrupt_data(handle, 0x81, 8, duration=5)
320
321
except usb1.USBError as e:
322
print(f"Error accessing HID device: {e}")
323
break
324
```
325
326
### High-Performance Bulk Transfer
327
328
```python
329
import usb1
330
import time
331
332
def high_speed_bulk_transfer(handle, endpoint_out, endpoint_in, chunk_size=64*1024):
333
"""
334
Demonstrate high-performance bulk transfer with large buffers.
335
"""
336
# Use bytearray for zero-copy operations
337
send_buffer = bytearray(chunk_size)
338
339
# Fill with test pattern
340
for i in range(len(send_buffer)):
341
send_buffer[i] = i & 0xFF
342
343
print(f"Starting high-speed bulk transfer, chunk size: {chunk_size}")
344
start_time = time.time()
345
total_sent = 0
346
total_received = 0
347
348
try:
349
for iteration in range(10): # 10 iterations
350
# Send data
351
sent = handle.bulkWrite(endpoint_out, send_buffer, timeout=5000)
352
total_sent += sent
353
354
# Receive response
355
response = handle.bulkRead(endpoint_in, chunk_size, timeout=5000)
356
total_received += len(response)
357
358
# Verify data integrity (example)
359
if len(response) >= 10:
360
if response[:10] != send_buffer[:10]:
361
print(f"Warning: Data mismatch in iteration {iteration}")
362
363
if iteration % 2 == 0:
364
print(f"Iteration {iteration}: sent {sent}, received {len(response)}")
365
366
except usb1.USBErrorTimeout as e:
367
print(f"Transfer timed out: {e}")
368
if hasattr(e, 'transferred'):
369
total_sent += e.transferred
370
if hasattr(e, 'received'):
371
total_received += len(e.received)
372
373
elapsed = time.time() - start_time
374
print(f"Transfer complete:")
375
print(f" Time: {elapsed:.2f} seconds")
376
print(f" Sent: {total_sent} bytes ({total_sent/elapsed/1024:.1f} KB/s)")
377
print(f" Received: {total_received} bytes ({total_received/elapsed/1024:.1f} KB/s)")
378
379
with usb1.USBContext() as context:
380
device = context.getByVendorIDAndProductID(0x1234, 0x5678)
381
if device:
382
with device.open() as handle:
383
with handle.claimInterface(0):
384
high_speed_bulk_transfer(handle, 0x02, 0x82)
385
```
386
387
### Error Recovery and Retry Logic
388
389
```python
390
import usb1
391
import time
392
393
def robust_transfer(handle, transfer_func, *args, max_retries=3, retry_delay=0.1):
394
"""
395
Perform transfer with automatic retry on certain errors.
396
"""
397
for attempt in range(max_retries + 1):
398
try:
399
return transfer_func(*args)
400
401
except usb1.USBErrorTimeout as e:
402
print(f"Attempt {attempt + 1}: Timeout")
403
if attempt == max_retries:
404
raise
405
time.sleep(retry_delay)
406
407
except usb1.USBErrorPipe as e:
408
print(f"Attempt {attempt + 1}: Pipe error - clearing halt")
409
# Extract endpoint from args (depends on transfer type)
410
if len(args) >= 1 and isinstance(args[0], int):
411
endpoint = args[0]
412
handle.clearHalt(endpoint)
413
if attempt == max_retries:
414
raise
415
time.sleep(retry_delay)
416
417
except usb1.USBErrorNoDevice:
418
print("Device disconnected")
419
raise
420
421
except usb1.USBError as e:
422
print(f"Attempt {attempt + 1}: USB error {e}")
423
if attempt == max_retries:
424
raise
425
time.sleep(retry_delay)
426
427
def transfer_with_recovery_example(handle):
428
"""Example using robust transfer wrapper."""
429
try:
430
# Robust bulk read
431
data = robust_transfer(
432
handle,
433
handle.bulkRead,
434
0x82, # endpoint
435
1024, # length
436
5000 # timeout
437
)
438
print(f"Successfully read {len(data)} bytes after retries")
439
440
# Robust control write
441
result = robust_transfer(
442
handle,
443
handle.controlWrite,
444
usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE | usb1.ENDPOINT_OUT,
445
0x01, # request
446
0, # value
447
0, # index
448
b'\x42\x00', # data
449
5000 # timeout
450
)
451
print(f"Successfully sent control command after retries")
452
453
except usb1.USBError as e:
454
print(f"Transfer failed after all retries: {e}")
455
456
with usb1.USBContext() as context:
457
device = context.getByVendorIDAndProductID(0x1234, 0x5678)
458
if device:
459
with device.open() as handle:
460
with handle.claimInterface(0):
461
transfer_with_recovery_example(handle)
462
```