Host API

Drive USB devices over USB/IP. open / attach return a Handle with libusb-shaped control / bulk / interrupt / isochronous transfers, or subclass Driver for a reusable host driver that binds by class code (use it via Driver.open(vid, pid, transport=...)). A Connection is the underlying imported-device session both are built on.

host

Host side - write host drivers that drive USB devices over USB/IP.

open()/attach() return a Handle with libusb-shaped transfers. A reusable host driver is a Driver subclass that auto-binds by class code (the mirror of a device-side Interface).

Handle

Handle(conn, device_descriptor)

libusb-shaped device handle.

Source code in usbip/host.py

def __init__(self, conn: Connection, device_descriptor: DeviceDescriptor):
    self.conn = conn
    self.device_descriptor = device_descriptor

conn instance-attribute

conn = conn

device_descriptor instance-attribute

device_descriptor = device_descriptor

control

control(*args)

Source code in usbip/host.py

def control(self, *args):
    return self.conn.control(*args)

bulk_in

bulk_in(addr, length)

Source code in usbip/host.py

def bulk_in(self, addr, length):
    return self.conn.transfer_in(addr & 0x0F, length)

bulk_out

bulk_out(addr, data)

Source code in usbip/host.py

def bulk_out(self, addr, data):
    return self.conn.transfer_out(addr & 0x0F, data)

clear_halt

clear_halt(addr)

Clear a halted (STALLed) endpoint: CLEAR_FEATURE(ENDPOINT_HALT).

The standard recovery after a transfer raises :class:Stall - a device halts a pipe to abandon a transfer it cannot complete, and the pipe stays halted until this clears it.

Source code in usbip/host.py

def clear_halt(self, addr):
    """Clear a halted (STALLed) endpoint: CLEAR_FEATURE(ENDPOINT_HALT).

    The standard recovery after a transfer raises :class:`Stall` - a device
    halts a pipe to abandon a transfer it cannot complete, and the pipe stays
    halted until this clears it."""
    return self.conn.control(0x02, 0x01, 0x0000, addr, b"")

interrupt_in

interrupt_in(addr, length)

Source code in usbip/host.py

def interrupt_in(self, addr, length):
    # interval=1: a polled endpoint, which is also how a capture tells an
    # interrupt transfer from a bulk one (same as the C host API).
    return self.conn.transfer_in(addr & 0x0F, length, interval=1)

interrupt_out

interrupt_out(addr, data)

Source code in usbip/host.py

def interrupt_out(self, addr, data):
    return self.conn.transfer_out(addr & 0x0F, data, interval=1)

iso_in

iso_in(addr, packet_len, num_packets, interval=0)

Isochronous IN. Returns a list of bytes objects, one per packet (each trimmed to its actual_length).

Source code in usbip/host.py

def iso_in(self, addr, packet_len, num_packets, interval=0):
    """Isochronous IN. Returns a list of bytes objects, one per packet (each
    trimmed to its actual_length)."""
    buf, pkts = self.conn.iso_transfer(
        (addr & 0x0F) | 0x80, [packet_len] * num_packets, interval=interval
    )
    return [buf[off : off + actual] for off, length, actual, status in pkts]

iso_out

iso_out(addr, packets, interval=0)

Isochronous OUT. packets is a list of bytes objects, one per packet. Returns the per-packet [offset, length, actual_length, status] list.

Source code in usbip/host.py

def iso_out(self, addr, packets, interval=0):
    """Isochronous OUT. `packets` is a list of bytes objects, one per packet.
    Returns the per-packet [offset, length, actual_length, status] list."""
    data = b"".join(packets)
    _buf, pkts = self.conn.iso_transfer(
        addr & 0x0F, [len(pkt) for pkt in packets], data=data, interval=interval
    )
    return pkts

close

close()

Source code in usbip/host.py

def close(self):
    self.conn.close()

Driver

Driver(handle)

Source code in usbip/host.py

def __init__(self, handle: Handle):
    self.handle = handle

matches class-attribute instance-attribute

matches = {}

handle instance-attribute

handle = handle

open classmethod

open(vid=None, pid=None, *, busid='1-1', transport=None)

Source code in usbip/host.py

@classmethod
def open(cls, vid=None, pid=None, *, busid="1-1", transport=None):
    return cls(open(vid, pid, busid=busid, transport=transport))

close

close()

Source code in usbip/host.py

def close(self):
    self.handle.close()

Connection

Connection(sock)

One imported device: owns the socket and submits URBs.

Source code in usbip/host.py

def __init__(self, sock):
    self.sock = sock
    self.info = None
    self.devid = 0x00010002
    self._seq = 0

sock instance-attribute

sock = sock

info instance-attribute

info = None

devid instance-attribute

devid = 65538

import_device

import_device(busid='1-1')

Source code in usbip/host.py

