psxrecomp/tools/pad_delivery.py

#!/usr/bin/env python3
"""
pad_delivery.py — Decode the pad polls the game actually received over SIO.

pad_trace.py measures what the runtime *presented*; this measures what the
game *read*.  A single physical tap must appear as one contiguous run of
polls with the button held.  If the delivered stream drops back to "released"
in the middle of a press and then reports it held again, the game's edge
detector fires twice and the menu moves twice — from one tap.

Reads the always-on SIO byte ring (sio_trace) and reconstructs each pad
transaction:

    tx 0x01 -> rx 0xFF     address the controller
    tx 0x42 -> rx id       0x41 digital, 0x73 analog, 0xF3 config
    tx 0x00 -> rx 0x5A
    tx 0x00 -> rx btn_lo   active low
    tx 0x00 -> rx btn_hi
    tx 0x00 -> rx rx,ry,lx,ly   (analog only)

Usage:
    python3 pad_delivery.py --seconds 8 --json /tmp/delivery.json

Tap the D-pad once per press you want classified.
"""

import argparse
import json
import sys
import time

sys.path.insert(0, __file__.rsplit("/", 1)[0])

PAD_ADDR = 0x01
MC_ADDR = 0x81
CMD_POLL = 0x42

BUTTONS = {
    "select": 0x0001, "l3": 0x0002, "r3": 0x0004, "start": 0x0008,
    "up": 0x0010, "right": 0x0020, "down": 0x0040, "left": 0x0080,
    "l2": 0x0100, "r2": 0x0200, "l1": 0x0400, "r1": 0x0800,
    "triangle": 0x1000, "circle": 0x2000, "cross": 0x4000, "square": 0x8000,
}
DPAD = ("up", "right", "down", "left")


def _b(v):
    """sio_trace reports bytes as '0x41' strings."""
    return int(v, 16) if isinstance(v, str) else int(v)


def decode_polls(entries):
    """Reconstruct pad polls from an ordered SIO byte stream.

    `entries` must be ordered by seq and may contain memory-card traffic,
    which is skipped.  Returns one dict per pad transaction that got far
    enough to carry a button word.
    """
    polls = []
    txn = None
    for e in entries:
        tx, rx = _b(e["tx"]), _b(e["rx"])
        if tx in (PAD_ADDR, MC_ADDR):
            # A new address byte always ends the previous transaction.
            if txn is not None:
                polls.append(txn) if txn["complete"] else None
            txn = {"seq": e["seq"], "addr": tx, "bytes": [], "frame": e.get("frame", -1),
                   "cmd": None, "id": None, "buttons": None, "sticks": None,
                   "complete": False} if tx == PAD_ADDR else None
            continue
        if txn is None:
            continue
        txn["bytes"].append(rx)
        n = len(txn["bytes"])
        if n == 1:
            txn["cmd"] = tx      # the command byte we clocked out
            txn["id"] = rx
        elif n == 4:
            txn["buttons"] = txn["bytes"][2] | (txn["bytes"][3] << 8)
            txn["complete"] = txn["cmd"] == CMD_POLL
        elif n == 8:
            txn["sticks"] = [txn["bytes"][6], txn["bytes"][7],
                             txn["bytes"][4], txn["bytes"][5]]  # lx,ly,rx,ry
    if txn is not None and txn["complete"]:
        polls.append(txn)
    return polls


def pressed(word, name):
    return (word & BUTTONS[name]) == 0


def poll_runs(polls, name):
    """Contiguous spans of polls where `name` is held. Returns (i0, i1) pairs."""
    runs, start = [], None
    for i, p in enumerate(polls):
        if p["buttons"] is not None and pressed(p["buttons"], name):
            if start is None:
                start = i
        elif start is not None:
            runs.append((start, i - 1))
            start = None
    if start is not None:
        runs.append((start, len(polls) - 1))
    return runs


def analyze(polls, bounce_polls=12):
    """Classify delivered pad polls.

    bounce_polls: two runs separated by fewer than this many polls came from
    one physical tap (a tap is ~4 polls; a deliberate second tap is ~30+).
    """
    runs, findings = [], []
    for name in BUTTONS:
        spans = poll_runs(polls, name)
        for i0, i1 in spans:
            runs.append({
                "button": name, "i0": i0, "i1": i1,
                "polls": i1 - i0 + 1,
                "seq0": polls[i0]["seq"], "seq1": polls[i1]["seq"],
                "frame0": polls[i0].get("frame", -1),
                "frame1": polls[i1].get("frame", -1),
            })
        for (a0, a1), (b0, _b1) in zip(spans, spans[1:]):
            gap = b0 - a1 - 1
            if gap <= bounce_polls:
                findings.append({
                    "kind": "delivered_bounce", "button": name,
                    "gap_polls": gap,
                    "detail": "the game read %s as released for %d poll(s) in "
                              "the middle of one press, then held again — two "
                              "rising edges, two menu moves" % (name, gap),
                })
    runs.sort(key=lambda r: r["i0"])
    return runs, findings


