|
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
|
|
113
|
110 buildfunc('libusb_get_device_descriptor',\
|
|
114
|
111 [libusb_device_p, libusb_device_descriptor_p])
|
|
|
112
|
|
|
113 # synchronous functions:
|
|
|
114 buildfunc('libusb_bulk_transfer', [libusb_device_handle_p, c_uint8, \
|
|
|
115 c_void_p, c_int, c_int_p, c_uint])
|
|
112
|
116
|
|
|
117 # pythonic API:
|
|
|
118
|
|
|
119 class UsbError(Exception):
|
|
|
120 def __init__(self, msg, errorcode):
|
|
|
121 if errorcode in libusb_error.reverse_mapping:
|
|
|
122 errorcode = libusb_error.reverse_mapping[errorcode]
|
|
|
123 msg = msg + 'Error code: {0}'.format(errorcode)
|
|
|
124 super().__init__(msg)
|
|
|
125
|
|
|
126 class UsbContext(object):
|
|
|
127 """ A usb context in case of multiple use """
|
|
|
128 def __init__(self):
|
|
|
129 self.context_p = libusb_context_p()
|
|
|
130 r = libusb_init(byref(self.context_p))
|
|
|
131 if r != 0:
|
|
|
132 raise UsbError('libusb_init error!', r)
|
|
|
133 def getDeviceList(self):
|
|
|
134 devlist = libusb_device_p_p()
|
|
|
135 count = libusb_get_device_list(self.context_p, byref(devlist))
|
|
|
136 if count < 0:
|
|
|
137 raise UsbError('Error getting device list', count)
|
|
|
138 l = [UsbDevice(self, device_p.contents) for device_p in devlist[0:count]]
|
|
|
139 libusb_free_device_list(devlist, 0)
|
|
|
140 return l
|
|
|
141 DeviceList = property(getDeviceList)
|
|
|
142
|
|
|
143 class UsbDevice:
|
|
|
144 """ A detected usb device """
|
|
|
145 def __init__(self, context, device_p):
|
|
|
146 self.context = context
|
|
|
147 self.dev_p = device_p
|
|
|
148 def __del__(self):
|
|
|
149 libusb_unref_device(self.dev_p)
|
|
|
150 def getBusNumber(self):
|
|
|
151 return libusb_get_bus_number(self.dev_p)
|
|
|
152 BusNumber = property(getBusNumber)
|
|
|
153 def getDeviceAddress(self):
|
|
|
154 return libusb_get_device_address(self.dev_p)
|
|
|
155 DeviceAddress = property(getDeviceAddress)
|
|
|
156 def getSpeed(self):
|
|
|
157 s = libusb_get_device_speed(self.dev_p)
|
|
|
158 if s in libusb_speed.reverse_mapping:
|
|
|
159 s = libusb_speed.reverse_mapping[s]
|
|
|
160 return s
|
|
|
161 Speed = property(getSpeed)
|
|
|
162 def getDescriptor(self):
|
|
|
163 descriptor = libusb_device_descriptor()
|
|
|
164 r = libusb_get_device_descriptor(self.dev_p, byref(descriptor))
|
|
|
165 if r != 0:
|
|
|
166 raise UsbError('Error getting descriptor', r)
|
|
|
167 return descriptor
|
|
|
168 Descriptor = property(getDescriptor)
|
|
|
169 VendorId = property(lambda self: self.Descriptor.idVendor)
|
|
|
170 ProductId = property(lambda self: self.Descriptor.idProduct)
|
|
|
171 NumConfigurations = property(lambda self: self.Descriptor.bNumConfigurations)
|
|
|
172 def open(self):
|
|
|
173 """ Opens this device and returns a handle """
|
|
|
174 handle_p = libusb_device_handle_p()
|
|
|
175 r = libusb_open(self.dev_p, byref(handle_p))
|
|
|
176 if r != 0:
|
|
|
177 raise UsbError('error opening device', r)
|
|
113
|
178 return UsbDeviceHandle(self, handle_p)
|
|
112
|
179 def __repr__(self):
|
|
113
|
180 r2 = 'Usb device: bus {0} address {1} {2:04X}:{3:04X} speed {4}' \
|
|
|
181 .format( \
|
|
|
182 self.BusNumber, self.DeviceAddress, self.VendorId, \
|
|
|
183 self.ProductId, self.Speed)
|
|
112
|
184 return r2
|
|
|
185
|
|
114
|
186 USB_ENDPOINT_DIR_MASK = 0x80
|
|
|
187 USB_ENDPOINT_IN = 0x80
|
|
|
188 USB_ENDPOINT_OUT = 0x0
|
|
|
189
|
|
112
|
190 class UsbDeviceHandle:
|
|
|
191 """ Handle to a detected usb device """
|
|
|
192 def __init__(self, device, handle_p):
|
|
|
193 self.device = device
|
|
|
194 self.handle_p = handle_p
|
|
130
|
195 def __del__(self):
|
|
|
196 self.close()
|
|
|
197 def close(self):
|
|
|
198 if self.handle_p:
|
|
|
199 libusb_close(self.handle_p)
|
|
|
200 self.handle_p = None
|
|
113
|
201 def getConfiguration(self):
|
|
|
202 config = c_int()
|
|
|
203 r = libusb_get_configuration(self.handle_p, byref(config))
|
|
|
204 if r != 0: raise UsbError('Error getting configuration', r)
|
|
|
205 return config.value
|
|
|
206 def setConfiguration(self, config):
|
|
|
207 r = libusb_set_configuration(self.handle_p, config)
|
|
|
208 if r != 0: raise UsbError('Error setting configuration', r)
|
|
|
209 Configuration = property(getConfiguration, setConfiguration)
|
|
|
210 def claimInterface(self, interface_number):
|
|
|
211 r = libusb_claim_interface(self.handle_p, interface_number)
|
|
|
212 if r != 0: raise UsbError('Error claiming interface', r)
|
|
114
|
213 def bulkWrite(self, endpoint, data, timeout=0):
|
|
|
214 """ Synchronous bulk write """
|
|
|
215 assert type(data) is bytes
|
|
|
216 # assure the endpoint indicates the correct:
|
|
|
217 endpoint = (endpoint & (~USB_ENDPOINT_DIR_MASK)) | USB_ENDPOINT_OUT
|
|
|
218 buf = create_string_buffer(data)
|
|
|
219 transferred = c_int()
|
|
|
220 r = libusb_bulk_transfer(self.handle_p, endpoint, buf, len(data), \
|
|
|
221 byref(transferred), timeout)
|
|
|
222 if r != 0:
|
|
|
223 raise UsbError('Bulk write failed', r)
|
|
|
224 if transferred.value != len(data):
|
|
|
225 raise UsbError('Not all {0} transferred {1}'.format(len(data), \
|
|
|
226 transferred.value))
|
|
|
227 def bulkRead(self, endpoint, numbytes, timeout=0):
|
|
|
228 """ Synchronous bulk read """
|
|
|
229 # assure the endpoint indicates the correct:
|
|
|
230 endpoint = (endpoint & (~USB_ENDPOINT_DIR_MASK)) | USB_ENDPOINT_IN
|
|
|
231 buf = create_string_buffer(numbytes)
|
|
|
232 transferred = c_int()
|
|
|
233 r = libusb_bulk_transfer(self.handle_p, endpoint, buf, numbytes, \
|
|
|
234 byref(transferred), timeout)
|
|
|
235 if r != 0:
|
|
|
236 raise UsbError('Bulk read failed', r)
|
|
|
237 if transferred.value != numbytes:
|
|
|
238 raise UsbError('Not all {0} transferred {1}'.format(numbytes, \
|
|
|
239 transferred.value))
|
|
|
240 data = buf.raw[0:numbytes]
|
|
|
241 return data
|
|
112
|
242
|
|
|
243 class UsbTransfer:
|
|
|
244 def __init__(self):
|
|
|
245 libusb_alloc_transfer(0)
|
