tools/usb/p9_fwd.py

Source file repositories/reference/linux-study-clean/tools/usb/p9_fwd.py

File Facts

System
Linux kernel
Corpus path
tools/usb/p9_fwd.py
Extension
.py
Size
8594 bytes
Lines
244
Domain
Support Tooling And Documentation
Bucket
tools
Inferred role
Support Tooling And Documentation: tools
Status
atlas-only

Why This File Exists

Repository support layer: documentation, build tooling, samples, user-space helper tools, generated initramfs support, licenses, and validation utilities.

Dependency Surface

Detected Declarations

Annotated Snippet

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0

import argparse
import errno
import logging
import socket
import struct
import time

import usb.core
import usb.util


def path_from_usb_dev(dev):
    """Takes a pyUSB device as argument and returns a string.
    The string is a Path representation of the position of the USB device on the USB bus tree.

    This path is used to find a USB device on the bus or all devices connected to a HUB.
    The path is made up of the number of the USB controller followed be the ports of the HUB tree."""
    if dev.port_numbers:
        dev_path = ".".join(str(i) for i in dev.port_numbers)
        return f"{dev.bus}-{dev_path}"
    return ""


HEXDUMP_FILTER = "".join(chr(x).isprintable() and chr(x) or "." for x in range(128)) + "." * 128


class Forwarder:
    @staticmethod
    def _log_hexdump(data):
        if not logging.root.isEnabledFor(logging.TRACE):
            return
        L = 16
        for c in range(0, len(data), L):
            chars = data[c : c + L]
            dump = " ".join(f"{x:02x}" for x in chars)
            printable = "".join(HEXDUMP_FILTER[x] for x in chars)
            line = f"{c:08x}  {dump:{L*3}s} |{printable:{L}s}|"
            logging.root.log(logging.TRACE, "%s", line)

    def __init__(self, server, vid, pid, path):
        self.stats = {
            "c2s packets": 0,
            "c2s bytes": 0,
            "s2c packets": 0,
            "s2c bytes": 0,
        }
        self.stats_logged = time.monotonic()

        def find_filter(dev):
            dev_path = path_from_usb_dev(dev)
            if path is not None:
                return dev_path == path
            return True

        dev = usb.core.find(idVendor=vid, idProduct=pid, custom_match=find_filter)
        if dev is None:
            raise ValueError("Device not found")

        logging.info(f"found device: {dev.bus}/{dev.address} located at {path_from_usb_dev(dev)}")

        # dev.set_configuration() is not necessary since g_multi has only one
        usb9pfs = None
        # g_multi adds 9pfs as last interface
        cfg = dev.get_active_configuration()
        for intf in cfg:
            # we have to detach the usb-storage driver from multi gadget since
            # stall option could be set, which will lead to spontaneous port

Annotation

Implementation Notes