def polls_per_frame(polls):
    """How many times the game read the pad in each frame it polled."""
    counts = {}
    for p in polls:
        f = p.get("frame", -1)
        if f >= 0:
            counts[f] = counts.get(f, 0) + 1
    return counts


def resolve_frame(prev_frame, frame):
    """Exact frame for entries seen between two reads, or -1 if unknowable.

    Entries newly visible in this batch were generated between the previous
    read and this one.  If the frame counter did not advance across that
    interval, every one of them belongs to that frame.  If it did advance,
    they straddle a boundary and this ring carries no per-entry stamp to
    separate them — so say -1 rather than invent one.
    """
    if prev_frame is None or frame < 0 or prev_frame != frame:
        return -1
    return frame


def capture(host, port, seconds, count):
    import pad_trace
    link = pad_trace.Link(host, port)
    seen_max, entries, gaps = -1, [], 0
    prev_frame = None
    t_end = time.time() + seconds
    while time.time() < t_end:
        r = link.cmd_with({"cmd": "sio_trace", "count": count})
        fr = link.cmd_with({"cmd": "frame"})
        frame = int(fr.get("frame", -1)) if fr else -1
        if not r or "entries" not in r:
            raise RuntimeError("sio_trace failed: %r" % (r,))
        batch = r["entries"]
        if not batch:
            continue
        first = batch[0]["seq"]
        if seen_max >= 0 and first > seen_max + 1:
            gaps += first - seen_max - 1
        stamp = resolve_frame(prev_frame, frame)
        prev_frame = frame
        for e in batch:
            if e["seq"] > seen_max:
                e["frame"] = stamp
                entries.append(e)
                seen_max = e["seq"]
    link.close()
    return entries, gaps, link


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=4370)
    ap.add_argument("--seconds", type=float, default=8.0)
    ap.add_argument("--count", type=int, default=192,
                    help="sio_trace entries per poll; raise if gaps appear")
    ap.add_argument("--json")
    args = ap.parse_args()

    print("Capturing %.0fs — tap the D-pad." % args.seconds)
    entries, gaps, link = capture(args.host, args.port, args.seconds, args.count)
    polls = decode_polls(entries)
    print("%d SIO bytes, %d pad polls%s"
          % (len(entries), len(polls),
             "" if not link.persistent else ""))
    if gaps:
        print("WARNING: %d bytes missed between reads — raise --count." % gaps)

    ppf = polls_per_frame(polls)
    unknown = sum(1 for p in polls if p.get("frame", -1) < 0)
    if ppf:
        vals = sorted(ppf.values())
        print("polls per frame: min=%d median=%d max=%d  (%d of %d polls "
              "straddled a batch boundary and are unstamped)"
              % (vals[0], vals[len(vals) // 2], vals[-1], unknown, len(polls)))
    # The pad is polled on a fixed cadence; a changing seq delta is the honest
    # way to see irregular polling, since it needs no frame stamp at all.
    deltas = sorted({polls[i + 1]["seq"] - polls[i]["seq"]
                     for i in range(len(polls) - 1)})
    print("poll seq deltas: %s" % (deltas if len(deltas) < 8
                                   else "%d distinct" % len(deltas)))

    runs, findings = analyze(polls)
    for r in runs:
        print("  %-8s %3d polls  seq %d..%d  frames %s..%s"
              % (r["button"], r["polls"], r["seq0"], r["seq1"],
                 r["frame0"], r["frame1"]))
    print()
    if findings:
        for f in findings:
            print("  [%s] %s: %s" % (f["kind"], f["button"], f["detail"]))
    else:
        print("Every press was delivered as one contiguous run.")
        print("The doubling is not in pad delivery — look at the game's own "
              "edge detection or its second input source.")

    if args.json:
        with open(args.json, "w") as fh:
            json.dump({"polls": polls, "runs": runs, "findings": findings,
                       "gaps": gaps}, fh, indent=2)
        print("\nwrote %s" % args.json)
    return 0


if __name__ == "__main__":
    sys.exit(main())