Device core API

Build a virtual USB device from descriptors and requests: create a USBDevice, author its configuration through DescriptorGroups (raw bytes or lazy blocks, with Endpoint pipes declared via In / Out), register per-interface control / SET_INTERFACE handlers and per-endpoint data callbacks, then plug() it onto a transport. The core has no class concept - the object-oriented authoring layer lives in the class layer; Components → USBDevice explains the split.

Most applications author devices with the class layer or the ready-made classes; the core API is for hand-built devices - see Components → USBDevice for what authoring one directly looks like.

device

The device core - build a virtual USB device from descriptors and requests.

The core knows only descriptors and requests: append descriptor content through :class:DescriptorGroup (raw bytes or lazy blocks), route :class:Endpoint pipes, and register per-interface control / SET_INTERFACE handlers plus per-endpoint data callbacks. Standard requests, Microsoft OS/BOS/WebUSB descriptors and the URB engine are answered here. It has no class concept - the object-oriented authoring layer (Interface / Function) lives in :mod:usbip.function and is built entirely on this module's public API.

USBDevice

USBDevice(vid, pid, *, product=None, manufacturer=None, serial=None, bcdDevice=256, bcdUSB=512, device_class=0, device_subclass=0, device_protocol=0)

Source code in usbip/device.py

def __init__(
    self,
    vid,
    pid,
    *,
    product=None,
    manufacturer=None,
    serial=None,
    bcdDevice=0x0100,
    bcdUSB=0x0200,
    device_class=0,
    device_subclass=0,
    device_protocol=0,
):
    self.vid = vid
    self.pid = pid
    self.bcdDevice = bcdDevice
    self.bcdUSB = bcdUSB
    self.device_class = device_class
    self.device_subclass = device_subclass
    self.device_protocol = device_protocol
    self.composite = False  # set by set_composite()
    self._msos = []  # Microsoft OS advertisements; see enable_msos()
    self._msos_v1 = 0x20  # vendor request codes, see set_msos_vendor_codes()
    self._msos_v2 = 0x21
    self._webusb = None  # set by enable_webusb()
    self.speed = core.SPEED_FULL  # reported speed; set by set_speed()
    self.iso_paced = False  # set by set_iso_pacing()
    self._pace_q = []  # heap of (deadline, seq, respond, urb) - iso pacer
    self._pace_cv = threading.Condition()
    self._pace_thread = None
    self._pace_seq = itertools.count()
    self._groups = []  # DescriptorGroup objects, in wire order
    self._ifnum_owners = []  # one entry per claimed bInterfaceNumber
    self._ctrl = {}  # ifnum (or None = device fallback) -> control handler
    self._set_alt = {}  # ifnum -> SET_INTERFACE handler
    self._reset_hooks = []  # run by reset_io() after flushing endpoints
    self._speed_hooks = []  # run by set_speed()
    self.config = 0
    self._ep_map = {}  # (number, dir) -> Endpoint
    self._transport = None
    # Wire identity on the bus, assigned by plug() unless set_busid() named it
    # first. devnum is unique per listener, which is what lets a host tell two
    # exported devices apart (`lsusb -s`, and each device's pcap stream).
    self.busid = None
    self.busnum = 0
    self.devnum = 0
    self._strings = [None]
    self.iManufacturer = self._add_string(manufacturer)
    self.iProduct = self._add_string(product)
    self.iSerialNumber = self._add_string(serial)

vid instance-attribute

vid = vid

pid instance-attribute

pid = pid

bcdDevice instance-attribute

bcdDevice = bcdDevice

bcdUSB instance-attribute

bcdUSB = bcdUSB

device_class instance-attribute

device_class = device_class

device_subclass instance-attribute

device_subclass = device_subclass

device_protocol instance-attribute

device_protocol = device_protocol

composite instance-attribute

composite = False

speed instance-attribute

speed = core.SPEED_FULL

iso_paced instance-attribute

iso_paced = False

config instance-attribute

config = 0

busid instance-attribute

busid = None

busnum instance-attribute

busnum = 0

devnum instance-attribute

devnum = 0

iManufacturer instance-attribute

iManufacturer = self._add_string(manufacturer)

iProduct instance-attribute

iProduct = self._add_string(product)

iSerialNumber instance-attribute

iSerialNumber = self._add_string(serial)

num_interfaces property

num_interfaces

Number of claimed interfaces == the next free bInterfaceNumber.

interfaces property

interfaces

The claim owners, one per bInterfaceNumber (class-built devices: the :class:Interface objects; raw-authored devices: whatever tag - possibly None - was passed to claim_interface).

functions property

functions

The group owners in wire order (class-built devices: the :class:Function objects).

add_string

add_string(text)

Register a string descriptor and return its index (for iInterface etc.).

Source code in usbip/device.py

def add_string(self, text: str) -> int:
    """Register a string descriptor and return its index (for iInterface etc.)."""
    return self._add_string(text)

add_group

add_group(owner=None)

Open a new :class:DescriptorGroup - one function's worth of configuration-descriptor content and the Microsoft OS scoping unit. owner is an opaque tag the layer above may attach (never interpreted here).

Source code in usbip/device.py

def add_group(self, owner=None) -> DescriptorGroup:
    """Open a new :class:`DescriptorGroup` - one function's worth of
    configuration-descriptor content and the Microsoft OS scoping unit. `owner` is an
    opaque tag the layer above may attach (never interpreted here)."""
    group = DescriptorGroup(self, owner)
    self._groups.append(group)
    return group

