Parse USBPcap captures of the Ulanzi Stream Controller D200X (vid 2207 pid 0019); decodes the 7c7c framing layer and flags host->device frames sent before the first input event as candidates for the firmware's input-enable opcode. Companion to https://github.com/doitian/ulanzi-studio-niri

parse_usbpcap.py

#!/usr/bin/env python3
"""Extract host->device packets to the Ulanzi D200X from a USBPcap capture.

Pure stdlib; runs on either Linux or Windows. Reads a PCAP/PCAPNG file
captured by USBPcap (https://desowin.org/usbpcap/) and prints every
URB submitted FROM the host TO an endpoint of vid:pid 2207:0019, decoding
the framing layer used by the device:

    magic 7c 7c | command_protocol (BE u16) | length (LE u32) | payload

Usage:
    python scripts/parse_usbpcap.py ulanzi.pcap

Capture procedure on Windows (no install on this repo's machine):
    1. Install USBPcap from https://desowin.org/usbpcap/  (admin, one-time;
       reboot if prompted; install Wireshark too if you want a GUI).
    2. Plug the D200X in. Open Device Manager -> View -> Devices by container
       and confirm which USB root hub it sits under, OR just try each
       USBPcap1, USBPcap2, ... interface until one produces data.
    3. From a regular cmd window:
           "C:\\Program Files\\USBPcap\\USBPcapCMD.exe" -d \\\\.\\USBPcap1 -o ulanzi.pcap
       (Choose the right number for your hub.)
    4. Launch the official Ulanzi Stream Controller app; let it connect.
    5. Press both 4th-row hardware buttons; click and twist each encoder.
    6. Quit the official app; Ctrl+C the USBPcapCMD window.
    7. Send the .pcap file back, OR run this script on it and paste the
       output of the FIRST 30 host->device frames (those before the first
       device->host BUTTON event are what we need).
"""

from __future__ import annotations

import struct
import sys
from pathlib import Path

VID = 0x2207
PID = 0x0019

# pcap link types
DLT_USBPCAP = 249  # USBPcap pseudo-header

# USBPcap pseudo-header (little endian):
#   u16 headerLen
#   u64 IRP id
#   u32 status
#   u16 function
#   u8  info       (bit 0 = 1 means PDO->FDO, i.e. device->host)
#   u16 bus
#   u16 device
#   u8  endpoint   (bit 7 = 1 means IN endpoint)
#   u8  transfer   (0=isochronous, 1=interrupt, 2=control, 3=bulk)
#   u32 dataLen
USBPCAP_HDR_FMT = "<HQIHBHHBBI"
USBPCAP_HDR_SIZE = struct.calcsize(USBPCAP_HDR_FMT)


def iter_pcap(path: Path):
    """Yield (timestamp_us, link_data) tuples from a classic PCAP file."""
    with path.open("rb") as fh:
        magic = fh.read(4)
        if magic == b"\xd4\xc3\xb2\xa1":
            endian = "<"
        elif magic == b"\xa1\xb2\xc3\xd4":
            endian = ">"
        elif magic in (b"\x0a\x0d\x0d\x0a",):
            raise SystemExit(
                "PCAPNG not supported by this tiny parser; please re-capture as PCAP, "
                "or open in Wireshark and File > Export Specified Packets > pcap."
            )
        else:
            raise SystemExit(f"not a pcap file: magic={magic!r}")
        # global header: version_major u16, version_minor u16, thiszone i32,
        # sigfigs u32, snaplen u32, network u32
        hdr = fh.read(20)
        _vmaj, _vmin, _tz, _sig, _snap, network = struct.unpack(endian + "HHiIII", hdr)
        if network != DLT_USBPCAP:
            raise SystemExit(
                f"link type {network} != USBPcap ({DLT_USBPCAP}); is this really a USBPcap capture?"
            )
        rec_fmt = endian + "IIII"
        rec_size = struct.calcsize(rec_fmt)
        while True:
            rec_hdr = fh.read(rec_size)
            if len(rec_hdr) < rec_size:
                return
            ts_sec, ts_usec, incl_len, _orig_len = struct.unpack(rec_fmt, rec_hdr)
            data = fh.read(incl_len)
            if len(data) < incl_len:
                return
            yield (ts_sec * 1_000_000 + ts_usec, data)


