ecm_interface.py

# MicroPython USB CDC-ECM (Ethernet Control Model) interface
# MIT license; Copyright (c) 2024
#
# Implements a CDC-ECM USB Ethernet adapter using the micropython-lib
# usb.device.core Interface base class. Designed for Raspberry Pi Pico 2 W.

import struct
from micropython import const
from usb.device.core import Interface, split_bmRequestType

_EP_IN_FLAG = const(1 << 7)

# Control transfer stages
_STAGE_SETUP = const(1)
_STAGE_DATA = const(2)
_STAGE_ACK = const(3)

# Request types
_REQ_TYPE_CLASS = const(0x1)

# CDC class/subclass for ECM
_CDC_CLASS = const(0x02)
_CDC_SUBCLASS_ECM = const(0x06)  # Ethernet Networking Control Model
_CDC_DATA_CLASS = const(0x0A)

# CS_INTERFACE descriptor type
_CS_DESC_TYPE = const(0x24)

# CDC functional descriptor subtypes
_CDC_FUNC_DESC_HEADER = const(0x00)
_CDC_FUNC_DESC_UNION = const(0x06)
_CDC_FUNC_DESC_ETHERNET = const(0x0F)

# CDC version
_CDC_VERSION = const(0x0120)

# ECM class request
_SET_ETHERNET_PACKET_FILTER = const(0x43)

# Endpoint and buffer sizes
_BULK_EP_LEN = const(64)
_NOTIF_EP_LEN = const(8)
_NOTIF_EP_INTERVAL = const(10)
_MAX_SEGMENT_SIZE = const(1514)
_RX_BUF_SIZE = const(1536)
_RX_QUEUE_MAX = const(16)

# Default MAC address as uppercase hex string (no separators)
_DEFAULT_MAC = "AABBCCDDEEF1"