route_endpoint

route_endpoint(ep, owner=None)

Route an endpoint into the device, relocating its address if taken.

The declared address is a preference: it is honoured when the number is free, so a lone function keeps the addresses it asks for, but when another endpoint already holds it the endpoint is moved to the lowest free number in the same direction. Without this a composite would emit a descriptor in which two interfaces claim one address, and the host would route both to whichever registered last. Mirrors ep_alloc_number() in the C library's src/usbip_device.c. owner tags the endpoint for diagnostics.

Source code in usbip/device.py

def route_endpoint(self, ep: Endpoint, owner=None) -> Endpoint:
    """Route an endpoint into the device, relocating its address if taken.

    The declared address is a *preference*: it is honoured when the number is
    free, so a lone function keeps the addresses it asks for, but when another
    endpoint already holds it the endpoint is moved to the lowest free number
    in the same direction. Without this a composite would emit a descriptor in
    which two interfaces claim one address, and the host would route both to
    whichever registered last. Mirrors ep_alloc_number() in
    the C library's src/usbip_device.c. `owner` tags the endpoint for diagnostics."""
    if (ep.number, ep.dir) in self._ep_map:
        for number in range(1, 16):
            if (number, ep.dir) not in self._ep_map:
                ep.number = number
                ep.addr = number | (0x80 if ep.dir == IN else 0x00)
                break
        else:
            raise RuntimeError(
                f"no free endpoint number left for {owner.__class__.__name__} "
                f"({'IN' if ep.dir == IN else 'OUT'})"
            )
    ep.owner = owner
    self._ep_map[(ep.number, ep.dir)] = ep
    return ep

on_control

on_control(ifnum, handler)

Register handler(setup, data=b"") -> bytes | None for class/vendor control requests addressed to interface ifnum (raise :class:Stall to reject). ifnum=None registers the device-level fallback: requests with a non-interface recipient, or for an interface nobody registered.

Source code in usbip/device.py

def on_control(self, ifnum, handler):
    """Register `handler(setup, data=b"") -> bytes | None` for class/vendor
    control requests addressed to interface `ifnum` (raise :class:`Stall` to
    reject). ``ifnum=None`` registers the device-level fallback: requests with
    a non-interface recipient, or for an interface nobody registered."""
    self._ctrl[ifnum] = handler

has_control_fallback

has_control_fallback()

Whether a device-level control fallback (on_control(None, ...)) is set.

Source code in usbip/device.py

def has_control_fallback(self) -> bool:
    """Whether a device-level control fallback (``on_control(None, ...)``) is set."""
    return None in self._ctrl

on_set_alt

on_set_alt(ifnum, handler)

Register handler(alt) for SET_INTERFACE on ifnum. Unregistered interfaces acknowledge SET_INTERFACE with no side effect.

Source code in usbip/device.py

def on_set_alt(self, ifnum, handler):
    """Register `handler(alt)` for SET_INTERFACE on `ifnum`. Unregistered
    interfaces acknowledge SET_INTERFACE with no side effect."""
    self._set_alt[ifnum] = handler

add_reset_hook

add_reset_hook(fn)

Run fn() on bus reset (host attach), after all endpoints are flushed.

Source code in usbip/device.py

def add_reset_hook(self, fn):
    """Run `fn()` on bus reset (host attach), after all endpoints are flushed."""
    self._reset_hooks.append(fn)

add_speed_hook

add_speed_hook(fn)

Run fn(speed) when the reported link speed changes (see set_speed).

Source code in usbip/device.py

def add_speed_hook(self, fn):
    """Run `fn(speed)` when the reported link speed changes (see set_speed)."""
    self._speed_hooks.append(fn)

set_device_triple

set_device_triple(cls, sub, proto)