def decode_command(payload: bytes) -> str | None:
    if len(payload) < 8 or payload[:2] != b"\x7c\x7c":
        return None
    cmd = (payload[2] << 8) | payload[3]
    length = struct.unpack("<I", payload[4:8])[0]
    body = payload[8 : 8 + min(length, 32)]
    name = {
        0x0001: "SET_BUTTONS",
        0x0003: "GET_DEVICE_INFO",
        0x0006: "SET_SMALL_WINDOW_DATA",
        0x000A: "SET_BRIGHTNESS",
        0x000B: "SET_LABEL_STYLE",
        0x000D: "PARTIALLY_UPDATE_BUTTONS",
        0x0101: "BUTTON_IN",
        0x0303: "DEVICE_INFO_IN",
    }.get(cmd, "UNKNOWN")
    return f"cmd=0x{cmd:04x} ({name}) len={length} body[:32]={body.hex()}"


def main() -> int:
    if len(sys.argv) != 2:
        print(__doc__, file=sys.stderr)
        return 2
    pcap = Path(sys.argv[1])
    if not pcap.exists():
        print(f"no such file: {pcap}", file=sys.stderr)
        return 1

    target_devices: set[int] = set()
    first_input_ts: int | None = None
    rows: list[tuple[int, str, int, str]] = []

    for ts, data in iter_pcap(pcap):
        if len(data) < USBPCAP_HDR_SIZE:
            continue
        (
            hdr_len,
            _irp,
            _status,
            _func,
            info,
            _bus,
            device,
            endpoint,
            _transfer,
            data_len,
        ) = struct.unpack(USBPCAP_HDR_FMT, data[:USBPCAP_HDR_SIZE])
        payload = data[hdr_len : hdr_len + data_len]
        # USBPcap places the USB control-transfer setup bytes inline for control;
        # for interrupt/bulk transfers it's just the raw HID report payload.

        # Filter: only devices we care about. We don't get vid/pid in the
        # per-packet header; we have to learn (device, vid, pid) from a
        # GET_DESCRIPTOR control transfer or by sniffing both directions
        # and matching framing magic. Simpler heuristic: any device that
        # ever emits a host->device frame starting with 7c7c is ours.
        is_in = bool(info & 0x01)
        starts_with_magic = payload[:2] == b"\x7c\x7c"
        if starts_with_magic:
            target_devices.add(device)

        if device not in target_devices:
            continue

        direction = "D->H" if is_in else "H->D"
        decoded = decode_command(payload) or f"raw[:32]={payload[:32].hex()}"

        if is_in and "BUTTON_IN" in decoded and first_input_ts is None:
            first_input_ts = ts

        rows.append((ts, direction, endpoint, decoded))

    if not target_devices:
        print("no D200X-style frames (magic 7c7c) found in capture", file=sys.stderr)
        return 1

    base = rows[0][0] if rows else 0
    print(f"# learned target USBPcap device id(s): {sorted(target_devices)}")
    if first_input_ts is not None:
        print(f"# first device->host BUTTON_IN at +{(first_input_ts - base) / 1000:.1f} ms")
        print("# host->device frames BEFORE first input (these are the unlock candidates):")

    for ts, direction, endpoint, decoded in rows:
        rel_ms = (ts - base) / 1000
        marker = ""
        if first_input_ts is not None and ts < first_input_ts and direction == "H->D":
            marker = " <-- candidate"
        print(f"+{rel_ms:8.1f}ms ep=0x{endpoint:02x} {direction} {decoded}{marker}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())