mirror of
https://github.com/RetroPortingToolKit/psxrecomp
synced 2026-10-25 18:05:12 -04:00
Matthew Stanley
432a44f64a
Supersedes the global HYBRID controller-mode shape from PR #233 with a trusted plugin callback that can choose analog or digital presentation per input sample. Keeps analog mode hardware-faithful by no longer folding physical D-pad input onto the left stick, while leaving debug injection able to steer stick-only paths. Includes the pad/cursor probe tooling from #233 for diagnosing input doubling.
260 lines
9.1 KiB
Python
Executable File
260 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
cursor_writers.py — Catch every write to the menu cursor and name the writer.
|
|
|
|
menu_scan.py located a variable that moved two steps from one D-pad tap.
|
|
This registers a write-trace range over it and records each write during a
|
|
capture: the value before and after, the PC that made it, and the frame.
|
|
|
|
That distinguishes the two ways one tap becomes two moves:
|
|
|
|
two writes, two PCs two independent code paths each acted on the input.
|
|
If one reads the button bits and the other the stick
|
|
bytes, the D-pad/stick fold is implicated.
|
|
two writes, one PC one path ran twice — the game polled or ticked twice.
|
|
one write of -2 one path computed a step of two; the doubling is in
|
|
the step calculation, and input is not to blame.
|
|
|
|
Usage:
|
|
python3 cursor_writers.py --seconds 12 --focus 0x8004A5F5
|
|
|
|
Registers a trace range, captures, then always removes the range it added.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, __file__.rsplit("/", 1)[0])
|
|
|
|
PHYS_MASK = 0x1FFFFFFF
|
|
|
|
|
|
def phys(addr):
|
|
"""The trace ring records physical addresses; KSEG0/KSEG1 map down."""
|
|
return addr & PHYS_MASK
|
|
|
|
|
|
def writes_in_range(entries, lo, hi):
|
|
"""Trace entries whose address falls in [lo, hi), oldest first."""
|
|
out = []
|
|
for e in entries:
|
|
a = int(e["addr"], 16)
|
|
if phys(lo) <= a < phys(hi):
|
|
out.append(e)
|
|
out.sort(key=lambda e: e["seq"])
|
|
return out
|
|
|
|
|
|
def covers(e, addr):
|
|
"""Does this write touch `addr`?
|
|
|
|
The ring records a write at its base address and width, so a halfword
|
|
store to 0x...F4 changes the byte at 0x...F5 without ever appearing at
|
|
that address. Matching on address equality misses those entirely.
|
|
"""
|
|
a = int(e["addr"], 16)
|
|
return a <= phys(addr) < a + e.get("w", 4)
|
|
|
|
|
|
def byte_value(e, key, addr):
|
|
"""The byte this write put at `addr` ("old" or "new")."""
|
|
shift = (phys(addr) - int(e["addr"], 16)) * 8
|
|
return (int(e[key], 16) >> shift) & 0xFF
|
|
|
|
|
|
def byte_delta(e, addr):
|
|
"""Signed change this write made to the single byte at `addr`."""
|
|
d = byte_value(e, "new", addr) - byte_value(e, "old", addr)
|
|
if d > 128:
|
|
d -= 256
|
|
elif d < -128:
|
|
d += 256
|
|
return d
|
|
|
|
|
|
def touching_writes(writes, addr):
|
|
"""Writes that actually changed the byte at `addr`.
|
|
|
|
A wide store may cover the byte while leaving it alone; that is not a
|
|
cursor move and must not be counted as one.
|
|
"""
|
|
return [e for e in writes if covers(e, addr) and byte_delta(e, addr) != 0]
|
|
|
|
|
|
def delta(e):
|
|
"""Signed change this write made, at the traced width."""
|
|
w = e.get("w", 4)
|
|
bits = w * 8
|
|
old = int(e["old"], 16) & ((1 << bits) - 1)
|
|
new = int(e["new"], 16) & ((1 << bits) - 1)
|
|
d = new - old
|
|
# Interpret a large jump as a small signed step (wrap at the width).
|
|
if d > (1 << (bits - 1)):
|
|
d -= (1 << bits)
|
|
elif d < -(1 << (bits - 1)):
|
|
d += (1 << bits)
|
|
return d
|
|
|
|
|
|
def summarize(writes, focus):
|
|
"""Per-address rollup, with the focus address reported in full."""
|
|
by_addr = {}
|
|
for e in writes:
|
|
a = int(e["addr"], 16)
|
|
s = by_addr.setdefault(a, {"addr": a, "writes": 0, "pcs": set(),
|
|
"funcs": set(), "deltas": []})
|
|
s["writes"] += 1
|
|
s["pcs"].add(e["pc"])
|
|
s["funcs"].add(e.get("func", "?"))
|
|
s["deltas"].append(delta(e))
|
|
for s in by_addr.values():
|
|
s["pcs"] = sorted(s["pcs"])
|
|
s["funcs"] = sorted(s["funcs"])
|
|
focus_writes = touching_writes(writes, focus)
|
|
return by_addr, focus_writes
|
|
|
|
|
|
def group_events(writes, focus, gap_frames=8):
|
|
"""Split writes into one group per press.
|
|
|
|
A press produces its writes within a frame or two; separate presses are
|
|
seconds apart. Grouping keeps a capture containing several presses from
|
|
being read as one confused event.
|
|
"""
|
|
events, cur = [], []
|
|
for e in sorted(writes, key=lambda x: x["seq"]):
|
|
if cur and (e.get("frame", 0) - cur[-1].get("frame", 0)) > gap_frames:
|
|
events.append(cur)
|
|
cur = []
|
|
cur.append(e)
|
|
if cur:
|
|
events.append(cur)
|
|
return events
|
|
|
|
|
|
def describe_event(ev, focus):
|
|
total = sum(byte_delta(e, focus) for e in ev)
|
|
pcs = sorted({e["pc"] for e in ev})
|
|
if len(ev) == 1:
|
|
kind = ("single_write_double_step" if abs(total) >= 2
|
|
else "single_write_single_step")
|
|
elif len(pcs) > 1:
|
|
kind = "two_paths"
|
|
else:
|
|
kind = "one_path_twice"
|
|
return {"kind": kind, "writes": len(ev), "total": total, "pcs": pcs,
|
|
"frame0": ev[0].get("frame"), "frame1": ev[-1].get("frame"),
|
|
"funcs": sorted({e.get("func", "?") for e in ev})}
|
|
|
|
|
|
def verdict(focus_writes, focus=None):
|
|
"""Name the mechanism, per press and overall."""
|
|
if not focus_writes:
|
|
return ("no_writes",
|
|
"the focus byte was never changed during the capture")
|
|
events = [describe_event(ev, focus)
|
|
for ev in group_events(focus_writes, focus)]
|
|
doubled = [e for e in events if abs(e["total"]) >= 2]
|
|
if not doubled:
|
|
return ("no_double_step",
|
|
"%d press(es) seen, none moved the cursor more than one step"
|
|
% len(events))
|
|
kinds = {e["kind"] for e in doubled}
|
|
if kinds == {"single_write_double_step"}:
|
|
return ("single_write_double_step",
|
|
"%d of %d press(es) moved the cursor %s in ONE write — the "
|
|
"step size was computed wrong; the input path is not at fault"
|
|
% (len(doubled), len(events),
|
|
"/".join(str(e["total"]) for e in doubled)))
|
|
if "two_paths" in kinds:
|
|
pcs = sorted({pc for e in doubled if e["kind"] == "two_paths"
|
|
for pc in e["pcs"]})
|
|
return ("two_paths",
|
|
"%d of %d press(es) were acted on by MULTIPLE code paths "
|
|
"(PCs: %s) — two handlers consumed one press"
|
|
% (len(doubled), len(events), ", ".join(pcs)))
|
|
return ("one_path_twice",
|
|
"%d of %d press(es) ran the same handler twice"
|
|
% (len(doubled), len(events)))
|
|
|
|
|
|
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("--lo", default="0x8004A500")
|
|
ap.add_argument("--hi", default="0x8004A600")
|
|
ap.add_argument("--focus", default="0x8004A5F5")
|
|
ap.add_argument("--seconds", type=float, default=12.0)
|
|
ap.add_argument("--count", type=int, default=4096)
|
|
ap.add_argument("--json")
|
|
args = ap.parse_args()
|
|
|
|
lo, hi, focus = int(args.lo, 16), int(args.hi, 16), int(args.focus, 16)
|
|
|
|
import pad_trace
|
|
link = pad_trace.Link(args.host, args.port)
|
|
slot = None
|
|
try:
|
|
r = link.cmd_with({"cmd": "wtrace_add",
|
|
"lo": "0x%08X" % lo, "hi": "0x%08X" % hi})
|
|
if not r or not r.get("ok"):
|
|
raise RuntimeError("wtrace_add failed: %r" % (r,))
|
|
slot = r["slot"]
|
|
print("tracing 0x%08X..0x%08X (slot %d) for %.0fs — tap the D-pad "
|
|
"ONCE, and note whether it doubles."
|
|
% (lo, hi, slot, args.seconds))
|
|
time.sleep(args.seconds)
|
|
d = link.cmd_with({"cmd": "wtrace_dump", "addr_lo": "0x%08X" % lo,
|
|
"addr_hi": "0x%08X" % hi, "count": args.count,
|
|
"newest": 1})
|
|
entries = (d or {}).get("entries", [])
|
|
finally:
|
|
if slot is not None:
|
|
link.cmd_with({"cmd": "wtrace_del", "slot": slot})
|
|
link.close()
|
|
|
|
writes = writes_in_range(entries, lo, hi)
|
|
print("\n%d write(s) in range during the capture" % len(writes))
|
|
by_addr, focus_writes = summarize(writes, focus)
|
|
|
|
for a in sorted(by_addr):
|
|
s = by_addr[a]
|
|
print(" 0x%08X writes=%-3d deltas=%-18s pc=%s"
|
|
% (0x80000000 | a, s["writes"],
|
|
",".join(str(x) for x in s["deltas"][:6]),
|
|
",".join(s["pcs"][:3])))
|
|
|
|
print("\nfocus 0x%08X:" % focus)
|
|
for e in focus_writes:
|
|
print(" frame=%-8s @0x%08X w=%s %s -> %s byte %d -> %d (%+d)"
|
|
" pc=%s func=%s ra=%s"
|
|
% (e.get("frame"), int(e["addr"], 16), e.get("w"),
|
|
e["old"], e["new"], byte_value(e, "old", focus),
|
|
byte_value(e, "new", focus), byte_delta(e, focus),
|
|
e["pc"], e.get("func"), e.get("ra")))
|
|
|
|
print("\nper-press breakdown:")
|
|
for ev in group_events(focus_writes, focus):
|
|
d = describe_event(ev, focus)
|
|
print(" frames %s..%s writes=%d step=%+d %s pc=%s"
|
|
% (d["frame0"], d["frame1"], d["writes"], d["total"],
|
|
d["kind"], ",".join(d["pcs"])))
|
|
|
|
kind, text = verdict(focus_writes, focus)
|
|
print("\n[%s] %s" % (kind, text))
|
|
|
|
if args.json:
|
|
with open(args.json, "w") as fh:
|
|
json.dump({"writes": writes, "verdict": kind, "detail": text},
|
|
fh, indent=2)
|
|
print("wrote %s" % args.json)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|