|
114
|
1 from ctypes import Structure, POINTER, CDLL, CFUNCTYPE
|
|
|
2 from ctypes import c_uint16, c_uint8, c_int, c_uint, c_ssize_t, c_void_p
|
|
|
3 from ctypes import byref, create_string_buffer
|
|
112
|
4
|
|
|
5 # libusb wrapper:
|
|
|
6 libusb = CDLL('libusb-1.0.so')
|
|
|
7
|
|
|
8 # helper:
|
|
114
|
9 def buildfunc(name, argtypes, restype=c_int):
|
|
112
|
10 f = getattr(libusb, name)
|
|
|
11 f.argtypes = argtypes
|
|
|
12 f.restype = restype
|
|
113
|
13 globals()[name] = f
|
|
112
|
14 return f
|
|
|
15 def enum(**enums):
|
|
|
16 reverse = dict((value, key) for key, value in enums.items())
|
|
|
17 enums['reverse_mapping'] = reverse
|
|
|
18 return type('enum', (), enums)
|
|
|
19
|
|
|
20 # enums
|
|
|
21 libusb_class_code = enum(PER_INTERFACE=0, AUDIO=1, COMM=2, HID=3, \
|
|
|
22 PHYSICAL=5, PRINTER=7, PTP=6, MASS_STORAGE=8, HUB=9, \
|
|
|
23 DATA=10, SMART_CARD=0xb, CONTENT_SECURITY=0xd, VIDEO=0xe, \
|
|
|
24 PERSONAL_HEALTHCARE=0xf, DIAGNOSTIC_DEVICE=0xdc, WIRELESS=0xe,\
|
|
|
25 APPLICATION=0xfe, VENDOR_SPEC=0xff)
|
|
|
26 libusb_speed = enum(UNKNOWN=0, LOW=1, FULL=2, HIGH=3, SUPER=4)
|
|
|
27 libusb_error = enum(SUCCES=0, ERROR_IO=-1, ERROR_INVALID_PARAM=-2, \
|
|
|
28 ERROR_ACCESS=-3, ERROR_NO_DEVICE=-4, ERROR_NOT_FOUND=-5, \
|
|
|
29 ERROR_BUSY=-6, ERROR_TIMEOUT=-7, ERROR_OVERFLOW=-8, \
|
|
|
30 ERROR_PIPE=-9, ERROR_INTERRUPTED=-10, ERROR_NO_MEM=-11, \
|
|
|
31 ERROR_NOT_SUPPORTED=-12, ERROR_OTHER=-99)
|
|
|
32 libusb_transfer_status = enum(\
|
|
|
33 COMPLETED=0, ERROR=1, TIMED_OUT=2, \
|
|
|
34 CANCELLED=3, STALL=4, NO_DEVICE=5, OVERFLOW=6)
|
|
|
35
|
|
|
36 # types
|
|
114
|
37 c_int_p = POINTER(c_int)
|
|
112
|
38 class libusb_context(Structure):
|
|
|
39 pass
|
|
|
40 libusb_context_p = POINTER(libusb_context)
|
|
|
41 libusb_context_p_p = POINTER(libusb_context_p)
|
|
|
42
|
|
|
43 class libusb_device(Structure):
|
|
|
44 pass
|
|
|
45 libusb_device_p = POINTER(libusb_device)
|
|
|
46 libusb_device_p_p = POINTER(libusb_device_p)
|
|
|
47 libusb_device_p_p_p = POINTER(libusb_device_p_p)
|
|
|
48
|
|
113
|
49 class libusb_device_handle(Structure):
|
|
|
50 pass
|
|
|
51 libusb_device_handle_p = POINTER(libusb_device_handle)
|
|
|
52 libusb_device_handle_p_p = POINTER(libusb_device_handle_p)
|
|
|
53
|
|
112
|
54 class libusb_device_descriptor(Structure):
|
|
|
55 _fields_ = [
|
|
|
56 ('bLength', c_uint8),
|
|
|
57 ('bDescriptorType', c_uint8),
|
|
|
58 ('bcdUSB', c_uint16),
|
|
|
59 ('bDeviceClass', c_uint8),
|
|
|
60 ('bDeviceSubClass', c_uint8),
|
|
|
61 ('bDeviceProtocol', c_uint8),
|
|
|
62 ('bMaxPacketSize0', c_uint8),
|
|
|
63 ('idVendor', c_uint16),
|
|
|
64 ('idProduct', c_uint16),
|
|
|
65 ('bcdDevice', c_uint16),
|
|
|
66 ('iManufacturer', c_uint8),
|
|
|
67 ('iProduct', c_uint8),
|
|
|
68 ('iSerialNumber', c_uint8),
|
|
|
69 ('iNumConfigurations', c_uint8)
|
|
|
70 ]
|
|
|
71 libusb_device_descriptor_p = POINTER(libusb_device_descriptor)
|
|
|
72
|
|
114
|
73 """
|
|
|
74 class libusb_transfer(Structure):
|
|
|
75 pass
|
|
|
76 libusb_transfer_p = POINTER(libusb_transfer)
|
|
|
77 libusb_transfer_cb_fn = CFUNCTYPE(None, libusb_transfer_p)
|
|
|
78 libusb_transfer._fields_ = [
|
|
|
79 ('dev_handle', libusb_device_handle_p),
|
|
|
80 ('flags', c_uint8),
|
|
|
81 ('endpoint', c_uchar),
|
|
|
82 ('type', c_uchar),
|
|
|
83 ('timeout', c_uint),
|
|
|
84 ('status', c_int), # enum libusb_transfer_status
|
|
|
85 ('length', c_int),
|
|
|
86 ('actual_length', c_int),
|
|
|
87 ('callback', libusb_transfer_cb_fn),
|
|
|
88 ('userdata', c_void_p),
|
|
|
89 ('buffer', c_void_p),
|
|
|
90 ('num_iso_packets', c_int),
|
|
|
91 ('iso_packet_desc', libusb_iso_packet_descriptor)
|
|
|
92 ]
|
|
|
93 """
|
|
112
|
94 # functions
|
|
113
|
95 buildfunc('libusb_init', [libusb_context_p_p], c_int)
|
|
112
|
96
|
|
113
|
97 buildfunc('libusb_get_device_list', \
|
|
112
|
98 [libusb_context_p, libusb_device_p_p_p], c_ssize_t)
|
|
113
|
99 buildfunc('libusb_free_device_list', [libusb_device_p_p, c_int], None)
|
|
|
100 buildfunc('libusb_get_bus_number', [libusb_device_p], c_uint8)
|
|
|
101 buildfunc('libusb_get_device_address', [libusb_device_p], c_uint8)
|
|
114
|
102 buildfunc('libusb_get_device_speed', [libusb_device_p])
|
|
113
|
103 buildfunc('libusb_unref_device', [libusb_device_p], None)
|
|
114
|
104 buildfunc('libusb_open', [libusb_device_p, libusb_device_handle_p_p])
|
|
113
|
105 buildfunc('libusb_close', [libusb_device_handle_p], None)
|
|
114
|
106 buildfunc('libusb_get_configuration',[libusb_device_handle_p,POINTER(c_int)])
|
|
|
107 buildfunc('libusb_set_configuration', [libusb_device_handle_p, c_int])
|
|
|
108 buildfunc('libusb_claim_interface', [libusb_device_handle_p, c_int])
|
|
112
|
109
|
|
|
110
|
|
113
|
111 buildfunc('libusb_get_device_descriptor',\
|
|
114
|
112 [libusb_device_p, libusb_device_descriptor_p])
|
|
|
113
|
|
|
114 # synchronous functions:
|
|
|
115 buildfunc('libusb_bulk_transfer', [libusb_device_handle_p, c_uint8, \
|
|
|
116 c_void_p, c_int, c_int_p, c_uint])
|
|
112
|
117
|
|
|
118 # pythonic API:
|
|
|
119
|
|
|
120 class UsbError(Exception):
|
|
|
121 def __init__(self, msg, errorcode):
|
|
|
122 if errorcode in libusb_error.reverse_mapping:
|
|
|
123 errorcode = libusb_error.reverse_mapping[errorcode]
|
|
|
124 msg = msg + 'Error code: {0}'.format(errorcode)
|
|
|
125 super().__init__(msg)
|
|
|
126
|
|
|
127 class UsbContext(object):
|
|
|
128 """ A usb context in case of multiple use """
|
|
|
129 def __init__(self):
|
|
|
130 self.context_p = libusb_context_p()
|
|
|
131 r = libusb_init(byref(self.context_p))
|
|
|
132 if r != 0:
|
|
|
133 raise UsbError('libusb_init error!', r)
|
|
|
134 def getDeviceList(self):
|
|
|
135 devlist = libusb_device_p_p()
|
|
|
136 count = libusb_get_device_list(self.context_p, byref(devlist))
|
|
|
137 if count < 0:
|
|
|
138 raise UsbError('Error getting device list', count)
|
|
|
139 l = [UsbDevice(self, device_p.contents) for device_p in devlist[0:count]]
|
|
|
140 libusb_free_device_list(devlist, 0)
|
|
|
141 return l
|
|
|
142 DeviceList = property(getDeviceList)
|
|
|
143
|
|
|
144 class UsbDevice:
|
|
|
145 """ A detected usb device """
|
|
|
146 def __init__(self, context, device_p):
|
|
|
147 self.context = context
|
|
|
148 self.dev_p = device_p
|
|
|
149 def __del__(self):
|
|
|
150 libusb_unref_device(self.dev_p)
|
|
|
151 def getBusNumber(self):
|
|
|
152 return libusb_get_bus_number(self.dev_p)
|
|
|
153 BusNumber = property(getBusNumber)
|
|
|
154 def getDeviceAddress(self):
|
|
|
155 return libusb_get_device_address(self.dev_p)
|
|
|
156 DeviceAddress = property(getDeviceAddress)
|
|
|
157 def getSpeed(self):
|
|
|
158 s = libusb_get_device_speed(self.dev_p)
|
|
|
159 if s in libusb_speed.reverse_mapping:
|
|
|
160 s = libusb_speed.reverse_mapping[s]
|
|
|
161 return s
|
|
|
162 Speed = property(getSpeed)
|
|
|
163 def getDescriptor(self):
|
|
|
164 descriptor = libusb_device_descriptor()
|
|
|
165 r = libusb_get_device_descriptor(self.dev_p, byref(descriptor))
|
|
|
166 if r != 0:
|
|
|
167 raise UsbError('Error getting descriptor', r)
|
|
|
168 return descriptor
|
|
|
169 Descriptor = property(getDescriptor)
|
|
|
170 VendorId = property(lambda self: self.Descriptor.idVendor)
|
|
|
171 ProductId = property(lambda self: self.Descriptor.idProduct)
|
|
|
172 NumConfigurations = property(lambda self: self.Descriptor.bNumConfigurations)
|
|
|
173 def open(self):
|
|
|
174 """ Opens this device and returns a handle """
|
|
|
175 handle_p = libusb_device_handle_p()
|
|
|
176 r = libusb_open(self.dev_p, byref(handle_p))
|
|
|
177 if r != 0:
|
|
|
178 raise UsbError('error opening device', r)
|
|
113
|
179 return UsbDeviceHandle(self, handle_p)
|
|
112
|
180 def __repr__(self):
|
|
113
|
181 r2 = 'Usb device: bus {0} address {1} {2:04X}:{3:04X} speed {4}' \
|
|
|
182 .format( \
|
|
|
183 self.BusNumber, self.DeviceAddress, self.VendorId, \
|
|
|
184 self.ProductId, self.Speed)
|
|
112
|
185 return r2
|
|
|
186
|
|
114
|
187 USB_ENDPOINT_DIR_MASK = 0x80
|
|
|
188 USB_ENDPOINT_IN = 0x80
|
|
|
189 USB_ENDPOINT_OUT = 0x0
|
|
|
190
|
|
112
|
191 class UsbDeviceHandle:
|
|
|
192 """ Handle to a detected usb device """
|
|
|
193 def __init__(self, device, handle_p):
|
|
|
194 self.device = device
|
|
|
195 self.handle_p = handle_p
|
|
113
|
196 def getConfiguration(self):
|
|
|
197 config = c_int()
|
|
|
198 r = libusb_get_configuration(self.handle_p, byref(config))
|
|
|
199 if r != 0: raise UsbError('Error getting configuration', r)
|
|
|
200 return config.value
|
|
|
201 def setConfiguration(self, config):
|
|
|
202 r = libusb_set_configuration(self.handle_p, config)
|
|
|
203 if r != 0: raise UsbError('Error setting configuration', r)
|
|
|
204 Configuration = property(getConfiguration, setConfiguration)
|
|
|
205 def claimInterface(self, interface_number):
|
|
|
206 r = libusb_claim_interface(self.handle_p, interface_number)
|
|
|
207 if r != 0: raise UsbError('Error claiming interface', r)
|
|
114
|
208 def bulkWrite(self, endpoint, data, timeout=0):
|
|
|
209 """ Synchronous bulk write """
|
|
|
210 assert type(data) is bytes
|
|
|
211 # assure the endpoint indicates the correct:
|
|
|
212 endpoint = (endpoint & (~USB_ENDPOINT_DIR_MASK)) | USB_ENDPOINT_OUT
|
|
|
213 buf = create_string_buffer(data)
|
|
|
214 transferred = c_int()
|
|
|
215 r = libusb_bulk_transfer(self.handle_p, endpoint, buf, len(data), \
|
|
|
216 byref(transferred), timeout)
|
|
|
217 if r != 0:
|
|
|
218 raise UsbError('Bulk write failed', r)
|
|
|
219 if transferred.value != len(data):
|
|
|
220 raise UsbError('Not all {0} transferred {1}'.format(len(data), \
|
|
|
221 transferred.value))
|
|
|
222 def bulkRead(self, endpoint, numbytes, timeout=0):
|
|
|
223 """ Synchronous bulk read """
|
|
|
224 # assure the endpoint indicates the correct:
|
|
|
225 endpoint = (endpoint & (~USB_ENDPOINT_DIR_MASK)) | USB_ENDPOINT_IN
|
|
|
226 buf = create_string_buffer(numbytes)
|
|
|
227 transferred = c_int()
|
|
|
228 r = libusb_bulk_transfer(self.handle_p, endpoint, buf, numbytes, \
|
|
|
229 byref(transferred), timeout)
|
|
|
230 if r != 0:
|
|
|
231 raise UsbError('Bulk read failed', r)
|
|
|
232 if transferred.value != numbytes:
|
|
|
233 raise UsbError('Not all {0} transferred {1}'.format(numbytes, \
|
|
|
234 transferred.value))
|
|
|
235 data = buf.raw[0:numbytes]
|
|
|
236 return data
|
|
112
|
237
|
|
|
238 class UsbTransfer:
|
|
|
239 def __init__(self):
|
|
|
240 libusb_alloc_transfer(0)
|
