# Copyright (C) 2009-2011 Wander Lairson Costa
#
# The following terms apply to all files associated
# with the software unless explicitly disclaimed in individual files.
#
# The authors hereby grant permission to use, copy, modify, distribute,
# and license this software and its documentation for any purpose, provided
# that existing copyright notices are retained in all copies and that this
# notice is included verbatim in any distributions. No written agreement,
# license, or royalty fee is required for any of the authorized uses.
# Modifications to this software may be copyrighted by their authors
# and need not follow the licensing terms described here, provided that
# the new terms are clearly indicated on the first page of each file where
# they apply.
#
# IN NO EVENT SHALL THE AUTHORS OR DISTRIBUTORS BE LIABLE TO ANY PARTY
# FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
# ARISING OUT OF THE USE OF THIS SOFTWARE, ITS DOCUMENTATION, OR ANY
# DERIVATIVES THEREOF, EVEN IF THE AUTHORS HAVE BEEN ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# THE AUTHORS AND DISTRIBUTORS SPECIFICALLY DISCLAIM ANY WARRANTIES,
# INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT. THIS SOFTWARE
# IS PROVIDED ON AN "AS IS" BASIS, AND THE AUTHORS AND DISTRIBUTORS HAVE
# NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR
# MODIFICATIONS.
from ctypes import *
import ctypes.util
import usb.util
from usb._debug import methodtrace
import logging
__author__ = 'Wander Lairson Costa'
__all__ = ['get_backend']
_logger = logging.getLogger('usb.backend.openusb')
class _usb_endpoint_desc(Structure):
_fields_ = [('bLength', c_uint8),
('bDescriptorType', c_uint8),
('bEndpointAddress', c_uint8),
('bmAttributes', c_uint8),
('wMaxPacketSize', c_uint16),
('bInterval', c_uint8),
('bRefresh', c_uint8),
('bSynchAddress', c_uint8)]
class _usb_interface_desc(Structure):
_fields_ = [('bLength', c_uint8),
('bDescriptorType', c_uint8),
('bInterfaceNumber', c_uint8),
('bAlternateSetting', c_uint8),
('bNumEndpoints', c_uint8),
('bInterfaceClass', c_uint8),
('bInterfaceSubClass', c_uint8),
('bInterfaceProtocol', c_uint8),
('iInterface', c_uint8)]
class _usb_config_desc(Structure):
_fields_ = [('bLength', c_uint8),
('bDescriptorType', c_uint8),
('wTotalLength', c_uint16),
('bNumInterfaces', c_uint8),
('bConfigurationValue', c_uint8),
('iConfiguration', c_uint8),
('bmAttributes', c_uint8),
('bMaxPower', c_uint8)]
class _usb_device_desc(Structure):
_fields_ = [('bLength', c_uint8),
('bDescriptorType', c_uint8),
('bcdUSB', c_uint16),
('bDeviceClass', c_uint8),
('bDeviceSubClass', c_uint8),
('bDeviceProtocol', c_uint8),
('bMaxPacketSize0', c_uint8),
('idVendor', c_uint16),
('idProduct', c_uint16),
('bcdDevice', c_uint16),
('iManufacturer', c_uint8),
('iProduct', c_uint8),
('iSerialNumber', c_uint8),
('bNumConfigurations', c_uint8)]
class _openusb_request_result(Structure):
_fields_ = [('status', c_int32),
('transfered_bytes', c_uint32)]
class _openusb_ctrl_request(Structure):
class _openusb_ctrl_setup(Structure):
_fields_ = [('bmRequestType', c_uint8),
('bRequest', c_uint8),
('wValue', c_uint16),
('wIndex', c_uint16)]
_fields_ = [('payload', POINTER(c_uint8)),
('length', c_uint32),
('timeout', c_uint32),
('flags', c_uint32),
('result', _openusb_request_result),
('next', c_void_p)]
class _openusb_intr_request(Structure):
_fields_ = [('interval', c_uint16),
('payload', POINTER(c_uint8)),
('length', c_uint32),
('timeout', c_uint32),
('flags', c_uint32),
('result', _openusb_request_result),
('next', c_void_p)]
class _openusb_bulk_request(Structure):
_fields_ = [('payload', POINTER(c_uint8)),
('length', c_uint32),
('timeout', c_uint32),
('flags', c_uint32),
('result', _openusb_request_result),
('next', c_void_p)]
class _openusb_isoc_pkts(Structure):
class _openusb_isoc_packet(Structure):
_fields_ = [('payload', POINTER(c_uint8)),
('length', c_uint32)]
_fields_ = [('num_packets', c_uint32),
('packets', POINTER(_openusb_isoc_packet))]
class _openusb_isoc_request(Structure):
_fields_ = [('start_frame', c_uint32),
('flags', c_uint32),
('pkts', _openusb_isoc_pkts),
('isoc_results', POINTER(_openusb_request_result)),
('isoc_status', c_int32),
('next', c_void_p)]
_openusb_devid = c_uint64
_openusb_busid = c_uint64
_openusb_handle = c_uint64
_openusb_dev_handle = c_uint64
_lib = None
_ctx = None
def _load_library():
libname = ctypes.util.find_library('openusb')
if libname is None:
raise OSError('USB library could not be found')
return CDLL(libname)
def _setup_prototypes(lib):
# int32_t openusb_init(uint32_t flags , openusb_handle_t *handle);
lib.openusb_init.argtypes = [c_uint32, POINTER(_openusb_handle)]
lib.openusb_init.restype = c_int32
# void openusb_fini(openusb_handle_t handle );
lib.openusb_fini.argtypes = [_openusb_handle]
# uint32_t openusb_get_busid_list(openusb_handle_t handle,
# openusb_busid_t **busids,
# uint32_t *num_busids);
lib.openusb_get_busid_list.argtypes = [
_openusb_handle,
POINTER(POINTER(_openusb_busid)),
POINTER(c_uint32)
]
# void openusb_free_busid_list(openusb_busid_t * busids);
lib.openusb_free_busid_list.argtypes = [POINTER(_openusb_busid)]
# uint32_t openusb_get_devids_by_bus(openusb_handle_t handle,
# openusb_busid_t busid,
# openusb_devid_t **devids,
# uint32_t *num_devids);
lib.openusb_get_devids_by_bus.argtypes = [
_openusb_handle,
_openusb_busid,
POINTER(POINTER(_openusb_devid)),
POINTER(c_uint32)
]
lib.openusb_get_devids_by_bus.restype = c_int32
# void openusb_free_devid_list(openusb_devid_t * devids);
lib.openusb_free_devid_list.argtypes = [POINTER(_openusb_devid)]
# int32_t openusb_open_device(openusb_handle_t handle,
# openusb_devid_t devid ,
# uint32_t flags,
# openusb_dev_handle_t *dev);
lib.openusb_open_device.argtypes = [
_openusb_handle,
_openusb_devid,
c_uint32,
POINTER(_openusb_dev_handle)
]
lib.openusb_open_device.restype = c_int32
# int32_t openusb_close_device(openusb_dev_handle_t dev);
lib.openusb_close_device.argtypes = [_openusb_dev_handle]
lib.openusb_close_device.restype = c_int32
# int32_t openusb_set_configuration(openusb_dev_handle_t dev,
# uint8_t cfg);
lib.openusb_set_configuration.argtypes = [_openusb_dev_handle, c_uint8]
lib.openusb_set_configuration.restype = c_int32
# int32_t openusb_get_configuration(openusb_dev_handle_t dev,
# uint8_t *cfg);
lib.openusb_get_configuration.argtypes = [_openusb_dev_handle, POINTER(c_uint8)]
lib.openusb_get_configuration.restype = c_int32
# int32_t openusb_claim_interface(openusb_dev_handle_t dev,
# uint8_t ifc,
# openusb_init_flag_t flags);
lib.openusb_claim_interface.argtypes = [
_openusb_dev_handle,
c_uint8,
c_int
]
lib.openusb_claim_interface.restype = c_int32
# int32_t openusb_release_interface(openusb_dev_handle_t dev,
# uint8_t ifc);
lib.openusb_release_interface.argtypes = [
_openusb_dev_handle,
c_uint8
]
lib.openusb_release_interface.restype = c_int32
# int32_topenusb_set_altsetting(openusb_dev_handle_t dev,
# uint8_t ifc,
# uint8_t alt);
lib.openusb_set_altsetting.argtypes = [
_openusb_dev_handle,
c_uint8,
c_uint8
]
lib.openusb_set_altsetting.restype = c_int32
# int32_t openusb_reset(openusb_dev_handle_t dev);
lib.openusb_reset.argtypes = [_openusb_dev_handle]
lib.openusb_reset.restype = c_int32
# int32_t openusb_parse_device_desc(openusb_handle_t handle,
# openusb_devid_t devid,
# uint8_t *buffer,
# uint16_t buflen,
# usb_device_desc_t *devdesc);
lib.openusb_parse_device_desc.argtypes = [
_openusb_handle,
_openusb_devid,
POINTER(c_uint8),
c_uint16,
POINTER(_usb_device_desc)
]
lib.openusb_parse_device_desc.restype = c_int32
# int32_t openusb_parse_config_desc(openusb_handle_t handle,
# openusb_devid_t devid,
# uint8_t *buffer,
# uint16_t buflen,
# uint8_t cfgidx,
# usb_config_desc_t *cfgdesc);
lib.openusb_parse_config_desc.argtypes = [
_openusb_handle,
_openusb_devid,
POINTER(c_uint8),
c_uint16,
c_uint8,
POINTER(_usb_config_desc)
]
lib.openusb_parse_config_desc.restype = c_int32
# int32_t openusb_parse_interface_desc(openusb_handle_t handle,
# openusb_devid_t devid,
# uint8_t *buffer,
# uint16_t buflen,
# uint8_t cfgidx,
# uint8_t ifcidx,
# uint8_t alt,
# usb_interface_desc_t *ifcdesc);
lib.openusb_parse_interface_desc.argtypes = [
_openusb_handle,
_openusb_devid,
POINTER(c_uint8),
c_uint16,
c_uint8,
c_uint8,
c_uint8,
POINTER(_usb_interface_desc)
]
lib.openusb_parse_interface_desc.restype = c_int32
# int32_t openusb_parse_endpoint_desc(openusb_handle_t handle,
# openusb_devid_t devid,
# uint8_t *buffer,
# uint16_t buflen,
# uint8_t cfgidx,
# uint8_t ifcidx,
# uint8_t alt,
# uint8_t eptidx,
# usb_endpoint_desc_t *eptdesc);
lib.openusb_parse_endpoint_desc.argtypes = [
_openusb_handle,
_openusb_devid,
POINTER(c_uint8),
c_uint16,
c_uint8,
c_uint8,
c_uint8,
c_uint8,
POINTER(_usb_endpoint_desc)
]
lib.openusb_parse_interface_desc.restype = c_int32
# const char *openusb_strerror(int32_t error );
lib.openusb_strerror.argtypes = [c_int32]
lib.openusb_strerror.restype = c_char_p
# int32_t openusb_ctrl_xfer(openusb_dev_handle_t dev,
# uint8_t ifc,
# uint8_t ept,
# openusb_ctrl_request_t *ctrl);
lib.openusb_ctrl_xfer.argtypes = [
_openusb_dev_handle,
c_uint8,
c_uint8,
POINTER(_openusb_ctrl_request)
]
lib.openusb_ctrl_xfer.restype = c_int32
# int32_t openusb_intr_xfer(openusb_dev_handle_t dev,
# uint8_t ifc,
# uint8_t ept,
# openusb_intr_request_t *intr);
lib.openusb_intr_xfer.argtypes = [
_openusb_dev_handle,
c_uint8,
c_uint8,
POINTER(_openusb_intr_request)
]
lib.openusb_bulk_xfer.restype = c_int32
# int32_t openusb_bulk_xfer(openusb_dev_handle_t dev,
# uint8_t ifc,
# uint8_t ept,
# openusb_bulk_request_t *bulk);
lib.openusb_bulk_xfer.argtypes = [
_openusb_dev_handle,
c_uint8,
c_uint8,
POINTER(_openusb_bulk_request)
]
lib.openusb_bulk_xfer.restype = c_int32
# int32_t openusb_isoc_xfer(openusb_dev_handle_t dev,
# uint8_t ifc,
# uint8_t ept,
# openusb_isoc_request_t *isoc);
lib.openusb_isoc_xfer.argtypes = [
_openusb_dev_handle,
c_uint8,
c_uint8,
POINTER(_openusb_isoc_request)
]
lib.openusb_isoc_xfer.restype = c_int32
def _check(retval):
if retval.value != 0:
from usb.core import USBError
raise USBError(_lib.openusb_strerror(retval).value)
return retval
class _Context(object):
def __init__(self):
self.handle = _openusb_handle()
_check(_lib.openusb_init(0, byref(self.handle)))
def __del__(self):
_lib.openusb_fini(self.handle)
class _BusIterator(object):
def __init__(self):
self.buslist = POINTER(openusb_busid)()
num_busids = c_uint32()
_check(_lib.openusb_get_busid_list(_ctx.handle,
byref(self.buslist),
byref(num_busids)))
self.num_busids = num_busids.value
def __iter__(self):
for i in range(self.num_busids):
yield self.buslist[i]
def __del__(self):
_lib.openusb_free_busid_list(self.buslist)
class _DevIterator(object):
def __init__(self, busid):
self.devlist = POINTER(_openusb_devid)()
num_devids = c_uint32()
_check(_lib.openusb_get_devids_by_bus(_ctx.handle,
busid,
byref(self.devlist),
byref(num_devids)))
self.num_devids = num_devids.value
def __iter__(self):
for i in range(self.num_devids):
yield self.devlist[i]
def __del__(self):
_lib.openusb_free_devid_list(self.devlist)
class _OpenUSB(usb.backend.IBackend):
@methodtrace(_logger)
def enumerate_devices(self):
for bus in _BusIterator():
for devid in _DevIterator(bus):
yield devid
@methodtrace(_logger)
def get_device_descriptor(self, dev):
desc = _usb_device_desc()
_check(_lib.openusb_parse_device_desc(_ctx.handle,
dev,
None,
0,
byref(desc)))
return desc
@methodtrace(_logger)
def get_configuration_descriptor(self, dev, config):
desc = _usb_config_desc()
_check(_lib.openusb_parse_config_desc(_ctx.handle,
dev,
None,
0,
config,
byref(desc)))
return desc
@methodtrace(_logger)
def get_interface_descriptor(self, dev, intf, alt, config):
desc = _usb_interface_desc()
_check(_lib.openusb_parse_interface_desc(_ctx.handle,
dev,
None,
0,
config,
intf,
alt,
byref(desc)))
return desc
@methodtrace(_logger)
def get_endpoint_descriptor(self, dev, ep, intf, alt, config):
desc = _usb_endpoint_desc()
_check(_lib.openusb_parse_endpoint_desc(_ctx.handle,
dev,
None,
0,
config,
intf,
alt,
ep,
byref(desc)))
return desc
@methodtrace(_logger)
def open_device(self, dev):
handle = _openusb_dev_handle()
_check(_lib.openusb_open_device(_ctx.handle, dev, 0, byref(handle)))
return handle
@methodtrace(_logger)
def close_device(self, dev_handle):
_lib.openusb_close_device(dev_handle)
@methodtrace(_logger)
def set_configuration(self, dev_handle, config_value):
_check(_lib.openusb_set_configuration(dev_handle, config_value))
@methodtrace(_logger)
def get_configuration(self, dev_handle):
config = c_uint8()
_check(_lib.openusb_get_configuration(dev_handle, byref(config)))
return config.value
@methodtrace(_logger)
def set_interface_altsetting(self, dev_handle, intf, altsetting):
_check(_lib.set_altsetting(dev_handle, intf, altsetting))
@methodtrace(_logger)
def claim_interface(self, dev_handle, intf):
_check(_lib.openusb_claim_interface(dev_handle, intf, 0))
@methodtrace(_logger)
def release_interface(self, dev_handle, intf):
_lib.openusb_release_interface(dev_handle, intf)
@methodtrace(_logger)
def bulk_write(self, dev_handle, ep, intf, data, timeout):
request = _openusb_bulk_request()
memset(byref(request), 0, sizeof(request))
request.payload, request.length = data.buffer_info()
request.timeout = timeout
_check(_lib.openusb_bulk_xfer(dev_handle, intf, ep, byref(request)))
return request.transfered_bytes.value
@methodtrace(_logger)
def bulk_read(self, dev_handle, ep, intf, size, timeout):
request = _openusb_bulk_request()
buffer = array.array('B', '\x00' * size)
memset(byref(request), 0, sizeof(request))
request.payload, request.length = buffer.buffer_info()
request.timeout = timeout
_check(_lib.openusb_bulk_xfer(dev_handle, intf, ep, byref(request)))
return buffer[:request.transfered_bytes.value]
@methodtrace(_logger)
def intr_write(self, dev_handle, ep, intf, data, timeout):
request = _openusb_intr_request()
memset(byref(request), 0, sizeof(request))
payload, request.length = data.buffer_info()
request.payload = cast(payload, POINTER(c_uint8))
request.timeout = timeout
_check(_lib.openusb_intr_xfer(dev_handle, intf, ep, byref(request)))
return request.transfered_bytes.value
@methodtrace(_logger)
def intr_read(self, dev_handle, ep, intf, size, timeout):
request = _openusb_intr_request()
buffer = array.array('B', '\x00' * size)
memset(byref(request), 0, sizeof(request))
payload, request.length = buffer.buffer_info()
request.payload = cast(payload, POINTER(c_uint8))
request.timeout = timeout
_check(_lib.openusb_intr_xfer(dev_handle, intf, ep, byref(request)))
return buffer[:request.transfered_bytes.value]
# TODO: implement isochronous
# @methodtrace(_logger)
# def iso_write(self, dev_handle, ep, intf, data, timeout):
# pass
# @methodtrace(_logger)
# def iso_read(self, dev_handle, ep, intf, size, timeout):
# pass
@methodtrace(_logger)
def ctrl_transfer(self,
dev_handle,
bmRequestType,
bRequest,
wValue,
wIndex,
data_or_wLength,
timeout):
request = _openusb_ctrl_request()
request.setup.bmRequestType = bmRequestType
request.setup.bRequest = bRequest
request.setup.wValue
request.setup.wIndex
request.timeout = timeout
direction = usb.util.ctrl_direction(bmRequestType)
if direction == ENDPOINT_OUT:
buffer = data_or_wLength
else:
buffer = array.array('B', '\x00' * data_or_wLength)
payload, request.length = buffer.buffer_info()
request.payload = cast(payload, POINTER(c_uint8))
ret = _check(_lib.openusb_ctrl_xfer(dev_handle, 0, 0, byref(request)))
if direction == ENDPOINT_OUT:
ret
else:
buffer[:ret]
@methodtrace(_logger)
def reset_device(self, dev_handle):
_check(_lib.openusb_reset(dev_handle))
def get_backend():
try:
global _lib, _ctx
if _lib is None:
_lib = _load_library()
_setup_prototypes(_lib)
_ctx = _Context()
return _OpenUSB()
except Exception:
_logger.error('Error loading OpenUSB backend', exc_info=True)
return None