From 95e250643c3711f62f810a827f8d29ee11ceeb54 Mon Sep 17 00:00:00 2001 From: Anne Barela <1911920+TheKitty@users.noreply.github.com> Date: Mon, 4 Aug 2025 20:49:17 -0500 Subject: [PATCH] Enhance number of functions for USB Host With multiple USB host ports on boards like Adafruit Fruit Jam, provide a robust library of functions for checking what kind of devices are attached. This is important when a game or program wants to check the correct hardware is enumerating on the USB bus. Two tests, a simpletest and a more thorough test are in examples. --- adafruit_usb_host_descriptors.py | 1237 +++++++++++++++++++++-- examples/usb_devices_detection.py | 892 ++++++++++++++++ examples/usb_host_devices_simpletest.py | 304 ++++++ 3 files changed, 2352 insertions(+), 81 deletions(-) create mode 100644 examples/usb_devices_detection.py create mode 100644 examples/usb_host_devices_simpletest.py diff --git a/adafruit_usb_host_descriptors.py b/adafruit_usb_host_descriptors.py index c02c1ba..00d3563 100644 --- a/adafruit_usb_host_descriptors.py +++ b/adafruit_usb_host_descriptors.py @@ -1,133 +1,1208 @@ +# SPDX-FileCopyrightText: 2017 Scott Shawcroft, written for Adafruit Industries # SPDX-FileCopyrightText: Copyright (c) 2023 Scott Shawcroft for Adafruit Industries +# SPDX-FileCopyrightText: Copyright (c) 2025 Anne Barela for Adafruit Industries # # SPDX-License-Identifier: MIT + """ `adafruit_usb_host_descriptors` ================================================================================ -Helpers for getting USB descriptors - -* Author(s): Scott Shawcroft """ -import struct +import usb.core -from micropython import const - -try: - from typing import Literal -except ImportError: - pass +# imports __version__ = "0.0.0+auto.0" __repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_USB_Host_Descriptors.git" +# USB Descriptor Type constants +DESC_DEVICE = 1 +DESC_CONFIGURATION = 2 +DESC_STRING = 3 +DESC_INTERFACE = 4 +DESC_ENDPOINT = 5 +DESC_DEVICE_QUALIFIER = 6 +DESC_OTHER_SPEED_CONFIGURATION = 7 +DESC_INTERFACE_POWER = 8 +DESC_OTG = 9 +DESC_DEBUG = 10 +DESC_INTERFACE_ASSOCIATION = 11 -# USB defines -# Use const for these internal values so that they are inlined with mpy-cross. -_DIR_OUT = const(0x00) -_DIR_IN = const(0x80) +# USB Class codes +CLASS_PER_INTERFACE = 0x00 +CLASS_AUDIO = 0x01 +CLASS_COMM = 0x02 +CLASS_HID = 0x03 +CLASS_PHYSICAL = 0x05 +CLASS_IMAGE = 0x06 +CLASS_PRINTER = 0x07 +CLASS_MASS_STORAGE = 0x08 +CLASS_HUB = 0x09 +CLASS_DATA = 0x0A +CLASS_SMART_CARD = 0x0B +CLASS_CONTENT_SECURITY = 0x0D +CLASS_VIDEO = 0x0E +CLASS_PERSONAL_HEALTHCARE = 0x0F +CLASS_DIAGNOSTIC_DEVICE = 0xDC +CLASS_WIRELESS = 0xE0 +CLASS_APPLICATION = 0xFE +CLASS_VENDOR_SPECIFIC = 0xFF -_REQ_RCPT_DEVICE = const(0) - -_REQ_TYPE_STANDARD = const(0x00) - -_REQ_GET_DESCRIPTOR = const(6) - -# No const because these are public -DESC_DEVICE = 0x01 -DESC_CONFIGURATION = 0x02 -DESC_STRING = 0x03 -DESC_INTERFACE = 0x04 -DESC_ENDPOINT = 0x05 - -INTERFACE_HID = 0x03 -SUBCLASS_BOOT = 0x01 -PROTOCOL_MOUSE = 0x02 -PROTOCOL_KEYBOARD = 0x01 +# Direction constants +DIR_OUT = 0x00 +DIR_IN = 0x80 def get_descriptor(device, desc_type, index, buf, language_id=0): """Fetch the descriptor from the device into buf.""" - # Allow capitalization that matches the USB spec. - # pylint: disable=invalid-name - wValue = desc_type << 8 | index - wIndex = language_id - device.ctrl_transfer( - _REQ_RCPT_DEVICE | _REQ_TYPE_STANDARD | _DIR_IN, - _REQ_GET_DESCRIPTOR, - wValue, - wIndex, - buf, + return device.ctrl_transfer( + DIR_IN | 0x00, # bmRequestType + 6, # bRequest (GET_DESCRIPTOR) + (desc_type << 8) | index, # wValue + language_id, # wIndex + buf, # data_or_wLength ) def get_device_descriptor(device): """Fetch the device descriptor and return it.""" - buf = bytearray(1) - get_descriptor(device, DESC_DEVICE, 0, buf) - full_buf = bytearray(buf[0]) - get_descriptor(device, DESC_DEVICE, 0, full_buf) - return full_buf + buf = bytearray(18) # Device descriptor is always 18 bytes + length = get_descriptor(device, DESC_DEVICE, 0, buf) + if length != 18: + raise ValueError("Invalid device descriptor length") + return buf def get_configuration_descriptor(device, index): """Fetch the configuration descriptor, its associated descriptors and return it.""" - # Allow capitalization that matches the USB spec. - # pylint: disable=invalid-name - buf = bytearray(4) - get_descriptor(device, DESC_CONFIGURATION, index, buf) - wTotalLength = struct.unpack(" old_count: + # Device added + for callback in _device_change_callbacks: + try: + # Remove 's' from plural + callback(device_type[:-1], 'added', new_count - 1) + except Exception as e: + print(f"Callback error: {e}") + elif new_count < old_count: + # Device removed + for callback in _device_change_callbacks: + try: + callback(device_type[:-1], 'removed', old_count - 1) + except Exception as e: + print(f"Callback error: {e}") + +def _refresh_device_cache(force_refresh=False): + """ + Refresh the internal device cache by scanning all USB devices. + + @param force_refresh: Force a refresh even if recently scanned + @type force_refresh: bool + @raises usb.core.USBError: If USB scanning fails + @raises ImportError: If time module is not available + """ + import time + + # Only refresh if forced or cache is empty/old + current_time = time.monotonic() + if not force_refresh and _device_cache['last_scan'] is not None: + if _cache_timeout > 0 and current_time - _device_cache['last_scan'] \ + < _cache_timeout: + return + + # Store old cache for change detection + old_cache = { + 'keyboards': _device_cache['keyboards'].copy(), + 'mice': _device_cache['mice'].copy(), + 'gamepads': _device_cache['gamepads'].copy() + } + + # Clear cache + _device_cache['keyboards'].clear() + _device_cache['mice'].clear() + _device_cache['gamepads'].clear() + _device_cache['composite_devices'].clear() + + # Scan all connected USB devices + try: + for device in usb.core.find(find_all=True): + _analyze_device(device) + except usb.core.USBError as e: + print(f"USB error scanning devices: {e}") + except usb.core.NoBackendError as e: + print(f"USB backend error: {e}") + + _device_cache['last_scan'] = current_time + + # Notify callbacks of changes + if _device_change_callbacks: + _notify_device_changes(old_cache, _device_cache) + +def _analyze_device(device): + """ + Analyze a USB device to determine if it's a keyboard, mouse, or gamepad. + Properly handles composite devices with multiple HID interfaces. + + @param device: USB device object from usb.core.find() + @type device: usb.core.Device + @raises usb.core.USBError: If device descriptor cannot be read + @raises IndexError: If descriptor parsing encounters invalid data + @raises AttributeError: If device lacks expected attributes + """ + try: + # Get configuration descriptor + config_descriptor = get_configuration_descriptor(device, 0) + + # Collect all HID interfaces for this device + hid_interfaces = [] + + # Parse descriptor to find HID interfaces + i = 0 + while i < len(config_descriptor): + descriptor_len = config_descriptor[i] + descriptor_type = config_descriptor[i + 1] + + if descriptor_type == DESC_INTERFACE: + interface_number = config_descriptor[i + 2] + interface_class = config_descriptor[i + 5] + interface_subclass = config_descriptor[i + 6] + interface_protocol = config_descriptor[i + 7] + + # Check if this is an HID interface + if interface_class == HID_CLASS: + # Find the corresponding input endpoint + endpoint_address = _find_input_endpoint(config_descriptor, + i + descriptor_len) + + if endpoint_address is not None: + interface_info = { + 'device': device, + 'interface_index': interface_number, + 'endpoint_address': endpoint_address, + 'interface_class': interface_class, + 'interface_subclass': interface_subclass, + 'interface_protocol': interface_protocol, + 'vid': device.idVendor, + 'pid': device.idProduct, + 'manufacturer': getattr(device, 'manufacturer', 'Unknown'), + 'product': getattr(device, 'product', 'Unknown') + } + + # Apply device filters + if _device_passes_filter(interface_info): + hid_interfaces.append(interface_info) + + i += descriptor_len + + # Process all HID interfaces for this device + if hid_interfaces: + device_id = f"{device.idVendor:04x}:{device.idProduct:04x}:\ + {getattr(device, 'serial_number', 'no_serial')}" + interfaces_by_type = {} + + # Classify each interface individually + for interface_info in hid_interfaces: + interface_type = _classify_hid_device(interface_info) + if interface_type not in interfaces_by_type: + interfaces_by_type[interface_type] = [] + interfaces_by_type[interface_type].append(interface_info) + + # Store device in ALL applicable categories + for device_type, interfaces in interfaces_by_type.items(): + # For each type, select the best interface (boot protocol preferred) + best_interface = _select_best_interface_for_type(interfaces, + device_type) + + if device_type == 'keyboard' and not _device_already_cached(device, + 'keyboards' + ): + _device_cache['keyboards'].append(best_interface) + elif device_type == 'mouse' and not _device_already_cached(device, + 'mice'): + _device_cache['mice'].append(best_interface) + elif device_type == 'gamepad' and not _device_already_cached(device, + 'gamepads' + ): + _device_cache['gamepads'].append(best_interface) + + # Track composite devices for reference + if len(interfaces_by_type) > 1: + _device_cache['composite_devices'][device_id] = { + 'device': device, + 'interfaces_by_type': interfaces_by_type, + 'product': getattr(device, 'product', 'Unknown'), + 'types': list(interfaces_by_type.keys()) + } + + except usb.core.USBError as e: + # Silently ignore USB errors for devices we can't access + pass + except IndexError as e: + print(f"Descriptor parsing error for device {device}: {e}") + except AttributeError as e: + print(f"Device attribute error for device {device}: {e}") + +def _select_best_interface_for_type(interfaces, device_type): + """ + Select the best interface for a specific device type. + Prefers boot protocol interfaces when available. + """ + boot_interfaces = [iface for iface in interfaces + if iface['interface_subclass'] == BOOT_SUBCLASS] + + if boot_interfaces: + # Prefer boot protocol interfaces + for iface in boot_interfaces: + expected_protocol = { + 'keyboard': KEYBOARD_PROTOCOL, + 'mouse': MOUSE_PROTOCOL + }.get(device_type, NO_PROTOCOL) + + if iface['interface_protocol'] == expected_protocol: + return iface + return boot_interfaces[0] # Any boot interface is better than non-boot + + return interfaces[0] # Fallback to first interface + +def _device_already_cached(device, device_type): + """ + Check if a device is already in the cache to avoid duplicates. + + @param device: USB device object + @type device: usb.core.Device + @param device_type: Type of device cache to check ('keyboards', 'mice', 'gamepads') + @type device_type: str + @return: True if device already cached, False otherwise + @rtype: bool + """ + try: + for cached_device_info in _device_cache[device_type]: + cached_device = cached_device_info['device'] + if (cached_device.idVendor == device.idVendor and + cached_device.idProduct == device.idProduct and + getattr(cached_device, 'serial_number', None) == + getattr(device, 'serial_number', None)): + return True + except (KeyError, AttributeError): + pass + return False + +def _find_input_endpoint(config_descriptor, start_index): + """ + Find the input endpoint address for an HID interface. + + @param config_descriptor: Configuration descriptor bytes + @type config_descriptor: bytes + @param start_index: Index to start searching from + @type start_index: int + @return: Endpoint address or None if not found + @rtype: int or None + @raises IndexError: If descriptor indices are out of bounds + """ + try: + i = start_index + while i < len(config_descriptor): + descriptor_len = config_descriptor[i] + descriptor_type = config_descriptor[i + 1] + + if descriptor_type == DESC_ENDPOINT: + endpoint_address = config_descriptor[i + 2] + # Return first input endpoint found + if endpoint_address & DIR_IN: + return endpoint_address + elif descriptor_type == DESC_INTERFACE: + # Reached next interface, stop searching + break + + i += descriptor_len + + except IndexError as e: + print(f"Index error finding endpoint: {e}") + + return None + +def _classify_hid_device(device_info): + """ + Classify an HID device as keyboard, mouse, or gamepad. + + @param device_info: Dictionary with device information + @type device_info: dict + @return: Device type classification + @rtype: str + @retval 'keyboard': Device is classified as a keyboard + @retval 'mouse': Device is classified as a mouse + @retval 'gamepad': Device is classified as a gamepad + @retval 'unknown': Device type could not be determined + """ + protocol = device_info['interface_protocol'] + subclass = device_info['interface_subclass'] + vid = device_info['vid'] + pid = device_info['pid'] + product = device_info['product'].lower() if device_info['product'] else '' + + # Boot protocol devices are easy to identify + if subclass == BOOT_SUBCLASS: + if protocol == KEYBOARD_PROTOCOL: + return 'keyboard' + elif protocol == MOUSE_PROTOCOL: + return 'mouse' + + # For non-boot HID devices, use heuristics + # Check product name for common keywords + keyboard_keywords = ['keyboard', 'kbd', 'keypad'] + mouse_keywords = ['mouse', 'pointer', 'trackball', 'touchpad', 'trackpad'] + gamepad_keywords = ['gamepad', 'controller', 'joystick', 'xbox', 'playstation', + 'ps3', 'ps4', 'ps5'] + + product_lower = product.lower() + + for keyword in keyboard_keywords: + if keyword in product_lower: + return 'keyboard' + + for keyword in mouse_keywords: + if keyword in product_lower: + return 'mouse' + + for keyword in gamepad_keywords: + if keyword in product_lower: + return 'gamepad' + + # Check known VID/PID combinations for common devices + known_keyboards = [ + (0x046D, 0xC52B), # Logitech keyboards + (0x413C, 0x2113), # Dell keyboards + (0x045E, 0x0750), # Microsoft keyboards + ] + + known_mice = [ + (0x046D, 0xC077), # Logitech mice + (0x1532, 0x0037), # Razer mice + (0x045E, 0x0040), # Microsoft mice + ] + + known_gamepads = [ + (0x045E, 0x028E), # Xbox 360 Controller + (0x045E, 0x02D1), # Xbox One Controller + (0x054C, 0x0268), # PS3 Controller + (0x054C, 0x05C4), # PS4 Controller + (0x054C, 0x0CE6), # PS5 Controller + ] + + if (vid, pid) in known_keyboards: + return 'keyboard' + elif (vid, pid) in known_mice: + return 'mouse' + elif (vid, pid) in known_gamepads: + return 'gamepad' + + # Default: if it's not boot protocol and we can't identify it, + # assume it's a gamepad (most non-boot HID devices are) + return 'gamepad' + +# ============================================================================ +# PUBLIC API FUNCTIONS +# ============================================================================ + +def count_keyboards(refresh=False): + """ + Count the number of keyboards connected to USB host ports. + + This function scans all connected USB devices and counts those identified + as keyboards using boot protocol detection and device classification heuristics. + + @param refresh: Force a device scan refresh instead of using cached results + @type refresh: bool + @return: Number of keyboards detected + @rtype: int + @retval 0: No keyboards detected + @retval 1+: Number of keyboards detected + @raises usb.core.USBError: If USB device scanning fails + + @par Example: + @code + num_keyboards = count_keyboards() + if num_keyboards > 0: + print(f"Found {num_keyboards} keyboard(s)") + @endcode + """ + _refresh_device_cache(refresh) + return len(_device_cache['keyboards']) + +def count_mice(refresh=False): + """ + Count the number of mice connected to USB host ports. + + This function scans all connected USB devices and counts those identified + as mice using boot protocol detection and device classification heuristics. + Includes trackpads from composite keyboard+trackpad devices. + + @param refresh: Force a device scan refresh instead of using cached results + @type refresh: bool + @return: Number of mice detected + @rtype: int + @retval 0: No mice detected + @retval 1+: Number of mice detected + @raises usb.core.USBError: If USB device scanning fails + + @par Example: + @code + num_mice = count_mice() + if num_mice > 0: + print(f"Found {num_mice} mouse/mice") + @endcode + """ + _refresh_device_cache(refresh) + return len(_device_cache['mice']) + +def count_gamepads(refresh=False): + """ + Count the number of gamepads connected to USB host ports. + + This function scans all connected USB devices and counts those identified + as gamepads including Xbox controllers, PlayStation controllers, and generic + USB game controllers. + + @param refresh: Force a device scan refresh instead of using cached results + @type refresh: bool + @return: Number of gamepads detected + @rtype: int + @retval 0: No gamepads detected + @retval 1+: Number of gamepads detected + @raises usb.core.USBError: If USB device scanning fails + + @par Example: + @code + num_gamepads = count_gamepads() + if num_gamepads > 0: + print(f"Found {num_gamepads} gamepad(s)") + @endcode + """ + _refresh_device_cache(refresh) + return len(_device_cache['gamepads']) + +def get_keyboard_info(index): + """ + Get interface index and endpoint address for a keyboard. + + Returns the USB interface index and input endpoint address needed to + communicate with the specified keyboard. This information is required + for reading input reports from the keyboard. + + @param index: Keyboard index (0 for first keyboard, 1 for second, etc.) + @type index: int + @return: Tuple of (interface_index, endpoint_address) or (None, None) if not found + @rtype: tuple[int, int] or tuple[None, None] + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + interface_index, endpoint_address = get_keyboard_info(0) # First keyboard + if interface_index is not None: + print(f"Keyboard interface: + {interface_index}, endpoint: 0x{endpoint_address:02x}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['keyboards']): + device_info = _device_cache['keyboards'][index] + return device_info['interface_index'], device_info['endpoint_address'] + except (IndexError, KeyError) as e: + print(f"Error accessing keyboard info: {e}") + + return None, None + +def get_mouse_info(index): + """ + Get interface index and endpoint address for a mouse. + + Returns the USB interface index and input endpoint address needed to + communicate with the specified mouse. This information is required + for reading input reports from the mouse. + + @param index: Mouse index (0 for first mouse, 1 for second, etc.) + @type index: int + @return: Tuple of (interface_index, endpoint_address) or (None, None) if not found + @rtype: tuple[int, int] or tuple[None, None] + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + interface_index, endpoint_address = get_mouse_info(0) # First mouse + if interface_index is not None: + print(f"Mouse interface: {interface_index}, endpoint: 0x{endpoint_address:02x}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['mice']): + device_info = _device_cache['mice'][index] + return device_info['interface_index'], device_info['endpoint_address'] + except (IndexError, KeyError) as e: + print(f"Error accessing mouse info: {e}") + + return None, None + +def get_gamepad_info(index): + """ + Get interface index and endpoint address for a gamepad. + + Returns the USB interface index and input endpoint address needed to + communicate with the specified gamepad. This information is required + for reading input reports from the gamepad. + + @param index: Gamepad index (0 for first gamepad, 1 for second, etc.) + @type index: int + @return: Tuple of (interface_index, endpoint_address) or (None, None) if not found + @rtype: tuple[int, int] or tuple[None, None] + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + interface_index, endpoint_address = get_gamepad_info(0) # First gamepad + if interface_index is not None: + print(f"Gamepad interface: {interface_index}, + endpoint: 0x{endpoint_address:02x}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['gamepads']): + device_info = _device_cache['gamepads'][index] + return device_info['interface_index'], device_info['endpoint_address'] + except (IndexError, KeyError) as e: + print(f"Error accessing gamepad info: {e}") + + return None, None + +def get_keyboard_device(index): + """ + Get the USB device object for a keyboard. + + Returns the underlying USB device object that can be used for direct + communication with the keyboard using PyUSB functions. + + @param index: Keyboard index (0 for first keyboard, 1 for second, etc.) + @type index: int + @return: USB device object or None if not found + @rtype: usb.core.Device or None + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + device = get_keyboard_device(0) + if device: + print(f"Keyboard VID:PID = {device.idVendor:04x}:{device.idProduct:04x}") + print(f"Manufacturer: {device.manufacturer}") + print(f"Product: {device.product}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['keyboards']): + return _device_cache['keyboards'][index]['device'] + except (IndexError, KeyError) as e: + print(f"Error accessing keyboard device: {e}") + + return None + +def get_mouse_device(index): + """ + Get the USB device object for a mouse. + + Returns the underlying USB device object that can be used for direct + communication with the mouse using PyUSB functions. + + @param index: Mouse index (0 for first mouse, 1 for second, etc.) + @type index: int + @return: USB device object or None if not found + @rtype: usb.core.Device or None + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + device = get_mouse_device(0) + if device: + print(f"Mouse VID:PID = {device.idVendor:04x}:{device.idProduct:04x}") + print(f"Manufacturer: {device.manufacturer}") + print(f"Product: {device.product}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['mice']): + return _device_cache['mice'][index]['device'] + except (IndexError, KeyError) as e: + print(f"Error accessing mouse device: {e}") + + return None + +def get_gamepad_device(index): + """ + Get the USB device object for a gamepad. + + Returns the underlying USB device object that can be used for direct + communication with the gamepad using PyUSB functions. + + @param index: Gamepad index (0 for first gamepad, 1 for second, etc.) + @type index: int + @return: USB device object or None if not found + @rtype: usb.core.Device or None + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Example: + @code + device = get_gamepad_device(0) + if device: + print(f"Gamepad VID:PID = {device.idVendor:04x}:{device.idProduct:04x}") + print(f"Manufacturer: {device.manufacturer}") + print(f"Product: {device.product}") + @endcode + """ + if index < 0: + raise IndexError("Device index cannot be negative") + + _refresh_device_cache() + + try: + if index < len(_device_cache['gamepads']): + return _device_cache['gamepads'][index]['device'] + except (IndexError, KeyError) as e: + print(f"Error accessing gamepad device: {e}") + + return None + +def list_all_hid_devices(): + """ + Get a comprehensive summary of all detected HID devices. + + Returns a dictionary containing counts and detailed information for all + keyboards, mice, and gamepads detected on the USB host ports. + + @return: Summary dictionary with device counts and lists + @rtype: dict + @raises usb.core.USBError: If USB device scanning fails + + @par Return Dictionary Structure: + @code + { + 'keyboards': { + 'count': int, + 'devices': [ + { + 'index': int, + 'vid': int, + 'pid': int, + 'manufacturer': str, + 'product': str, + 'interface_index': int, + 'endpoint_address': int + }, + ... + ] + }, + 'mice': { ... }, + 'gamepads': { ... } + } + @endcode + + @par Example: + @code + summary = list_all_hid_devices() + print(f"Found {summary['keyboards']['count']} keyboards") + print(f"Found {summary['mice']['count']} mice") + print(f"Found {summary['gamepads']['count']} gamepads") + + for device in summary['keyboards']['devices']: + print(f"Keyboard[{device['index']}]: {device['manufacturer']} + {device['product']}") + @endcode + """ + _refresh_device_cache(force_refresh=True) + + try: + return { + 'keyboards': { + 'count': len(_device_cache['keyboards']), + 'devices': [ + { + 'index': i, + 'vid': dev['vid'], + 'pid': dev['pid'], + 'manufacturer': dev['manufacturer'], + 'product': dev['product'], + 'interface_index': dev['interface_index'], + 'endpoint_address': dev['endpoint_address'] + } + for i, dev in enumerate(_device_cache['keyboards']) + ] + }, + 'mice': { + 'count': len(_device_cache['mice']), + 'devices': [ + { + 'index': i, + 'vid': dev['vid'], + 'pid': dev['pid'], + 'manufacturer': dev['manufacturer'], + 'product': dev['product'], + 'interface_index': dev['interface_index'], + 'endpoint_address': dev['endpoint_address'] + } + for i, dev in enumerate(_device_cache['mice']) + ] + }, + 'gamepads': { + 'count': len(_device_cache['gamepads']), + 'devices': [ + { + 'index': i, + 'vid': dev['vid'], + 'pid': dev['pid'], + 'manufacturer': dev['manufacturer'], + 'product': dev['product'], + 'interface_index': dev['interface_index'], + 'endpoint_address': dev['endpoint_address'] + } + for i, dev in enumerate(_device_cache['gamepads']) + ] + } + } + except KeyError as e: + print(f"Error building device summary: {e}") + return {'keyboards': {'count': 0, 'devices': []}, + 'mice': {'count': 0, 'devices': []}, + 'gamepads': {'count': 0, 'devices': []}} + +def get_composite_device_info(): + """ + Get information about composite devices (devices with multiple HID interfaces). + + @return: Dictionary of composite devices + @rtype: dict + + @par Example: + @code + composite_devices = get_composite_device_info() + for device_id, info in composite_devices.items(): + print(f"Composite device: {info['product']}") + print(f" Types: {', '.join(info['types'])}") + @endcode + """ + _refresh_device_cache() + return _device_cache['composite_devices'].copy() + +def is_composite_device(device_type, index): + """ + Check if a device at given index is part of a composite device. + + @param device_type: 'keyboard', 'mouse', or 'gamepad' + @param index: Device index + @return: Tuple of (is_composite, other_types) + @rtype: tuple[bool, list] + + @par Example: + @code + is_composite, other_types = is_composite_device('keyboard', 0) + if is_composite: + print(f"Keyboard also has: {other_types}") + @endcode + """ + try: + if device_type == 'keyboard': + device = get_keyboard_device(index) + elif device_type == 'mouse': + device = get_mouse_device(index) + elif device_type == 'gamepad': + device = get_gamepad_device(index) + else: + return False, [] + + if device is None: + return False, [] + + device_id = f"{device.idVendor:04x}:{device.idProduct:04x}: \ + {getattr(device, 'serial_number', 'no_serial')}" + + if device_id in _device_cache['composite_devices']: + composite_info = _device_cache['composite_devices'][device_id] + other_types = [t for t in composite_info['types'] if t != device_type] + return True, other_types + + return False, [] + + except Exception: + return False, [] + +def get_companion_interfaces(device_type, index): + """ + Get companion interfaces for a composite device. + For example, if you have a keyboard+trackpad, calling this with + ('keyboard', 0) will return the mouse interface info. + + @param device_type: Current device type ('keyboard', 'mouse', or 'gamepad') + @param index: Device index + @return: Dictionary mapping companion types to their interface info + @rtype: dict + + @par Example: + @code + companions = get_companion_interfaces('keyboard', 0) + if 'mouse' in companions: + print("This keyboard has a trackpad!") + trackpad_info = companions['mouse'] + print(f"Trackpad interface: {trackpad_info['interface_index']}") + @endcode + """ + try: + # Get the device + if device_type == 'keyboard': + device = get_keyboard_device(index) + elif device_type == 'mouse': + device = get_mouse_device(index) + elif device_type == 'gamepad': + device = get_gamepad_device(index) + else: + return {} + + if device is None: + return {} + + device_id = f"{device.idVendor:04x}:{device.idProduct:04x}: \ + {getattr(device, 'serial_number', 'no_serial')}" + + if device_id in _device_cache['composite_devices']: + composite_info = _device_cache['composite_devices'][device_id] + companions = {} + + for companion_type, interfaces in \ + composite_info['interfaces_by_type'].items(): + if companion_type != device_type: + best_interface = _select_best_interface_for_type(interfaces, + companion_type) + companions[companion_type] = { + 'interface_index': best_interface['interface_index'], + 'endpoint_address': best_interface['endpoint_address'], + 'interface_info': best_interface + } + + return companions + + return {} + + except Exception as e: + print(f"Error getting companion interfaces: {e}") + return {} + +def force_refresh(): + """ + Force a complete refresh of the device cache. + + Forces an immediate rescan of all connected USB devices, bypassing the + normal cache timeout. Call this when you know devices have been connected + or disconnected and need immediate updated results. + + @raises usb.core.USBError: If USB device scanning fails + @raises usb.core.NoBackendError: If USB backend is not available + + @par Example: + @code + # Connect a new keyboard, then force refresh + force_refresh() + keyboards = count_keyboards() # Will reflect newly connected device + @endcode + """ + _refresh_device_cache(force_refresh=True) + +def is_device_connected(device_type, index): + """ + Check if a device is still connected and responding. + + @param device_type: 'keyboard', 'mouse', or 'gamepad' + @param index: Device index + @return: True if device is connected and responding + @rtype: bool + + @par Example: + @code + if is_device_connected('keyboard', 0): + print("Keyboard is still connected") + else: + print("Keyboard has been disconnected") + @endcode + """ + try: + device = None + if device_type == 'keyboard': + device = get_keyboard_device(index) + elif device_type == 'mouse': + device = get_mouse_device(index) + elif device_type == 'gamepad': + device = get_gamepad_device(index) + + if device is None: + return False + + # Try to read device descriptor to verify connectivity + try: + desc = get_device_descriptor(device) + return len(desc) == 18 + except usb.core.USBError: + return False + + except Exception: + return False + +# Convenience function for generic access +def get_info(device_type, index): + """ + Generic function to get interface_index and endpoint_address for any device type. + + This is a convenience function that wraps the specific device info functions + and provides a unified interface for getting device communication parameters. + + @param device_type: Type of device to query + @type device_type: str + @param index: Device index (0, 1, etc.) + @type index: int + @return: Tuple of (interface_index, endpoint_address) or (None, None) + @rtype: tuple[int, int] or tuple[None, None] + @raises ValueError: If device_type is not valid + @raises IndexError: If index is negative + @raises usb.core.USBError: If device information cannot be retrieved + + @par Valid device_type values: + - 'keyboard': Query keyboard devices + - 'mouse': Query mouse devices + - 'gamepad': Query gamepad devices + + @par Example: + @code + interface_index, endpoint_address = get_info('keyboard', 0) + interface_index, endpoint_address = get_info('mouse', 1) + interface_index, endpoint_address = get_info('gamepad', 0) + @endcode + """ + if device_type == 'keyboard': + return get_keyboard_info(index) + elif device_type == 'mouse': + return get_mouse_info(index) + elif device_type == 'gamepad': + return get_gamepad_info(index) + else: + raise ValueError(f"Invalid device_type: {device_type}. Must be 'keyboard', \ + 'mouse', or 'gamepad'") diff --git a/examples/usb_devices_detection.py b/examples/usb_devices_detection.py new file mode 100644 index 0000000..06166d4 --- /dev/null +++ b/examples/usb_devices_detection.py @@ -0,0 +1,892 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 Anne Barela for Adafruit Industries +# +# SPDX-License-Identifier: MIT + +""" +Enhanced USB HID Device Detection Example for CircuitPython +=========================================================== + +Comprehensive demonstration of the enhanced adafruit_usb_host_descriptors library +with advanced HID device detection, composite device support, and real-time monitoring. + +This example demonstrates: +- Basic device counting and enumeration +- Composite device detection (keyboard+trackpad combos) +- Enhanced input parsing with human-readable output +- Real-time device change monitoring +- Device filtering and configuration +- Performance benchmarking + +Hardware Requirements: +- Adafruit board with USB host support (Feather RP2040 USB Host, Fruit Jam, etc.) +- USB keyboards, mice, gamepads, and composite devices for testing + +Software Requirements: +- CircuitPython 10.0.0 or highher +- Enhanced adafruit_usb_host_descriptors library + +""" +# pylint: disable=imported-but-not-used + +import time +import array +import board +import sys +import usb.core + +# Import the enhanced USB host descriptors library +try: + from adafruit_usb_host_descriptors import ( + # Basic device counting + count_keyboards, + count_mice, + count_gamepads, + # Device information retrieval + get_keyboard_info, + get_mouse_info, + get_gamepad_info, + get_keyboard_device, + get_mouse_device, + get_gamepad_device, + # Advanced features + list_all_hid_devices, + force_refresh, + get_composite_device_info, + is_composite_device, + get_companion_interfaces, + # Configuration + set_cache_timeout, + register_device_change_callback, + set_device_filter, + # Utility functions + is_device_connected, + get_info, + ) + + ENHANCED_API_AVAILABLE = True + print("Enhanced HID detection API loaded successfully!") +except ImportError as e: + print(f"Enhanced HID API not available: {e}") + print("Please ensure you have the enhanced adafruit_usb_host_descriptors library") + ENHANCED_API_AVAILABLE = False + sys.exit(1) +# ============================================================================ +# HID REPORT PARSING UTILITIES +# ============================================================================ + +# HID Usage IDs for common keys (Boot Keyboard Report) +HID_KEY_NAMES = { + 0x04: "A", + 0x05: "B", + 0x06: "C", + 0x07: "D", + 0x08: "E", + 0x09: "F", + 0x0A: "G", + 0x0B: "H", + 0x0C: "I", + 0x0D: "J", + 0x0E: "K", + 0x0F: "L", + 0x10: "M", + 0x11: "N", + 0x12: "O", + 0x13: "P", + 0x14: "Q", + 0x15: "R", + 0x16: "S", + 0x17: "T", + 0x18: "U", + 0x19: "V", + 0x1A: "W", + 0x1B: "X", + 0x1C: "Y", + 0x1D: "Z", + 0x1E: "1", + 0x1F: "2", + 0x20: "3", + 0x21: "4", + 0x22: "5", + 0x23: "6", + 0x24: "7", + 0x25: "8", + 0x26: "9", + 0x27: "0", + 0x28: "ENTER", + 0x29: "ESC", + 0x2A: "BACKSPACE", + 0x2B: "TAB", + 0x2C: "SPACE", + 0x2D: "-", + 0x2E: "=", + 0x2F: "[", + 0x30: "]", + 0x31: "\\", + 0x33: ";", + 0x34: "'", + 0x35: "`", + 0x36: ",", + 0x37: ".", + 0x38: "/", + 0x39: "CAPS_LOCK", + 0x3A: "F1", + 0x3B: "F2", + 0x3C: "F3", + 0x3D: "F4", + 0x3E: "F5", + 0x3F: "F6", + 0x40: "F7", + 0x41: "F8", + 0x42: "F9", + 0x43: "F10", + 0x44: "F11", + 0x45: "F12", +} + +MODIFIER_NAMES = { + 0x01: "L_CTRL", + 0x02: "L_SHIFT", + 0x04: "L_ALT", + 0x08: "L_GUI", + 0x10: "R_CTRL", + 0x20: "R_SHIFT", + 0x40: "R_ALT", + 0x80: "R_GUI", +} + + +def parse_keyboard_report(report_data): + """ + Parse a keyboard HID report into human-readable format. + + @param report_data: Raw HID report bytes (8 bytes for boot keyboard) + @return: Dictionary with parsed key information + """ + if len(report_data) < 8: + return {"error": "Report too short", "raw": list(report_data)} + modifier_byte = report_data[0] + # Byte 1 is reserved + key_codes = report_data[2:8] # Up to 6 simultaneous keys + + # Parse modifiers + active_modifiers = [] + for bit, name in MODIFIER_NAMES.items(): + if modifier_byte & bit: + active_modifiers.append(name) + # Parse key codes + active_keys = [] + for key_code in key_codes: + if key_code != 0: + key_name = HID_KEY_NAMES.get(key_code, f"KEY_{key_code:02X}") + active_keys.append(key_name) + return { + "modifiers": active_modifiers, + "keys": active_keys, + "raw_modifier": modifier_byte, + "raw_keys": [k for k in key_codes if k != 0], + "has_input": bool(active_modifiers or active_keys), + } + + +def parse_mouse_report(report_data): + """ + Parse a mouse HID report into human-readable format. + + @param report_data: Raw HID report bytes (3-4 bytes for boot mouse) + @return: Dictionary with parsed mouse information + """ + if len(report_data) < 3: + return {"error": "Report too short", "raw": list(report_data)} + buttons = report_data[0] + x_movement = ( + report_data[1] if report_data[1] < 128 else report_data[1] - 256 + ) # Convert to signed + y_movement = ( + report_data[2] if report_data[2] < 128 else report_data[2] - 256 + ) # Convert to signed + wheel = 0 + + if len(report_data) > 3: + wheel = report_data[3] if report_data[3] < 128 else report_data[3] - 256 + # Parse button names + button_names = [] + if buttons & 0x01: + button_names.append("LEFT") + if buttons & 0x02: + button_names.append("RIGHT") + if buttons & 0x04: + button_names.append("MIDDLE") + if buttons & 0x08: + button_names.append("BTN4") + if buttons & 0x10: + button_names.append("BTN5") + return { + "buttons": button_names, + "x_movement": x_movement, + "y_movement": y_movement, + "wheel": wheel, + "raw_buttons": buttons, + "has_input": bool(buttons or x_movement or y_movement or wheel), + } + + +# ============================================================================ +# DEVICE CHANGE MONITORING +# ============================================================================ + + +class DeviceMonitor: + """Device change monitoring with callbacks and statistics""" + + def __init__(self): + self.device_history = [] + self.change_count = 0 + register_device_change_callback(self._on_device_change) + + def _on_device_change(self, device_type, action, index): + """Internal callback for device changes""" + self.change_count += 1 + timestamp = time.monotonic() + + change_info = { + "timestamp": timestamp, + "device_type": device_type, + "action": action, + "index": index, + } + + self.device_history.append(change_info) + + # Keep only last 50 changes + if len(self.device_history) > 50: + self.device_history.pop(0) + print(f"[{timestamp:.1f}s] Device {action}: {device_type}[{index}]") + + def get_statistics(self): + """Get monitoring statistics""" + return { + "total_changes": self.change_count, + "recent_changes": len(self.device_history), + "history": self.device_history.copy(), + } + + +# Global device monitor instance +device_monitor = DeviceMonitor() + +# ============================================================================ +# ENHANCED INPUT READING FUNCTIONS +# ============================================================================ + + +def safe_device_operation(operation_func, *args, **kwargs): + """ + Safely perform device operations with automatic retry and error handling. + """ + max_retries = 3 + for attempt in range(max_retries): + try: + return operation_func(*args, **kwargs) + except usb.core.USBError as e: + if attempt < max_retries - 1: + print(f"USB error (attempt {attempt + 1}): {e}, retrying...") + time.sleep(0.1) + force_refresh() + else: + print(f"USB operation failed after {max_retries} attempts: {e}") + raise + except Exception as e: + print(f"Unexpected error in device operation: {e}") + raise + + +def enhanced_read_keyboard_input(keyboard_index, duration=5): + """ + Enhanced keyboard reading with better error handling and parsing. + """ + print(f"\n=== Reading from Keyboard[{keyboard_index}] ===") + + try: + device = safe_device_operation(get_keyboard_device, keyboard_index) + interface_index, endpoint_address = safe_device_operation( + get_keyboard_info, keyboard_index + ) + + if device is None or endpoint_address is None: + print(f"Keyboard[{keyboard_index}] not available") + return + print(f"Device: {device.manufacturer} {device.product}") + print(f"VID:PID: {device.idVendor:04x}:{device.idProduct:04x}") + print(f"Interface: {interface_index}, Endpoint: 0x{endpoint_address:02x}") + + # Check if this is part of a composite device + is_composite, other_types = is_composite_device("keyboard", keyboard_index) + if is_composite: + print(f"Composite device - also has: {', '.join(other_types)}") + # Setup device + device.set_configuration() + if device.is_kernel_driver_active(interface_index): + device.detach_kernel_driver(interface_index) + buf = array.array("B", [0] * 8) + print(f"\nPress keys for {duration} seconds (parsed output will appear):") + + start_time = time.monotonic() + last_report = None + key_press_count = 0 + + while time.monotonic() - start_time < duration: + try: + count = device.read(endpoint_address, buf, timeout=100) + if count > 0: + # Only process if report changed (avoid spam from key repeat) + current_report = bytes(buf[:count]) + if current_report != last_report: + parsed = parse_keyboard_report(buf) + + if parsed.get("has_input"): + key_press_count += 1 + output_parts = [] + + if parsed["modifiers"]: + output_parts.append( + f"Modifiers: {'+'.join(parsed['modifiers'])}" + ) + if parsed["keys"]: + output_parts.append(f"Keys: {'+'.join(parsed['keys'])}") + print( + f" [{key_press_count:2d}] {' | '.join(output_parts)}" + ) + last_report = current_report + except usb.core.USBTimeoutError: + pass # Normal timeout + except usb.core.USBError as e: + print(f" USB read error: {e}") + break + print(f"Keyboard reading complete. Detected {key_press_count} key events.") + except Exception as e: + print(f"Error in enhanced keyboard reading: {e}") + + +def enhanced_read_mouse_input(mouse_index, duration=5): + """ + Enhanced mouse reading with better error handling and parsing. + """ + print(f"\n=== Reading from Mouse[{mouse_index}] ===") + + try: + device = safe_device_operation(get_mouse_device, mouse_index) + interface_index, endpoint_address = safe_device_operation( + get_mouse_info, mouse_index + ) + + if device is None or endpoint_address is None: + print(f"Mouse[{mouse_index}] not available") + return + print(f"Device: {device.manufacturer} {device.product}") + print(f"VID:PID: {device.idVendor:04x}:{device.idProduct:04x}") + print(f"Interface: {interface_index}, Endpoint: 0x{endpoint_address:02x}") + + # Check if this is part of a composite device + is_composite, other_types = is_composite_device("mouse", mouse_index) + if is_composite: + print(f"Composite device - also has: {', '.join(other_types)}") + print("(This might be a trackpad from a keyboard+trackpad combo)") + # Setup device + device.set_configuration() + if device.is_kernel_driver_active(interface_index): + device.detach_kernel_driver(interface_index) + buf = array.array("B", [0] * 4) + print(f"\nMove mouse/trackpad and click buttons for {duration} seconds:") + + start_time = time.monotonic() + last_report = None + mouse_event_count = 0 + total_x_movement = 0 + total_y_movement = 0 + + while time.monotonic() - start_time < duration: + try: + count = device.read(endpoint_address, buf, timeout=100) + if count > 0: + current_report = bytes(buf[:count]) + if current_report != last_report: + parsed = parse_mouse_report(buf) + + if parsed.get("has_input"): + mouse_event_count += 1 + total_x_movement += abs(parsed["x_movement"]) + total_y_movement += abs(parsed["y_movement"]) + + output_parts = [] + + if parsed["buttons"]: + output_parts.append( + f"Buttons: {'+'.join(parsed['buttons'])}" + ) + if parsed["x_movement"] or parsed["y_movement"]: + output_parts.append( + f"Move: X{parsed['x_movement']:+d} \ + Y{parsed['y_movement']:+d}" + ) + if parsed["wheel"]: + output_parts.append(f"Wheel: {parsed['wheel']:+d}") + print( + f" [{mouse_event_count:2d}] {' | '.join(output_parts)}" + ) + last_report = current_report + except usb.core.USBTimeoutError: + pass # Normal timeout + except usb.core.USBError as e: + print(f" USB read error: {e}") + break + print(f"Mouse reading complete. Detected {mouse_event_count} mouse events.") + print(f"Total movement: X={total_x_movement} Y={total_y_movement}") + except Exception as e: + print(f"Error in enhanced mouse reading: {e}") + + +def read_composite_device(keyboard_index, duration=5): + """ + Read from both keyboard and trackpad interfaces of a composite device. + """ + print(f"\n=== Reading Composite Device (Keyboard[{keyboard_index}]) ===") + + # Verify this is a composite device with mouse interface + is_composite, other_types = is_composite_device("keyboard", keyboard_index) + if not is_composite or "mouse" not in other_types: + print(f"Keyboard[{keyboard_index}] is not a composite device with trackpad") + return + try: + # Get keyboard interfaces + kbd_device = get_keyboard_device(keyboard_index) + kbd_interface, kbd_endpoint = get_keyboard_info(keyboard_index) + + # Get trackpad interface from companion interfaces + companions = get_companion_interfaces("keyboard", keyboard_index) + trackpad_info = companions["mouse"] + trackpad_interface = trackpad_info["interface_index"] + trackpad_endpoint = trackpad_info["endpoint_address"] + + print(f"Composite device: {kbd_device.product}") + print(f" Keyboard interface: {kbd_interface}, endpoint: 0x{kbd_endpoint:02x}") + print( + f" Trackpad interface: {trackpad_interface}, \ + endpoint: 0x{trackpad_endpoint:02x}" + ) + + # Setup device + kbd_device.set_configuration() + + # Detach kernel drivers for both interfaces + if kbd_device.is_kernel_driver_active(kbd_interface): + kbd_device.detach_kernel_driver(kbd_interface) + if trackpad_interface != kbd_interface and kbd_device.is_kernel_driver_active( + trackpad_interface + ): + kbd_device.detach_kernel_driver(trackpad_interface) + kbd_buf = array.array("B", [0] * 8) + mouse_buf = array.array("B", [0] * 4) + + print(f"\nUse both keyboard and trackpad for {duration} seconds:") + print("(Events from both interfaces will be shown)") + + start_time = time.monotonic() + last_kbd_report = None + last_mouse_report = None + kbd_events = 0 + mouse_events = 0 + + while time.monotonic() - start_time < duration: + try: + # Try to read keyboard + try: + count = kbd_device.read(kbd_endpoint, kbd_buf, timeout=10) + if count > 0: + current_kbd_report = bytes(kbd_buf[:count]) + if current_kbd_report != last_kbd_report: + parsed = parse_keyboard_report(kbd_buf) + if parsed.get("has_input"): + kbd_events += 1 + output_parts = [] + if parsed["modifiers"]: + output_parts.append( + f"Mods: {'+'.join(parsed['modifiers'])}" + ) + if parsed["keys"]: + output_parts.append( + f"Keys: {'+'.join(parsed['keys'])}" + ) + print( + f" šŸŽ¹ KEYBOARD[{kbd_events:2d}]: \ + {' | '.join(output_parts)}" + ) + last_kbd_report = current_kbd_report + except usb.core.USBTimeoutError: + pass + # Try to read trackpad + try: + count = kbd_device.read(trackpad_endpoint, mouse_buf, timeout=10) + if count > 0: + current_mouse_report = bytes(mouse_buf[:count]) + if current_mouse_report != last_mouse_report: + parsed = parse_mouse_report(mouse_buf) + if parsed.get("has_input"): + mouse_events += 1 + output_parts = [] + if parsed["buttons"]: + output_parts.append( + f"Btns: {'+'.join(parsed['buttons'])}" + ) + if parsed["x_movement"] or parsed["y_movement"]: + output_parts.append( + f"Move: X{parsed['x_movement']:+d} \ + Y{parsed['y_movement']:+d}" + ) + if parsed["wheel"]: + output_parts.append(f"Wheel: {parsed['wheel']:+d}") + print( + f" šŸ–±ļø TRACKPAD[{mouse_events:2d}]: \ + {' | '.join(output_parts)}" + ) + last_mouse_report = current_mouse_report + except usb.core.USBTimeoutError: + pass + except usb.core.USBError as e: + print(f"USB error: {e}") + break + print("\nComposite device reading complete:") + print(f" Keyboard events: {kbd_events}") + print(f" Trackpad events: {mouse_events}") + except Exception as e: + print(f"Error reading composite device: {e}") + +# ============================================================================ +# DEVICE DISCOVERY AND ANALYSIS +# ============================================================================ + +def analyze_all_devices(): + """ + Comprehensive analysis of all connected HID devices. + """ + print("\n" + "=" * 60) + print("COMPREHENSIVE HID DEVICE ANALYSIS") + print("=" * 60) + + # Force fresh scan + force_refresh() + + # Get device counts + keyboards = count_keyboards() + mice = count_mice() + gamepads = count_gamepads() + + print("\nDevice Summary:") + print(f" Keyboards: {keyboards}") + print(f" Mice: {mice}") + print(f" Gamepads: {gamepads}") + print(f" Total: {keyboards + mice + gamepads}") + + # Detailed device information + summary = list_all_hid_devices() + + print("\nDetailed Device Information:") + print("-" * 40) + + # Analyze keyboards + if summary["keyboards"]["count"] > 0: + print(f"\nšŸŽ¹ KEYBOARDS ({summary['keyboards']['count']}):") + for device in summary["keyboards"]["devices"]: + print(f" [{device['index']}] {device['manufacturer']} {device['product']}") + print(f" VID:PID = {device['vid']:04x}:{device['pid']:04x}") + print( + f" Interface: {device['interface_index']}, \ + Endpoint: 0x{device['endpoint_address']:02x}" + ) + + # Check for composite capabilities + is_composite, other_types = is_composite_device("keyboard", device["index"]) + if is_composite: + print(f" šŸ”— Composite device with: {', '.join(other_types)}") + # Check connectivity + connected = is_device_connected("keyboard", device["index"]) + print(f" Status: {'🟢 Connected' if connected else 'šŸ”“ Disconnected'}") + # Analyze mice + if summary["mice"]["count"] > 0: + print(f"\nšŸ–±ļø MICE ({summary['mice']['count']}):") + for device in summary["mice"]["devices"]: + print(f" [{device['index']}] {device['manufacturer']} {device['product']}") + print(f" VID:PID = {device['vid']:04x}:{device['pid']:04x}") + print( + f" Interface: {device['interface_index']}, \ + Endpoint: 0x{device['endpoint_address']:02x}" + ) + + # Check for composite capabilities + is_composite, other_types = is_composite_device("mouse", device["index"]) + if is_composite: + print(f" šŸ”— Composite device with: {', '.join(other_types)}") + print(" (This might be a trackpad from a keyboard+trackpad combo)") + # Check connectivity + connected = is_device_connected("mouse", device["index"]) + print(f" Status: {'🟢 Connected' if connected else 'šŸ”“ Disconnected'}") + # Analyze gamepads + if summary["gamepads"]["count"] > 0: + print(f"\nšŸŽ® GAMEPADS ({summary['gamepads']['count']}):") + for device in summary["gamepads"]["devices"]: + print(f" [{device['index']}] {device['manufacturer']} {device['product']}") + print(f" VID:PID = {device['vid']:04x}:{device['pid']:04x}") + print( + f" Interface: {device['interface_index']}, \ + Endpoint: 0x{device['endpoint_address']:02x}" + ) + + # Check connectivity + connected = is_device_connected("gamepad", device["index"]) + print(f" Status: {'🟢 Connected' if connected else 'šŸ”“ Disconnected'}") + # Analyze composite devices + composite_devices = get_composite_device_info() + if composite_devices: + print(f"\nšŸ”— COMPOSITE DEVICES ({len(composite_devices)}):") + for device_id, info in composite_devices.items(): + print(f" {info['product']}") + print(f" Device ID: {device_id}") + print(f" Interface types: {', '.join(info['types'])}") + + # Show how to access each interface + for device_type in info["types"]: + if device_type == "keyboard": + for i in range(keyboards): + kbd_device = get_keyboard_device(i) + kbd_id = f"{kbd_device.idVendor:04x}: \ + {kbd_device.idProduct:04x}: \ + {getattr(kbd_device, 'serial_number', 'no_serial')}" + if kbd_id == device_id: + print(f" -> Access keyboard via: get_keyboard_info({i})") + break + elif device_type == "mouse": + for i in range(mice): + mouse_device = get_mouse_device(i) + # pylint: disable=line-too-long + mouse_id = f"{mouse_device.idVendor:04x}: \ + {mouse_device.idProduct:04x}: \ + {getattr(mouse_device, 'serial_number', 'no_serial')}" + if mouse_id == device_id: + print( + f" -> Access mouse/trackpad via: get_mouse_info({i})" + ) + break + + +def benchmark_performance(): + """ + Benchmark the performance of device detection operations. + """ + print("\nšŸ” PERFORMANCE BENCHMARK") + print("-" * 30) + + import gc + + # Warm up + force_refresh() + + # Benchmark full device scan + gc.collect() + start_time = time.monotonic() + force_refresh() + scan_time = time.monotonic() - start_time + + # Benchmark cached access + start_time = time.monotonic() + keyboards = count_keyboards() + mice = count_mice() + gamepads = count_gamepads() + cache_time = time.monotonic() - start_time + + # Benchmark device info access + start_time = time.monotonic() + for i in range(keyboards): + get_keyboard_info(i) + for i in range(mice): + get_mouse_info(i) + for i in range(gamepads): + get_gamepad_info(i) + info_time = time.monotonic() - start_time + + # Benchmark composite device operations + start_time = time.monotonic() + get_composite_device_info() + for i in range(keyboards): + is_composite_device("keyboard", i) + composite_time = time.monotonic() - start_time + + print(f"Full device scan: {scan_time*1000:6.1f} ms") + print(f"Cached device counts: {cache_time*1000:6.3f} ms") + print(f"Device info access: {info_time*1000:6.1f} ms") + print(f"Composite operations: {composite_time*1000:6.1f} ms") + print(f"Devices found: {keyboards} kbd, {mice} mouse, {gamepads} gamepad") + + +def demonstrate_device_filtering(): + """ + Demonstrate device filtering capabilities. + """ + print("\nšŸ” DEVICE FILTERING DEMONSTRATION") + print("-" * 40) + + # Show all devices first + print("Before filtering:") + summary = list_all_hid_devices() + total_before = sum(info["count"] for info in summary.values()) + print(f" Total devices: {total_before}") + + if total_before == 0: + print(" No devices to filter. Connect some devices first.") + return + # Get a VID to filter by (use first keyboard if available) + if summary["keyboards"]["count"] > 0: + test_vid = summary["keyboards"]["devices"][0]["vid"] + print(f"\nFiltering to only allow VID 0x{test_vid:04x}...") + + # Apply filter + set_device_filter(allowed_vids={test_vid}) + + # Force refresh and check results + force_refresh() + filtered_summary = list_all_hid_devices() + total_after = sum(info["count"] for info in filtered_summary.values()) + + print("After filtering:") + print(f" Total devices: {total_after}") + for device_type, info in filtered_summary.items(): + if info["count"] > 0: + print(f" {device_type}: {info['count']}") + # Clear filter + print("\nClearing filter...") + set_device_filter(allowed_vids=None) + force_refresh() + + restored_summary = list_all_hid_devices() + total_restored = sum(info["count"] for info in restored_summary.values()) + print("After clearing filter:") + print(f" Total devices: {total_restored}") + + +def continuous_monitoring(duration=30): + """ + Demonstrate continuous device monitoring. + """ + print("\nšŸ“” CONTINUOUS DEVICE MONITORING") + print(f"Duration: {duration} seconds") + print("-" * 40) + print("Connect and disconnect devices to see real-time detection!") + print("(Device changes will be logged automatically)") + + start_time = time.monotonic() + # last_summary = None + check_interval = 1.0 # Check every second + + try: + while time.monotonic() - start_time < duration: + current_time = time.monotonic() - start_time + + # Check for device changes + force_refresh() + current_summary = list_all_hid_devices() + + # Show periodic status updates + if int(current_time) % 10 == 0 and int(current_time) > 0: + total_devices = sum(info["count"] for info in current_summary.values()) + print( + f"[{current_time:5.0f}s] Status: {total_devices} \ + total devices connected" + ) + # last_summary = current_summary + time.sleep(check_interval) + except KeyboardInterrupt: + print("\nMonitoring stopped by user") + # Show monitoring statistics + stats = device_monitor.get_statistics() + print("\nMonitoring Statistics:") + print(f" Total device changes detected: {stats['total_changes']}") + print(f" Recent changes in history: {stats['recent_changes']}") + + if stats["recent_changes"] > 0: + print(" Recent changes:") + for change in stats["history"][-5:]: # Show last 5 changes + print( + f" {change['timestamp']:.1f}s: {change['device_type']} \ + {change['action']}" + ) + +# ============================================================================ +# MAIN DEMONSTRATION FUNCTION +# ============================================================================ + +def main(): + """ + Main demonstration function showcasing all enhanced features. + """ + print("Enhanced USB HID Device Detection Example") + print("=========================================") + print(f"Running on: {board.board_id}") + print(f"CircuitPython version: {board.__name__}") + + # Configure library settings + print("\nConfiguring library settings...") + set_cache_timeout(0.5) # 500ms cache timeout for responsive demo + + # Initial device analysis + analyze_all_devices() + + # Performance benchmark + benchmark_performance() + + # Quick input demos if devices are available + keyboards = count_keyboards() + mice = count_mice() + + if keyboards > 0: + print("\nāŒØļø Quick keyboard demo (first 3 seconds)...") + enhanced_read_keyboard_input(0, duration=3) + if mice > 0: + print("\nšŸ–±ļø Quick mouse demo (first 3 seconds)...") + enhanced_read_mouse_input(0, duration=3) + # Check for composite devices + composite_found = False + for i in range(keyboards): + is_composite, other_types = is_composite_device("keyboard", i) + if is_composite and "mouse" in other_types: + print("\nšŸ”— Quick composite device demo (3 seconds)...") + read_composite_device(i, duration=3) + composite_found = True + break + if not composite_found and keyboards > 0: + print( + "\nšŸ’” Tip: Try connecting a keyboard with integrated trackpad \ + to see composite device features!" + ) + # Device filtering demo + demonstrate_device_filtering() + + print("\nšŸŽ‰ Basic demonstration complete!") + print("\nFor extended testing:") + print(" - Call analyze_all_devices() for detailed analysis") + print(" - Call continuous_monitoring(duration) to monitor device changes") + print(" - Call enhanced_read_keyboard_input(index) to test keyboard input") + print(" - Call enhanced_read_mouse_input(index) to test mouse input") + print(" - Call read_composite_device(index) for composite device testing") + + +if __name__ == "__main__": + if not ENHANCED_API_AVAILABLE: + print("Enhanced API not available. Cannot run demonstration.") + else: + try: + main() + except KeyboardInterrupt: + print("\nDemo interrupted by user") + except Exception as e: + print(f"Demo error: {e}") + import traceback + + traceback.print_exc() diff --git a/examples/usb_host_devices_simpletest.py b/examples/usb_host_devices_simpletest.py new file mode 100644 index 0000000..f5ebfc0 --- /dev/null +++ b/examples/usb_host_devices_simpletest.py @@ -0,0 +1,304 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 Anne Barela for Adafruit Industries +# +# SPDX-License-Identifier: MIT + +""" +Simple USB Device Monitor +========================= + +A simple program that continuously monitors USB host ports for connected devices. +Automatically starts monitoring and detects when devices are plugged in or unplugged. + +This program: +- Loops continuously checking for USB devices every 5 seconds +- Detects device additions and removals automatically +- Shows basic device information +- Handles composite devices (keyboard+trackpad combos) +- Runs until stopped with Ctrl+C + +Hardware Requirements: +- Adafruit board with USB host support (Feather RP2040 USB Host, Fruit Jam, etc.) +- USB devices can be connected/disconnected during operation + +Software Requirements: +- CircuitPython 9.0+ with usb_host support +- Enhanced adafruit_usb_host_descriptors library + +Usage: +- Simply run this program +- Connect and disconnect USB devices to see detection +- Press Ctrl+C to stop monitoring + +""" + +import time +import board + +# Import the enhanced USB host descriptors library +try: + from adafruit_usb_host_descriptors import ( + count_keyboards, count_mice, count_gamepads, + list_all_hid_devices, get_composite_device_info, + force_refresh, is_device_connected + ) + print("Enhanced USB host descriptors library loaded") +except ImportError as e: + print("Library import failed: {}".format(e)) + print("Please ensure you have the enhanced " + + "adafruit_usb_host_descriptors library") + exit(1) + +class SimpleDeviceMonitor: + """Simple USB device monitor with change detection""" + + def __init__(self): + self.last_device_state = None + self.loop_count = 0 + self.total_changes = 0 + + def get_current_device_state(self): + """Get current device state as a comparable dictionary""" + try: + # Force fresh scan to detect changes + force_refresh() + + # Get device counts + keyboards = count_keyboards() + mice = count_mice() + gamepads = count_gamepads() + + # Get detailed device info + devices = list_all_hid_devices() + + # Create a state snapshot + state = { + 'counts': { + 'keyboards': keyboards, + 'mice': mice, + 'gamepads': gamepads, + 'total': keyboards + mice + gamepads + }, + 'devices': {} + } + + # Store device details for comparison + for device_type in ['keyboards', 'mice', 'gamepads']: + state['devices'][device_type] = [] + for device in devices[device_type]['devices']: + # Create a simple identifier for each device + device_id = "{:04x}:{:04x}:{}:{}".format( + device['vid'], device['pid'], + device['manufacturer'], device['product']) + state['devices'][device_type].append({ + 'id': device_id, + 'index': device['index'], + 'manufacturer': device['manufacturer'], + 'product': device['product'], + 'vid': device['vid'], + 'pid': device['pid'] + }) + + return state + + except Exception as e: + # Return empty state on error to prevent crash + print(" WARNING: Could not get device state: {}".format(e)) + return { + 'counts': {'keyboards': 0, 'mice': 0, 'gamepads': 0, + 'total': 0}, + 'devices': {'keyboards': [], 'mice': [], 'gamepads': []} + } + + def detect_changes(self, current_state): + """Detect and report changes between current and last state""" + if self.last_device_state is None: + # First run - no change detection needed, just note it's first run + return + + changes_detected = False + + # Proper singular forms mapping + singular_forms = { + 'keyboards': 'keyboard', + 'mice': 'mouse', + 'gamepads': 'gamepad' + } + + # Check for count changes + for device_type in ['keyboards', 'mice', 'gamepads']: + current_count = current_state['counts'][device_type] + last_count = self.last_device_state['counts'][device_type] + + if current_count != last_count: + changes_detected = True + change = current_count - last_count + if change > 0: + print(" + {}: +{} (now {})".format( + device_type, change, current_count)) + else: + print(" - {}: {} (now {})".format( + device_type, change, current_count)) + + # Detect specific device changes + for device_type in ['keyboards', 'mice', 'gamepads']: + current_devices = {d['id']: d for d in + current_state['devices'][device_type]} + last_devices = {d['id']: d for d in + self.last_device_state['devices'][device_type]} + + device_type_singular = singular_forms[device_type] + + # Find new devices + for device_id, device in current_devices.items(): + if device_id not in last_devices: + changes_detected = True + print(" CONNECTED: {} {} ({})".format( + device['manufacturer'], device['product'], + device_type_singular)) + + # Find removed devices + for device_id, device in last_devices.items(): + if device_id not in current_devices: + changes_detected = True + print(" DISCONNECTED: {} {} ({})".format( + device['manufacturer'], device['product'], + device_type_singular)) + + if changes_detected: + self.total_changes += 1 + + def report_current_devices(self, state): + """Report current connected devices""" + total = state['counts']['total'] + + if total == 0: + print(" No HID devices connected") + else: + print(" Current devices ({} total):".format(total)) + + for device_type in ['keyboards', 'mice', 'gamepads']: + devices = state['devices'][device_type] + if devices: + # Manual capitalization for CircuitPython compatibility + type_name = device_type[0].upper() + device_type[1:] + print(" {}: {}".format(type_name, len(devices))) + for device in devices: + print(" [{}] {} {}".format( + device['index'], device['manufacturer'], + device['product'])) + + # Check for composite devices + composite_devices = get_composite_device_info() + if composite_devices: + print(" Composite devices: {}".format( + len(composite_devices))) + for device_id, info in composite_devices.items(): + print(" {} ({})".format( + info['product'], ', '.join(info['types']))) + + def check_device_connectivity(self, state): + """Check if previously detected devices are still responding""" + connectivity_issues = [] + + # Proper singular forms mapping + singular_forms = { + 'keyboards': 'keyboard', + 'mice': 'mouse', # Fix: was becoming 'mic' incorrectly + 'gamepads': 'gamepad' + } + + for device_type in ['keyboards', 'mice', 'gamepads']: + for device in state['devices'][device_type]: + device_type_singular = singular_forms[device_type] + connected = is_device_connected(device_type_singular, + device['index']) + if not connected: + connectivity_issues.append("{}[{}]".format( + device_type_singular, device['index'])) + + if connectivity_issues: + print(" WARNING: Connectivity issues: {}".format( + ', '.join(connectivity_issues))) + + def run(self): + """Main monitoring loop""" + print("Simple USB Device Monitor") + print("=" * 25) + print("Running on: {}".format(board.board_id)) + print("Monitoring USB host ports for device changes...") + print("Connect/disconnect devices to see detection in action") + print() + print("IMPORTANT: Press Ctrl+C to stop monitoring at any time") + print(" Monitoring will continue until interrupted") + print() + + start_time = time.monotonic() + + try: + while True: + self.loop_count += 1 + + print("Loop #{} - Checking devices...".format( + self.loop_count)) + + try: + # Get current device state + current_state = self.get_current_device_state() + + # Detect and report changes + self.detect_changes(current_state) + + # Always show current device info each loop + self.report_current_devices(current_state) + + # Check device connectivity (optional diagnostic) + if current_state['counts']['total'] > 0: + self.check_device_connectivity(current_state) + + # Update state for next iteration + self.last_device_state = current_state + + except KeyboardInterrupt: + # Break out of inner loop on Ctrl+C + raise + except Exception as e: + print(" ERROR: Error during device check: {}".format(e)) + + # Show statistics periodically + if self.loop_count % 10 == 0: + print(" Statistics: {} checks, {} changes detected".format( + self.loop_count, self.total_changes)) + + print() # Empty line for readability + + # Wait 5 seconds before next check + time.sleep(5.0) + + except KeyboardInterrupt: + # Clean shutdown on Ctrl+C + runtime = time.monotonic() - start_time + print("\nMonitoring stopped by user (Ctrl+C pressed)") + print("Final statistics:") + print(" Total checks: {}".format(self.loop_count)) + print(" Changes detected: {}".format(self.total_changes)) + print(" Runtime: {:.1f} seconds".format(runtime)) + print("Thank you for using USB Device Monitor!") + +# Main execution +if __name__ == "__main__": + print("Starting USB Device Monitor...") + print("Press Ctrl+C at any time to stop monitoring") + print() + + try: + # Create and run the continuous monitor + monitor = SimpleDeviceMonitor() + monitor.run() + + except KeyboardInterrupt: + print("\nProgram interrupted by user (Ctrl+C)") + print("Goodbye!") + except Exception as e: + print("\nError starting monitor: {}".format(e)) + print("Please check your USB host hardware and library installation") + print("Program terminated.")