tessl/pypi-libusb1

Pure-python wrapper for libusb-1.0 providing comprehensive USB device access with support for all transfer types

—

Pending

Overview

Eval results

Files

USB Context Management

Name: tessl/pypi-libusb1
Author: tessl

The USBContext class provides the entry point for all USB operations, managing system resources, device enumeration, and global USB settings. All USB operations require an active context, which handles low-level libusb-1.0 initialization and cleanup.

Capabilities

Context Creation and Lifecycle

Create and manage USB contexts with optional configuration for logging, Windows backend selection, and device discovery behavior.

class USBContext:
    def __init__(self, log_level=None, use_usbdk=False, with_device_discovery=True, log_callback=None):
        """
        Create a new USB context.
        
        Args:
            log_level: Log level constant (LOG_LEVEL_NONE, LOG_LEVEL_ERROR, etc.)
            use_usbdk (bool): Windows only - use UsbDk backend if available
            with_device_discovery (bool): Linux only - enable device scan during init
            log_callback: Function accepting (context, level, message) for log messages
        """

    def __enter__(self):
        """Context manager entry - initializes the context."""

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit - closes the context."""

    def open(self):
        """
        Finish context initialization manually.
        
        Returns:
            USBContext: Self for method chaining
        """

    def close(self):
        """Close and destroy the USB context, releasing all resources."""

Device Enumeration

Discover and iterate over USB devices connected to the system with flexible filtering and error handling options.

def getDeviceIterator(self, skip_on_error=False):
    """
    Return iterator over all USB devices currently connected.
    
    Args:
        skip_on_error (bool): If True, ignore devices that raise USBError
        
    Yields:
        USBDevice: Device instances for each connected device
    """

def getDeviceList(self, skip_on_access_error=False, skip_on_error=False):
    """
    Return list of all USB devices currently connected.
    
    Args:
        skip_on_error (bool): If True, ignore devices that raise USBError
        skip_on_access_error (bool): Deprecated alias for skip_on_error
        
    Returns:
        list[USBDevice]: List of device instances
    """

def getByVendorIDAndProductID(self, vendor_id, product_id, skip_on_access_error=False, skip_on_error=False):
    """
    Find first device matching vendor and product IDs.
    
    Args:
        vendor_id (int): USB vendor ID (16-bit)
        product_id (int): USB product ID (16-bit)
        skip_on_error (bool): If True, ignore devices that raise USBError
        skip_on_access_error (bool): Deprecated alias for skip_on_error
        
    Returns:
        USBDevice or None: Device instance or None if not found
    """

def openByVendorIDAndProductID(self, vendor_id, product_id, skip_on_access_error=False, skip_on_error=False):
    """
    Find and open first device matching vendor and product IDs.
    
    Args:
        vendor_id (int): USB vendor ID (16-bit)
        product_id (int): USB product ID (16-bit)
        skip_on_error (bool): If True, ignore devices that raise USBError
        skip_on_access_error (bool): Deprecated alias for skip_on_error
        
    Returns:
        USBDeviceHandle or None: Opened device handle or None if not found
    """

Event Handling

Handle USB events for asynchronous operations with timeout control and interrupt capabilities.

def handleEvents(self):
    """Handle pending USB events (blocking with internal timeout)."""

def handleEventsTimeout(self, tv=0):
    """
    Handle pending USB events with specified timeout.
    
    Args:
        tv (float): Timeout in seconds, 0 for immediate return
    """

def interruptEventHandler(self):
    """Interrupt active event handling thread."""

def getNextTimeout(self):
    """
    Get next internal timeout libusb needs to handle.
    
    Returns:
        float or None: Timeout in seconds or None if no timeout needed
    """

Hotplug Device Detection

def hotplugRegisterCallback(self, callback, 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):
    """
    Register hotplug callback for device arrival/departure events.
    
    Args:
        callback: Function accepting (context, device, event) returning bool
        events (int): Event mask (HOTPLUG_EVENT_DEVICE_ARRIVED | HOTPLUG_EVENT_DEVICE_LEFT)
        flags (int): Callback flags (HOTPLUG_ENUMERATE to get existing devices)
        vendor_id (int): Vendor ID filter or HOTPLUG_MATCH_ANY
        product_id (int): Product ID filter or HOTPLUG_MATCH_ANY
        dev_class (int): Device class filter or HOTPLUG_MATCH_ANY
        
    Returns:
        int: Opaque handle for hotplugDeregisterCallback
        
    Note:
        Callback must return True to unregister, False to stay registered.
        Callback cannot call synchronous libusb functions.
    """

def hotplugDeregisterCallback(self, handle):
    """
    Deregister hotplug callback.
    
    Args:
        handle (int): Handle returned by hotplugRegisterCallback
    """

System Device Wrapping

Wrap existing system device file descriptors for USB communication without device enumeration.

def wrapSysDevice(self, sys_device):
    """
    Wrap system device file descriptor as USBDeviceHandle.
    
    Args:
        sys_device: File object or file descriptor of sys device node
        
    Returns:
        USBDeviceHandle: Handle for wrapped device
        
    Note:
        Keep the file open while using the device handle.
    """

Advanced Event Loop Integration

Low-level file descriptor polling support for integration with select/poll-based event loops.

def getPollFDList(self):
    """
    Get file descriptors for USB event polling.
    
    Returns:
        list[tuple]: List of (fd, events) tuples for polling
        
    Raises:
        NotImplementedError: If libusb doesn't support pollable FDs
    """

def setPollFDNotifiers(self, added_cb=None, removed_cb=None, user_data=None):
    """
    Set callbacks for file descriptor addition/removal.
    
    Args:
        added_cb: Function called when FD is added: (fd, events, user_data)
        removed_cb: Function called when FD is removed: (fd, user_data)
        user_data: User data passed to callbacks
    """

Logging and Debugging

Configure context-specific logging and debug output for troubleshooting USB operations.

def setLogCallback(self, log_callback):
    """
    Set context-specific log callback.
    
    Args:
        log_callback: Function accepting (context, level, message) or None to disable
    """

def setDebug(self, level):
    """
    Set debug level for this context.
    
    Args:
        level (int): Log level constant (LOG_LEVEL_NONE through LOG_LEVEL_DEBUG)
    """

Usage Examples

Basic Context Usage

import usb1

# Simple context manager usage
with usb1.USBContext() as context:
    devices = context.getDeviceList()
    print(f"Found {len(devices)} USB devices")

# Manual context management
context = usb1.USBContext()
context.open()
try:
    device = context.getByVendorIDAndProductID(0x1234, 0x5678)
    if device:
        print(f"Found device: {device}")
finally:
    context.close()

Device Enumeration with Error Handling

import usb1

with usb1.USBContext() as context:
    # Iterate with automatic error skipping
    for device in context.getDeviceIterator(skip_on_error=True):
        try:
            print(f"Device {device.getVendorID():04x}:{device.getProductID():04x}")
            print(f"  Manufacturer: {device.getManufacturer()}")
            print(f"  Product: {device.getProduct()}")
        except usb1.USBError as e:
            print(f"  Error accessing device: {e}")

Hotplug Event Monitoring

import usb1
import threading
import time

def hotplug_callback(context, device, event):
    if event == usb1.HOTPLUG_EVENT_DEVICE_ARRIVED:
        print(f"Device connected: {device.getVendorID():04x}:{device.getProductID():04x}")
    elif event == usb1.HOTPLUG_EVENT_DEVICE_LEFT:
        print(f"Device disconnected: {device.getVendorID():04x}:{device.getProductID():04x}")
    return False  # Keep callback registered

with usb1.USBContext() as context:
    # Register hotplug callback
    handle = context.hotplugRegisterCallback(hotplug_callback)
    
    # Event handling loop
    try:
        while True:
            context.handleEventsTimeout(1.0)  # 1 second timeout
    except KeyboardInterrupt:
        print("Stopping...")
    finally:
        context.hotplugDeregisterCallback(handle)

Event Loop Integration

import usb1
import select

with usb1.USBContext() as context:
    # Get file descriptors for polling
    try:
        poll_fds = context.getPollFDList()
        print(f"Monitoring {len(poll_fds)} file descriptors")
        
        # Integration with select
        read_fds = [fd for fd, events in poll_fds if events & select.POLLIN]
        write_fds = [fd for fd, events in poll_fds if events & select.POLLOUT]
        
        # Poll with timeout
        ready_read, ready_write, _ = select.select(read_fds, write_fds, [], 1.0)
        
        if ready_read or ready_write:
            context.handleEventsTimeout(0)  # Process immediately
            
    except NotImplementedError:
        print("Platform doesn't support pollable file descriptors")
        # Fall back to timeout-based event handling
        context.handleEventsTimeout(1.0)

Install with Tessl CLI