usbd: Add basic HID keypad (with status LED detection), basic mass storage

micropython/usbd/device.py

@@ -2,7 +2,7 @@
 # MIT license; Copyright (c) 2022 Angus Gratton
 from micropython import const
 import machine
-import ustruct
+import struct
 
 from .utils import split_bmRequestType, EP_IN_FLAG
 
@@ -78,7 +78,8 @@ def __init__(self):
             descriptor_device_cb=self._descriptor_device_cb,
             descriptor_config_cb=self._descriptor_config_cb,
             descriptor_string_cb=self._descriptor_string_cb,
-            open_driver_cb=self._open_driver_cb,
+            open_cb=self._open_cb,
+            reset_cb=self._reset_cb,
             control_xfer_cb=self._control_xfer_cb,
             xfer_cb=self._xfer_cb,
         )
@@ -118,7 +119,7 @@ def _descriptor_device_cb(self):
 
         FMT = "<BBHBBBBHHHBBBB"
         # static descriptor fields
-        f = ustruct.unpack(FMT, self._usbd.static.desc_device)
+        f = struct.unpack(FMT, self._usbd.static.desc_device)
 
         def maybe_set(value, idx):
             # Override a numeric descriptor value or keep static value f[idx] if 'value' is None
@@ -134,7 +135,7 @@ def maybe_set_str(s, idx):
 
         # Either copy each descriptor field directly from the static device descriptor, or 'maybe'
         # override if a custom value has been set on this object