def import_device(self, busid="1-1") -> dict:
    self.info = protocol.client_import(self.sock, busid.encode())
    self.devid = (self.info["busnum"] << 16) | self.info["devnum"]
    return self.info

control

control(bmRequestType, bRequest, wValue, wIndex, data_or_len)

Source code in usbip/host.py

def control(self, bmRequestType, bRequest, wValue, wIndex, data_or_len):
    if bmRequestType & 0x80:  # IN
        length = int(data_or_len)
        setup = Setup(bmRequestType, bRequest, wValue, wIndex, length).pack()
        urb = self._submit(Urb(direction=IN, ep=0, setup=setup, length=length))
        if urb.status != 0:
            raise Stall(f"control status {urb.status}")
        return urb.buffer
    data = data_or_len or b""
    setup = Setup(bmRequestType, bRequest, wValue, wIndex, len(data)).pack()
    request = Urb(direction=OUT, ep=0, setup=setup, length=len(data), buffer=data)
    urb = self._submit(request)
    if urb.status != 0:
        raise Stall(f"control status {urb.status}")
    return urb.actual

transfer_in

transfer_in(ep_number, length, interval=0)

Source code in usbip/host.py

def transfer_in(self, ep_number, length, interval=0):
    urb = self._submit(Urb(direction=IN, ep=ep_number, length=length, interval=interval))
    if urb.status != 0:
        raise Stall(f"transfer status {urb.status}")
    return urb.buffer

transfer_out

transfer_out(ep_number, data, interval=0)

Source code in usbip/host.py

def transfer_out(self, ep_number, data, interval=0):
    request = Urb(direction=OUT, ep=ep_number, length=len(data), buffer=data, interval=interval)
    urb = self._submit(request)
    if urb.status != 0:
        raise Stall(f"transfer status {urb.status}")
    return urb.actual

iso_transfer

iso_transfer(addr, packet_lengths, data=None, interval=0)

One isochronous transfer (libusb-shaped, synchronous). packet_lengths is the requested length per packet; for OUT data holds the packets laid out at their slot offsets. Returns (buffer, packets) where packets is a list of [offset, length, actual_length, status].

Source code in usbip/host.py

def iso_transfer(self, addr, packet_lengths, data=None, interval=0):
    """One isochronous transfer (libusb-shaped, synchronous). `packet_lengths`
    is the requested length per packet; for OUT `data` holds the packets laid
    out at their slot offsets. Returns (buffer, packets) where packets is a
    list of [offset, length, actual_length, status]."""
    iso, off = [], 0
    for length in packet_lengths:
        iso.append([off, length, 0, 0])
        off += length
    direction = IN if (addr & 0x80) else OUT
    urb = Urb(
        direction=direction, ep=addr & 0x0F, interval=interval, length=off, buffer=data or b""
    )
    urb.iso_packets = iso
    urb = self._submit(urb)
    return urb.buffer, urb.iso_packets

close

close()

Source code in usbip/host.py

def close(self):
    try:
        self.sock.close()
    except OSError:
        pass

open

open(vid=None, pid=None, *, busid=None, transport=None, set_config=1)

Import a device and return a Handle. With vid/pid given, verifies the match; without, accepts whatever busid resolves to (real devices).

When a server exports several devices, vid/pid alone is enough: the busid is looked up with :func:list_devices. Passing busid skips that and imports it directly.

Source code in usbip/host.py

def open(vid=None, pid=None, *, busid=None, transport=None, set_config=1):
    """Import a device and return a Handle. With vid/pid given, verifies the
    match; without, accepts whatever `busid` resolves to (real devices).

    When a server exports several devices, vid/pid alone is enough: the busid is
    looked up with :func:`list_devices`. Passing `busid` skips that and imports
    it directly.
    """
    from .transport import default_transport

    resolved = transport or default_transport()
    if busid is None:
        busid = _resolve_busid(resolved, vid, pid)
    conn = Connection(resolved.connect())
    info = conn.import_device(busid)
    if vid is not None and (info["idVendor"] != vid or info["idProduct"] != pid):
        conn.close()
        raise NotFound(
            f"{vid:04x}:{pid:04x} not found (got {info['idVendor']:04x}:{info['idProduct']:04x})"
        )
    dd = DeviceDescriptor.parse(conn.control(0x80, 0x06, 0x0100, 0, 18))
    if set_config is not None:
        try:
            conn.control(0x00, 0x09, set_config, 0, b"")  # SET_CONFIGURATION
        except Stall:
            pass
    return Handle(conn, dd)

attach

attach(remote_host, busid, *, port=3240, vid=None, pid=None)

Convenience for real remote devices: attach(host, '1-1.4').

Source code in usbip/host.py

def attach(remote_host, busid, *, port=3240, vid=None, pid=None):
    """Convenience for real remote devices: attach(host, '1-1.4')."""
    from .transport import USBIP

    transport = USBIP(remote_host, port)
    return open(vid, pid, busid=busid, transport=transport)