Set the device-descriptor class triple - unless the device is composite, where 0xEF/0x02/0x01 is pinned (a function's own triple must not silently unmake the composite) and the request is ignored with a diagnostic.

Source code in usbip/device.py

def set_device_triple(self, cls, sub, proto):
    """Set the device-descriptor class triple - unless the device is composite,
    where 0xEF/0x02/0x01 is pinned (a function's own triple must not silently
    unmake the composite) and the request is ignored with a diagnostic."""
    if self.composite:
        sys.stderr.write(
            f"[usbip] device class {cls:02x}/{sub:02x}/{proto:02x} ignored: composite pins "
            "the triple to EF/02/01\n"
        )
    else:
        self.device_class, self.device_subclass, self.device_protocol = cls, sub, proto

set_composite

set_composite()

Declare the device composite: several independent class functions on one device. Call BEFORE adding any class.

Sets the device-descriptor triple to 0xEF/0x02/0x01 (Miscellaneous / Common Class / Interface Association) and asks every multi-interface class to emit an Interface Association Descriptor grouping its own interfaces.

Linux does not need this - cdc_acm and friends group their interfaces from the class-specific descriptors (CDC's Union descriptor). Windows does: usbccgp splits a composite into one child devnode per function, and without an IAD it splits per interface, handing CDC's data interface to a different devnode from its communications interface, so no COM port forms.

A single-function device must NOT set this. The triple is pinned from here on: a class's own device_triple (e.g. Bluetooth's 0xE0/0x01/0x01) is ignored with a diagnostic, since overwriting 0xEF/0x02/0x01 would silently unmake the composite. Mirrors the C usbip_device_set_composite().

Source code in usbip/device.py

def set_composite(self):
    """Declare the device composite: several independent class functions on one
    device. Call BEFORE adding any class.

    Sets the device-descriptor triple to 0xEF/0x02/0x01 (Miscellaneous / Common
    Class / Interface Association) and asks every multi-interface class to emit
    an Interface Association Descriptor grouping its own interfaces.

    Linux does not need this - cdc_acm and friends group their interfaces from
    the class-specific descriptors (CDC's Union descriptor). Windows does:
    usbccgp splits a composite into one child devnode per *function*, and
    without an IAD it splits per *interface*, handing CDC's data interface to a
    different devnode from its communications interface, so no COM port forms.

    A single-function device must NOT set this. The triple is pinned from
    here on: a class's own ``device_triple`` (e.g. Bluetooth's
    0xE0/0x01/0x01) is ignored with a diagnostic, since overwriting
    0xEF/0x02/0x01 would silently unmake the composite. Mirrors the C
    ``usbip_device_set_composite()``."""
    self.composite = True
    self.device_class, self.device_subclass, self.device_protocol = 0xEF, 0x02, 0x01

enable_msos

enable_msos(compatible=b'WINUSB', guid=None, function=None)

Advertise Microsoft OS 1.0 + 2.0 descriptors with the given Compatible ID so Windows auto-binds a function driver without an .inf: "WINUSB" (so libusb apps like dfu-util work without Zadig) or "MTP" (Media Transfer Protocol). guid adds a DeviceInterfaceGUID - WinUSB wants one, MTP leaves it None. Bumps bcdUSB to 0x0210 so the host fetches the BOS (Microsoft OS 2.0).

function scopes the advertisement to one :class:DescriptorGroup (a :class:Function or :class:Interface is accepted and resolved to its group). Leave it None to describe the whole device, which is right for a single-function device but not for a composite: there Windows gives each function its own devnode, and a device-wide Compatible ID would bind one driver over all of them. Prefer :meth:Function.enable_msos. Re-enabling the same scope replaces it.

Source code in usbip/device.py

def enable_msos(self, compatible=b"WINUSB", guid=None, function=None):
    """Advertise Microsoft OS 1.0 + 2.0 descriptors with the given Compatible ID so
    Windows auto-binds a function driver without an .inf: "WINUSB" (so libusb
    apps like dfu-util work without Zadig) or "MTP" (Media Transfer Protocol).
    `guid` adds a DeviceInterfaceGUID - WinUSB wants one, MTP leaves it None.
    Bumps bcdUSB to 0x0210 so the host fetches the BOS (Microsoft OS 2.0).

    `function` scopes the advertisement to one :class:`DescriptorGroup` (a
    :class:`Function` or :class:`Interface` is accepted and resolved to its
    group). Leave it None to describe the whole device, which is right for a
    single-function device but not for a composite: there Windows gives each
    function its own devnode, and a device-wide Compatible ID would bind one
    driver over all of them. Prefer :meth:`Function.enable_msos`. Re-enabling
    the same scope replaces it."""
    group = self._resolve_group(function)
    entry = {"group": group, "compatible": bytes(compatible), "guid": guid}
    for i, existing in enumerate(self._msos):
        if existing["group"] is group:
            self._msos[i] = entry
            break
    else:
        self._msos.append(entry)
    self.bcdUSB = 0x0210

enable_winusb

enable_winusb(guid=None, function=None)

Advertise WinUSB (Compatible ID "WINUSB" + a DeviceInterfaceGUID).

Source code in usbip/device.py

def enable_winusb(self, guid=None, function=None):
    """Advertise WinUSB (Compatible ID "WINUSB" + a DeviceInterfaceGUID)."""
    self.enable_msos(b"WINUSB", guid or core.DEFAULT_WINUSB_GUID, function)

set_msos_vendor_codes

set_msos_vendor_codes(v1=None, v2=None)

Override the vendor request codes for the Microsoft OS requests (default 0x20 for Microsoft OS 1.0, 0x21 for Microsoft OS 2.0). The device answers those codes itself before any interface sees them, so change them if a function on this device uses vendor request 0x20/0x21 with wIndex 0x0004, 0x0005 or 0x0007.

Source code in usbip/device.py

def set_msos_vendor_codes(self, v1=None, v2=None):
    """Override the vendor request codes for the Microsoft OS requests (default 0x20 for
    Microsoft OS 1.0, 0x21 for Microsoft OS 2.0). The device answers those codes itself before
    any interface sees them, so change them if a function on this device uses
    vendor request 0x20/0x21 with wIndex 0x0004, 0x0005 or 0x0007."""
    if v1:
        self._msos_v1 = v1
    if v2:
        self._msos_v2 = v2

enable_webusb

enable_webusb(vendor_code, url)

Advertise WebUSB: add a WebUSB platform-capability descriptor to the BOS and answer the bVendorCode/GET_URL (wIndex 0x02) vendor request with url, so a WebUSB-capable browser surfaces the device and can open it. Pick vendor_code distinct from WinUSB's 0x20/0x21 if both are enabled. Pair with enable_winusb() so the device also binds WinUSB on Windows. Bumps bcdUSB to 0x0210 so the host fetches the BOS. iLandingPage is derived from the URL (present when non-empty), matching the C usbip_device_enable_webusb().

Source code in usbip/device.py

def enable_webusb(self, vendor_code, url):
    """Advertise WebUSB: add a WebUSB platform-capability descriptor to the BOS
    and answer the bVendorCode/GET_URL (wIndex 0x02) vendor request with `url`,
    so a WebUSB-capable browser surfaces the device and can open it. Pick
    vendor_code distinct from WinUSB's 0x20/0x21 if both are enabled. Pair with
    enable_winusb() so the device also binds WinUSB on Windows. Bumps bcdUSB to
    0x0210 so the host fetches the BOS. iLandingPage is derived from the URL
    (present when non-empty), matching the C usbip_device_enable_webusb()."""
    self._webusb = {"vendor": vendor_code, "url": url, "landing": 1 if url else 0}
    self.bcdUSB = 0x0210

set_speed

set_speed(speed)

Report a link speed to the importer (default SPEED_FULL). USB/IP is URB-level, so "high speed" is just the reported speed plus speed-correct endpoint sizing. Call BEFORE add() so speed-aware classes (cdc_acm, msc, mtp, uac, uvc) size their endpoints; for robustness any already-added interface is re-sized here too.

Source code in usbip/device.py

def set_speed(self, speed: int):
    """Report a link speed to the importer (default SPEED_FULL). USB/IP is URB-level, so
    "high speed" is just the reported speed plus speed-correct endpoint sizing. Call BEFORE
    add() so speed-aware classes (cdc_acm, msc, mtp, uac, uvc) size their endpoints; for
    robustness any already-added interface is re-sized here too."""
    self.speed = speed
    for fn in self._speed_hooks:
        fn(speed)

add

add(obj)

Add a :class:Function (or a bare :class:Interface, wrapped in an anonymous single-interface Function) - the one way to add anything, bundled class or your own. Composition itself lives in the class layer: any object implementing _attach(dev) can be added; the device only provides the primitives (add_group, claim_interface, route_endpoint, on_control, ...).

Returns the class's data-plane handle - _attach's return value, which for a :class:Function is its :attr:~usbip.function.Function.primary (CDCACM -> the Data port) and for a bare interface is the interface itself. An _attach that returns nothing yields the object passed in.

Source code in usbip/device.py

def add(self, obj):
    """Add a :class:`Function` (or a bare :class:`Interface`, wrapped in an anonymous
    single-interface Function) - the one way to add anything, bundled class or your
    own. Composition itself lives in the class layer: any object implementing
    ``_attach(dev)`` can be added; the device only provides the primitives
    (`add_group`, `claim_interface`, `route_endpoint`, `on_control`, ...).

    Returns the class's data-plane handle - ``_attach``'s return value, which for a
    :class:`Function` is its :attr:`~usbip.function.Function.primary` (``CDCACM`` ->
    the Data port) and for a bare interface is the interface itself. An ``_attach``
    that returns nothing yields the object passed in."""
    handle = obj._attach(self)
    return obj if handle is None else handle

is_iso

is_iso(ep_number, direction)

Source code in usbip/device.py

def is_iso(self, ep_number, direction) -> bool:
    ep = self._ep_map.get((ep_number, direction))
    return bool(ep and ep.type == "iso")

set_iso_pacing

set_iso_pacing(enabled=True)

Pace isochronous completions to real time. USB/IP has no SOF clock, so by default iso transfers complete instantly and the host's audio/video engine free-runs (a UAC speaker plays many times too fast). When enabled, each iso completion is scheduled for the wall-clock time its packet schedule would really take (a per-endpoint deadline) and delivered by a background pacer thread, so the serve thread never blocks - full-duplex (speaker + mic at once) stays real time, like real hardware.

Source code in usbip/device.py

def set_iso_pacing(self, enabled=True):
    """Pace isochronous completions to real time. USB/IP has no SOF clock, so by
    default iso transfers complete instantly and the host's audio/video engine
    free-runs (a UAC speaker plays many times too fast). When enabled, each iso
    completion is scheduled for the wall-clock time its packet schedule would
    really take (a per-endpoint deadline) and delivered by a background pacer
    thread, so the serve thread never blocks - full-duplex (speaker + mic at
    once) stays real time, like real hardware."""
    self.iso_paced = enabled

pace_cancel

pace_cancel(seqnum)

Drop a queued iso completion whose URB was unlinked (so the host's stream releases cleanly instead of getting a late completion).

Source code in usbip/device.py

def pace_cancel(self, seqnum):
    """Drop a queued iso completion whose URB was unlinked (so the host's stream
    releases cleanly instead of getting a late completion)."""
    with self._pace_cv:
        self._pace_q = [entry for entry in self._pace_q if entry[3].seqnum != seqnum]
        heapq.heapify(self._pace_q)

cancel_urb

cancel_urb(seqnum)

Handle a host CMD_UNLINK for seqnum: drop any parked IN URB (real kernels pipeline interrupt-IN URBs and unlink the spares; completing an unlinked seqnum makes vhci 'cannot find urb' and tear the device down) and any queued iso completion.

Source code in usbip/device.py

def cancel_urb(self, seqnum):
    """Handle a host CMD_UNLINK for `seqnum`: drop any parked IN URB (real
    kernels pipeline interrupt-IN URBs and unlink the spares; completing an
    unlinked seqnum makes vhci 'cannot find urb' and tear the device down) and
    any queued iso completion."""
    self.pace_cancel(seqnum)
    for ep in self._ep_map.values():
        if ep.cancel(seqnum):
            return

device_descriptor

device_descriptor()

Source code in usbip/device.py

def device_descriptor(self) -> core.DeviceDescriptor:
    return core.DeviceDescriptor(
        bcdUSB=self.bcdUSB,
        idVendor=self.vid,
        idProduct=self.pid,
        bcdDevice=self.bcdDevice,
        bDeviceClass=self.device_class,
        bDeviceSubClass=self.device_subclass,
        bDeviceProtocol=self.device_protocol,
        iManufacturer=self.iManufacturer,
        iProduct=self.iProduct,
        iSerialNumber=self.iSerialNumber,
        bNumConfigurations=1,
    )

config_bytes

config_bytes()

The full configuration descriptor, exactly as served to the host: the groups' blocks rendered in order (callables re-render every time, so late-bound content is always current).

Source code in usbip/device.py

def config_bytes(self) -> bytes:
    """The full configuration descriptor, exactly as served to the host: the
    groups' blocks rendered in order (callables re-render every time, so
    late-bound content is always current)."""
    body = b"".join(group._render() for group in self._groups)
    cfg = core.ConfigurationDescriptor(
        wTotalLength=9 + len(body), bNumInterfaces=self.num_interfaces
    )
    return cfg.pack() + body

interface_triples

interface_triples()

(bInterfaceClass, bInterfaceSubClass, bInterfaceProtocol) per interface, in bInterfaceNumber order, read from the rendered configuration (alternate setting 0) - used for the USB/IP device-list reply.

Source code in usbip/device.py

def interface_triples(self):
    """(bInterfaceClass, bInterfaceSubClass, bInterfaceProtocol) per interface,
    in bInterfaceNumber order, read from the rendered configuration (alternate
    setting 0) - used for the USB/IP device-list reply."""
    blob = self.config_bytes()
    triples = {}
    for desc in core.iter_descriptors(blob, offset=9):
        if desc[1] == core.DT_INTERFACE and len(desc) >= 8 and desc[3] == 0:  # alt 0
            triples[desc[2]] = (desc[5], desc[6], desc[7])
    return [triples[ifnum] for ifnum in sorted(triples)]

bos_bytes

bos_bytes()

The BOS descriptor (WebUSB / Microsoft OS 2.0 capabilities), as served to the host.

Source code in usbip/device.py

def bos_bytes(self) -> bytes:
    """The BOS descriptor (WebUSB / Microsoft OS 2.0 capabilities), as served to the host."""
    caps = []
    if self._webusb:  # WebUSB platform capability
        caps.append(core.webusb_platform_cap(self._webusb["vendor"], self._webusb["landing"]))
    if self._msos:  # Microsoft OS 2.0 platform capability
        caps.append(core.msos2_platform_cap(self._msos_v2, len(self._msos2_bytes())))
    return core.bos(*caps)

vendor_descriptor

vendor_descriptor(setup)

Answer a core-handled vendor request - Microsoft OS (WinUSB) or WebUSB GET_URL - or None if it isn't one of ours (so an interface can handle it).

Source code in usbip/device.py

def vendor_descriptor(self, setup):
    """Answer a core-handled vendor request - Microsoft OS (WinUSB) or WebUSB
    GET_URL - or None if it isn't one of ours (so an interface can handle it)."""
    if self._msos and (resp := self._winusb_vendor(setup)) is not None:
        return resp
    if self._webusb and setup.bRequest == self._webusb["vendor"] and setup.wIndex == 0x02:
        return core.webusb_url(self._webusb["url"])  # WebUSB GET_URL
    return None

reset_io

reset_io()

Flush every endpoint (buffered IN data + parked URBs) and let each interface reset its own buffers. The USB/IP server calls this when a host attaches, so a re-attach behaves like a fresh plug (a real device sees a bus reset). Without it, events that piled up after a previous host left get delivered to the new host and corrupt its init.

Source code in usbip/device.py

def reset_io(self):
    """Flush every endpoint (buffered IN data + parked URBs) and let each
    interface reset its own buffers. The USB/IP server calls this when a host
    attaches, so a re-attach behaves like a fresh plug (a real device sees a
    bus reset). Without it, events that piled up after a previous host left
    get delivered to the new host and corrupt its init."""
    for ep in self._ep_map.values():
        ep.reset()
    for fn in self._reset_hooks:
        fn()

handle_urb

handle_urb(urb, respond=None)

Process one URB. Returns the urb to send its response now, or None if the URB was PARKED (an IN endpoint with no data yet) - it will be completed asynchronously via respond when the device writes data, so the server's read loop never blocks. respond is supplied by the USB/IP server; when absent we fall back to a short blocking wait.

Source code in usbip/device.py

def handle_urb(self, urb, respond=None):
    """Process one URB. Returns the urb to send its response now, or None if
    the URB was PARKED (an IN endpoint with no data yet) - it will be
    completed asynchronously via `respond` when the device writes data, so
    the server's read loop never blocks. `respond` is supplied by the USB/IP
    server; when absent we fall back to a short blocking wait."""
    if urb.ep == 0:
        return self._handle_control(urb)
    ep = self._ep_map.get((urb.ep, urb.direction))
    if ep is None:
        urb.status = -19  # -ENODEV
        _dbg_xfer(urb.direction, urb.ep, None, urb.length, "-> no such endpoint")
        return urb
    if ep.halted:  # until CLEAR_FEATURE(ENDPOINT_HALT)
        urb.buffer, urb.actual, urb.status = b"", 0, STALL_STATUS
        _dbg_xfer(urb.direction, urb.ep, ep.type, urb.length, "-> STALL")
        return urb
    if ep.type == "iso" and urb.direction == IN and urb.iso_packets is not None:
        lengths = [pkt[1] for pkt in urb.iso_packets]
        chunks = ep.on_iso(ep, lengths) if ep.on_iso else []
        buf = bytearray()
        for i, pkt in enumerate(urb.iso_packets):
            ch = chunks[i] if i < len(chunks) else b""
            pkt[2], pkt[3] = len(ch), 0  # actual_length, status
            buf += ch
        urb.buffer, urb.actual, urb.status = bytes(buf), len(buf), 0
        _dbg_xfer(IN, urb.ep, ep.type, len(buf), f"{len(urb.iso_packets)} pkts")
        if self.iso_paced and respond is not None:  # deliver at real time via the pacer
            deadline = self._iso_deadline(ep, len(urb.iso_packets))
            self._pace_schedule(deadline, respond, urb)
            return None
        return urb
    if ep.type == "iso" and urb.direction == OUT and urb.iso_packets is not None:
        packets = [urb.buffer[off : off + length] for off, length, _a, _s in urb.iso_packets]
        if ep.on_iso:
            ep.on_iso(ep, packets)
        for pkt in urb.iso_packets:
            pkt[2], pkt[3] = pkt[1], 0  # actual_length = length, status ok
        urb.actual, urb.status = 0, 0  # OUT RET carries descriptors, no data
        _dbg_xfer(
            OUT, urb.ep, ep.type, sum(len(pkt) for pkt in packets), f"{len(urb.iso_packets)} pkts"
        )
        if self.iso_paced and respond is not None:  # deliver at real time via the pacer
            self._pace_schedule(self._iso_deadline(ep, len(urb.iso_packets)), respond, urb)
            return None
        return urb
    if urb.direction == IN:
        length = urb.length or ep.mps
        if respond is None:  # synchronous fallback (no async server)
            data = ep._take_in(length, max(0.5, urb.interval / 1000 or 2.0))
            urb.buffer, urb.actual, urb.status = data, len(data), 0
            _dbg_xfer(IN, urb.ep, ep.type, len(data))
            return urb

        def complete(data, status=0):  # runs when ep.write() feeds the parked URB
            urb.buffer, urb.actual, urb.status = data, len(data), status
            note = "-> STALL" if status else "parked URB"
            _dbg_xfer(IN, urb.ep, ep.type, len(data), note)
            respond(urb)

        data = ep.take_or_park(length, complete, urb.seqnum)
        if data is None:
            _dbg_xfer(IN, urb.ep, ep.type, length, "-> parked, no data yet")
            return None  # parked: hold the URB until data (NAK)
        urb.buffer, urb.actual, urb.status = data, len(data), 0
        _dbg_xfer(IN, urb.ep, ep.type, len(data))
        return urb
    # OUT: the endpoint's hook consumes it; without one, queue it on the
    # endpoint for Endpoint.read
    try:
        if ep.on_out:
            ep.on_out(ep, urb.buffer)
        else:
            ep._put_out(urb.buffer)
    except Stall:
        urb.status = -32
        _dbg_xfer(OUT, urb.ep, ep.type, len(urb.buffer), "-> STALL")
        return urb
    urb.actual, urb.status = len(urb.buffer), 0
    _dbg_xfer(OUT, urb.ep, ep.type, len(urb.buffer))
    return urb

plug

plug(via=None)

Serve this device and return immediately (a context manager that unplugs on exit).

Plugging another device onto the same transport exports it alongside this one rather than failing to bind: the listener is shared, and each device is named by its own busid - 1-1, 1-2, ... in plug order unless :meth:set_busid named it. The importer picks one (usbip attach -b 1-2), and every imported device gets its own connection. Distinct from a composite device (:meth:set_composite), which is one device with several functions.

Source code in usbip/device.py

def plug(self, via=None):
    """Serve this device and return immediately (a context manager that
    unplugs on exit).

    Plugging another device onto the same transport exports it alongside this
    one rather than failing to bind: the listener is shared, and each device
    is named by its own busid - ``1-1``, ``1-2``, ... in plug order unless
    :meth:`set_busid` named it. The importer picks one (``usbip attach -b
    1-2``), and every imported device gets its own connection. Distinct from a
    *composite* device (:meth:`set_composite`), which is one device with
    several functions.
    """
    from .transport import default_transport

    self._check_endpoint_conflicts()
    self._check_composite_iads()
    self._transport = via or default_transport()
    self._transport.serve(self)
    return _PlugContext(self)

set_busid

set_busid(busid)

Name this device on the wire instead of the 1-<n> assigned by :meth:plug. Only useful when a process exports several devices and the importer wants stable names regardless of plug order. Call before plug().

Source code in usbip/device.py

def set_busid(self, busid: str):
    """Name this device on the wire instead of the ``1-<n>`` assigned by
    :meth:`plug`. Only useful when a process exports several devices and the
    importer wants stable names regardless of plug order. Call before plug()."""
    self.busid = busid
    return self

unplug

unplug()

Source code in usbip/device.py

def unplug(self):
    if self._transport:
        # Per device: a listener shared with other plugged devices keeps
        # serving them, and closes when the last one unplugs.
        self._transport.stop(self)
        self._transport = None

DescriptorGroup

DescriptorGroup(dev, owner=None)

One function's worth of configuration-descriptor content: an ordered run of descriptor blocks plus the interface numbers claimed through it. Created with :meth:USBDevice.add_group; the groups render in creation order to form the configuration descriptor.

A block is bytes, or a zero-argument callable returning bytes - callables are rendered on every GET_DESCRIPTOR, so content that depends on late-bound state (relocated endpoint addresses, speed-sized wMaxPacketSize, alternate settings) is always current.

The group is also the scoping key for the Microsoft OS descriptors (:meth:enable_msos): on a composite each group is advertised as its own Windows function. A device that never creates a group is treated as one implicit device-wide group.

Source code in usbip/device.py

def __init__(self, dev, owner=None):
    self._dev = dev
    self.owner = owner  # opaque back-pointer for the layer above
    self.interface_numbers = []  # bInterfaceNumbers claimed through this group
    self._blocks = []

owner instance-attribute

owner = owner

interface_numbers instance-attribute

interface_numbers = []

add

add(block)

Append a descriptor block: bytes or callable() -> bytes.

Source code in usbip/device.py

def add(self, block):
    """Append a descriptor block: ``bytes`` or ``callable() -> bytes``."""
    self._blocks.append(block)

claim_interface

claim_interface(owner=None)

Claim the next free bInterfaceNumber for this group and return it.

Source code in usbip/device.py

def claim_interface(self, owner=None) -> int:
    """Claim the next free bInterfaceNumber for this group and return it."""
    ifnum = self._dev._claim_interface(owner)
    self.interface_numbers.append(ifnum)
    return ifnum

add_endpoint

add_endpoint(spec)

Create an :class:Endpoint from an :func:In/:func:Out spec, route it (relocating the address if taken), and append its endpoint descriptor as a lazy block - the descriptor always reflects the final address and size.

Source code in usbip/device.py

def add_endpoint(self, spec: _EPSpec) -> Endpoint:
    """Create an :class:`Endpoint` from an :func:`In`/:func:`Out` spec, route it
    (relocating the address if taken), and append its endpoint descriptor as a
    lazy block - the descriptor always reflects the final address and size."""
    ep = Endpoint(spec)
    self._dev.route_endpoint(ep, owner=self.owner or self)

    def render():
        return core.EndpointDescriptor(
            bEndpointAddress=ep.addr,
            bmAttributes=core.XFER_BY_NAME[ep.type],
            wMaxPacketSize=ep.mps,
            bInterval=ep.interval,
        ).pack()

    self._blocks.append(render)
    return ep

enable_msos

enable_msos(compatible=b'WINUSB', guid=None)

Advertise Microsoft OS descriptors scoped to this group's interfaces (see :meth:USBDevice.enable_msos).

Source code in usbip/device.py

def enable_msos(self, compatible=b"WINUSB", guid=None):
    """Advertise Microsoft OS descriptors scoped to this group's interfaces (see
    :meth:`USBDevice.enable_msos`)."""
    self._dev.enable_msos(compatible, guid, function=self)

enable_winusb

enable_winusb(guid=None)

:meth:enable_msos with the "WINUSB" Compatible ID.

Source code in usbip/device.py

def enable_winusb(self, guid=None):
    """:meth:`enable_msos` with the "WINUSB" Compatible ID."""
    self.enable_msos(b"WINUSB", guid or core.DEFAULT_WINUSB_GUID)

Endpoint

Endpoint(spec)

A byte pipe. Device writes IN data and reads OUT data; the device's URB handler is the other end. Backed by a queue so reads/writes decouple.

Source code in usbip/device.py

def __init__(self, spec: _EPSpec):
    self.addr = spec.addr
    self.number = spec.addr & 0x0F
    self.dir = IN if spec.addr & 0x80 else OUT
    self.type = spec.type
    self.mps = spec.mps
    self.interval = spec.interval
    self.owner = None  # opaque tag set by USBDevice.route_endpoint()
    # data-path callbacks; the device consults these instead of knowing who
    # built the endpoint. Bound methods/closures carry any state they need.
    self.on_out = None  # fn(ep, data);      None -> queue for ep.read()
    # iso callback, contract by endpoint direction:
    #   IN:  fn(ep, lengths) -> list[bytes]  (requested maxima in); None -> no data
    #   OUT: fn(ep, packets)                 (received bytes in);   None -> discard
    self.on_iso = None
    self.halted = False  # STALL until CLEAR_FEATURE(ENDPOINT_HALT)
    self._q: deque = deque()
    self._cv = threading.Condition()
    self._pending: deque = deque()  # IN URBs parked waiting for data
    self.iso_deadline = 0.0  # iso pacing: monotonic time the next frame is due

addr instance-attribute

addr = spec.addr

number instance-attribute

number = spec.addr & 15

dir instance-attribute

dir = IN if spec.addr & 128 else OUT

type instance-attribute

type = spec.type

mps instance-attribute

mps = spec.mps

interval instance-attribute

interval = spec.interval

owner instance-attribute

owner = None

on_out instance-attribute

on_out = None

on_iso instance-attribute

on_iso = None

halted instance-attribute

halted = False

iso_deadline instance-attribute

iso_deadline = 0.0

write

write(data)

Queue IN data and complete as many parked URBs as it covers, in FIFO order.

Draining a WHILE loop here (not a single pop) matters once a client keeps more than one IN URB outstanding: completing only one URB per write left the remainder in _q with further URBs still parked - nothing ever re-matched them, so the stream stalled and the NEXT write (e.g. an MSC CSW) was handed to the wrong URB, completing them out of order. Each parked URB takes up to its length from the FRONT chunk only (_pop_locked), preserving write/short-packet boundaries.

Source code in usbip/device.py

def write(self, data: bytes):  # device -> host (IN)
    """Queue IN data and complete as many parked URBs as it covers, in FIFO order.

    Draining a WHILE loop here (not a single pop) matters once a client keeps
    more than one IN URB outstanding: completing only one URB per write left
    the remainder in _q with further URBs still parked - nothing ever
    re-matched them, so the stream stalled and the NEXT write (e.g. an MSC
    CSW) was handed to the wrong URB, completing them out of order. Each
    parked URB takes up to its length from the FRONT chunk only (_pop_locked),
    preserving write/short-packet boundaries."""
    done = []
    with self._cv:
        self._q.append(bytes(data))
        while self._pending and self._q:
            length, cb, _seq = self._pending.popleft()
            done.append((cb, self._pop_locked(length)))
        if self._q:
            self._cv.notify_all()
    for cb, chunk in done:
        cb(chunk)  # complete parked URBs outside the lock

read

read(timeout=1.0)

Source code in usbip/device.py

def read(self, timeout=1.0) -> bytes:  # device reads host OUT data
    with self._cv:
        if not self._q and not self._cv.wait_for(lambda: bool(self._q), timeout):
            return b""
        return self._q.popleft()

take_or_park

take_or_park(length, complete, seqnum=None)

Return queued IN data if any; otherwise register complete(data) to run when data arrives (NAK-until-data) and return None. seqnum lets the server cancel this parked URB if the host later unlinks it.

Source code in usbip/device.py

def take_or_park(self, length, complete, seqnum=None):
    """Return queued IN data if any; otherwise register `complete(data)` to
    run when data arrives (NAK-until-data) and return None. `seqnum` lets the
    server cancel this parked URB if the host later unlinks it."""
    with self._cv:
        if self._q:
            return self._pop_locked(length)
        self._pending.append((length, complete, seqnum))
        return None

cancel

cancel(seqnum)

Drop a parked IN URB the host has UNLINKed, so it is never completed (completing an unlinked seqnum desyncs the kernel's vhci). Returns True if one was removed.

Source code in usbip/device.py

def cancel(self, seqnum):
    """Drop a parked IN URB the host has UNLINKed, so it is never completed
    (completing an unlinked seqnum desyncs the kernel's vhci). Returns True if
    one was removed."""
    with self._cv:
        for i, entry in enumerate(self._pending):
            if entry[2] == seqnum:
                del self._pending[i]
                return True
    return False

stall

stall()

Halt the endpoint: every URB STALLs until the host clears the halt.

Any IN URB the host already has outstanding is STALLed immediately - leaving it parked is the hang this exists to avoid, since the host would otherwise sit on the read until its own timeout fires. Queued data survives the halt and is delivered once it is cleared: mass storage halts bulk-IN to abandon a failed data phase, then queues the CSW.

Source code in usbip/device.py

def stall(self):
    """Halt the endpoint: every URB STALLs until the host clears the halt.

    Any IN URB the host already has outstanding is STALLed immediately -
    leaving it parked is the hang this exists to avoid, since the host would
    otherwise sit on the read until its own timeout fires. Queued data
    survives the halt and is delivered once it is cleared: mass storage halts
    bulk-IN to abandon a failed data phase, then queues the CSW."""
    with self._cv:
        self.halted = True
        parked, self._pending = list(self._pending), deque()
    for _length, cb, _seq in parked:
        cb(b"", STALL_STATUS)  # complete outside the lock

clear_halt

clear_halt()

Clear a halt, as CLEAR_FEATURE(ENDPOINT_HALT) does. Queued data is kept.

Source code in usbip/device.py

def clear_halt(self):
    """Clear a halt, as CLEAR_FEATURE(ENDPOINT_HALT) does. Queued data is kept."""
    with self._cv:
        self.halted = False

reset

reset()

Drop buffered data and any parked URBs - called when a host (re)attaches so leftovers from a previous session can't corrupt the new one.

Source code in usbip/device.py

def reset(self):
    """Drop buffered data and any parked URBs - called when a host (re)attaches
    so leftovers from a previous session can't corrupt the new one."""
    with self._cv:
        self._q.clear()
        self._pending.clear()
        self.halted = False

In

In(addr, type='bulk', mps=64, interval=0)

Declare a device->host endpoint, e.g. In(0x81, 'interrupt', mps=8).

Source code in usbip/device.py

def In(addr, type="bulk", mps=64, interval=0):
    """Declare a device->host endpoint, e.g. In(0x81, 'interrupt', mps=8)."""
    return _EPSpec(addr | 0x80, type, mps, interval)

Out

Out(addr, type='bulk', mps=64, interval=0)

Declare a host->device endpoint, e.g. Out(0x01, 'bulk').

Source code in usbip/device.py

def Out(addr, type="bulk", mps=64, interval=0):
    """Declare a host->device endpoint, e.g. Out(0x01, 'bulk')."""
    return _EPSpec(addr & 0x7F, type, mps, interval)