| # Copyright 2014 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Generic USB gadget functionality. |
| """ |
| |
| import struct |
| |
| import usb_constants |
| |
| |
| class Gadget(object): |
| """Basic functionality for a USB device. |
| |
| Implements standard control requests assuming that a subclass will handle |
| class- or vendor-specific requests. |
| """ |
| |
| def __init__(self, device_desc, fs_config_desc, hs_config_desc): |
| """Create a USB gadget device. |
| |
| Args: |
| device_desc: USB device descriptor. |
| fs_config_desc: Low/full-speed device descriptor. |
| hs_config_desc: High-speed device descriptor. |
| """ |
| self._speed = usb_constants.Speed.UNKNOWN |
| self._chip = None |
| self._device_desc = device_desc |
| self._fs_config_desc = fs_config_desc |
| self._hs_config_desc = hs_config_desc |
| # dict mapping language codes to a dict mapping indexes to strings |
| self._strings = {} |
| # dict mapping interface numbers to a set of endpoint addresses |
| self._active_endpoints = {} |
| |
| def GetDeviceDescriptor(self): |
| return self._device_desc |
| |
| def GetFullSpeedConfigurationDescriptor(self): |
| return self._fs_config_desc |
| |
| def GetHighSpeedConfigurationDescriptor(self): |
| return self._hs_config_desc |
| |
| def GetConfigurationDescriptor(self): |
| if self._speed == usb_constants.Speed.FULL: |
| return self._fs_config_desc |
| elif self._speed == usb_constants.Speed.HIGH: |
| return self._hs_config_desc |
| else: |
| raise RuntimeError('Device is not connected.') |
| |
| def GetSpeed(self): |
| return self._speed |
| |
| def AddStringDescriptor(self, index, value, lang=0x0409): |
| """Add a string descriptor to this device. |
| |
| Args: |
| index: String descriptor index (matches 'i' fields in descriptors). |
| value: The string. |
| lang: Language code (default: English). |
| |
| Raises: |
| ValueError: The index or language code is invalid. |
| """ |
| if index < 1 or index > 255: |
| raise ValueError('String descriptor index out of range.') |
| if lang < 0 or lang > 0xffff: |
| raise ValueError('String descriptor language code out of range.') |
| |
| lang_strings = self._strings.setdefault(lang, {}) |
| lang_strings[index] = value |
| |
| def Connected(self, chip, speed): |
| """The device has been connected to a USB host. |
| |
| Args: |
| chip: USB controller. |
| speed: Connection speed. |
| """ |
| self._speed = speed |
| self._chip = chip |
| |
| def Disconnected(self): |
| """The device has been disconnected from the USB host.""" |
| self._speed = usb_constants.Speed.UNKNOWN |
| self._chip = None |
| self._active_endpoints.clear() |
| |
| def IsConnected(self): |
| return self._chip is not None |
| |
| def ControlRead(self, request_type, request, value, index, length): |
| """Handle a read on the control pipe (endpoint zero). |
| |
| Args: |
| request_type: bmRequestType field of the setup packet. |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| A buffer to return to the USB host with len <= length on success or |
| None to stall the pipe. |
| """ |
| assert request_type & usb_constants.Dir.IN |
| typ = request_type & usb_constants.Type.MASK |
| recipient = request_type & usb_constants.Recipient.MASK |
| if typ == usb_constants.Type.STANDARD: |
| return self.StandardControlRead( |
| recipient, request, value, index, length) |
| elif typ == usb_constants.Type.CLASS: |
| return self.ClassControlRead( |
| recipient, request, value, index, length) |
| elif typ == usb_constants.Type.VENDOR: |
| return self.VendorControlRead( |
| recipient, request, value, index, length) |
| |
| def ControlWrite(self, request_type, request, value, index, data): |
| """Handle a write to the control pipe (endpoint zero). |
| |
| Args: |
| request_type: bmRequestType field of the setup packet. |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| data: Data stage of the request. |
| |
| Returns: |
| True on success, None to stall the pipe. |
| """ |
| assert not request_type & usb_constants.Dir.IN |
| typ = request_type & usb_constants.Type.MASK |
| recipient = request_type & usb_constants.Recipient.MASK |
| if typ == usb_constants.Type.STANDARD: |
| return self.StandardControlWrite( |
| recipient, request, value, index, data) |
| elif typ == usb_constants.Type.CLASS: |
| return self.ClassControlWrite( |
| recipient, request, value, index, data) |
| elif typ == usb_constants.Type.VENDOR: |
| return self.VendorControlWrite( |
| recipient, request, value, index, data) |
| |
| def SendPacket(self, endpoint, data): |
| """Send a data packet on the given endpoint. |
| |
| Args: |
| endpoint: Endpoint address. |
| data: Data buffer. |
| |
| Raises: |
| ValueError: If the endpoint address is not valid. |
| RuntimeError: If the device is not connected. |
| """ |
| if self._chip is None: |
| raise RuntimeError('Device is not connected.') |
| if not endpoint & usb_constants.Dir.IN: |
| raise ValueError('Cannot write to non-input endpoint.') |
| self._chip.SendPacket(endpoint, data) |
| |
| def ReceivePacket(self, endpoint, data): |
| """Handle an incoming data packet on one of the device's active endpoints. |
| |
| This method should be overridden by a subclass implementing endpoint-based |
| data transfers. |
| |
| Args: |
| endpoint: Endpoint address. |
| data: Data buffer. |
| """ |
| pass |
| |
| def HaltEndpoint(self, endpoint): |
| """Signals a STALL condition to the host on the given endpoint. |
| |
| Args: |
| endpoint: Endpoint address. |
| """ |
| self._chip.HaltEndpoint(endpoint) |
| |
| def StandardControlRead(self, recipient, request, value, index, length): |
| """Handle standard control transfers. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| A buffer to return to the USB host with len <= length on success or |
| None to stall the pipe. |
| """ |
| if request == usb_constants.Request.GET_DESCRIPTOR: |
| desc_type = value >> 8 |
| desc_index = value & 0xff |
| desc_lang = index |
| |
| print 'GetDescriptor(recipient={}, type={}, index={}, lang={})'.format( |
| recipient, desc_type, desc_index, desc_lang) |
| |
| return self.GetDescriptor(recipient, desc_type, desc_index, desc_lang, |
| length) |
| |
| def GetDescriptor(self, recipient, typ, index, lang, length): |
| """Handle a standard GET_DESCRIPTOR request. |
| |
| See Universal Serial Bus Specification Revision 2.0 section 9.4.3. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| typ: Descriptor type. |
| index: Descriptor index. |
| lang: Descriptor language code. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| The value of the descriptor or None to stall the pipe. |
| """ |
| if recipient == usb_constants.Recipient.DEVICE: |
| if typ == usb_constants.DescriptorType.STRING: |
| return self.GetStringDescriptor(index, lang, length) |
| |
| def ClassControlRead(self, recipient, request, value, index, length): |
| """Handle class-specific control transfers. |
| |
| This function should be overridden by a subclass implementing a particular |
| device class. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| A buffer to return to the USB host with len <= length on success or |
| None to stall the pipe. |
| """ |
| _ = recipient, request, value, index, length |
| return None |
| |
| def VendorControlRead(self, recipient, request, value, index, length): |
| """Handle vendor-specific control transfers. |
| |
| This function should be overridden by a subclass if implementing a device |
| that responds to vendor-specific requests. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| A buffer to return to the USB host with len <= length on success or |
| None to stall the pipe. |
| """ |
| _ = recipient, request, value, index, length |
| return None |
| |
| def StandardControlWrite(self, recipient, request, value, index, data): |
| """Handle standard control transfers. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| data: Data stage of the request. |
| |
| Returns: |
| True on success, None to stall the pipe. |
| """ |
| _ = data |
| |
| if request == usb_constants.Request.SET_CONFIGURATION: |
| if recipient == usb_constants.Recipient.DEVICE: |
| return self.SetConfiguration(value) |
| elif request == usb_constants.Request.SET_INTERFACE: |
| if recipient == usb_constants.Recipient.INTERFACE: |
| return self.SetInterface(index, value) |
| |
| def ClassControlWrite(self, recipient, request, value, index, data): |
| """Handle class-specific control transfers. |
| |
| This function should be overridden by a subclass implementing a particular |
| device class. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| data: Data stage of the request. |
| |
| Returns: |
| True on success, None to stall the pipe. |
| """ |
| _ = recipient, request, value, index, data |
| return None |
| |
| def VendorControlWrite(self, recipient, request, value, index, data): |
| """Handle vendor-specific control transfers. |
| |
| This function should be overridden by a subclass if implementing a device |
| that responds to vendor-specific requests. |
| |
| Args: |
| recipient: Request recipient (device, interface, endpoint, etc.) |
| request: bRequest field of the setup packet. |
| value: wValue field of the setup packet. |
| index: wIndex field of the setup packet. |
| data: Data stage of the request. |
| |
| Returns: |
| True on success, None to stall the pipe. |
| """ |
| _ = recipient, request, value, index, data |
| return None |
| |
| def GetStringDescriptor(self, index, lang, length): |
| """Handle a GET_DESCRIPTOR(String) request from the host. |
| |
| Descriptor index 0 returns the set of languages supported by the device. |
| All other indices return the string descriptors registered with those |
| indices. |
| |
| See Universal Serial Bus Specification Revision 2.0 section 9.6.7. |
| |
| Args: |
| index: Descriptor index. |
| lang: Descriptor language code. |
| length: Maximum amount of data the host expects the device to return. |
| |
| Returns: |
| The string descriptor or None to stall the pipe if the descriptor is not |
| found. |
| """ |
| if index == 0: |
| length = 2 + len(self._strings) * 2 |
| header = struct.pack('<BB', length, usb_constants.DescriptorType.STRING) |
| lang_codes = [struct.pack('<H', lang) |
| for lang in self._strings.iterkeys()] |
| buf = header + ''.join(lang_codes) |
| assert len(buf) == length |
| return buf[:length] |
| elif lang not in self._strings: |
| return None |
| elif index not in self._strings[lang]: |
| return None |
| else: |
| string = self._strings[lang][index].encode('UTF-16LE') |
| header = struct.pack( |
| '<BB', 2 + len(string), usb_constants.DescriptorType.STRING) |
| buf = header + string |
| return buf[:length] |
| |
| def SetConfiguration(self, index): |
| """Handle a SET_CONFIGURATION request from the host. |
| |
| See Universal Serial Bus Specification Revision 2.0 section 9.4.7. |
| |
| Args: |
| index: Configuration index selected. |
| |
| Returns: |
| True on success, None on error to stall the pipe. |
| """ |
| print 'SetConfiguration({})'.format(index) |
| |
| for endpoint_addrs in self._active_endpoints.values(): |
| for endpoint_addr in endpoint_addrs: |
| self._chip.StopEndpoint(endpoint_addr) |
| endpoint_addrs.clear() |
| |
| if index == 0: |
| # SET_CONFIGRATION(0) puts the device into the Address state which |
| # Windows does before suspending the port. |
| return True |
| elif index != 1: |
| return None |
| |
| config_desc = self.GetConfigurationDescriptor() |
| for interface_desc in config_desc.GetInterfaces(): |
| if interface_desc.bAlternateSetting != 0: |
| continue |
| endpoint_addrs = self._active_endpoints.setdefault( |
| interface_desc.bInterfaceNumber, set()) |
| for endpoint_desc in interface_desc.GetEndpoints(): |
| self._chip.StartEndpoint(endpoint_desc) |
| endpoint_addrs.add(endpoint_desc.bEndpointAddress) |
| return True |
| |
| def SetInterface(self, interface, alt_setting): |
| """Handle a SET_INTERFACE request from the host. |
| |
| See Universal Serial Bus Specification Revision 2.0 section 9.4.10. |
| |
| Args: |
| interface: Interface number to configure. |
| alt_setting: Alternate setting to select. |
| |
| Returns: |
| True on success, None on error to stall the pipe. |
| """ |
| print 'SetInterface({}, {})'.format(interface, alt_setting) |
| |
| config_desc = self.GetConfigurationDescriptor() |
| interface_desc = None |
| for interface_option in config_desc.GetInterfaces(): |
| if (interface_option.bInterfaceNumber == interface and |
| interface_option.bAlternateSetting == alt_setting): |
| interface_desc = interface_option |
| if interface_desc is None: |
| return None |
| |
| endpoint_addrs = self._active_endpoints.setdefault(interface, set()) |
| for endpoint_addr in endpoint_addrs: |
| self._chip.StopEndpoint(endpoint_addr) |
| for endpoint_desc in interface_desc.GetEndpoints(): |
| self._chip.StartEndpoint(endpoint_desc) |
| endpoint_addrs.add(endpoint_desc.bEndpointAddress) |
| return True |