-        return ustruct.pack(
+        return struct.pack(
             FMT,
             f[0],  # bLength
             f[1],  # bDescriptorType
@@ -257,7 +258,7 @@ def _update_configuration_descriptor(self, desc):
         bNumInterfaces = self._usbd.static.itf_max if self.include_static else 0
         bNumInterfaces += len(self._itfs)
 
-        ustruct.pack_into(
+        struct.pack_into(
             "<BBHBBBBB",
             desc,
             0,
@@ -286,9 +287,37 @@ def _descriptor_string_cb(self, index):
         except IndexError:
             return None
 
-    def _open_driver_cb(self, interface_desc_view):
-        # Singleton callback from TinyUSB custom class driver
-        pass
+    def _open_cb(self, interface_desc_view):
+        # Singleton callback from TinyUSB custom class driver, when USB host does
+        # Set Configuration. The "runtime class device" accepts all interfaces that
+        # it has sent in descriptors, and calls this callback.
+
+        # Walk the view of the "claimed" descriptor data provided in the
+        # callback and call handle_open() on each claimed interface
+        #
+        # ... this may be unnecessary at the moment, as only one configuration is supported so we
+        # can probably assume all the interfaces will be included.
+        i = 0
+        while i < len(interface_desc_view):
+            # descriptor length, type, and index (if it's an interface descriptor)
+            dl, dt, di = interface_desc_view[i : i + 3]
+            if dt == _STD_DESC_INTERFACE_TYPE:
+                if di >= self._usbd.static.itf_max:
+                    di -= self._usbd.static.itf_max
+                self._itfs[di].handle_open()
+            i += dl
+            assert dl
+
+    def _reset_cb(self):
+        # Callback when the USB device is reset by the host
+
+        # Cancel outstanding transfer callbacks
+        for k in self._ep_cbs.keys():
+            self._ep_cbs[k] = None
+
+        # Allow interfaces to respond to the reset
+        for itf in self._itfs:
+            itf.handle_reset()
 
     def _submit_xfer(self, ep_addr, data, done_cb=None):
         # Singleton function to submit a USB transfer (of any type except control).
@@ -387,6 +416,7 @@ def __init__(
         self.bInterfaceSubClass = bInterfaceSubClass
         self.bInterfaceProtocol = bInterfaceProtocol
         self.interface_str = interface_str
+        self._open = False
 
     def get_itf_descriptor(self, num_eps, itf_idx, str_idx):
         # Return the interface descriptor binary data and associated other
@@ -421,7 +451,7 @@ def get_itf_descriptor(self, num_eps, itf_idx, str_idx):
         #   (indexes in the descriptor data should start from 'str_idx'.)
         #
         # See USB 2.0 specification section 9.6.5 p267 for standard interface descriptors.
-        desc = ustruct.pack(
+        desc = struct.pack(
             "<" + "B" * _STD_DESC_INTERFACE_LEN,
             _STD_DESC_INTERFACE_LEN,  # bLength
             _STD_DESC_INTERFACE_TYPE,  # bDescriptorType
@@ -466,6 +496,30 @@ def get_endpoint_descriptors(self, ep_addr, str_idx):
         #   start from ep_addr, optionally with the utils.EP_IN_FLAG bit set.)
         return (b"", [], [])
 
+    def handle_open(self):
+        # Callback called when the USB host accepts the device configuration.
+        #
+        # Override this function to initiate any operations that the USB interface
+        # should do when the USB device is configured to the host.
+        self._open = True
+
+    def handle_reset(self):
+        # Callback called on every registered interface when the USB device is
+        # reset by the host. This can happen when the USB device is unplugged,
+        # or if the host triggers a reset for some other reason.
+        #
+        # Override this function to cancel any pending operations specific to
+        # the interface (outstanding USB transfers are already cancelled).
+        #
+        # At this point, no USB functionality is available - handle_open() will
+        # be called later if/when the USB host re-enumerates and configures the
+        # interface.
+        self._open = False
+
+    def is_open(self):
+        # Returns True if the interface is in use
+        return self._open
+
     def handle_device_control_xfer(self, stage, request):
         # Control transfer callback. Override to handle a non-standard device
         # control transfer where bmRequestType Recipient is Device, Type is
@@ -486,11 +540,11 @@ def handle_device_control_xfer(self, stage, request):
         # The function can call split_bmRequestType() to split bmRequestType into
         # (Recipient, Type, Direction).
         #
-        # Result:
+        # Result, any of:
         #
-        # - True to continue the request False to STALL the endpoint A buffer
-        # - interface object to provide a buffer to the host as part of the
-        # - transfer, if possible.
+        # - True to continue the request, False to STALL the endpoint.
+        # - Buffer interface object to provide a buffer to the host as part of the
+        #   transfer, if possible.
         return False
 
     def handle_interface_control_xfer(self, stage, request):
@@ -512,7 +566,8 @@ def handle_interface_control_xfer(self, stage, request):
     def handle_endpoint_control_xfer(self, stage, request):
         # Control transfer callback. Override to handle a device
         # control transfer where bmRequestType Recipient is Endpoint and
-        # the lower byte of wIndex indicates an endpoint address associated with this interface.
+        # the lower byte of wIndex indicates an endpoint address associated
+        # with this interface.
         #
         # bmRequestType Type will generally have any value except
         # utils.REQ_TYPE_STANDARD, as Standard endpoint requests are handled by
@@ -546,4 +601,25 @@ def submit_xfer(self, ep_addr, data, done_cb=None):
         #
         # Note that done_cb may be called immediately, possibly before this
         # function has returned to the caller.
+        if not self._open:
+            raise RuntimeError
         return get_usbdevice()._submit_xfer(ep_addr, data, done_cb)
+
+    def set_ep_stall(self, ep_addr, stall):
+        # Set or clear endpoint STALL state, according to the bool "stall" parameter.
+        #
+        # Generally endpoint STALL is handled automatically by TinyUSB, but
+        # there are some device classes that need to explicitly stall or unstall
+        # an endpoint under certain conditions.
+        if not self._open or ep_addr not in get_usbdevice()._eps:
+            raise RuntimeError
+        get_usbdevice()._usbd.set_ep_stall(ep_addr, stall)
+
+    def get_ep_stall(self, ep_addr):
+        # Get the current endpoint STALL state.
+        #
+        # Endpoint can be stalled/unstalled by host, TinyUSB stack, or calls to
+        # set_ep_stall().
+        if not self._open or ep_addr not in get_usbdevice()._eps:
+            raise RuntimeError
+        return get_usbdevice()._usbd.get_ep_stall(ep_addr)

micropython/usbd/hid.py

@@ -8,11 +8,12 @@
     split_bmRequestType,
     EP_IN_FLAG,
     STAGE_SETUP,
+    STAGE_DATA,
     REQ_TYPE_STANDARD,
     REQ_TYPE_CLASS,
 )
 from micropython import const
-import ustruct
+import struct
 
 _DESC_HID_TYPE = const(0x21)
 _DESC_REPORT_TYPE = const(0x22)
@@ -43,6 +44,7 @@ def __init__(
         self,
         report_descriptor,
         extra_descriptors=[],
+        set_report_buf=None,
         protocol=_INTERFACE_PROTOCOL_NONE,
         interface_str=None,
         use_out_ep=False,
@@ -58,6 +60,12 @@ def __init__(
         #   descriptors, to append after the mandatory report descriptor. Most
         #   HID devices do not use these.
         #
+        # - set_report_buf is an optional writable buffer object (i.e.
+        #   bytearray), where SET_REPORT requests from the host can be
+        #   written. Only necessary if the report_descriptor contains Output
+        #   entries. If set, the size must be at least the size of the largest
+        #   Output entry.
+        #
         # - protocol can be set to a specific value as per HID v1.11 section 4.3 Protocols, p9.
         #
         # - interface_str is an optional string descriptor to associate with the HID USB interface.
@@ -67,16 +75,23 @@ def __init__(
         super().__init__(_INTERFACE_CLASS, _INTERFACE_SUBCLASS_NONE, protocol, interface_str)
         self.extra_descriptors = extra_descriptors
         self.report_descriptor = report_descriptor
+        self._set_report_buf = set_report_buf
         self._int_ep = None  # set during enumeration
         self._out_ep = None
         self.use_out_ep = use_out_ep
 
     def get_report(self):
         return False
 
-    def set_report(self):
-        # Override this if you are expecting reports from the host
-        return False
+    def handle_set_report(self, report_data, report_id, report_type):
+        # Override this function in order to handle SET REPORT requests from the host,
+        # where it sends data to the HID device.
+        #
+        # This function will only be called if the Report descriptor contains at least one Output entry,
+        # and the set_report_buf argument is provided to the constructor.
+        #
+        # Return True to complete the control transfer normally, False to abort it.
+        return True
 
     def send_report(self, report_data):
         # Helper function to send a HID report in the typical USB interrupt
@@ -109,7 +124,7 @@ def get_hid_descriptor(self):
         # and optional additional descriptors.
         #
         # See HID Specification Version 1.1, Section 6.2.1 HID Descriptor p22
-        result = ustruct.pack(
+        result = struct.pack(
             "<BBHBBBH",
             9 + 3 * len(self.extra_descriptors),  # bLength
             _DESC_HID_TYPE,  # bDescriptorType
@@ -125,46 +140,70 @@ def get_hid_descriptor(self):
         # support in base class
         if self.extra_descriptors:
             result += b"".join(
-                ustruct.pack("<BH", dt, len(dd)) for (dt, dd) in self.extra_descriptors
+                struct.pack("<BH", dt, len(dd)) for (dt, dd) in self.extra_descriptors
             )
 
         return result
 
     def handle_interface_control_xfer(self, stage, request):
         # Handle standard and class-specific interface control transfers for HID devices.
-        bmRequestType, bRequest, wValue, _, _ = request
+        bmRequestType, bRequest, wValue, _, wLength = request
 
         recipient, req_type, _ = split_bmRequestType(bmRequestType)
 
-        if stage != STAGE_SETUP:
-            return True  # allow request DATA/ACK stages to complete normally
-
-        if req_type == REQ_TYPE_STANDARD:
-            # HID Spec p48: 7.1 Standard Requests
-            if bRequest == _REQ_CONTROL_GET_DESCRIPTOR:
-                desc_type = wValue >> 8
-                if desc_type == _DESC_HID_TYPE:
-                    return self.get_hid_descriptor()
-                if desc_type == _DESC_REPORT_TYPE:
-                    return self.report_descriptor
-        elif req_type == REQ_TYPE_CLASS:
-            # HID Spec p50: 7.2 Class-Specific Requests
-            if bRequest == _REQ_CONTROL_GET_REPORT:
-                return False  # Unsupported for now
-            if bRequest == _REQ_CONTROL_GET_IDLE:
-                return bytes([self.idle_rate])
-            if bRequest == _REQ_CONTROL_GET_PROTOCOL:
-                return bytes([self.protocol])
-            if bRequest == _REQ_CONTROL_SET_IDLE:
-                self.idle_rate = wValue >> 8
-                return b""
-            if bRequest == _REQ_CONTROL_SET_PROTOCOL:
-                self.protocol = wValue
-                return b""
-            if bRequest == _REQ_CONTROL_SET_REPORT:
-                return self.set_report()
-
-        return False  # Unsupported
+        if stage == STAGE_SETUP:
+            if req_type == REQ_TYPE_STANDARD:
+                # HID Spec p48: 7.1 Standard Requests
+                if bRequest == _REQ_CONTROL_GET_DESCRIPTOR:
+                    desc_type = wValue >> 8
+                    if desc_type == _DESC_HID_TYPE:
+                        return self.get_hid_descriptor()
+                    if desc_type == _DESC_REPORT_TYPE:
+                        return self.report_descriptor
+            elif req_type == REQ_TYPE_CLASS:
+                # HID Spec p50: 7.2 Class-Specific Requests
+                if bRequest == _REQ_CONTROL_GET_REPORT:
+                    print("GET_REPORT?")
+                    return False  # Unsupported for now
+                if bRequest == _REQ_CONTROL_GET_IDLE:
+                    return bytes([self.idle_rate])
+                if bRequest == _REQ_CONTROL_GET_PROTOCOL:
+                    return bytes([self.protocol])
+                if bRequest == _REQ_CONTROL_SET_IDLE:
+                    self.idle_rate = wValue >> 8
+                    return b""
+                if bRequest == _REQ_CONTROL_SET_PROTOCOL:
+                    self.protocol = wValue
+                    return b""
+                if bRequest == _REQ_CONTROL_SET_REPORT:
+                    # Return the _set_report_buf to be filled with the
+                    # report data
+                    if not self._set_report_buf:
+                        return False
+                    elif wLength >= len(self._set_report_buf):
+                        # Saves an allocation if the size is exactly right (or will be a short read)
+                        return self._set_report_buf
+                    else:
+                        # Otherwise, need to wrap the buffer in a memoryview of the correct length
+                        #
+                        # TODO: check this is correct, maybe TinyUSB won't mind if we ask for more
+                        # bytes than the host has offered us.
+                        return memoryview(self._set_report_buf)[:wLength]
+            return False  # Unsupported
+
+        if stage == STAGE_DATA:
+            if req_type == REQ_TYPE_CLASS:
+                if bRequest == _REQ_CONTROL_SET_REPORT and self._set_report_buf:
+                    report_id = wValue & 0xFF
+                    report_type = wValue >> 8
+                    report_data = self._set_report_buf
+                    if wLength < len(report_data):
+                        # as above, need to truncate the buffer if we read less
+                        # bytes than what was provided
+                        report_data = memoryview(self._set_report_buf)[:wLength]
+                    self.handle_set_report(report_data, report_id, report_type)
+
+        return True  # allow DATA/ACK stages to complete normally
 
 
 # Basic 3-button mouse HID Report Descriptor.
@@ -254,7 +293,7 @@ def send_report(self, dx=0, dy=0):
         # transfer after it's submitted. So reusing a bytearray() creates a risk
         # of a race condition if a new report transfer is submitted using the
         # same buffer, before the previous one has completed.
-        report = ustruct.pack("Bbb", b, dx, dy)
+        report = struct.pack("Bbb", b, dx, dy)
 
         super().send_report(report)