# Copyright (C) 2010-2016 Vincent Pelletier # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; either # version 2.1 of the License, or (at your option) any later version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA # pylint: disable=invalid-name, too-many-locals, too-many-arguments # pylint: disable=too-many-public-methods, too-many-instance-attributes # pylint: disable=missing-docstring """ Pythonic wrapper for libusb-1.0. The first thing you must do is to get an "USB context". To do so, create an USBContext instance. Then, you can use it to browse available USB devices and open the one you want to talk to. At this point, you should have a USBDeviceHandle instance (as returned by USBContext or USBDevice instances), and you can start exchanging with the device. Features: - Basic device settings (configuration & interface selection, ...) - String descriptor lookups (ASCII & unicode), and list supported language codes - Synchronous I/O (control, bulk, interrupt) - Asynchronous I/O (control, bulk, interrupt, isochronous) Note: Isochronous support is not well tested. See USBPoller, USBTransfer and USBTransferHelper. All LIBUSB_* constants are available in this module, without the LIBUSB_ prefix - with one exception: LIBUSB_5GBPS_OPERATION is available as SUPER_SPEED_OPERATION, so it is a valid python identifier. All LIBUSB_ERROR_* constants are available in this module as exception classes, subclassing USBError. """ from __future__ import division from ctypes import byref, create_string_buffer, c_int, sizeof, POINTER, \ cast, c_uint8, c_uint16, c_ubyte, string_at, c_void_p, cdll, addressof, \ c_char from ctypes.util import find_library import sys import threading import warnings import weakref import collections import functools import contextlib import inspect import libusb1 if sys.version_info[:2] >= (2, 6): # pylint: disable=wrong-import-order,ungrouped-imports if sys.platform == 'win32': from ctypes import get_last_error as get_errno else: from ctypes import get_errno # pylint: enable=wrong-import-order,ungrouped-imports else: def get_errno(): raise NotImplementedError( 'Your python version does not support errno/last_error' ) __all__ = [ 'USBContext', 'USBDeviceHandle', 'USBDevice', 'hasCapability', 'USBPoller', 'USBTransfer', 'USBTransferHelper', 'EVENT_CALLBACK_SET', 'USBPollerThread', 'USBEndpoint', 'USBInterfaceSetting', 'USBInterface', 'USBConfiguration', 'DoomedTransferError', 'getVersion', 'USBError', ] # Bind libusb1 constants and libusb1.USBError to this module, so user does not # have to import two modules. USBError = libusb1.USBError STATUS_TO_EXCEPTION_DICT = {} def __bindConstants(): global_dict = globals() PREFIX = 'LIBUSB_' for name, value in libusb1.__dict__.items(): if name.startswith(PREFIX): name = name[len(PREFIX):] # Gah. if name == '5GBPS_OPERATION': name = 'SUPER_SPEED_OPERATION' assert name not in global_dict global_dict[name] = value __all__.append(name) # Finer-grained exceptions. for name, value in libusb1.libusb_error.forward_dict.items(): if value: assert name.startswith(PREFIX + 'ERROR_'), name if name == 'LIBUSB_ERROR_IO': name = 'ErrorIO' else: name = ''.join(x.capitalize() for x in name.split('_')[1:]) name = 'USB' + name assert name not in global_dict, name assert value not in STATUS_TO_EXCEPTION_DICT STATUS_TO_EXCEPTION_DICT[value] = global_dict[name] = type( name, (USBError, ), {'value': value}, ) __all__.append(name) __bindConstants() del __bindConstants def raiseUSBError(value): raise STATUS_TO_EXCEPTION_DICT.get(value, USBError)(value) def mayRaiseUSBError(value): if value < 0: raiseUSBError(value) try: namedtuple = collections.namedtuple except AttributeError: Version = tuple else: Version = namedtuple( 'Version', ['major', 'minor', 'micro', 'nano', 'rc', 'describe'], ) if sys.version_info[0] == 3: BYTE = bytes([0]) # pylint: disable=redefined-builtin xrange = range long = int # pylint: enable=redefined-builtin else: BYTE = '\x00' # pylint: disable=undefined-variable CONTROL_SETUP = BYTE * CONTROL_SETUP_SIZE # pylint: enable=undefined-variable __libc_name = find_library('c') if __libc_name is None: # Of course, will leak memory. # Should we warn user ? How ? _free = lambda x: None else: _free = getattr(cdll, __libc_name).free del __libc_name try: WeakSet = weakref.WeakSet except AttributeError: # Python < 2.7: tiny wrapper around WeakKeyDictionary class WeakSet(object): def __init__(self): self.__dict = weakref.WeakKeyDictionary() def add(self, item): self.__dict[item] = None def pop(self): return self.__dict.popitem()[0] # Default string length # From a comment in libusb-1.0: "Some devices choke on size > 255" STRING_LENGTH = 255 # As of v3 of USB specs, there cannot be more than 7 hubs from controller to # device. PATH_MAX_DEPTH = 7 EVENT_CALLBACK_SET = frozenset(( # pylint: disable=undefined-variable TRANSFER_COMPLETED, TRANSFER_ERROR, TRANSFER_TIMED_OUT, TRANSFER_CANCELLED, TRANSFER_STALL, TRANSFER_NO_DEVICE, TRANSFER_OVERFLOW, # pylint: enable=undefined-variable )) DEFAULT_ASYNC_TRANSFER_ERROR_CALLBACK = lambda x: False def create_binary_buffer(init_or_size): """ ctypes.create_string_buffer variant which does not add a trailing null when init_or_size is not a size. """ # As per ctypes.create_string_buffer, as of python 2.7.10 at least: # - int or long is a length # - str or unicode is an initialiser # Testing the latter confuses 2to3, so test the former. if isinstance(init_or_size, (int, long)): result = create_string_buffer(init_or_size) else: result = create_string_buffer(init_or_size, len(init_or_size)) return result class DoomedTransferError(Exception): """Exception raised when altering/submitting a doomed transfer.""" pass class USBTransfer(object): """ USB asynchronous transfer control & data. All modification methods will raise if called on a submitted transfer. Methods noted as "should not be called on a submitted transfer" will not prevent you from reading, but returned value is unspecified. Note on user_data: because of pypy's current ctype restrictions, user_data is not provided to C level, but is managed purely in python. It should change nothing for you, unless you are looking at underlying C transfer structure - which you should never have to. """ # Prevent garbage collector from freeing the free function before our # instances, as we need it to property destruct them. __libusb_free_transfer = libusb1.libusb_free_transfer __libusb_cancel_transfer = libusb1.libusb_cancel_transfer __USBError = USBError # pylint: disable=undefined-variable __USBErrorNotFound = USBErrorNotFound # pylint: enable=undefined-variable __transfer = None __initialized = False __submitted = False __callback = None __ctypesCallbackWrapper = None __doomed = False __user_data = None __transfer_buffer = None def __init__(self, handle, iso_packets, before_submit, after_completion): """ You should not instanciate this class directly. Call "getTransfer" method on an USBDeviceHandle instance to get instances of this class. """ if iso_packets < 0: raise ValueError( 'Cannot request a negative number of iso packets.' ) self.__handle = handle self.__before_submit = before_submit self.__after_completion = after_completion self.__num_iso_packets = iso_packets result = libusb1.libusb_alloc_transfer(iso_packets) if not result: # pylint: disable=undefined-variable raise USBErrorNoMem # pylint: enable=undefined-variable self.__transfer = result self.__ctypesCallbackWrapper = libusb1.libusb_transfer_cb_fn_p( self.__callbackWrapper) def close(self): """ Break reference cycles to allow instance to be garbage-collected. Raises if called on a submitted transfer. """ if self.__submitted: raise ValueError('Cannot close a submitted transfer') self.doom() self.__initialized = False # Break possible external reference cycles self.__callback = None self.__user_data = None # Break libusb_transfer reference cycles self.__ctypesCallbackWrapper = None # For some reason, overwriting callback is not enough to remove this # reference cycle - though sometimes it works: # self -> self.__dict__ -> libusb_transfer -> dict[x] -> dict[x] -> # CThunkObject -> __callbackWrapper -> self # So free transfer altogether. if self.__transfer is not None: self.__libusb_free_transfer(self.__transfer) self.__transfer = None self.__transfer_buffer = None # Break USBDeviceHandle reference cycle self.__before_submit = None self.__after_completion = None def doom(self): """ Prevent transfer from being submitted again. """ self.__doomed = True def __del__(self): if self.__transfer is not None: try: # If this doesn't raise, we're doomed; transfer was submitted, # still python decided to garbage-collect this instance. # Stick to libusb's documentation, and don't free the # transfer. If interpreter is shutting down, kernel will # reclaim memory anyway. # Note: we can't prevent transfer's buffer from being # garbage-collected as soon as there will be no remaining # reference to transfer, so a segfault might happen anyway. # Should we warn user ? How ? self.cancel() except self.__USBErrorNotFound: # Transfer was not submitted, we can free it. self.__libusb_free_transfer(self.__transfer) # pylint: disable=unused-argument def __callbackWrapper(self, transfer_p): """ Makes it possible for user-provided callback to alter transfer when fired (ie, mark transfer as not submitted upon call). """ self.__submitted = False self.__after_completion(self) callback = self.__callback if callback is not None: callback(self) if self.__doomed: self.close() # pylint: enable=unused-argument def setCallback(self, callback): """ Change transfer's callback. """ self.__callback = callback def getCallback(self): """ Get currently set callback. """ return self.__callback def setControl( self, request_type, request, value, index, buffer_or_len, callback=None, user_data=None, timeout=0): """ Setup transfer for control use. request_type, request, value, index See USBDeviceHandle.controlWrite. request_type defines transfer direction (see ENDPOINT_OUT and ENDPOINT_IN)). buffer_or_len Either a string (when sending data), or expected data length (when receiving data). callback Callback function to be invoked on transfer completion. Called with transfer as parameter, return value ignored. user_data User data to pass to callback function. timeout Transfer timeout in milliseconds. 0 to disable. """ if self.__submitted: raise ValueError('Cannot alter a submitted transfer') if self.__doomed: raise DoomedTransferError('Cannot reuse a doomed transfer') if isinstance(buffer_or_len, (int, long)): length = buffer_or_len # pylint: disable=undefined-variable string_buffer = create_binary_buffer(length + CONTROL_SETUP_SIZE) # pylint: enable=undefined-variable else: length = len(buffer_or_len) string_buffer = create_binary_buffer(CONTROL_SETUP + buffer_or_len) self.__initialized = False self.__transfer_buffer = string_buffer self.__user_data = user_data libusb1.libusb_fill_control_setup( string_buffer, request_type, request, value, index, length) libusb1.libusb_fill_control_transfer( self.__transfer, self.__handle, string_buffer, self.__ctypesCallbackWrapper, None, timeout) self.__callback = callback self.__initialized = True def setBulk( self, endpoint, buffer_or_len, callback=None, user_data=None, timeout=0): """ Setup transfer for bulk use. endpoint Endpoint to submit transfer to. Defines transfer direction (see ENDPOINT_OUT and ENDPOINT_IN)). buffer_or_len Either a string (when sending data), or expected data length (when receiving data) callback Callback function to be invoked on transfer completion. Called with transfer as parameter, return value ignored. user_data User data to pass to callback function. timeout Transfer timeout in milliseconds. 0 to disable. """ if self.__submitted: raise ValueError('Cannot alter a submitted transfer') if self.__doomed: raise DoomedTransferError('Cannot reuse a doomed transfer') string_buffer = create_binary_buffer(buffer_or_len) self.__initialized = False self.__transfer_buffer = string_buffer self.__user_data = user_data libusb1.libusb_fill_bulk_transfer( self.__transfer, self.__handle, endpoint, string_buffer, sizeof(string_buffer), self.__ctypesCallbackWrapper, None, timeout) self.__callback = callback self.__initialized = True def setInterrupt( self, endpoint, buffer_or_len, callback=None, user_data=None, timeout=0): """ Setup transfer for interrupt use. endpoint Endpoint to submit transfer to. Defines transfer direction (see ENDPOINT_OUT and ENDPOINT_IN)). buffer_or_len Either a string (when sending data), or expected data length (when receiving data) callback Callback function to be invoked on transfer completion. Called with transfer as parameter, return value ignored. user_data User data to pass to callback function. timeout Transfer timeout in milliseconds. 0 to disable. """ if self.__submitted: raise ValueError('Cannot alter a submitted transfer') if self.__doomed: raise DoomedTransferError('Cannot reuse a doomed transfer') string_buffer = create_binary_buffer(buffer_or_len) self.__initialized = False self.__transfer_buffer = string_buffer self.__user_data = user_data libusb1.libusb_fill_interrupt_transfer( self.__transfer, self.__handle, endpoint, string_buffer, sizeof(string_buffer), self.__ctypesCallbackWrapper, None, timeout) self.__callback = callback self.__initialized = True def setIsochronous( self, endpoint, buffer_or_len, callback=None, user_data=None, timeout=0, iso_transfer_length_list=None): """ Setup transfer for isochronous use. endpoint Endpoint to submit transfer to. Defines transfer direction (see ENDPOINT_OUT and ENDPOINT_IN)). buffer_or_len Either a string (when sending data), or expected data length (when receiving data) callback Callback function to be invoked on transfer completion. Called with transfer as parameter, return value ignored. user_data User data to pass to callback function. timeout Transfer timeout in milliseconds. 0 to disable. iso_transfer_length_list List of individual transfer sizes. If not provided, buffer_or_len will be divided evenly among available transfers if possible, and raise ValueError otherwise. """ if self.__submitted: raise ValueError('Cannot alter a submitted transfer') num_iso_packets = self.__num_iso_packets if num_iso_packets == 0: raise TypeError( 'This transfer canot be used for isochronous I/O. ' 'You must get another one with a non-zero iso_packets ' 'parameter.' ) if self.__doomed: raise DoomedTransferError('Cannot reuse a doomed transfer') string_buffer = create_binary_buffer(buffer_or_len) buffer_length = sizeof(string_buffer) if iso_transfer_length_list is None: iso_length, remainder = divmod(buffer_length, num_iso_packets) if remainder: raise ValueError( 'Buffer size %i cannot be evenly distributed among %i ' 'transfers' % ( buffer_length, num_iso_packets, ) ) iso_transfer_length_list = [iso_length] * num_iso_packets configured_iso_packets = len(iso_transfer_length_list) if configured_iso_packets > num_iso_packets: raise ValueError( 'Too many ISO transfer lengths (%i), there are ' 'only %i ISO transfers available' % ( configured_iso_packets, num_iso_packets, ) ) if sum(iso_transfer_length_list) > buffer_length: raise ValueError( 'ISO transfers too long (%i), there are only ' '%i bytes available' % ( sum(iso_transfer_length_list), buffer_length, ) ) transfer_p = self.__transfer self.__initialized = False self.__transfer_buffer = string_buffer self.__user_data = user_data libusb1.libusb_fill_iso_transfer( transfer_p, self.__handle, endpoint, string_buffer, buffer_length, configured_iso_packets, self.__ctypesCallbackWrapper, None, timeout) for length, iso_packet_desc in zip( iso_transfer_length_list, libusb1.get_iso_packet_list(transfer_p)): if length <= 0: raise ValueError( 'Negative/null length transfers are not possible.' ) iso_packet_desc.length = length self.__callback = callback self.__initialized = True def getType(self): """ Get transfer type. Returns one of: TRANSFER_TYPE_CONTROL TRANSFER_TYPE_ISOCHRONOUS TRANSFER_TYPE_BULK TRANSFER_TYPE_INTERRUPT """ return self.__transfer.contents.type def getEndpoint(self): """ Get endpoint. """ return self.__transfer.contents.endpoint def getStatus(self): """ Get transfer status. Should not be called on a submitted transfer. """ return self.__transfer.contents.status def getActualLength(self): """ Get actually transfered data length. Should not be called on a submitted transfer. """ return self.__transfer.contents.actual_length def getBuffer(self): """ Get data buffer content. Should not be called on a submitted transfer. """ transfer_p = self.__transfer transfer = transfer_p.contents # pylint: disable=undefined-variable if transfer.type == TRANSFER_TYPE_CONTROL: # pylint: enable=undefined-variable result = libusb1.libusb_control_transfer_get_data(transfer_p) else: result = string_at(transfer.buffer, transfer.length) return result def getUserData(self): """ Retrieve user data provided on setup. """ return self.__user_data def setUserData(self, user_data): """ Change user data. """ self.__user_data = user_data def getISOBufferList(self): """ Get individual ISO transfer's buffer. Returns a list with one item per ISO transfer, with their individually-configured sizes. Returned list is consistent with getISOSetupList return value. Should not be called on a submitted transfer. See also iterISO. """ transfer_p = self.__transfer transfer = transfer_p.contents # pylint: disable=undefined-variable if transfer.type != TRANSFER_TYPE_ISOCHRONOUS: # pylint: enable=undefined-variable raise TypeError( 'This method cannot be called on non-iso transfers.' ) return libusb1.get_iso_packet_buffer_list(transfer_p) def getISOSetupList(self): """ Get individual ISO transfer's setup. Returns a list of dicts, each containing an individual ISO transfer parameters: - length - actual_length - status (see libusb1's API documentation for their signification) Returned list is consistent with getISOBufferList return value. Should not be called on a submitted transfer (except for 'length' values). """ transfer_p = self.__transfer transfer = transfer_p.contents # pylint: disable=undefined-variable if transfer.type != TRANSFER_TYPE_ISOCHRONOUS: # pylint: enable=undefined-variable raise TypeError( 'This method cannot be called on non-iso transfers.' ) return [ { 'length': x.length, 'actual_length': x.actual_length, 'status': x.status, } for x in libusb1.get_iso_packet_list(transfer_p) ] def iterISO(self): """ Generator yielding (status, buffer) for each isochornous transfer. buffer is truncated to actual_length. This is more efficient than calling both getISOBufferList and getISOSetupList when receiving data. Should not be called on a submitted transfer. """ transfer_p = self.__transfer transfer = transfer_p.contents # pylint: disable=undefined-variable if transfer.type != TRANSFER_TYPE_ISOCHRONOUS: # pylint: enable=undefined-variable raise TypeError( 'This method cannot be called on non-iso transfers.' ) buffer_position = transfer.buffer for iso_transfer in libusb1.get_iso_packet_list(transfer_p): yield ( iso_transfer.status, string_at(buffer_position, iso_transfer.actual_length), ) buffer_position += iso_transfer.length def setBuffer(self, buffer_or_len): """ Replace buffer with a new one. Allows resizing read buffer and replacing data sent. Note: resizing is not allowed for isochronous buffer (use setIsochronous). Note: disallowed on control transfers (use setControl). """ if self.__submitted: raise ValueError('Cannot alter a submitted transfer') transfer = self.__transfer.contents # pylint: disable=undefined-variable if transfer.type == TRANSFER_TYPE_CONTROL: # pylint: enable=undefined-variable raise ValueError( 'To alter control transfer buffer, use setControl' ) buff = create_binary_buffer(buffer_or_len) # pylint: disable=undefined-variable if transfer.type == TRANSFER_TYPE_ISOCHRONOUS and \ sizeof(buff) != transfer.length: # pylint: enable=undefined-variable raise ValueError( 'To alter isochronous transfer buffer length, use ' 'setIsochronous' ) self.__transfer_buffer = buff transfer.buffer = cast(buff, c_void_p) transfer.length = sizeof(buff) def isSubmitted(self): """ Tells if this transfer is submitted and still pending. """ return self.__submitted def submit(self): """ Submit transfer for asynchronous handling. """ if self.__submitted: raise ValueError('Cannot submit a submitted transfer') if not self.__initialized: raise ValueError( 'Cannot submit a transfer until it has been initialized' ) if self.__doomed: raise DoomedTransferError('Cannot submit doomed transfer') self.__before_submit(self) self.__submitted = True result = libusb1.libusb_submit_transfer(self.__transfer) if result: self.__after_completion(self) self.__submitted = False raiseUSBError(result) def cancel(self): """ Cancel transfer. Note: cancellation happens asynchronously, so you must wait for TRANSFER_CANCELLED. """ if not self.__submitted: # XXX: Workaround for a bug reported on libusb 1.0.8: calling # libusb_cancel_transfer on a non-submitted transfer might # trigger a segfault. raise self.__USBErrorNotFound result = self.__libusb_cancel_transfer(self.__transfer) if result: raise self.__USBError(result) class USBTransferHelper(object): """ Simplifies subscribing to the same transfer over and over, and callback handling: - no need to read event status to execute apropriate code, just setup different functions for each status code - just return True instead of calling submit - no need to check if transfer is doomed before submitting it again, DoomedTransferError is caught. Callbacks used in this class must follow the callback API described in USBTransfer, and are expected to return a boolean: - True if transfer is to be submitted again (to receive/send more data) - False otherwise Note: as per libusb1 specifications, isochronous transfer global state might be TRANSFER_COMPLETED although some individual packets might have an error status. You can check individual packet status by calling getISOSetupList on transfer object in your callback. """ def __init__(self, transfer=None): """ Create a transfer callback dispatcher. transfer parameter is deprecated. If provided, it will be equivalent to: helper = USBTransferHelper() transfer.setCallback(helper) and also allows using deprecated methods on this class (otherwise, they raise AttributeError). """ if transfer is not None: # Deprecated: to drop self.__transfer = transfer transfer.setCallback(self) self.__event_callback_dict = {} self.__errorCallback = DEFAULT_ASYNC_TRANSFER_ERROR_CALLBACK def submit(self): """ Submit the asynchronous read request. Deprecated. Use submit on transfer. """ # Deprecated: to drop self.__transfer.submit() def cancel(self): """ Cancel a pending read request. Deprecated. Use cancel on transfer. """ # Deprecated: to drop self.__transfer.cancel() def setEventCallback(self, event, callback): """ Set a function to call for a given event. event must be one of: TRANSFER_COMPLETED TRANSFER_ERROR TRANSFER_TIMED_OUT TRANSFER_CANCELLED TRANSFER_STALL TRANSFER_NO_DEVICE TRANSFER_OVERFLOW """ if event not in EVENT_CALLBACK_SET: raise ValueError('Unknown event %r.' % (event, )) self.__event_callback_dict[event] = callback def setDefaultCallback(self, callback): """ Set the function to call for event which don't have a specific callback registered. The initial default callback does nothing and returns False. """ self.__errorCallback = callback def getEventCallback(self, event, default=None): """ Return the function registered to be called for given event identifier. """ return self.__event_callback_dict.get(event, default) def __call__(self, transfer): """ Callback to set on transfers. """ if self.getEventCallback(transfer.getStatus(), self.__errorCallback)( transfer): try: transfer.submit() except DoomedTransferError: pass def isSubmited(self): """ Returns whether this reader is currently waiting for an event. Deprecatd. Use isSubmitted on transfer. """ # Deprecated: to drop return self.__transfer.isSubmitted() class USBPollerThread(threading.Thread): """ Implements libusb1 documentation about threaded, asynchronous applications. In short, instanciate this class once (...per USBContext instance), call start() on the instance, and do whatever you need. This thread will be used to execute transfer completion callbacks, and you are free to use libusb1's synchronous API in another thread, and can forget about libusb1 file descriptors. See http://libusb.sourceforge.net/api-1.0/mtasync.html . """ _can_run = True def __init__(self, context, poller, exc_callback=None): """ Create a poller thread for given context. Warning: it will not check if another poller instance was already present for that context, and will replace it. poller (same as USBPoller.__init__ "poller" parameter) exc_callback (callable) Called with a libusb_error value as single parameter when event handling fails. If not given, an USBError will be raised, interrupting the thread. """ super(USBPollerThread, self).__init__() self.daemon = True self.__context = context self.__poller = poller self.__fd_set = set() if exc_callback is not None: self.exceptionHandler = exc_callback def stop(self): """ Stop & join thread. Allows stopping event thread before context gets closed. """ self._can_run = False self.join() # pylint: disable=method-hidden @staticmethod def exceptionHandler(exc): raise exc # pylint: enable=method-hidden def run(self): # We expect quite some spinning in below loop, so move any unneeded # operation out of it. context = self.__context poll = self.__poller.poll try_lock_events = context.tryLockEvents lock_event_waiters = context.lockEventWaiters wait_for_event = context.waitForEvent unlock_event_waiters = context.unlockEventWaiters event_handling_ok = context.eventHandlingOK unlock_events = context.unlockEvents handle_events_locked = context.handleEventsLocked event_handler_active = context.eventHandlerActive getNextTimeout = context.getNextTimeout exceptionHandler = self.exceptionHandler fd_set = self.__fd_set context.setPollFDNotifiers(self._registerFD, self._unregisterFD) for fd, events in context.getPollFDList(): self._registerFD(fd, events, None) try: while fd_set and self._can_run: if try_lock_events(): lock_event_waiters() while event_handler_active(): wait_for_event() unlock_event_waiters() else: try: while event_handling_ok(): if poll(getNextTimeout()): try: handle_events_locked() except USBError: exceptionHandler(sys.exc_info()[1]) finally: unlock_events() finally: context.setPollFDNotifiers(None, None) def _registerFD(self, fd, events, _): self.__poller.register(fd, events) self.__fd_set.add(fd) def _unregisterFD(self, fd, _): self.__fd_set.discard(fd) self.__poller.unregister(fd) class USBPoller(object): """ Class allowing integration of USB event polling in a file-descriptor monitoring event loop. WARNING: Do not call "poll" from several threads concurently. Do not use synchronous USB transfers in a thread while "poll" is running. Doing so will result in unnecessarily long pauses in some threads. Opening and/or closing devices while polling can cause race conditions to occur. """ def __init__(self, context, poller): """ Create a poller for given context. Warning: it will not check if another poller instance was already present for that context, and will replace it. poller is a polling instance implementing the following methods: - register(fd, event_flags) event_flags have the same meaning as in poll API (POLLIN & POLLOUT) - unregister(fd) - poll(timeout) timeout being a float in seconds, or negative/None if there is no timeout. It must return a list of (descriptor, event) pairs. Note: USBPoller is itself a valid poller. Note2: select.poll uses a timeout in milliseconds, for some reason (all other select.* classes use seconds for timeout), so you should wrap it to convert & round/truncate timeout. """ self.__context = context self.__poller = poller self.__fd_set = set() context.setPollFDNotifiers(self._registerFD, self._unregisterFD) for fd, events in context.getPollFDList(): self._registerFD(fd, events) def __del__(self): self.__context.setPollFDNotifiers(None, None) def poll(self, timeout=None): """ Poll for events. timeout can be a float in seconds, or None for no timeout. Returns a list of (descriptor, event) pairs. """ next_usb_timeout = self.__context.getNextTimeout() if timeout is None or timeout < 0: usb_timeout = next_usb_timeout elif next_usb_timeout: usb_timeout = min(next_usb_timeout, timeout) else: usb_timeout = timeout event_list = self.__poller.poll(usb_timeout) if event_list: fd_set = self.__fd_set result = [(x, y) for x, y in event_list if x not in fd_set] if len(result) != len(event_list): self.__context.handleEventsTimeout() else: result = event_list self.__context.handleEventsTimeout() return result def register(self, fd, events): """ Register an USB-unrelated fd to poller. Convenience method. """ if fd in self.__fd_set: raise ValueError( 'This fd is a special USB event fd, it cannot be polled.' ) self.__poller.register(fd, events) def unregister(self, fd): """ Unregister an USB-unrelated fd from poller. Convenience method. """ if fd in self.__fd_set: raise ValueError( 'This fd is a special USB event fd, it must stay registered.' ) self.__poller.unregister(fd) # pylint: disable=unused-argument def _registerFD(self, fd, events, user_data=None): self.register(fd, events) self.__fd_set.add(fd) # pylint: enable=unused-argument # pylint: disable=unused-argument def _unregisterFD(self, fd, user_data=None): self.__fd_set.discard(fd) self.unregister(fd) # pylint: enable=unused-argument class _ReleaseInterface(object): def __init__(self, handle, interface): self._handle = handle self._interface = interface def __enter__(self): # USBDeviceHandle.claimInterface already claimed the interface. pass def __exit__(self, exc_type, exc_val, exc_tb): self._handle.releaseInterface(self._interface) class USBDeviceHandle(object): """ Represents an opened USB device. """ __handle = None __libusb_close = libusb1.libusb_close __USBError = USBError # pylint: disable=undefined-variable __USBErrorNoDevice = USBErrorNoDevice __USBErrorNotFound = USBErrorNotFound __USBErrorInterrupted = USBErrorInterrupted # pylint: enable=undefined-variable __set = set __KeyError = KeyError __sys = sys def __init__(self, context, handle, device): """ You should not instanciate this class directly. Call "open" method on an USBDevice instance to get an USBDeviceHandle instance. """ self.__context = context # Weak reference to transfers about this device so we can clean up # before closing device. self.__transfer_set = WeakSet() # Strong references to inflight transfers so they do not get freed # even if user drops all strong references to them. If this instance # is garbage-collected, we close all transfers, so it's fine. self.__inflight = inflight = set() # XXX: For some reason, doing self.__inflight.{add|remove} inside # getTransfer causes extra intermediate python objects for each # allocated transfer. Storing them as properties solves this. Found # with objgraph. self.__inflight_add = inflight.add self.__inflight_remove = inflight.remove self.__handle = handle self.__device = device def __del__(self): self.close() def close(self): """ Close this handle. If not called explicitely, will be called by destructor. This method cancels any in-flight transfer when it is called. As cancellation is not immediate, this method needs to let libusb handle events until transfers are actually cancelled. In multi-threaded programs, this can lead to stalls. To avoid this, do not close nor let GC collect a USBDeviceHandle which has in-flight transfers. """ handle = self.__handle if handle is None: return # Build a strong set from weak self.__transfer_set so we can doom # and close all contained transfers. # Because of backward compatibility, self.__transfer_set might be a # wrapper around WeakKeyDictionary. As it might be modified by gc, # we must pop until there is not key left instead of iterating over # it. weak_transfer_set = self.__transfer_set transfer_set = self.__set() while True: try: transfer = weak_transfer_set.pop() except self.__KeyError: break transfer_set.add(transfer) transfer.doom() inflight = self.__inflight for transfer in inflight: try: transfer.cancel() except (self.__USBErrorNotFound, self.__USBErrorNoDevice): pass while inflight: try: self.__context.handleEvents() except self.__USBErrorInterrupted: pass for transfer in transfer_set: transfer.close() self.__libusb_close(handle) self.__handle = None def getDevice(self): """ Get an USBDevice instance for the device accessed through this handle. Useful for example to query its configurations. """ return self.__device def getConfiguration(self): """ Get the current configuration number for this device. """ configuration = c_int() mayRaiseUSBError(libusb1.libusb_get_configuration( self.__handle, byref(configuration), )) return configuration.value def setConfiguration(self, configuration): """ Set the configuration number for this device. """ mayRaiseUSBError( libusb1.libusb_set_configuration(self.__handle, configuration), ) def claimInterface(self, interface): """ Claim (= get exclusive access to) given interface number. Required to receive/send data. Can be used as a context manager: with handle.claimInterface(0): # do stuff # handle.releaseInterface(0) gets automatically called """ mayRaiseUSBError( libusb1.libusb_claim_interface(self.__handle, interface), ) return _ReleaseInterface(self, interface) def releaseInterface(self, interface): """ Release interface, allowing another process to use it. """ mayRaiseUSBError( libusb1.libusb_release_interface(self.__handle, interface), ) def setInterfaceAltSetting(self, interface, alt_setting): """ Set interface's alternative setting (both parameters are integers). """ mayRaiseUSBError(libusb1.libusb_set_interface_alt_setting( self.__handle, interface, alt_setting, )) def clearHalt(self, endpoint): """ Clear a halt state on given endpoint number. """ mayRaiseUSBError(libusb1.libusb_clear_halt(self.__handle, endpoint)) def resetDevice(self): """ Reinitialise current device. Attempts to restore current configuration & alt settings. If this fails, will result in a device disconnect & reconnect, so you have to close current device and rediscover it (notified by a ERROR_NOT_FOUND error code). """ mayRaiseUSBError(libusb1.libusb_reset_device(self.__handle)) def kernelDriverActive(self, interface): """ Tell whether a kernel driver is active on given interface number. """ result = libusb1.libusb_kernel_driver_active(self.__handle, interface) if result == 0: return False elif result == 1: return True raiseUSBError(result) def detachKernelDriver(self, interface): """ Ask kernel driver to detach from given interface number. """ mayRaiseUSBError( libusb1.libusb_detach_kernel_driver(self.__handle, interface), ) def attachKernelDriver(self, interface): """ Ask kernel driver to re-attach to given interface number. """ mayRaiseUSBError( libusb1.libusb_attach_kernel_driver(self.__handle, interface), ) def setAutoDetachKernelDriver(self, enable): """ Control automatic kernel driver detach. enable (bool) True to enable auto-detach, False to disable it. """ mayRaiseUSBError(libusb1.libusb_set_auto_detach_kernel_driver( self.__handle, bool(enable), )) def getSupportedLanguageList(self): """ Return a list of USB language identifiers (as integers) supported by current device for its string descriptors. Note: language identifiers seem (I didn't check them all...) very similar to windows language identifiers, so you may want to use locales.windows_locale to get an rfc3066 representation. The 5 standard HID language codes are missing though. """ descriptor_string = create_binary_buffer(STRING_LENGTH) result = libusb1.libusb_get_string_descriptor( self.__handle, 0, 0, descriptor_string, sizeof(descriptor_string), ) # pylint: disable=undefined-variable if result == ERROR_PIPE: # pylint: enable=undefined-variable # From libusb_control_transfer doc: # control request not supported by the device return [] mayRaiseUSBError(result) langid_list = cast(descriptor_string, POINTER(c_uint16)) return [ libusb1.libusb_le16_to_cpu(langid_list[offset].value) for offset in xrange(1, cast(descriptor_string, POINTER(c_ubyte))[0] // 2) ] def getStringDescriptor(self, descriptor, lang_id): """ Fetch description string for given descriptor and in given language. Use getSupportedLanguageList to know which languages are available. Return value is an unicode string. Return None if there is no such descriptor on device. """ descriptor_string = create_binary_buffer(STRING_LENGTH) try: mayRaiseUSBError(libusb1.libusb_get_string_descriptor( self.__handle, descriptor, lang_id, descriptor_string, sizeof(descriptor_string), )) # pylint: disable=undefined-variable except USBErrorNotFound: # pylint: enable=undefined-variable return None return descriptor_string.value.decode('UTF-16-LE') def getASCIIStringDescriptor(self, descriptor): """ Fetch description string for given descriptor in first available language. Return value is an ASCII string. Return None if there is no such descriptor on device. """ descriptor_string = create_binary_buffer(STRING_LENGTH) try: mayRaiseUSBError(libusb1.libusb_get_string_descriptor_ascii( self.__handle, descriptor, descriptor_string, sizeof(descriptor_string), )) # pylint: disable=undefined-variable except USBErrorNotFound: # pylint: enable=undefined-variable return None return descriptor_string.value.decode('ASCII') # Sync I/O def _controlTransfer( self, request_type, request, value, index, data, length, timeout): result = libusb1.libusb_control_transfer( self.__handle, request_type, request, value, index, data, length, timeout, ) mayRaiseUSBError(result) return result def controlWrite( self, request_type, request, value, index, data, timeout=0): """ Synchronous control write. request_type: request type bitmask (bmRequestType), see constants TYPE_* and RECIPIENT_*. request: request id (some values are standard). value, index, data: meaning is request-dependent. timeout: in milliseconds, how long to wait for device acknowledgement. Set to 0 to disable. Returns the number of bytes actually sent. """ # pylint: disable=undefined-variable request_type = (request_type & ~ENDPOINT_DIR_MASK) | ENDPOINT_OUT # pylint: enable=undefined-variable data = (c_char * len(data))(*data) return self._controlTransfer(request_type, request, value, index, data, sizeof(data), timeout) def controlRead( self, request_type, request, value, index, length, timeout=0): """ Synchronous control read. timeout: in milliseconds, how long to wait for data. Set to 0 to disable. See controlWrite for other parameters description. Returns received data. """ # pylint: disable=undefined-variable request_type = (request_type & ~ENDPOINT_DIR_MASK) | ENDPOINT_IN # pylint: enable=undefined-variable data = create_binary_buffer(length) transferred = self._controlTransfer( request_type, request, value, index, data, length, timeout, ) return data.raw[:transferred] def _bulkTransfer(self, endpoint, data, length, timeout): transferred = c_int() mayRaiseUSBError(libusb1.libusb_bulk_transfer( self.__handle, endpoint, data, length, byref(transferred), timeout, )) return transferred.value def bulkWrite(self, endpoint, data, timeout=0): """ Synchronous bulk write. endpoint: endpoint to send data to. data: data to send. timeout: in milliseconds, how long to wait for device acknowledgement. Set to 0 to disable. Returns the number of bytes actually sent. """ # pylint: disable=undefined-variable endpoint = (endpoint & ~ENDPOINT_DIR_MASK) | ENDPOINT_OUT # pylint: enable=undefined-variable data = (c_char * len(data))(*data) return self._bulkTransfer(endpoint, data, sizeof(data), timeout) def bulkRead(self, endpoint, length, timeout=0): """ Synchronous bulk read. timeout: in milliseconds, how long to wait for data. Set to 0 to disable. See bulkWrite for other parameters description. Returns received data. """ # pylint: disable=undefined-variable endpoint = (endpoint & ~ENDPOINT_DIR_MASK) | ENDPOINT_IN # pylint: enable=undefined-variable data = create_binary_buffer(length) transferred = self._bulkTransfer(endpoint, data, length, timeout) # pylint: disable=invalid-slice-index return data.raw[:transferred] # pylint: enable=invalid-slice-index def _interruptTransfer(self, endpoint, data, length, timeout): transferred = c_int() mayRaiseUSBError(libusb1.libusb_interrupt_transfer( self.__handle, endpoint, data, length, byref(transferred), timeout, )) return transferred.value def interruptWrite(self, endpoint, data, timeout=0): """ Synchronous interrupt write. endpoint: endpoint to send data to. data: data to send. timeout: in milliseconds, how long to wait for device acknowledgement. Set to 0 to disable. Returns the number of bytes actually sent. """ # pylint: disable=undefined-variable endpoint = (endpoint & ~ENDPOINT_DIR_MASK) | ENDPOINT_OUT # pylint: enable=undefined-variable data = (c_char * len(data))(*data) return self._interruptTransfer(endpoint, data, sizeof(data), timeout) def interruptRead(self, endpoint, length, timeout=0): """ Synchronous interrupt write. timeout: in milliseconds, how long to wait for data. Set to 0 to disable. See interruptRead for other parameters description. Returns received data. """ # pylint: disable=undefined-variable endpoint = (endpoint & ~ENDPOINT_DIR_MASK) | ENDPOINT_IN # pylint: enable=undefined-variable data = create_binary_buffer(length) transferred = self._interruptTransfer(endpoint, data, length, timeout) # pylint: disable=invalid-slice-index return data.raw[:transferred] # pylint: enable=invalid-slice-index def getTransfer(self, iso_packets=0): """ Get an USBTransfer instance for asynchronous use. iso_packets: the number of isochronous transfer descriptors to allocate. """ result = USBTransfer( self.__handle, iso_packets, self.__inflight_add, self.__inflight_remove, ) self.__transfer_set.add(result) return result class USBConfiguration(object): def __init__(self, context, config): """ You should not instanciate this class directly. Call USBDevice methods to get instances of this class. """ if not isinstance(config, libusb1.libusb_config_descriptor): raise TypeError('Unexpected descriptor type.') self.__config = config self.__context = context def getNumInterfaces(self): return self.__config.bNumInterfaces __len__ = getNumInterfaces def getConfigurationValue(self): return self.__config.bConfigurationValue def getDescriptor(self): return self.__config.iConfiguration def getAttributes(self): return self.__config.bmAttributes def getMaxPower(self): """ Returns device's power consumption in mW. Beware of unit: USB descriptor uses 2mW increments, this method converts it to mW units. """ return self.__config.MaxPower * 2 def getExtra(self): """ Returns a list of extra (non-basic) descriptors (DFU, HID, ...). """ return libusb1.get_extra(self.__config) def __iter__(self): """ Iterates over interfaces available in this configuration, yielding USBInterface instances. """ context = self.__context interface_list = self.__config.interface for interface_num in xrange(self.getNumInterfaces()): yield USBInterface(context, interface_list[interface_num]) # BBB iterInterfaces = __iter__ def __getitem__(self, interface): """ Returns an USBInterface instance. """ if not isinstance(interface, int): raise TypeError('interface parameter must be an integer') if not 0 <= interface < self.getNumInterfaces(): raise IndexError('No such interface: %r' % (interface, )) return USBInterface(self.__context, self.__config.interface[interface]) class USBInterface(object): def __init__(self, context, interface): """ You should not instanciate this class directly. Call USBConfiguration methods to get instances of this class. """ if not isinstance(interface, libusb1.libusb_interface): raise TypeError('Unexpected descriptor type.') self.__interface = interface self.__context = context def getNumSettings(self): return self.__interface.num_altsetting __len__ = getNumSettings def __iter__(self): """ Iterates over settings in this insterface, yielding USBInterfaceSetting instances. """ context = self.__context alt_setting_list = self.__interface.altsetting for alt_setting_num in xrange(self.getNumSettings()): yield USBInterfaceSetting( context, alt_setting_list[alt_setting_num]) # BBB iterSettings = __iter__ def __getitem__(self, alt_setting): """ Returns an USBInterfaceSetting instance. """ if not isinstance(alt_setting, int): raise TypeError('alt_setting parameter must be an integer') if not 0 <= alt_setting < self.getNumSettings(): raise IndexError('No such setting: %r' % (alt_setting, )) return USBInterfaceSetting( self.__context, self.__interface.altsetting[alt_setting]) class USBInterfaceSetting(object): def __init__(self, context, alt_setting): """ You should not instanciate this class directly. Call USBDevice or USBInterface methods to get instances of this class. """ if not isinstance(alt_setting, libusb1.libusb_interface_descriptor): raise TypeError('Unexpected descriptor type.') self.__alt_setting = alt_setting self.__context = context def getNumber(self): return self.__alt_setting.bInterfaceNumber def getAlternateSetting(self): return self.__alt_setting.bAlternateSetting def getNumEndpoints(self): return self.__alt_setting.bNumEndpoints __len__ = getNumEndpoints def getClass(self): return self.__alt_setting.bInterfaceClass def getSubClass(self): return self.__alt_setting.bInterfaceSubClass def getClassTuple(self): """ For convenience: class and subclass are probably often matched simultaneously. """ alt_setting = self.__alt_setting return (alt_setting.bInterfaceClass, alt_setting.bInterfaceSubClass) # BBB getClassTupple = getClassTuple def getProtocol(self): return self.__alt_setting.bInterfaceProtocol def getDescriptor(self): return self.__alt_setting.iInterface def getExtra(self): return libusb1.get_extra(self.__alt_setting) def __iter__(self): """ Iterates over endpoints in this interface setting , yielding USBEndpoint instances. """ context = self.__context endpoint_list = self.__alt_setting.endpoint for endpoint_num in xrange(self.getNumEndpoints()): yield USBEndpoint(context, endpoint_list[endpoint_num]) # BBB iterEndpoints = __iter__ def __getitem__(self, endpoint): """ Returns an USBEndpoint instance. """ if not isinstance(endpoint, int): raise TypeError('endpoint parameter must be an integer') if not 0 <= endpoint < self.getNumEndpoints(): raise ValueError('No such endpoint: %r' % (endpoint, )) return USBEndpoint( self.__context, self.__alt_setting.endpoint[endpoint]) class USBEndpoint(object): def __init__(self, context, endpoint): if not isinstance(endpoint, libusb1.libusb_endpoint_descriptor): raise TypeError('Unexpected descriptor type.') self.__endpoint = endpoint self.__context = context def getAddress(self): return self.__endpoint.bEndpointAddress def getAttributes(self): return self.__endpoint.bmAttributes def getMaxPacketSize(self): return self.__endpoint.wMaxPacketSize def getInterval(self): return self.__endpoint.bInterval def getRefresh(self): return self.__endpoint.bRefresh def getSyncAddress(self): return self.__endpoint.bSynchAddress def getExtra(self): return libusb1.get_extra(self.__endpoint) class USBDevice(object): """ Represents a USB device. """ __configuration_descriptor_list = () __libusb_unref_device = libusb1.libusb_unref_device __libusb_free_config_descriptor = libusb1.libusb_free_config_descriptor __byref = byref __KeyError = KeyError def __init__(self, context, device_p, can_load_configuration=True): """ You should not instanciate this class directly. Call USBContext methods to receive instances of this class. """ self.__context = context self.__close_set = WeakSet() libusb1.libusb_ref_device(device_p) self.device_p = device_p # Fetch device descriptor device_descriptor = libusb1.libusb_device_descriptor() result = libusb1.libusb_get_device_descriptor( device_p, byref(device_descriptor)) mayRaiseUSBError(result) self.device_descriptor = device_descriptor if can_load_configuration: self.__configuration_descriptor_list = descriptor_list = [] append = descriptor_list.append device_p = self.device_p for configuration_id in xrange( self.device_descriptor.bNumConfigurations): config = libusb1.libusb_config_descriptor_p() result = libusb1.libusb_get_config_descriptor( device_p, configuration_id, byref(config)) # pylint: disable=undefined-variable if result == ERROR_NOT_FOUND: # pylint: enable=undefined-variable # Some devices (ex windows' root hubs) tell they have # one configuration, but they have no configuration # descriptor. continue mayRaiseUSBError(result) append(config.contents) def __del__(self): self.close() def close(self): pop = self.__close_set.pop while True: try: closable = pop() except self.__KeyError: break closable.close() if not self.device_p: return self.__libusb_unref_device(self.device_p) # pylint: disable=redefined-outer-name byref = self.__byref # pylint: enable=redefined-outer-name descriptor_list = self.__configuration_descriptor_list while descriptor_list: self.__libusb_free_config_descriptor( byref(descriptor_list.pop()), ) self.device_p = None def __str__(self): return 'Bus %03i Device %03i: ID %04x:%04x' % ( self.getBusNumber(), self.getDeviceAddress(), self.getVendorID(), self.getProductID(), ) def __len__(self): return len(self.__configuration_descriptor_list) def __getitem__(self, index): return USBConfiguration( self.__context, self.__configuration_descriptor_list[index]) def __key(self): return ( id(self.__context), self.getBusNumber(), self.getDeviceAddress(), self.getVendorID(), self.getProductID(), ) def __hash__(self): return hash(self.__key()) def __eq__(self, other): # pylint: disable=unidiomatic-typecheck return type(self) == type(other) and ( # pylint: enable=unidiomatic-typecheck self.device_p == other.device_p or # pylint: disable=protected-access self.__key() == other.__key() # pylint: enable=protected-access ) def iterConfigurations(self): context = self.__context for config in self.__configuration_descriptor_list: yield USBConfiguration(context, config) # BBB iterConfiguations = iterConfigurations def iterSettings(self): for config in self.iterConfigurations(): for interface in config: for setting in interface: yield setting def getBusNumber(self): """ Get device's bus number. """ return libusb1.libusb_get_bus_number(self.device_p) def getPortNumber(self): """ Get device's port number. """ return libusb1.libusb_get_port_number(self.device_p) def getPortNumberList(self): """ Get the port number of each hub toward device. """ port_list = (c_uint8 * PATH_MAX_DEPTH)() result = libusb1.libusb_get_port_numbers( self.device_p, port_list, len(port_list)) mayRaiseUSBError(result) return list(port_list[:result]) # TODO: wrap libusb_get_parent when/if libusb removes the need to be inside # a libusb_(get|free)_device_list block. def getDeviceAddress(self): """ Get device's address on its bus. """ return libusb1.libusb_get_device_address(self.device_p) def getbcdUSB(self): """ Get the USB spec version device complies to, in BCD format. """ return self.device_descriptor.bcdUSB def getDeviceClass(self): """ Get device's class id. """ return self.device_descriptor.bDeviceClass def getDeviceSubClass(self): """ Get device's subclass id. """ return self.device_descriptor.bDeviceSubClass def getDeviceProtocol(self): """ Get device's protocol id. """ return self.device_descriptor.bDeviceProtocol def getMaxPacketSize0(self): """ Get device's max packet size for endpoint 0 (control). """ return self.device_descriptor.bMaxPacketSize0 def getMaxPacketSize(self, endpoint): """ Get device's max packet size for given endpoint. Warning: this function will not always give you the expected result. See https://libusb.org/ticket/77 . You should instead consult the endpoint descriptor of current configuration and alternate setting. """ result = libusb1.libusb_get_max_packet_size(self.device_p, endpoint) mayRaiseUSBError(result) return result def getMaxISOPacketSize(self, endpoint): """ Get the maximum size for a single isochronous packet for given endpoint. Warning: this function will not always give you the expected result. See https://libusb.org/ticket/77 . You should instead consult the endpoint descriptor of current configuration and alternate setting. """ result = libusb1.libusb_get_max_iso_packet_size(self.device_p, endpoint) mayRaiseUSBError(result) return result def getVendorID(self): """ Get device's vendor id. """ return self.device_descriptor.idVendor def getProductID(self): """ Get device's product id. """ return self.device_descriptor.idProduct def getbcdDevice(self): """ Get device's release number. """ return self.device_descriptor.bcdDevice def getSupportedLanguageList(self): """ Get the list of language ids device has string descriptors for. """ return self.open().getSupportedLanguageList() def _getStringDescriptor(self, descriptor, lang_id): if descriptor: return self.open().getStringDescriptor(descriptor, lang_id) def _getASCIIStringDescriptor(self, descriptor): if descriptor: return self.open().getASCIIStringDescriptor(descriptor) def getManufacturer(self): """ Get device's manufaturer name. Note: opens the device temporarily. """ return self._getASCIIStringDescriptor( self.device_descriptor.iManufacturer) def getProduct(self): """ Get device's product name. Note: opens the device temporarily. """ return self._getASCIIStringDescriptor(self.device_descriptor.iProduct) def getSerialNumber(self): """ Get device's serial number. Note: opens the device temporarily. """ return self._getASCIIStringDescriptor( self.device_descriptor.iSerialNumber) def getNumConfigurations(self): """ Get device's number of possible configurations. """ return self.device_descriptor.bNumConfigurations def getDeviceSpeed(self): """ Get device's speed. Returns one of: SPEED_UNKNOWN SPEED_LOW SPEED_FULL SPEED_HIGH SPEED_SUPER """ return libusb1.libusb_get_device_speed(self.device_p) def open(self): """ Open device. Returns an USBDeviceHandle instance. """ handle = libusb1.libusb_device_handle_p() mayRaiseUSBError(libusb1.libusb_open(self.device_p, byref(handle))) result = USBDeviceHandle(self.__context, handle, self) self.__close_set.add(result) return result _zero_tv = libusb1.timeval(0, 0) _zero_tv_p = byref(_zero_tv) class USBContext(object): """ libusb1 USB context. Provides methods to enumerate & look up USB devices. Also provides access to global (device-independent) libusb1 functions. """ __libusb_exit = libusb1.libusb_exit __context_p = None __added_cb = None __removed_cb = None __poll_cb_user_data = None __libusb_set_pollfd_notifiers = libusb1.libusb_set_pollfd_notifiers __null_pointer = POINTER(None) __KeyError = KeyError __auto_open = True # pylint: disable=no-self-argument,protected-access def _validContext(func): # Defined inside USBContext so we can access "self.__*". @contextlib.contextmanager def refcount(self): with self.__context_cond: if not self.__context_p and self.__auto_open: # BBB warnings.warn( 'Use "with USBContext() as context:" for safer cleanup' ' on interpreter shutdown. See also USBContext.open().', DeprecationWarning, ) self.open() self.__context_refcount += 1 try: yield finally: with self.__context_cond: self.__context_refcount -= 1 if not self.__context_refcount: self.__context_cond.notifyAll() if inspect.isgeneratorfunction(func): def wrapper(self, *args, **kw): with refcount(self): if self.__context_p: # pylint: disable=not-callable for value in func(self, *args, **kw): # pylint: enable=not-callable yield value else: def wrapper(self, *args, **kw): with refcount(self): if self.__context_p: # pylint: disable=not-callable return func(self, *args, **kw) # pylint: enable=not-callable functools.update_wrapper(wrapper, func) return wrapper # pylint: enable=no-self-argument,protected-access def __init__(self): """ Create a new USB context. """ # Used to prevent an exit to cause a segfault if a concurrent thread # is still in libusb. self.__context_refcount = 0 self.__context_cond = threading.Condition() self.__context_p = libusb1.libusb_context_p() self.__hotplug_callback_dict = {} self.__close_set = WeakSet() def __del__(self): # Avoid locking. # XXX: Assumes __del__ should not normally be called while any # instance's method is being executed. It seems unlikely (they hold a # reference to their instance). self._exit() def __enter__(self): return self.open() def __exit__(self, exc_type, exc_val, exc_tb): self.close() def open(self): """ Finish context initialisation, as is normally done in __enter__ . This happens automatically on the first method call needing access to the uninitialised properties, but with a warning. Call this method ONLY if your usage pattern prevents you from using the with USBContext() as contewt: form: this means there are ways to avoid calling close(), which can cause issues particularly hard to debug (ex: interpreter hangs on exit). """ assert self.__context_refcount == 0 mayRaiseUSBError(libusb1.libusb_init(byref(self.__context_p))) return self def close(self): """ Close (destroy) this USB context, and all related instances. When this method has been called, methods on its instance will become mosty no-ops, returning None until explicitly re-opened (by calling open() or __enter__()). Note: "exit" is a deprecated alias of "close". """ self.__auto_open = False self.__context_cond.acquire() try: while self.__context_refcount and self.__context_p: self.__context_cond.wait() self._exit() finally: self.__context_cond.notifyAll() self.__context_cond.release() def _exit(self): context_p = self.__context_p if context_p: for handle in self.__hotplug_callback_dict.keys(): self.hotplugDeregisterCallback(handle) pop = self.__close_set.pop while True: try: closable = pop() except self.__KeyError: break closable.close() self.__libusb_exit(context_p) self.__context_p = libusb1.libusb_context_p() self.__added_cb = self.__null_pointer self.__removed_cb = self.__null_pointer # BBB exit = close @_validContext def getDeviceIterator(self, skip_on_error=False): """ Return an iterator over all USB devices currently plugged in, as USBDevice instances. skip_on_error (bool) If True, ignore devices which raise USBError. """ device_p_p = libusb1.libusb_device_p_p() libusb_device_p = libusb1.libusb_device_p device_list_len = libusb1.libusb_get_device_list(self.__context_p, byref(device_p_p)) mayRaiseUSBError(device_list_len) try: for device_p in device_p_p[:device_list_len]: try: # Instanciate our own libusb_device_p object so we can free # libusb-provided device list. Is this a bug in ctypes that # it doesn't copy pointer value (=pointed memory address) ? # At least, it's not so convenient and forces using such # weird code. device = USBDevice(self, libusb_device_p(device_p.contents)) except USBError: if not skip_on_error: raise else: self.__close_set.add(device) yield device finally: libusb1.libusb_free_device_list(device_p_p, 1) def getDeviceList(self, skip_on_access_error=False, skip_on_error=False): """ Return a list of all USB devices currently plugged in, as USBDevice instances. skip_on_error (bool) If True, ignore devices which raise USBError. skip_on_access_error (bool) DEPRECATED. Alias for skip_on_error. """ return list( self.getDeviceIterator( skip_on_error=skip_on_access_error or skip_on_error, ), ) def getByVendorIDAndProductID( self, vendor_id, product_id, skip_on_access_error=False, skip_on_error=False): """ Get the first USB device matching given vendor and product ids. Returns an USBDevice instance, or None if no present device match. skip_on_error (bool) (see getDeviceList) skip_on_access_error (bool) (see getDeviceList) """ for device in self.getDeviceIterator( skip_on_error=skip_on_access_error or skip_on_error, ): if device.getVendorID() == vendor_id and \ device.getProductID() == product_id: return device def openByVendorIDAndProductID( self, vendor_id, product_id, skip_on_access_error=False, skip_on_error=False): """ Get the first USB device matching given vendor and product ids. Returns an USBDeviceHandle instance, or None if no present device match. skip_on_error (bool) (see getDeviceList) skip_on_access_error (bool) (see getDeviceList) """ result = self.getByVendorIDAndProductID( vendor_id, product_id, skip_on_access_error=skip_on_access_error, skip_on_error=skip_on_error) if result is not None: return result.open() @_validContext def getPollFDList(self): """ Return file descriptors to be used to poll USB events. You should not have to call this method, unless you are integrating this class with a polling mechanism. """ pollfd_p_p = libusb1.libusb_get_pollfds(self.__context_p) if not pollfd_p_p: errno = get_errno() if errno: raise OSError(errno) else: # Assume not implemented raise NotImplementedError( 'Your libusb does not seem to implement pollable FDs') try: result = [] append = result.append fd_index = 0 while pollfd_p_p[fd_index]: append(( pollfd_p_p[fd_index].contents.fd, pollfd_p_p[fd_index].contents.events, )) fd_index += 1 finally: _free(pollfd_p_p) return result @_validContext def handleEvents(self): """ Handle any pending event (blocking). See libusb1 documentation for details (there is a timeout, so it's not "really" blocking). """ mayRaiseUSBError( libusb1.libusb_handle_events(self.__context_p), ) # TODO: handleEventsCompleted @_validContext def handleEventsTimeout(self, tv=0): """ Handle any pending event. If tv is 0, will return immediately after handling already-pending events. Otherwise, defines the maximum amount of time to wait for events, in seconds. """ if tv is None: tv = 0 tv_s = int(tv) real_tv = libusb1.timeval(tv_s, int((tv - tv_s) * 1000000)) mayRaiseUSBError( libusb1.libusb_handle_events_timeout( self.__context_p, byref(real_tv), ), ) # TODO: handleEventsTimeoutCompleted @_validContext def setPollFDNotifiers( self, added_cb=None, removed_cb=None, user_data=None): """ Give libusb1 methods to call when it should add/remove file descriptor for polling. You should not have to call this method, unless you are integrating this class with a polling mechanism. """ if added_cb is None: added_cb = self.__null_pointer else: added_cb = libusb1.libusb_pollfd_added_cb_p(added_cb) if removed_cb is None: removed_cb = self.__null_pointer else: removed_cb = libusb1.libusb_pollfd_removed_cb_p(removed_cb) if user_data is None: user_data = self.__null_pointer self.__added_cb = added_cb self.__removed_cb = removed_cb self.__poll_cb_user_data = user_data self.__libusb_set_pollfd_notifiers( self.__context_p, added_cb, removed_cb, user_data) @_validContext def getNextTimeout(self): """ Returns the next internal timeout that libusb needs to handle, in seconds, or None if no timeout is needed. You should not have to call this method, unless you are integrating this class with a polling mechanism. """ timeval = libusb1.timeval() result = libusb1.libusb_get_next_timeout( self.__context_p, byref(timeval)) if result == 0: return None elif result == 1: return timeval.tv_sec + (timeval.tv_usec * 0.000001) raiseUSBError(result) @_validContext def setDebug(self, level): """ Set debugging level. Note: depending on libusb compilation settings, this might have no effect. """ libusb1.libusb_set_debug(self.__context_p, level) @_validContext def tryLockEvents(self): """ See libusb_try_lock_events doc. """ return libusb1.libusb_try_lock_events(self.__context_p) @_validContext def lockEvents(self): """ See libusb_lock_events doc. """ libusb1.libusb_lock_events(self.__context_p) @_validContext def lockEventWaiters(self): """ See libusb_lock_event_waiters doc. """ libusb1.libusb_lock_event_waiters(self.__context_p) @_validContext def waitForEvent(self, tv=0): """ See libusb_wait_for_event doc. """ if tv is None: tv = 0 tv_s = int(tv) real_tv = libusb1.timeval(tv_s, int((tv - tv_s) * 1000000)) libusb1.libusb_wait_for_event(self.__context_p, byref(real_tv)) @_validContext def unlockEventWaiters(self): """ See libusb_unlock_event_waiters doc. """ libusb1.libusb_unlock_event_waiters(self.__context_p) @_validContext def eventHandlingOK(self): """ See libusb_event_handling_ok doc. """ return libusb1.libusb_event_handling_ok(self.__context_p) @_validContext def unlockEvents(self): """ See libusb_unlock_events doc. """ libusb1.libusb_unlock_events(self.__context_p) @_validContext def handleEventsLocked(self): """ See libusb_handle_events_locked doc. """ # XXX: does tv parameter need to be exposed ? mayRaiseUSBError(libusb1.libusb_handle_events_locked( self.__context_p, _zero_tv_p, )) @_validContext def eventHandlerActive(self): """ See libusb_event_handler_active doc. """ return libusb1.libusb_event_handler_active(self.__context_p) @staticmethod def hasCapability(capability): """ Backward-compatibility alias for module-level hasCapability. """ return hasCapability(capability) @_validContext def hotplugRegisterCallback( self, callback, # pylint: disable=undefined-variable events=HOTPLUG_EVENT_DEVICE_ARRIVED | HOTPLUG_EVENT_DEVICE_LEFT, flags=HOTPLUG_ENUMERATE, vendor_id=HOTPLUG_MATCH_ANY, product_id=HOTPLUG_MATCH_ANY, dev_class=HOTPLUG_MATCH_ANY, # pylint: enable=undefined-variable ): """ Registers an hotplug callback. On success, returns an opaque value which can be passed to hotplugDeregisterCallback. Callback must accept the following positional arguments: - this USBContext instance - an USBDevice instance If device has left, configuration descriptors may not be available. Its device descriptor will be available. - event type, one of: HOTPLUG_EVENT_DEVICE_ARRIVED HOTPLUG_EVENT_DEVICE_LEFT Callback must return whether it must be unregistered (any true value to be unregistered, any false value to be kept registered). """ def wrapped_callback(context_p, device_p, event, _): assert addressof(context_p.contents) == addressof( self.__context_p.contents), (context_p, self.__context_p) device = USBDevice( self, device_p, # pylint: disable=undefined-variable event != HOTPLUG_EVENT_DEVICE_LEFT, # pylint: enable=undefined-variable ) self.__close_set.add(device) unregister = bool(callback( self, device, event, )) if unregister: del self.__hotplug_callback_dict[handle] return unregister handle = c_int() callback_p = libusb1.libusb_hotplug_callback_fn_p(wrapped_callback) mayRaiseUSBError(libusb1.libusb_hotplug_register_callback( self.__context_p, events, flags, vendor_id, product_id, dev_class, callback_p, None, byref(handle), )) handle = handle.value # Keep strong references assert handle not in self.__hotplug_callback_dict, ( handle, self.__hotplug_callback_dict, ) self.__hotplug_callback_dict[handle] = (callback_p, wrapped_callback) return handle @_validContext def hotplugDeregisterCallback(self, handle): """ Deregisters an hotplug callback. handle (opaque) Return value of a former hotplugRegisterCallback call. """ del self.__hotplug_callback_dict[handle] libusb1.libusb_hotplug_deregister_callback(self.__context_p, handle) del USBContext._validContext def getVersion(): """ Returns underlying libusb's version information as a 6-namedtuple (or 6-tuple if namedtuples are not avaiable): - major - minor - micro - nano - rc - describe Returns (0, 0, 0, 0, '', '') if libusb doesn't have required entry point. """ version = libusb1.libusb_get_version().contents return Version( version.major, version.minor, version.micro, version.nano, version.rc, version.describe, ) def hasCapability(capability): """ Tests feature presence. capability should be one of: CAP_HAS_CAPABILITY CAP_HAS_HOTPLUG CAP_HAS_HID_ACCESS CAP_SUPPORTS_DETACH_KERNEL_DRIVER """ return libusb1.libusb_has_capability(capability) class LibUSBContext(USBContext): """ Backward-compatibility alias for USBContext. """ def __init__(self): warnings.warn( 'LibUSBContext is being renamed to USBContext', DeprecationWarning, ) super(LibUSBContext, self).__init__()