class ECMInterface(Interface):
    """CDC-ECM (Ethernet Control Model) USB interface for MicroPython.

    Presents a USB Ethernet adapter to the host. Frames can be sent and
    received using send_frame() and recv_frame().
    """

    def __init__(self):
        Interface.__init__(self)
        self._ep_notif = None
        self._ep_in = None
        self._ep_out = None
        self._configured = False
        self._tx_busy = False
        self._tx_buf = None
        self._rx_buf = None
        self._rx_queue = []
        self._itf_num = None
        self._tx_queue = []  # Retry queue for frames that couldn't be sent

    def desc_cfg(self, desc, itf_num, ep_num, strs):
        # Save interface number for control transfer filtering
        self._itf_num = itf_num

        # --- Interface Association Descriptor ---
        # Groups the communication and data interfaces together
        desc.interface_assoc(
            itf_num,  # bFirstInterface
            2,  # bInterfaceCount
            _CDC_CLASS,  # bFunctionClass
            _CDC_SUBCLASS_ECM,  # bFunctionSubClass
        )

        # --- CDC Communication Interface ---
        # 1 endpoint (notification interrupt IN)
        desc.interface(
            itf_num,  # bInterfaceNumber
            1,  # bNumEndpoints
            _CDC_CLASS,  # bInterfaceClass
            _CDC_SUBCLASS_ECM,  # bInterfaceSubClass
            0x00,  # bInterfaceProtocol (ECM uses 0x00)
        )

        # --- Header Functional Descriptor ---
        # 5 bytes: bFunctionLength, bDescriptorType (CS_INTERFACE),
        # bDescriptorSubtype (Header), bcdCDC (1.20)
        desc.pack(
            "<BBBH",
            5,  # bFunctionLength
            _CS_DESC_TYPE,  # bDescriptorType
            _CDC_FUNC_DESC_HEADER,  # bDescriptorSubtype
            _CDC_VERSION,  # bcdCDC
        )

        # --- Union Functional Descriptor ---
        # 5 bytes: links control and data interfaces
        desc.pack(
            "<BBBBB",
            5,  # bFunctionLength
            _CS_DESC_TYPE,  # bDescriptorType
            _CDC_FUNC_DESC_UNION,  # bDescriptorSubtype
            itf_num,  # bControlInterface
            itf_num + 1,  # bSubordinateInterface0
        )

        # --- Ethernet Networking Functional Descriptor ---
        # 13 bytes total. The iMACAddress field is a string descriptor index
        # pointing to the MAC address hex string. Use struct.pack + extend
        # to avoid padding issues with the mixed field sizes.
        i_mac = len(strs)
        strs.append(_DEFAULT_MAC)

        desc.extend(
            struct.pack(
                "<BBBB",
                13,  # bFunctionLength
                _CS_DESC_TYPE,  # bDescriptorType
                _CDC_FUNC_DESC_ETHERNET,  # bDescriptorSubtype
                i_mac,  # iMACAddress (string descriptor index)
            )
        )
        desc.extend(
            struct.pack(
                "<I",
                0,  # bmEthernetStatistics (4 bytes, none supported)
            )
        )
        desc.extend(
            struct.pack(
                "<H",
                _MAX_SEGMENT_SIZE,  # wMaxSegmentSize
            )
        )
        desc.extend(
            struct.pack(
                "<HB",
                0,  # wNumberMCFilters
                0,  # bNumberPowerFilters
            )
        )

        # --- Notification Endpoint (interrupt IN) ---
        self._ep_notif = ep_num | _EP_IN_FLAG
        desc.endpoint(self._ep_notif, "interrupt", _NOTIF_EP_LEN, _NOTIF_EP_INTERVAL)

        # --- CDC Data Interface ---
        # 2 endpoints (bulk OUT + bulk IN), class 0x0A (CDC Data)
        desc.interface(
            itf_num + 1,  # bInterfaceNumber
            2,  # bNumEndpoints
            _CDC_DATA_CLASS,  # bInterfaceClass
            0x00,  # bInterfaceSubClass (CDC Data = 0)
            0x00,  # bInterfaceProtocol (CDC Data = 0)
        )

        # --- Bulk OUT Endpoint ---
        self._ep_out = ep_num + 1
        desc.endpoint(self._ep_out, "bulk", _BULK_EP_LEN, 0)

        # --- Bulk IN Endpoint ---
        self._ep_in = (ep_num + 1) | _EP_IN_FLAG
        desc.endpoint(self._ep_in, "bulk", _BULK_EP_LEN, 0)

    def num_itfs(self):
        return 2

    def num_eps(self):
        return 2

    def on_open(self):
        super().on_open()
        self._configured = True
        self._rx_buf = bytearray(_RX_BUF_SIZE)
        self._tx_busy = False
        self._rx_queue = []
        # Start listening for incoming frames on the bulk OUT endpoint
        self.submit_xfer(self._ep_out, self._rx_buf, self._rx_done)

    def on_reset(self):
        super().on_reset()
        self._configured = False
        self._tx_busy = False
        self._rx_queue = []

    def _rx_done(self, ep, result, xferred_bytes):
        if not result and xferred_bytes > 0:  # result == 0 means success
            # Copy received data into the queue
            if len(self._rx_queue) >= _RX_QUEUE_MAX:
                self._rx_queue.pop(0)  # Drop oldest frame
            self._rx_queue.append(bytes(self._rx_buf[:xferred_bytes]))
        # Re-submit read for next frame
        self.submit_xfer(self._ep_out, self._rx_buf, self._rx_done)

    def _tx_done(self, ep, result, xferred_bytes):
        self._tx_busy = False
        # Drain TX queue if pending
        while self._tx_queue and not self._tx_busy:
            frame = self._tx_queue.pop(0)
            if self._configured:
                self._tx_busy = True
                self._tx_buf = bytes(frame)
                self.submit_xfer(self._ep_in, self._tx_buf, self._tx_done)
        # Drain TX queue if pending
        while self._tx_queue and not self._tx_busy:
            frame = self._tx_queue.pop(0)
            if self._configured:
                self._tx_busy = True
                self._tx_buf = bytes(frame)
                self.submit_xfer(self._ep_in, self._tx_buf, self._tx_done)

    def send_frame(self, frame):
        """Send an Ethernet frame to the USB host.

        Returns True if the frame was queued, False if not configured.
        Frames are queued if USB is busy and sent in order.
        """
        if not self._configured:
            return False
        if self._tx_busy:
            # Queue for later — limit queue to prevent memory exhaustion
            if len(self._tx_queue) < 32:
                self._tx_queue.append(bytes(frame))
            return True
        self._tx_busy = True
        self._tx_buf = bytes(frame)
        self.submit_xfer(self._ep_in, self._tx_buf, self._tx_done)
        return True
        self._tx_busy = True
        self._tx_buf = bytes(frame)
        self.submit_xfer(self._ep_in, self._tx_buf, self._tx_done)
        return True

    def recv_frame(self):
        """Receive the next Ethernet frame from the USB host.

        Returns a bytes object, or None if no frames are queued.
        """
        if self._rx_queue:
            return self._rx_queue.pop(0)
        return None

    def has_frames(self):
        """Return True if there are received frames waiting in the queue."""
        return len(self._rx_queue) > 0

    def on_interface_control_xfer(self, stage, request):
        bmRequestType, bRequest, wValue, wIndex, wLength = struct.unpack(
            "BBHHH", request
        )
        recipient, req_type, req_dir = split_bmRequestType(bmRequestType)

        # Handle SET_ETHERNET_PACKET_FILTER (bmRequestType=0x21, bRequest=0x43)
        if req_type == _REQ_TYPE_CLASS:
            if bRequest == _SET_ETHERNET_PACKET_FILTER:
                return True  # Accept for all stages (SETUP, DATA, ACK)
            # Accept other class-specific requests silently
            return True

        return False