### Complete Multi-Function Gadget with ConfigFS using functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt This example shows how to create a multi-function USB gadget using functionfs and ConfigFS. It defines custom functions, wraps them in ConfigFunctionFFS, and configures multiple gadget configurations with different functions and attributes. Dependencies include 'functionfs' and standard Python libraries like 'signal'. ```python from functionfs.gadget import Gadget, ConfigFunctionFFS import functionfs class SimpleFunction(functionfs.Function): def __init__(self, path): fs_list, hs_list, ss_list = functionfs.getInterfaceInAllSpeeds( interface={'bInterfaceClass': functionfs.ch9.USB_CLASS_VENDOR_SPEC}, endpoint_list=[{ 'endpoint': { 'bEndpointAddress': functionfs.ch9.USB_DIR_IN, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, }], ) super().__init__(path, fs_list=fs_list, hs_list=hs_list, ss_list=ss_list) class FunctionWrapper(ConfigFunctionFFS): def getFunction(self, path): return SimpleFunction(path=path) # Create gadget with multiple configurations gadget = Gadget( config_list=[ { 'function_list': [ FunctionWrapper(name='func0'), FunctionWrapper(name='func1'), ], 'bmAttributes': 0x80 | 0x40, # Bus-powered, remote wakeup 'MaxPower': 250, # 500mA 'lang_dict': { 0x0409: { 'configuration': 'Primary Configuration', }, }, }, { 'function_list': [ FunctionWrapper(name='func2'), ], 'MaxPower': 100, 'lang_dict': { 0x0409: { 'configuration': 'Alternate Configuration', }, }, }, ], idVendor=0x1d6b, idProduct=0x0104, bcdDevice=0x0100, bcdUSB=0x0200, bDeviceClass=0x00, bDeviceSubClass=0x00, bDeviceProtocol=0x00, lang_dict={ 0x0409: { 'manufacturer': 'Custom Manufacturer', 'product': 'Multi-Function Device', 'serialnumber': 'SN123456', }, }, name='custom_gadget', udc='fe980000.usb', # Specific UDC, or None for auto-detect ) with gadget: # Access functions func = gadget.getFunction(configuration_index=0, function_index=0) # Check if UDC is still registered if gadget.isUDCRegistered(): print("Device is connected") # Wait for events import signal signal.pause() ``` -------------------------------- ### Handling OUT Endpoint Data Reception with Python functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Implements an OUT endpoint to receive data from a USB host. The `onComplete` method processes received data and passes it to a specified data handler. This example also includes an `EchoFunction` that echoes received data back to the IN endpoint. ```python import functionfs class CustomOUTEndpoint(functionfs.EndpointOUTFile): def __init__(self, data_handler, *args, **kw): self.data_handler = data_handler super().__init__(*args, **kw) def onComplete(self, data, status): """Called when OUT endpoint receives data""" if data is None: print(f"Receive error: {-status}") return # Process received data print(f"Received {len(data)} bytes") self.data_handler(data.tobytes()) class EchoFunction(functionfs.Function): def __init__(self, path): fs_list, hs_list, ss_list = functionfs.getInterfaceInAllSpeeds( interface={'bInterfaceClass': functionfs.ch9.USB_CLASS_VENDOR_SPEC}, endpoint_list=[ { 'endpoint': { 'bEndpointAddress': 1 | functionfs.ch9.USB_DIR_IN, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, }, { 'endpoint': { 'bEndpointAddress': 2 | functionfs.ch9.USB_DIR_OUT, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, }, ], ) super().__init__( path, fs_list=fs_list, hs_list=hs_list, ss_list=ss_list, out_aio_blocks_per_endpoint=4, # Number of buffers per OUT endpoint out_aio_blocks_max_packet_count=10, # Buffer size multiplier ) def getEndpointClass(self, is_in, descriptor): if not is_in: return lambda path, submit, release, aio_block_list: \ CustomOUTEndpoint(self.echo_data, path, submit, release, aio_block_list) return super().getEndpointClass(is_in, descriptor) def echo_data(self, data): """Echo received data back to IN endpoint""" in_endpoint = self.getEndpoint(1) buffer = bytearray(data) in_endpoint.submit([buffer]) ``` -------------------------------- ### HID Mouse Emulation with functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Emulates a USB HID mouse using the functionfs library. It defines a `MouseINEndpoint` for handling data transfers and a `HIDMouse` class that inherits from `functionfs.HIDFunction`. The `HIDMouse` class manages report descriptors, endpoint configuration, and processing of standard HID control requests like GET_REPORT and SET_IDLE. The example also shows how to integrate this mouse function into a USB gadget using `GadgetSubprocessManager`. ```python MOVE_RIGHT_REPORT = bytearray(b'\x00\x01\x00') class MouseINEndpoint(functionfs.EndpointINFile): def onComplete(self, buffer_list, user_data, status): if status < 0: if status == -errno.ESHUTDOWN: return False raise IOError(-status) # Continuously resubmit to keep mouse moving return True class HIDMouse(functionfs.HIDFunction): def __init__(self, path): super().__init__( path=path, report_descriptor=REPORT_DESCRIPTOR, in_report_max_length=len(MOVE_RIGHT_REPORT), out_report_max_length=0, # No OUT endpoint needed full_speed_interval=10, # Poll interval in ms high_speed_interval=4, # Poll interval as 2^(n-1) * 125us ) def getEndpointClass(self, is_in, descriptor): if is_in: return MouseINEndpoint return super().getEndpointClass(is_in, descriptor) def getHIDReport(self, value, index, length): """Handle GET_REPORT control request""" report_type = (value >> 8) & 0xFF report_id = value & 0xFF self.ep0.write(MOVE_RIGHT_REPORT[:length]) def setHIDIdle(self, value, index, length): """Handle SET_IDLE control request""" duration = (value >> 8) & 0xFF report_id = value & 0xFF # Acknowledge the request self.ep0.read(0) def onEnable(self): super().onEnable() # Start sending HID reports self.getEndpoint(1).submit([MOVE_RIGHT_REPORT]) # Usage with gadget from functionfs.gadget import GadgetSubprocessManager, ConfigFunctionFFSSubprocess class MouseSubprocess(ConfigFunctionFFSSubprocess): def getFunction(self, path): return HIDMouse(path=path) with GadgetSubprocessManager( args=GadgetSubprocessManager.getArgumentParser().parse_args(), config_list=[{ 'function_list': [MouseSubprocess], 'MaxPower': 100, }], idVendor=0x1d6b, idProduct=0x0104, ) as gadget: gadget.waitForever() ``` -------------------------------- ### Manual USB Descriptor Construction with functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Demonstrates how to manually construct USB descriptors using the `functionfs` library. This snippet shows the creation of a `USBInterfaceDescriptor` object, specifying various attributes such as interface number, number of endpoints, and interface class. This manual approach provides fine-grained control over the USB device's descriptor table. ```python import functionfs # Build interface descriptor interface_desc = functionfs.getDescriptor( functionfs.USBInterfaceDescriptor, bInterfaceNumber=0, bAlternateSetting=0, bNumEndpoints=2, bInterfaceClass=functionfs.ch9.USB_CLASS_VENDOR_SPEC, bInterfaceSubClass=0, bInterfaceProtocol=0, iInterface=1, ) ``` -------------------------------- ### Build Endpoint Descriptors using functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt This snippet demonstrates how to build USB endpoint descriptors for different USB speeds (Full-Speed, High-Speed, SuperSpeed) using the functionfs library. It specifies endpoint addresses, attributes, packet sizes, and intervals. Dependencies include the 'functionfs' library. ```python in_endpoint_desc = functionfs.getDescriptor( functionfs.USBEndpointDescriptorNoAudio, bEndpointAddress=0x81, # EP1 IN bmAttributes=functionfs.ch9.USB_ENDPOINT_XFER_BULK, wMaxPacketSize=512, bInterval=0, ) out_endpoint_desc = functionfs.getDescriptor( functionfs.USBEndpointDescriptorNoAudio, bEndpointAddress=0x02, # EP2 OUT bmAttributes=functionfs.ch9.USB_ENDPOINT_XFER_BULK, wMaxPacketSize=512, bInterval=0, ) # SuperSpeed companion descriptor ss_companion_desc = functionfs.getDescriptor( functionfs.USBSSEPCompDescriptor, bMaxBurst=15, # Max burst size bmAttributes=0, wBytesPerInterval=0, ) # Create complete descriptor lists fs_list = [interface_desc, in_endpoint_desc, out_endpoint_desc] hs_list = [interface_desc, in_endpoint_desc, out_endpoint_desc] ss_list = [ interface_desc, in_endpoint_desc, ss_companion_desc, out_endpoint_desc, ss_companion_desc, ] # String descriptors lang_dict = { 0x0409: [ # US English "Custom USB Function", "Configuration String", "Interface String", ], } # Create function with manual descriptors function = functionfs.Function( path="/tmp/ffs_mount", fs_list=fs_list, hs_list=hs_list, ss_list=ss_list, lang_dict=lang_dict, ) ``` -------------------------------- ### Creating an HID Device with Python functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Defines a report descriptor for a Human Interface Device (HID), specifically a mouse, as per the HID 1.11 specification. This descriptor is used when initializing a functionfs device to present it as an HID device to the host system. The descriptor specifies the device's capabilities, such as button states and movement axes. ```python import functionfs import errno # HID mouse report descriptor from HID 1.11 spec REPORT_DESCRIPTOR = ( b'\x05\x01\x09\x02\xa1\x01\x09\x01' b'\xa1\x00\x05\x09\x19\x01\x29\x03' b'\x15\x00\x25\x01\x95\x03\x75\x01' b'\x81\x02\x95\x01\x75\x05\x81\x01' b'\x05\x01\x09\x30\x09\x31\x15\x81' b'\x25\x7f\x75\x08\x95\x02\x81\x06' b'\xc0\xc0' ) ``` -------------------------------- ### Python: Create USB Bulk Transfer Device with FunctionFS Source: https://context7.com/vpelletier/python-functionfs/llms.txt Implements a USB bulk transfer device using the python-functionfs library. It defines device descriptors for various USB speeds and handles device enablement and event processing. Requires the functionfs library and a Linux kernel with FunctionFS support. ```python import functionfs from functionfs.gadget import GadgetSubprocessManager, ConfigFunctionFFSSubprocess class BulkFunction(functionfs.Function): def __init__(self, path): # Define descriptors for all USB speeds fs_list, hs_list, ss_list = functionfs.getInterfaceInAllSpeeds( interface={ 'bInterfaceClass': functionfs.ch9.USB_CLASS_VENDOR_SPEC, 'iInterface': 1, }, endpoint_list=[ { 'endpoint': { 'bEndpointAddress': functionfs.ch9.USB_DIR_IN, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, }, { 'endpoint': { 'bEndpointAddress': functionfs.ch9.USB_DIR_OUT, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, }, ], ) super().__init__( path, fs_list=fs_list, hs_list=hs_list, ss_list=ss_list, lang_dict={0x0409: ["Bulk Transfer Device"]}, ) def onEnable(self): super().onEnable() # Device is now enabled and ready for transfers print("USB device enabled") class BulkSubprocess(ConfigFunctionFFSSubprocess): def getFunction(self, path): return BulkFunction(path=path) def run(self): try: self.function.processEventsForever() except KeyboardInterrupt: pass # Main program parser = GadgetSubprocessManager.getArgumentParser( description='USB bulk transfer device' ) args = parser.parse_args() with GadgetSubprocessManager( args=args, config_list=[ { 'function_list': [BulkSubprocess], 'MaxPower': 500, 'lang_dict': {0x409: {'configuration': 'Bulk configuration'}}, } ], idVendor=0x1d6b, idProduct=0x0104, lang_dict={ 0x409: { 'product': 'Python USB Device', 'manufacturer': 'python-functionfs', }, }, ) as gadget: gadget.waitForever() ``` -------------------------------- ### Custom USB Control Request Handling with functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Implements a custom USB function that handles vendor-specific control requests. The `CustomFunction` class inherits from `functionfs.Function` and overrides the `onSetup` method to process incoming control transfers. It defines constants for vendor requests (`VENDOR_REQUEST_GET_DATA`, `VENDOR_REQUEST_SET_CONFIG`) and manages a `device_data` buffer for data exchange with the host. This allows for unique communication protocols over USB. ```python import functionfs import struct class CustomFunction(functionfs.Function): VENDOR_REQUEST_GET_DATA = 0x01 VENDOR_REQUEST_SET_CONFIG = 0x02 def __init__(self, path): fs_list, hs_list, ss_list = functionfs.getInterfaceInAllSpeeds( interface={ 'bInterfaceClass': functionfs.ch9.USB_CLASS_VENDOR_SPEC, }, endpoint_list=[], # No endpoints needed for control-only ) super().__init__( path, fs_list=fs_list, hs_list=hs_list, ss_list=ss_list, all_ctrl_recip=True, # Receive all control transfers config0_setup=True, # Receive setup before configuration ) self.device_data = bytearray(64) def onSetup(self, request_type, request, value, index, length): """Handle USB control transfers""" is_in = (request_type & functionfs.ch9.USB_DIR_IN) == functionfs.ch9.USB_DIR_IN req_type = request_type & functionfs.ch9.USB_TYPE_MASK if req_type == functionfs.ch9.USB_TYPE_VENDOR: if request == self.VENDOR_REQUEST_GET_DATA and is_in: # Send data to host data_to_send = self.device_data[:length] self.ep0.write(bytes(data_to_send)) return elif request == self.VENDOR_REQUEST_SET_CONFIG and not is_in: # Receive data from host if length > 0: data = self.ep0.read(length) self.device_data[:len(data)] = data else: self.ep0.read(0) # Acknowledge zero-length return # Let parent class handle standard requests super().onSetup(request_type, request, value, index, length) def onBind(self): print("USB function bound to UDC") def onUnbind(self): print("USB function unbound from UDC") def onEnable(self): super().onEnable() print("USB configuration enabled") def onDisable(self): super().onDisable() print("USB configuration disabled") def onSuspend(self): print("USB bus suspended") def onResume(self): print("USB bus resumed") ``` -------------------------------- ### Asynchronous IN Endpoint Transfers with Python functionfs Source: https://context7.com/vpelletier/python-functionfs/llms.txt Implements an asynchronous IN endpoint for sending data to a USB host. It handles transfer completion, errors (like device disconnection), and submission errors when the AIO queue is full. The `onComplete` method is called upon transfer completion, and `onSubmitEAGAIN` is called when the asynchronous I/O queue is full, allowing for data caching and retry. ```python import functionfs import errno class CustomINEndpoint(functionfs.EndpointINFile): def onComplete(self, buffer_list, user_data, status): """Called when an IN transfer completes""" if status < 0: if status == -errno.ESHUTDOWN: # Device disconnected, stop resubmitting return False print(f"Transfer error: {-status}") return False print(f"Sent {status} bytes successfully") # Resubmit with same buffer return True # Or resubmit with new buffer: # new_data = prepare_next_data() # return [memoryview(new_data)] def onSubmitEAGAIN(self, buffer_list, user_data): """Called when AIO queue is full""" print("AIO queue full, caching transfer") # Store buffer_list for retry later self.pending_transfers.append(buffer_list) class AsyncFunction(functionfs.Function): def __init__(self, path): fs_list, hs_list, ss_list = functionfs.getInterfaceInAllSpeeds( interface={'bInterfaceClass': functionfs.ch9.USB_CLASS_VENDOR_SPEC}, endpoint_list=[ { 'endpoint': { 'bEndpointAddress': 1 | functionfs.ch9.USB_DIR_IN, 'bmAttributes': functionfs.ch9.USB_ENDPOINT_XFER_BULK, }, } ], ) super().__init__( path, fs_list=fs_list, hs_list=hs_list, ss_list=ss_list, in_aio_blocks_max=64, # Max concurrent transfers ) self.pending_transfers = [] def getEndpointClass(self, is_in, descriptor): if is_in: return CustomINEndpoint return super().getEndpointClass(is_in, descriptor) def onEnable(self): super().onEnable() # Submit initial transfer in_endpoint = self.getEndpoint(1) data = bytearray(b"Hello from USB device\n") in_endpoint.submit([data], user_data={'seq': 1}) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.