tools: add DuckStation oracle and GPU frame tools

This commit is contained in:
Matthew Stanley
2026-08-24 22:52:33 -07:00
parent 1ea6a45066
commit d849485ab2
14 changed files with 4191 additions and 2 deletions
+302
View File
@@ -0,0 +1,302 @@
# GP0 frame tools — who drew this pixel?
A set of tools that answer one question about a rendering bug: **which guest
function issued this primitive, and with which blend mode?**
```
tools/psx_gpu_frame.py transport + GP0 decode + attribution (the library)
tools/gpu_frame_capture.py capture a frame from a running game
tools/gpu_frame_layers.py render one image per guest function
tools/gpu_frame_diff.py diff a good frame against a bad one
tools/gpu_parity.py frame-locked image parity vs DuckStation
tools/tests/test_gpu_frame.py
```
RetComM Studio's **Frames** tab is a viewer and launcher over exactly these
artifacts. It never decodes a GP0 packet itself, so a headless capture and the
GUI can never disagree about what a frame contained.
---
## Why this exists
`gpu_frame_dump` on the debug server has always stamped every GP0 packet with
the guest code that issued it — `func`, `pc`, `ra`, plus the linked-list OT rank
(`handle_gpu_frame_dump` in `runtime/src/debug_server.c`, `GpuGp0RingEntry` in
`runtime/include/gpu.h`). Nothing consumed that attribution, so "the glow is
opaque and the vignette is missing" stayed a description of a screenshot rather
than a pointer at a function.
---
## The provenance rule
Everything these tools report is **observed** from one execution of one frame,
and is labelled as such. It is never merged into a static claim.
A `func` in a frame dump proves that code ran and issued a primitive. It says
nothing about the call graph — the analyser's static coverage gap
(`FUNCTION_DISCOVERY.md`) stays exactly as wide as it was. Studio shows observed
counts and static names in different columns for the same reason.
---
## What is decoded faithfully, and what is not
**Faithful:**
* Vertices, including the 11-bit signed coordinate and the running GP0(E5) draw
offset — the same offset `gp0_exec_*` applies in `gpu.c`.
* Packet word layouts for every polygon, line, rectangle, fill and copy form.
* The texpage latch. A textured polygon's own tpage word overwrites draw-mode
state (`set_tpage_from_poly` in `gpu.c`), and later untextured polygons and
rectangles — which carry no tpage word — inherit it. Getting this wrong
mislabels precisely the semi-transparency mode you are usually chasing.
* All four semi-transparency blends, when layers are rendered:
`0.5B+0.5F`, `B+F`, `B-F`, `B+0.25F` (GPUSTAT bits 5-6).
**Not:**
* Textures are never sampled. Texture pages, CLUTs and UVs are decoded and
reported, so a wrong-CLUT hypothesis is testable, but a textured primitive
renders as its command colour. Re-implementing texture sampling would buy
fidelity the question does not need and add a whole new way to be wrong.
* The ring truncates each packet to `GPU_GP0_RING_MAX_WORDS` (12). Longer
packets are decoded as far as recorded and flagged `truncated` — never
guessed at.
---
## The build must actually have a debug server
**Release + `-DPSX_DEBUG_TOOLS=ON` is the combination you want.** Not a Debug
build: a Debug build of a recomp is far too slow to reach the frame you are
chasing.
`runtime/src/debug_server.c` is always compiled, but `debug_server_init()` is
only *called* when `PSX_NO_DEBUG_TOOLS` is undefined (`runtime/src/main.cpp`).
`PSX_DEBUG_TOOLS` defaults **ON** for `Debug` / `RelWithDebInfo` and **OFF** for
`Release` / `MinSizeRel` (`runtime/runtime.cmake`), so a plain
`cmake -DCMAKE_BUILD_TYPE=Release` produces a lean binary that opens no port at
all. Every "the tool won't connect" starts here.
```bash
cmake -S . -B build-release -G Ninja \
-DCMAKE_BUILD_TYPE=Release -DPSX_DEBUG_TOOLS=ON
cmake --build build-release --target psx-runtime
```
Check an existing build dir without reconfiguring it:
```bash
grep -E '^(PSX_DEBUG_TOOLS|CMAKE_BUILD_TYPE):' build-release/CMakeCache.txt
```
`PSX_DEBUG_TOOLS` is an `option()`, so it lives in the cache: changing
`CMAKE_BUILD_TYPE` on an **existing** build dir does *not* flip it. Pass the
`-D` explicitly, or start a fresh build dir.
In RetComM Studio this is the **Build tab → Debug tools** row: a selector that
injects the flag, a **Configure for debugging** button that sets Release + ON in
one click, and a status line that reads the configured build dir and says
whether the server will be there. The Frames tab shows the same line when it
cannot connect, so "not connected" never has to be guessed at.
### The port is a runtime setting, not a build flag
The debug port comes from `game.toml`:
```toml
[runtime]
debug_port = 4370
```
If that key is absent the runtime uses the compiled `DEFAULT_DEBUG_PORT` (4370
for `psx-runtime`, 4380 for `psx-beetle`), and `--debug-port <n>` on the command
line overrides both. Studio's Frames tab defaults its port field to whatever the
selected project's `game.toml` says.
---
## You cannot pause the game, and you do not need to
`pause`, `continue`, `step` and `run_to_frame` were **removed** from
psx-runtime. `runtime/src/debug_server.c` still registers them, but only as
handlers that return an error explaining the migration — pause-step-read
synthesizes a snapshot ("what is state right NOW") instead of reading the
history the runtime already records continuously, and in this codebase it forced
the runtime into a wait loop where a dropped client looked like a freeze.
The replacement is better for chasing a render bug. The GP0 ring holds
`1 << 20` packets (`gpu.c`, `GP0_RING_CAP`) — several hundred frames — so the
workflow runs the other way round:
> Play until the bug is on screen. **Then** capture that frame, and the frames
> leading up to it.
Reaching backwards beats freezing forwards, and it is the only thing that works
at all for a glitch you cannot reliably stop on.
```bash
python3 $P/gpu_frame_capture.py --ring # what can I still ask for?
# GP0 ring: 812344 packet(s) seen, capacity 1048576, 12 words/packet
# capturable frames: 40918..41230
```
`gpu_frame_dump` on a frame that has been evicted returns an **empty** dump, not
an error — which reads exactly like "that frame drew nothing", a conclusion you
might act on. So `capture()` checks the span first and refuses, and Studio's
Frames tab shows the span next to the frame selector and marks an out-of-ring
frame in red.
What replaces each removed command:
| Gone | Use instead |
|---|---|
| `run_to_frame N` | `--frame N` against the ring (any frame it still holds) |
| `step` | `--frames N`, walking backwards from `--frame` |
| `pause` + read state | `frame_range` / `get_frame` (per-frame records), `fn_entry_dump`, `wtrace_dump` |
| `pause` + inspect drawing | `gpu_frame_dump frame=N` — what this whole toolchain does |
Savestates are unaffected: `savestate` is a real handler, and a savestate plus a
fixed frame number still makes a run reproducible.
---
## Transport
**One request per connection.** `io_thread_main()` in `debug_server.c` accepts,
reads a single line, replies, and closes. A client that holds the socket open
and sends a second command gets silence and then EOF, which is indistinguishable
from a hung emulator. `psx_gpu_frame.DebugConn` opens a socket per command;
`tools/debug_client.py`'s `query()` documents the same contract.
---
## Typical session
```bash
cd <your recomp project>
P=psxrecomp/tools
# 1. Play until the bug is on screen. Capture the newest frame as the suspect,
# then reach BACK into the ring for one that still looked right.
python3 $P/gpu_frame_capture.py --tag bad --out analysis/frames --summary
python3 $P/gpu_frame_capture.py --frame 41100 --tag good --out analysis/frames --summary
# 2. What changed?
python3 $P/gpu_frame_diff.py analysis/frames/good.json analysis/frames/bad.json
# 3. Look at the bad frame one function at a time.
python3 $P/gpu_frame_layers.py analysis/frames/bad.json --out analysis/frames/bad-layers
# 4. Did the guest even run the same? (needs a patched DuckStation on 4371)
python3 $P/gpu_parity.py --frame 41230 --out analysis/frames/parity
```
`--frames N` also grabs the N-1 frames *before* `--frame`, walking backwards
through the ring — the frames leading up to the one you noticed are usually the
ones that explain it. A savestate plus a fixed frame number still makes the whole
thing reproducible.
A screenshot is only written for the live frame: the framebuffer holds the
present, not the historical frame you just dumped, and saving it beside an older
dump would label it as something it is not.
### Artifacts
| File | Contents |
|---|---|
| `<tag>.json` | the full dump: every decoded primitive, with `func`/`pc`/`ra`/`ot` |
| `<tag>.summary.json` | totals, opcode + blend-mode histograms, per-function attribution |
| `<tag>.opcodes.json` | the server's own `gpu_opcodes` histogram |
| `<tag>.png` | the presented frame |
| `<tag>-layers/composite.png` | every primitive, in issue order |
| `<tag>-layers/layer-<func>.png` | one function's contribution, RGBA, alpha = coverage |
| `<tag>-layers/sheet.png` | labelled contact sheet of all layers |
| `<tag>-layers/layers.json` | the layer index: per-function stats and file names |
| `diff.json` | machine-readable diff, `--json` |
A busy frame's dump runs to tens of megabytes. Anything that only wants to know
*who drew what* should read `<tag>.summary.json` instead — which is what Studio
does.
### Layers render over grey, on purpose
An isolated layer is drawn over neutral grey (`--layer-backdrop`, default 128),
not black. A `B-F` subtractive vignette against black is black, and an additive
glow against black is just the glow: both blends go invisible in exactly the
view meant to show them. The composite still starts from black, as the GPU does.
---
## Reading a diff
The report leads with the two findings that name a bug on their own:
* **a blend mode that disappeared** — if `B-F` drew ten primitives in the good
frame and none in the bad one, whatever layer used that blend (a vignette, a
shadow) is no longer being drawn at all;
* **a function that stopped drawing** — it stopped running, or stopped reaching
its emit path.
Then, per function: `stopped drawing`, `lost semi-transparency`,
`started drawing`, `gained semi-transparency`, `count changed`.
The diff is over **counts and flags per (function, opcode)**, never over exact
geometry. An animating effect legitimately moves its vertices every frame;
geometry appears only as a sample, for context.
---
## Turning an address into a name
Every function the tools report is an address you can name:
```bash
python3 -m project_studio analyze set-symbol --root . --pc 0x8004ABCD \
--name LandEffect_DrawRays --status guessed \
--note "observed issuing GP0 primitives (frame capture)"
python3 tools/sync_symbols.py # regenerates psx_symbols.h / PSX_FN_*
```
Studio's Frames tab has a **Save name** box that calls exactly this, so
`symbols.toml` keeps a single writer.
---
## Parity scope
`gpu_parity.py` compares **images**, not GP0 streams. Only DuckStation is
driven: it implements `run_to_frame`, psx-runtime does not, so the native side is
sampled where it already is and DuckStation is advanced to meet it. A residual
frame mismatch is reported rather than papered over. The DuckStation oracle
patch (`tools/duckstation/psxrecomp_oracle.patch`) implements `screenshot`,
`read_vram`, `gpu_state`, `run_to_frame`, `step` and `pause` — but not
`gpu_frame_dump`, so there is no packet stream to compare on that side. Adding
one to the patch is the obvious next step if image parity keeps saying "the
streams must differ" without saying where; until then, `tools/cosim.py` brackets
the divergence.
Getting an oracle on Linux: `python3 tools/duckstation_oracle.py all` builds and
installs a patched DuckStation into `~/.local/share/retcomm/oracle/duckstation/`
— outside any game repo, shared by every title — and `start --disc <cue>` runs it
headless on 4371. On Arch-family and NixOS hosts the build runs inside
`ubuntu:22.04` via podman, because upstream's CMake refuses those hosts outright
and its build scripts may not be patched. See `tools/duckstation/README.md`.
Both instances must be driven identically — same disc, same starting state.
Parity between two runs that were not driven the same way means nothing.
---
## Tests
```bash
python3 -m unittest discover -s tools/tests -p 'test_gpu_frame.py'
```
Covers the packet word layouts against `gpu.c`'s command lengths, the signed
vertex, the E5 offset, the textured-polygon tpage latch, the four blends,
truncation flagging, the attribution rollup, the diff verdicts, and a stub
server that reproduces the one-request-per-connection contract.
+23
View File
@@ -337,6 +337,29 @@ def build_cmd(args):
if len(args) > 1:
d["path"] = args[1]
return d, pretty_json
elif cmd in ("screenshot_file", "shot_file"):
# Canonical native 15-bit VRAM frame (pre-compositor, pre-stretch).
# Takes a positional path like its two sibling capture commands.
d = {"cmd": "screenshot_file"}
if len(args) > 1:
d["path"] = args[1]
return d, pretty_json
elif cmd in ("present_shot", "shot_present"):
# The composed renderer output -- what the window actually shows,
# including the logical-size aspect fit the buffer captures miss.
# Use this to verify anything aspect-shaped (widescreen, letterbox).
d = {"cmd": "present_shot"}
if len(args) > 1:
d["path"] = args[1]
return d, pretty_json
elif cmd == "present_shot_seq":
return {"cmd": "present_shot_seq"}, pretty_json
elif cmd == "turbo":
if len(args) < 2:
return None, lambda _: "Usage: turbo <0|1>"
return {"cmd": "turbo", "enabled": int(args[1])}, pretty_json
elif cmd == "turbo_state":
return {"cmd": "turbo_state"}, pretty_json
elif cmd == "bios_trace":
d = {"cmd": "bios_trace"}
if len(args) > 1:
+39
View File
@@ -0,0 +1,39 @@
# Builder image for the DuckStation oracle.
#
# Why a container at all: DuckStation's CMake refuses to configure on
# Arch-family and NixOS hosts (CMakeModules/DuckStationBuildSummary.cmake), and
# upstream explicitly forbids distributing patches that modify its build system.
# So instead of working around the check, we build somewhere it genuinely
# supports -- ubuntu:22.04, the exact image its own Linux CI uses
# (.github/workflows/linux-appimage-build.yml).
#
# Ubuntu 22.04's glibc is older than any modern host's, so the binary produced
# here runs natively on the host afterwards. That is the same property upstream
# relies on to ship one AppImage for every distro.
#
# The package list is upstream's own scripts/appimage/install-packages.sh,
# copied verbatim out of the pinned checkout by duckstation_oracle.py. It is not
# reproduced or modified here -- it is the authority on what the build needs.
FROM ubuntu:22.04
ENV DEBIAN_FRONTEND=noninteractive
# linuxdeploy and appimagetool ship as AppImages and want FUSE, which a
# container generally cannot give them. This is the AppImage runtime's own
# supported escape hatch, not a modification of upstream's packaging script.
ENV APPIMAGE_EXTRACT_AND_RUN=1
# What upstream's install-packages.sh assumes is already present: it calls sudo
# and apt-add-repository, neither of which a bare ubuntu image has.
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
ca-certificates curl wget gnupg sudo software-properties-common \
git xz-utils file desktop-file-utils \
&& rm -rf /var/lib/apt/lists/*
COPY install-packages.sh /tmp/install-packages.sh
RUN bash /tmp/install-packages.sh \
&& rm -f /tmp/install-packages.sh \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /oracle
+65 -1
View File
@@ -10,7 +10,71 @@ PSXRecomp v4 uses a patched build of [stenzek/duckstation](https://github.com/st
- **`tools/duckstation/build.sh`** — runs CMake (Visual Studio 17 2022, x64, Release) and MSBuild on `duckstation-qt`. Requires the Visual Studio 2022 "Desktop development with C++" workload (CMake + MSBuild come with it).
- **`tools/fix_duckstation_deps_paths.py`** — helper used by `setup.sh` to rewrite stale `_IMPORT_PREFIX` values in the extracted prebuilt deps.
## First-time setup
## First-time setup -- Linux / macOS
Use `tools/duckstation_oracle.py`. It installs outside any game repo, into the
shared RetComM data root, so one build serves every title:
```text
~/.local/share/retcomm/oracle/duckstation/ ($RETCOMM_DATA_DIR, or $RETCOMM_ORACLE_DIR)
src/ pinned upstream checkout with the oracle patch applied
build/ cmake/ninja tree
app/ the portable install that actually gets run
oracle.json
```
```bash
P=psxrecomp/tools
python3 $P/duckstation_oracle.py doctor # can this machine build it?
python3 $P/duckstation_oracle.py all # fetch + patch + build + install
python3 $P/duckstation_oracle.py start --disc disc/game.cue
python3 $P/duckstation_oracle.py status
python3 $P/duckstation_oracle.py stop
```
`gpu_parity.py --start-oracle --disc <cue>` will do the `start` step for you
when nothing is answering on 4371.
### Arch, CachyOS, NixOS: the build runs in a container
DuckStation's CMake refuses to configure on Arch-family and NixOS hosts, and
the note above that check states you do not have permission to distribute
patches modifying its build system. Its build scripts are CC-BY-NC-ND-4.0.
So the tool detects the refusal up front and builds in an environment upstream
supports: `ubuntu:22.04` via podman/docker, installing packages with upstream's
own `scripts/appimage/install-packages.sh`, copied verbatim out of the pinned
checkout. On a distro upstream accepts, the build runs directly on the host.
Force either path with `build --container` / `build --no-container`.
### Notes that cost an afternoon to find
* The binary's RUNPATH points at the build environment, and Qt needs its plugin
tree. `install` copies `dep/prebuilt/<platform>/{lib,plugins}` next to the
binary and writes a `run-oracle` launcher that sets `LD_LIBRARY_PATH` and
`QT_PLUGIN_PATH`, then verifies with `ldd` that nothing is unresolved.
* An unofficial build shows a modal warning dialog before the core thread
starts. Under `-nogui` it is invisible, so the process sits blocked with no
port and no error. `install` sets DuckStation's own
`[UI] UnofficialBuildWarningConfirmed = true` key instead of patching that
behavior.
* `install` seeds DuckStation's own default `settings.ini` by running it
briefly, then merges only the handful of keys an oracle needs.
* Wayland support is off by default (`-DENABLE_WAYLAND=OFF`) because an
offscreen oracle never opens a Wayland surface. `build --wayland` re-enables
it.
* The oracle server only starts when a system boots. An idle DuckStation with
nothing loaded will never open the port.
### The pin
`tools/duckstation/pin.json` is the single source of truth for the upstream
base. Both this tool and the Windows `setup.sh` read it, so the two platforms
cannot drift onto different trees. The prebuilt-dependency version and SHA-256
come from the checkout's own `dep/PREBUILT-VERSION` and
`dep/PREBUILT-SHA256SUMS`.
## First-time setup -- Windows
```bash
cd <psxrecomp-v4>
+13
View File
@@ -0,0 +1,13 @@
{
"_comment": [
"Single source of truth for the DuckStation oracle pin.",
"setup.sh (Windows) and duckstation_oracle.py (Linux/macOS) both read this,",
"so the two platforms can never drift onto different upstream bases.",
"Bump upstream_base only together with a regenerated psxrecomp_oracle.patch:",
" cd <checkout> && git diff <new base> > tools/duckstation/psxrecomp_oracle.patch"
],
"upstream_url": "https://github.com/stenzek/duckstation.git",
"upstream_base": "ffb33c281d196eb8ee0f559085ca285de7cdd51b",
"patch": "psxrecomp_oracle.patch",
"oracle_port": 4371
}
+3 -1
View File
@@ -16,7 +16,9 @@ set -euo pipefail
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
DUCK="$REPO_ROOT/duckstation"
PATCH="$REPO_ROOT/tools/duckstation/psxrecomp_oracle.patch"
UPSTREAM_BASE="ffb33c281d196eb8ee0f559085ca285de7cdd51b" # stenzek/master at time of daa1808
# The pin lives in pin.json so this script and duckstation_oracle.py (the
# Linux/macOS path) can never drift onto different upstream bases.
UPSTREAM_BASE="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["upstream_base"])' "$REPO_ROOT/tools/duckstation/pin.json")"
log() { echo "[duckstation-setup] $*"; }
File diff suppressed because it is too large Load Diff
+173
View File
@@ -0,0 +1,173 @@
#!/usr/bin/env python3
"""gpu_frame_capture.py -- capture one frame's GP0 stream + screenshot, decoded
and attributed to the guest functions that issued it.
# the newest frame the ring holds, saved as the "bad" reference
python3 gpu_frame_capture.py --tag bad --out analysis/frames
# a specific past frame, and the four before it
python3 gpu_frame_capture.py --frame 41230 --frames 5 --tag bad
# what can I still ask for?
python3 gpu_frame_capture.py --ring
You do not pause the game to use this, and you cannot: pause / step /
run_to_frame were removed from psx-runtime (runtime/src/debug_server.c) because
pause-step-read synthesizes a snapshot instead of reading the history the
runtime already keeps. The GP0 ring holds ~1M packets -- several hundred frames
-- so the workflow is the other way round: play until the bug is on screen, then
capture that frame AND the ones leading up to it. Reaching backwards beats
freezing forwards, and it is the only thing that works for a glitch you cannot
stop on.
Writes <out>/<tag>.json (a psx-gpu-frame dump), <out>/<tag>.summary.json (the
compact attribution RetComM Studio reads), <out>/<tag>.png (the presented
frame), and <out>/<tag>.opcodes.json, so a later diff or layer render needs
nothing still running.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from psx_gpu_frame import ( # noqa: E402
DEFAULT_NATIVE_PORT, DebugConn, DebugError, capture, save_dump, save_summary,
)
def summarise(dump, top=25, out=sys.stdout):
"""Per-function attribution table -- the thing you actually read."""
funcs = dump["funcs"]
order = sorted(funcs.values(), key=lambda r: (-r["drawing"], r["func"]))
total_draw = sum(r["drawing"] for r in funcs.values())
print(f"frame {dump['frame']} packets={dump['raw_count']} "
f"drawing={total_draw} functions={len(funcs)}", file=out)
if dump.get("capped"):
print(f" !! ring returned the full {dump['requested']} requested -- the frame "
f"may be truncated; re-run with a larger --count", file=out)
trunc = sum(1 for p in dump["prims"] if p.get("truncated"))
if trunc:
print(f" note: {trunc} packet(s) longer than the ring's word cap; "
f"decoded as far as recorded", file=out)
print(file=out)
print(f" {'func':<12} {'draw':>6} {'semi':>6} {'tex':>6} {'OT':>11} ops", file=out)
for r in order[:top]:
ot = "-" if r["ot_min"] is None else (
f"{r['ot_min']}" if r["ot_min"] == r["ot_max"] else f"{r['ot_min']}..{r['ot_max']}")
ops = ", ".join(f"{k}x{v}" for k, v in
sorted(r["ops"].items(), key=lambda kv: -kv[1])[:4])
print(f" {r['func']:<12} {r['drawing']:>6} {r['semi']:>6} "
f"{r['textured']:>6} {ot:>11} {ops}", file=out)
if len(order) > top:
print(f" ... {len(order) - top} more function(s)", file=out)
modes = {}
for r in funcs.values():
for m, n in r["stp_modes"].items():
modes[m] = modes.get(m, 0) + n
print(file=out)
if modes:
print(" semi-transparency modes in use: " +
", ".join(f"{m} x{n}" for m, n in sorted(modes.items())), file=out)
else:
print(" no semi-transparent primitives in this frame -- if the effect "
"you are chasing is a glow or a vignette, that is the finding",
file=out)
def main(argv=None):
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=DEFAULT_NATIVE_PORT)
ap.add_argument("--timeout", type=float, default=20.0)
ap.add_argument("--frame", type=int, default=None,
help="frame to dump (default: the newest one the ring holds)")
ap.add_argument("--frames", type=int, default=1,
help="also capture the N-1 frames BEFORE --frame, walking "
"backwards through the ring (tag gets a -NN suffix, "
"-00 being the newest)")
ap.add_argument("--ring", action="store_true",
help="print which frames the GP0 ring can still be asked "
"for, then exit")
ap.add_argument("--count", type=int, default=65536, help="max GP0 packets to pull")
ap.add_argument("--tag", default="frame", help="output basename")
ap.add_argument("--out", default="analysis/frames", help="output directory")
ap.add_argument("--no-save", action="store_true", help="print only, write nothing")
ap.add_argument("--no-shot", action="store_true", help="skip the screenshot")
ap.add_argument("--summary", action="store_true", help="print the attribution table")
ap.add_argument("--top", type=int, default=25)
args = ap.parse_args(argv)
try:
with DebugConn(args.host, args.port, args.timeout) as conn:
span = conn.ring_span()
if args.ring:
print(f"GP0 ring: {span['total']} packet(s) seen, capacity "
f"{span['capacity']}, {span['max_words']} words/packet")
if span["total"] == 0:
print(" nothing recorded yet -- is a game running?")
else:
print(f" capturable frames: {span['oldest']}..{span['newest']}")
return 0
outdir = os.path.abspath(args.out)
if not args.no_save:
os.makedirs(outdir, exist_ok=True)
base_frame = args.frame if args.frame is not None else span["newest"]
n = max(1, args.frames)
print(f"ring holds frames {span['oldest']}..{span['newest']}; "
f"capturing {n} frame(s) ending at {base_frame}")
for i in range(n):
# Walk BACKWARDS: the ring is history, and the frames before the
# one you noticed are the ones that explain it.
frame = base_frame - i
if frame < span["oldest"]:
print(f" stopping at {frame}: older than the ring holds "
f"({span['oldest']})", file=sys.stderr)
break
tag = args.tag if n == 1 else f"{args.tag}-{i:02d}"
dump = capture(conn, frame=frame, count=args.count, label=tag,
verify_ring=False)
if not args.no_save:
jp = os.path.join(outdir, f"{tag}.json")
save_dump(dump, jp)
save_summary(dump, os.path.join(outdir, f"{tag}.summary.json"),
dump_name=f"{tag}.json")
print(f"wrote {jp} ({dump['raw_count']} packets, "
f"{len(dump['funcs'])} functions)")
try:
oc = conn.cmd("gpu_opcodes")
with open(os.path.join(outdir, f"{tag}.opcodes.json"), "w",
encoding="utf-8") as f:
json.dump(oc.get("opcodes", {}), f, indent=1)
except DebugError as e:
print(f" (gpu_opcodes unavailable: {e})", file=sys.stderr)
if not args.no_shot:
if frame != span["newest"]:
print(" (no screenshot: the framebuffer holds the "
"live frame, not this historical one)")
else:
try:
r = conn.screenshot(os.path.join(outdir, f"{tag}.png"))
print(f"wrote {r.get('path')}")
except DebugError as e:
print(f" (screenshot failed: {e})", file=sys.stderr)
if args.summary or args.no_save:
summarise(dump, top=args.top)
except DebugError as e:
print(f"error: {e}", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
+212
View File
@@ -0,0 +1,212 @@
#!/usr/bin/env python3
"""gpu_frame_diff.py -- diff two captured frames by the guest function that drew them.
python3 gpu_frame_diff.py analysis/frames/good.json analysis/frames/bad.json
The report leads with the findings that actually name a bug:
* a function that drew in A and drew nothing in B -> it stopped running, or
stopped reaching its emit path
* a function whose semi-transparent primitive count collapsed -> the STP bit
or the draw-mode latch feeding it changed
* a semi-transparency MODE that disappeared entirely -> whatever layer used
that blend (a B-F vignette, a B+F glow) is no longer being drawn
Frame-to-frame jitter is real: an animating effect legitimately moves vertices
every frame. So the diff is over counts and flags per (function, opcode), never
over exact geometry, and geometry is only ever quoted as a sample for context.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from psx_gpu_frame import STP_MODES, load_dump # noqa: E402
def bucket(dump):
"""Counts keyed (func, op_name) plus per-function and global rollups."""
per_key = {}
per_func = {}
ops = {}
modes = {}
for p in dump["prims"]:
if p["kind"] not in ("poly", "rect", "line", "fill"):
continue
fn = p.get("func", "0x00000000")
key = (fn, p["op_name"])
b = per_key.setdefault(key, {"n": 0, "semi": 0, "tex": 0, "sample": None})
b["n"] += 1
if p.get("semi"):
b["semi"] += 1
if p.get("textured"):
b["tex"] += 1
if b["sample"] is None:
b["sample"] = {"verts": p.get("verts"), "colors": p.get("colors"),
"stp": p.get("stp"), "ot": p.get("ot"),
"ra": p.get("ra"), "seq": p.get("seq")}
f = per_func.setdefault(fn, {"n": 0, "semi": 0, "tex": 0, "ras": set()})
f["n"] += 1
if p.get("semi"):
f["semi"] += 1
if p.get("textured"):
f["tex"] += 1
if p.get("ra"):
f["ras"].add(p["ra"])
ops[p["op_name"]] = ops.get(p["op_name"], 0) + 1
if p.get("semi"):
m = STP_MODES.get(p.get("stp", 0), "?")
modes[m] = modes.get(m, 0) + 1
for f in per_func.values():
f["ras"] = sorted(f["ras"])
return {"keys": per_key, "funcs": per_func, "ops": ops, "modes": modes}
def _delta_map(a, b):
out = {}
for k in set(a) | set(b):
av, bv = a.get(k, 0), b.get(k, 0)
if av != bv:
out[k] = {"a": av, "b": bv, "delta": bv - av}
return out
def diff(dump_a, dump_b):
A, B = bucket(dump_a), bucket(dump_b)
funcs = []
for fn in sorted(set(A["funcs"]) | set(B["funcs"])):
fa = A["funcs"].get(fn, {"n": 0, "semi": 0, "tex": 0, "ras": []})
fb = B["funcs"].get(fn, {"n": 0, "semi": 0, "tex": 0, "ras": []})
if fa["n"] == fb["n"] and fa["semi"] == fb["semi"] and fa["tex"] == fb["tex"]:
continue
if fa["n"] and not fb["n"]:
verdict = "stopped drawing"
elif fb["n"] and not fa["n"]:
verdict = "started drawing"
elif fa["semi"] and not fb["semi"]:
verdict = "lost semi-transparency"
elif fb["semi"] and not fa["semi"]:
verdict = "gained semi-transparency"
else:
verdict = "count changed"
funcs.append({
"func": fn, "verdict": verdict,
"a": {"prims": fa["n"], "semi": fa["semi"], "tex": fa["tex"]},
"b": {"prims": fb["n"], "semi": fb["semi"], "tex": fb["tex"]},
"ras": sorted(set(fa["ras"]) | set(fb["ras"]))[:8],
})
keys = []
for k in sorted(set(A["keys"]) | set(B["keys"])):
ka = A["keys"].get(k, {"n": 0, "semi": 0, "tex": 0, "sample": None})
kb = B["keys"].get(k, {"n": 0, "semi": 0, "tex": 0, "sample": None})
if ka["n"] == kb["n"] and ka["semi"] == kb["semi"]:
continue
keys.append({
"func": k[0], "op": k[1],
"a": {"n": ka["n"], "semi": ka["semi"], "sample": ka["sample"]},
"b": {"n": kb["n"], "semi": kb["semi"], "sample": kb["sample"]},
})
order = {"stopped drawing": 0, "lost semi-transparency": 1, "started drawing": 2,
"gained semi-transparency": 3, "count changed": 4}
funcs.sort(key=lambda f: (order.get(f["verdict"], 9),
-abs(f["b"]["prims"] - f["a"]["prims"])))
return {
"kind": "psx-gpu-frame-diff",
"version": 1,
"a": {"label": dump_a.get("label", ""), "frame": dump_a["frame"],
"packets": dump_a["raw_count"]},
"b": {"label": dump_b.get("label", ""), "frame": dump_b["frame"],
"packets": dump_b["raw_count"]},
"funcs": funcs,
"keys": keys,
"ops": _delta_map(A["ops"], B["ops"]),
"modes": _delta_map(A["modes"], B["modes"]),
"modes_a": A["modes"],
"modes_b": B["modes"],
}
def report(d, out=sys.stdout, max_keys=40):
print(f"A {d['a']['label'] or 'a'}: frame {d['a']['frame']}, "
f"{d['a']['packets']} packets", file=out)
print(f"B {d['b']['label'] or 'b'}: frame {d['b']['frame']}, "
f"{d['b']['packets']} packets", file=out)
print(file=out)
gone = [m for m, v in d["modes"].items() if v["b"] == 0 and v["a"] > 0]
new = [m for m, v in d["modes"].items() if v["a"] == 0 and v["b"] > 0]
if gone:
print("HEADLINE: semi-transparency mode(s) present in A and absent in B: "
+ ", ".join(f"{m} (was {d['modes'][m]['a']} prims)" for m in gone), file=out)
if new:
print("HEADLINE: semi-transparency mode(s) new in B: "
+ ", ".join(f"{m} ({d['modes'][m]['b']} prims)" for m in new), file=out)
stopped = [f for f in d["funcs"] if f["verdict"] == "stopped drawing"]
if stopped:
print(f"HEADLINE: {len(stopped)} function(s) drew in A and nothing in B: "
+ ", ".join(f["func"] for f in stopped[:6])
+ (" ..." if len(stopped) > 6 else ""), file=out)
if not (gone or new or stopped):
print("No function stopped drawing and no blend mode disappeared.", file=out)
print(file=out)
print("semi-transparency modes A B", file=out)
for m in sorted(set(d["modes_a"]) | set(d["modes_b"])):
print(f" {m:<22} {d['modes_a'].get(m, 0):>5} {d['modes_b'].get(m, 0):>5}",
file=out)
print(file=out)
if d["ops"]:
print("opcode count changes", file=out)
for op, v in sorted(d["ops"].items(), key=lambda kv: -abs(kv[1]["delta"]))[:20]:
print(f" {op:<22} {v['a']:>5} -> {v['b']:>5} ({v['delta']:+d})", file=out)
print(file=out)
print(f"per-function changes ({len(d['funcs'])})", file=out)
print(f" {'func':<12} {'verdict':<24} {'A prims/semi':>14} {'B prims/semi':>14}",
file=out)
for f in d["funcs"][:max_keys]:
a = f"{f['a']['prims']}/{f['a']['semi']}"
b = f"{f['b']['prims']}/{f['b']['semi']}"
print(f" {f['func']:<12} {f['verdict']:<24} {a:>14} {b:>14}", file=out)
if len(d["funcs"]) > max_keys:
print(f" ... {len(d['funcs']) - max_keys} more", file=out)
print(file=out)
print(f"per (function, opcode) changes ({len(d['keys'])})", file=out)
for k in d["keys"][:max_keys]:
print(f" {k['func']:<12} {k['op']:<20} "
f"{k['a']['n']:>5} -> {k['b']['n']:>5} "
f"semi {k['a']['semi']:>4} -> {k['b']['semi']:>4}", file=out)
if len(d["keys"]) > max_keys:
print(f" ... {len(d['keys']) - max_keys} more", file=out)
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("a", help="reference (good) psx-gpu-frame dump")
ap.add_argument("b", help="suspect (bad) psx-gpu-frame dump")
ap.add_argument("--json", default=None, help="also write the diff as JSON here")
ap.add_argument("--max", type=int, default=40)
ap.add_argument("--quiet", action="store_true", help="write JSON only")
args = ap.parse_args(argv)
d = diff(load_dump(args.a), load_dump(args.b))
if args.json:
with open(args.json, "w", encoding="utf-8") as f:
json.dump(d, f, indent=1)
if not args.quiet:
report(d, max_keys=args.max)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+371
View File
@@ -0,0 +1,371 @@
#!/usr/bin/env python3
"""gpu_frame_layers.py -- render a captured frame as one image per guest function.
python3 gpu_frame_layers.py analysis/frames/bad.json --out analysis/frames/bad-layers
You get:
composite.png every primitive, in issue order
layer-<func>.png that function's contribution alone (RGBA, alpha=coverage)
sheet.png a labelled contact sheet of all layers
layers.json the index: per-function stats, bbox, file names
What this is and is not
-----------------------
It rasterises geometry and shading, and it applies the four real semi-
transparency blends, because "the glow is opaque" and "the vignette never got
drawn" are both blend-state findings and both invisible in a wireframe.
It does NOT sample textures. A textured primitive is filled with its command
colour (or flat grey when the raw-texture bit says the command colour is
ignored). Texture pages, CLUTs and UVs are decoded and reported in layers.json,
so a wrong-CLUT hypothesis is testable, but this is not a second GPU.
That trade is deliberate: the question these layers answer is *which function
drew what, and with which blend*, and re-implementing texture sampling would
buy fidelity the question does not need while adding a whole new way to be
wrong.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import numpy as np # noqa: E402
from PIL import Image, ImageDraw # noqa: E402
from psx_gpu_frame import STP_MODES, draw_area, load_dump, prim_bbox # noqa: E402
RAW_TEXTURE_GREY = (128.0, 128.0, 128.0)
TEX_TINT = (255.0, 0.0, 255.0)
# Isolated layers render over a neutral grey, not black. A B-F subtractive
# vignette against black is black, and an additive glow against black is just
# the glow -- both blends become invisible in exactly the view meant to show
# them. The composite still starts from black, which is what the GPU sees.
LAYER_BACKDROP = 128.0
class Canvas:
"""Float RGB canvas in VRAM coordinates, offset to the frame's draw area."""
def __init__(self, x0: int, y0: int, w: int, h: int, backdrop: float = 0.0):
self.x0, self.y0 = x0, y0
self.w, self.h = w, h
self.backdrop = float(backdrop)
self.rgb = np.full((h, w, 3), float(backdrop), dtype=np.float32)
self.cov = np.zeros((h, w), dtype=bool)
def _blend(self, mask, src, semi, stp):
"""Apply one primitive's pixels. src is (...,3) float or a 3-vector.
The four modes are GPUSTAT bits 5-6 exactly: 0.5B+0.5F, B+F, B-F,
B+0.25F. Getting these right is the point of rendering at all.
"""
dst = self.rgb[mask]
if not semi:
out = src
elif stp == 0:
out = 0.5 * dst + 0.5 * src
elif stp == 1:
out = dst + src
elif stp == 2:
out = dst - src
else:
out = dst + 0.25 * src
self.rgb[mask] = np.clip(out, 0.0, 255.0)
self.cov[mask] = True
def triangle(self, pts, cols, semi, stp):
(ax, ay), (bx, by), (cx, cy) = pts
den = (by - cy) * (ax - cx) + (cx - bx) * (ay - cy)
if den == 0:
return
minx = max(0, int(np.floor(min(ax, bx, cx))) - self.x0)
maxx = min(self.w - 1, int(np.ceil(max(ax, bx, cx))) - self.x0)
miny = max(0, int(np.floor(min(ay, by, cy))) - self.y0)
maxy = min(self.h - 1, int(np.ceil(max(ay, by, cy))) - self.y0)
if minx > maxx or miny > maxy:
return
# Oversize primitives are rejected by the hardware too (see
# psx_gpu_triangle_oversize in gpu.c); a 1000px triangle here is almost
# always a decode or a data bug, and filling it would hide the real ones.
if (maxx - minx) > 1024 or (maxy - miny) > 512:
return
xs = np.arange(minx, maxx + 1, dtype=np.float32) + self.x0
ys = np.arange(miny, maxy + 1, dtype=np.float32) + self.y0
X, Y = np.meshgrid(xs, ys)
l0 = ((by - cy) * (X - cx) + (cx - bx) * (Y - cy)) / den
l1 = ((cy - ay) * (X - cx) + (ax - cx) * (Y - cy)) / den
l2 = 1.0 - l0 - l1
inside = (l0 >= 0) & (l1 >= 0) & (l2 >= 0)
if not inside.any():
return
c = np.asarray(cols, dtype=np.float32)
src = (l0[..., None] * c[0] + l1[..., None] * c[1] + l2[..., None] * c[2])[inside]
full = np.zeros((self.h, self.w), dtype=bool)
full[miny:maxy + 1, minx:maxx + 1] = inside
self._blend(full, src, semi, stp)
def rect(self, x, y, w, h, color, semi, stp):
x0 = max(0, x - self.x0)
y0 = max(0, y - self.y0)
x1 = min(self.w, x - self.x0 + max(0, w))
y1 = min(self.h, y - self.y0 + max(0, h))
if x1 <= x0 or y1 <= y0:
return
mask = np.zeros((self.h, self.w), dtype=bool)
mask[y0:y1, x0:x1] = True
self._blend(mask, np.asarray(color, dtype=np.float32), semi, stp)
def line(self, p0, p1, c0, c1, semi, stp):
(x0, y0), (x1, y1) = p0, p1
n = int(max(abs(x1 - x0), abs(y1 - y0))) + 1
if n > 4096:
return
t = np.linspace(0.0, 1.0, n)
xs = np.rint(x0 + (x1 - x0) * t).astype(int) - self.x0
ys = np.rint(y0 + (y1 - y0) * t).astype(int) - self.y0
ok = (xs >= 0) & (xs < self.w) & (ys >= 0) & (ys < self.h)
if not ok.any():
return
cols = (np.asarray(c0, dtype=np.float32) * (1 - t)[:, None] +
np.asarray(c1, dtype=np.float32) * t[:, None])
mask = np.zeros((self.h, self.w), dtype=bool)
mask[ys[ok], xs[ok]] = True
# Duplicated pixels along a steep line would blend twice; take the
# first colour per pixel instead, which is what a scanline does.
acc = np.zeros((self.h, self.w, 3), dtype=np.float32)
acc[ys[ok], xs[ok]] = cols[ok]
self._blend(mask, acc[mask], semi, stp)
def to_rgb_image(self):
return Image.fromarray(self.rgb.astype(np.uint8), "RGB")
def to_rgba_image(self):
rgba = np.zeros((self.h, self.w, 4), dtype=np.uint8)
rgba[..., :3] = self.rgb.astype(np.uint8)
rgba[..., 3] = np.where(self.cov, 255, 0)
return Image.fromarray(rgba, "RGBA")
def prim_color(p, tex_tint: bool):
if p.get("textured"):
if tex_tint:
return TEX_TINT
if p.get("raw_tex"):
return RAW_TEXTURE_GREY
return None
def draw_prim(canvas: Canvas, p, tex_tint: bool):
"""Rasterise one decoded primitive. Returns True if it put down pixels."""
kind = p["kind"]
if kind not in ("poly", "rect", "line", "fill"):
return False
verts = p.get("verts") or []
cols = p.get("colors") or []
if not verts:
return False
override = prim_color(p, tex_tint)
if override is not None:
cols = [list(override)] * len(verts)
while len(cols) < len(verts):
cols.append(cols[-1] if cols else [128, 128, 128])
semi = bool(p.get("semi"))
stp = int(p.get("stp", 0))
if kind == "poly":
if len(verts) >= 3:
canvas.triangle(verts[0:3], cols[0:3], semi, stp)
if len(verts) >= 4:
canvas.triangle([verts[1], verts[2], verts[3]],
[cols[1], cols[2], cols[3]], semi, stp)
return len(verts) >= 3
if kind in ("rect", "fill"):
size = p.get("size") or [1, 1]
# GP0(02) fill is opaque and ignores blending entirely.
f_semi = semi and kind != "fill"
canvas.rect(verts[0][0], verts[0][1], size[0], size[1], cols[0], f_semi, stp)
return True
if kind == "line":
drew = False
for i in range(len(verts) - 1):
canvas.line(verts[i], verts[i + 1], cols[i], cols[i + 1], semi, stp)
drew = True
return drew
return False
def safe_name(func: str) -> str:
return re.sub(r"[^0-9A-Za-z_.-]", "_", func)
def render(dump, out_dir, scale=1, tex_tint=False, order="issue",
max_layers=64, only=None, exclude=None, sheet=True,
layer_backdrop=LAYER_BACKDROP):
x0, y0, x1, y1 = draw_area(dump)
w, h = min(1024, x1 - x0), min(512, y1 - y0)
if w <= 0 or h <= 0:
raise SystemExit("frame has no drawable area")
prims = list(dump["prims"])
if order == "ot":
prims.sort(key=lambda p: (p.get("ot", 0xFFFF), p.get("seq", 0)))
if only:
prims = [p for p in prims if p.get("func") in only]
if exclude:
prims = [p for p in prims if p.get("func") not in exclude]
os.makedirs(out_dir, exist_ok=True)
composite = Canvas(x0, y0, w, h)
per_func = {}
for p in prims:
drew = draw_prim(composite, p, tex_tint)
if drew:
per_func.setdefault(p.get("func", "0x00000000"), []).append(p)
def save(img, name):
if scale > 1:
img = img.resize((img.width * scale, img.height * scale), Image.NEAREST)
path = os.path.join(out_dir, name)
img.save(path)
return path
save(composite.to_rgb_image(), "composite.png")
src_funcs = dump.get("funcs", {})
ranked = sorted(per_func.items(), key=lambda kv: -len(kv[1]))
index = {
"kind": "psx-gpu-layers",
"version": 1,
"frame": dump["frame"],
"label": dump.get("label", ""),
"area": [x0, y0, x1, y1],
"scale": scale,
"order": order,
"tex_tint": tex_tint,
"layer_backdrop": layer_backdrop,
"composite": "composite.png",
"sheet": "sheet.png" if sheet else None,
"rendered": sum(len(v) for v in per_func.values()),
"non_drawing": len(prims) - sum(len(v) for v in per_func.values()),
"layers": [],
}
thumbs = []
for func, plist in ranked[:max_layers]:
c = Canvas(x0, y0, w, h, backdrop=layer_backdrop)
for p in plist:
draw_prim(c, p, tex_tint)
name = f"layer-{safe_name(func)}.png"
save(c.to_rgba_image(), name)
stats = src_funcs.get(func, {})
bbox = None
for p in plist:
b = prim_bbox(p)
if b:
bbox = b if bbox is None else [min(bbox[0], b[0]), min(bbox[1], b[1]),
max(bbox[2], b[2]), max(bbox[3], b[3])]
modes = stats.get("stp_modes", {})
index["layers"].append({
"func": func,
"file": name,
"prims": len(plist),
"semi": sum(1 for p in plist if p.get("semi")),
"textured": sum(1 for p in plist if p.get("textured")),
"pixels": int(c.cov.sum()),
"bbox": bbox,
"ot_min": stats.get("ot_min"),
"ot_max": stats.get("ot_max"),
"ops": stats.get("ops", {}),
"stp_modes": modes,
"ras": stats.get("ras", []),
})
thumbs.append((func, c, len(plist), sum(1 for p in plist if p.get("semi"))))
if len(ranked) > max_layers:
index["truncated_layers"] = len(ranked) - max_layers
if sheet and thumbs:
save_sheet(thumbs, os.path.join(out_dir, "sheet.png"), w, h,
backdrop=layer_backdrop)
with open(os.path.join(out_dir, "layers.json"), "w", encoding="utf-8") as f:
json.dump(index, f, indent=1)
return index
def save_sheet(thumbs, path, w, h, cols=4, pad=8, label_h=16, backdrop=LAYER_BACKDROP):
"""Labelled contact sheet -- the whole frame's authorship on one page."""
tw = max(96, min(240, w))
th = max(1, int(round(h * tw / max(1, w))))
rows = (len(thumbs) + cols - 1) // cols
W = cols * (tw + pad) + pad
H = rows * (th + label_h + pad) + pad
sheet = Image.new("RGB", (W, H), (18, 20, 24))
d = ImageDraw.Draw(sheet)
for i, (func, c, n, semi) in enumerate(thumbs):
r, col = divmod(i, cols)
x = pad + col * (tw + pad)
y = pad + r * (th + label_h + pad)
bg = int(round(backdrop * 0.35))
tile = Image.new("RGB", (c.w, c.h), (bg, bg, bg))
rgba = c.to_rgba_image()
tile.paste(rgba.convert("RGB"), (0, 0), rgba.split()[3])
sheet.paste(tile.resize((tw, th), Image.NEAREST), (x, y))
d.rectangle([x, y, x + tw - 1, y + th - 1], outline=(70, 78, 90))
d.text((x + 2, y + th + 2), f"{func} {n}p" + (f" {semi} semi" if semi else ""),
fill=(200, 210, 220))
sheet.save(path)
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("dump", help="a psx-gpu-frame JSON written by gpu_frame_capture.py")
ap.add_argument("--out", default=None, help="output directory (default: <dump>-layers)")
ap.add_argument("--scale", type=int, default=1)
ap.add_argument("--order", choices=("issue", "ot"), default="issue",
help="issue order (what the GPU saw) or OT rank order")
ap.add_argument("--tex-tint", action="store_true",
help="draw textured prims flat magenta so texture coverage stands out")
ap.add_argument("--max-layers", type=int, default=64)
ap.add_argument("--only", action="append", default=None, metavar="0x8004ABCD")
ap.add_argument("--exclude", action="append", default=None, metavar="0x8004ABCD")
ap.add_argument("--no-sheet", action="store_true")
ap.add_argument("--layer-backdrop", type=float, default=LAYER_BACKDROP,
help="grey level layers are rendered over so additive and "
"subtractive blends stay visible (0 = black, as the "
"composite is drawn)")
args = ap.parse_args(argv)
dump = load_dump(args.dump)
out = args.out or os.path.splitext(args.dump)[0] + "-layers"
idx = render(dump, out, scale=args.scale, tex_tint=args.tex_tint, order=args.order,
max_layers=args.max_layers,
only=set(args.only) if args.only else None,
exclude=set(args.exclude) if args.exclude else None,
sheet=not args.no_sheet, layer_backdrop=args.layer_backdrop)
print(f"frame {idx['frame']} area={idx['area']} "
f"rendered={idx['rendered']} non-drawing={idx['non_drawing']} "
f"layer backdrop={idx['layer_backdrop']:g}")
print(f" {len(idx['layers'])} layer(s) -> {out}")
for l in idx["layers"][:20]:
modes = ", ".join(f"{k}x{v}" for k, v in l["stp_modes"].items()) or "-"
print(f" {l['func']:<12} {l['prims']:>5}p {l['semi']:>4}semi "
f"{l['pixels']:>8}px bbox={l['bbox']} stp={modes}")
if idx.get("truncated_layers"):
print(f" ... {idx['truncated_layers']} more function(s) not rendered "
f"(raise --max-layers)")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+246
View File
@@ -0,0 +1,246 @@
#!/usr/bin/env python3
"""gpu_parity.py -- frame-locked image parity between psx-runtime and DuckStation.
python3 gpu_parity.py --frame 41230 --out analysis/frames/parity-41230
Drives both emulators to the SAME guest frame, screenshots both, and reports how
far apart the presented images are. That answers the first question any render
bug has to answer:
images match -> the game code executed the same on both; a divergence
you can see on screen is in the renderer
images differ -> the guest ran differently; the GP0 stream is the place
to look, and tools/cosim.py will bracket where
Scope, stated honestly: this is IMAGE parity, not GP0-stream parity. The
DuckStation oracle patch (tools/duckstation/psxrecomp_oracle.patch) implements
screenshot / read_vram / gpu_state / run_to_frame / step / pause, but not
gpu_frame_dump, so there is no packet stream to compare on that side. Adding one
to the patch is the obvious next step if image parity keeps saying "the streams
must differ" without saying where.
Only DuckStation gets driven. psx-runtime removed pause / step / run_to_frame
(runtime/src/debug_server.c) -- it is designed to be read from, not steered --
so the native side is sampled where it already is, and DuckStation is advanced
to meet it. If the native run has already passed the frame DuckStation is on,
there is nothing to do but restart the DuckStation side; that is reported rather
than papered over.
Getting the oracle: `python3 duckstation_oracle.py all` builds and installs a
patched DuckStation into the RetComM data root (once, ~10 minutes), and
`duckstation_oracle.py start --disc <cue>` runs it headless on 4371. Pass
--start-oracle here to have this tool do that for you when nothing is listening.
Both instances must already be running with their debug servers up, on the same
disc, from the same starting state -- typically both loaded from the same
savestate, or both run from boot with the same input script. Parity between two
runs that were not driven identically means nothing.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import numpy as np # noqa: E402
from PIL import Image # noqa: E402
from psx_gpu_frame import ( # noqa: E402
DEFAULT_DUCKSTATION_PORT, DEFAULT_NATIVE_PORT, DebugConn, DebugError,
)
GPU_STATE_FIELDS = ("display_area_x", "display_area_y", "disp_w", "disp_h",
"da_x", "da_y", "display_enable")
def ensure_oracle(args) -> int:
"""Start the installed DuckStation oracle if nothing is answering yet.
Opt-in (--start-oracle), because launching an emulator as a side effect of
running a diff tool is a surprise nobody asked for. When it is not
requested, a missing oracle just reports how to get one.
"""
import subprocess as sp
probe = DebugConn(args.host, args.ds_port, timeout=2.0)
try:
probe.cmd("ping")
return 0
except DebugError:
pass
if not args.disc:
print("error: --start-oracle needs --disc (the same image psx-runtime "
"is running)", file=sys.stderr)
return 2
tool = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"duckstation_oracle.py")
cmd = [sys.executable, tool, "start", "--disc", args.disc,
"--port", str(args.ds_port), "--wait", "90"]
print("starting the DuckStation oracle: " + " ".join(cmd))
if sp.run(cmd).returncode != 0:
print("error: could not start the oracle. `python3 duckstation_oracle.py "
"status` will say what is missing.", file=sys.stderr)
return 2
return 0
def compare_images(pa, pb, out_path):
a = Image.open(pa).convert("RGB")
b = Image.open(pb).convert("RGB")
note = None
if a.size != b.size:
note = f"size mismatch: {a.size} vs {b.size}; compared over the common region"
w, h = min(a.width, b.width), min(a.height, b.height)
a, b = a.crop((0, 0, w, h)), b.crop((0, 0, w, h))
na = np.asarray(a, dtype=np.int16)
nb = np.asarray(b, dtype=np.int16)
d = np.abs(na - nb)
per_px = d.max(axis=2)
differing = int((per_px > 0).sum())
total = int(per_px.size)
# The PSX writes 5 bits per channel; the runtime and DuckStation both
# expand to 8 by shifting left 3, so an exact match is achievable and a
# 1-2 level difference is still a real difference, not rounding.
vis = np.zeros((*per_px.shape, 3), dtype=np.uint8)
vis[..., 0] = np.clip(per_px * 4, 0, 255)
vis[..., 1] = np.clip(per_px, 0, 255) // 2
Image.fromarray(vis, "RGB").save(out_path)
ys, xs = np.nonzero(per_px)
bbox = [int(xs.min()), int(ys.min()), int(xs.max()), int(ys.max())] if differing else None
return {
"width": a.width, "height": a.height,
"differing_pixels": differing,
"total_pixels": total,
"differing_pct": round(100.0 * differing / max(1, total), 4),
"max_channel_delta": int(d.max()),
"mean_abs_delta": round(float(d.mean()), 4),
"diff_bbox": bbox,
"note": note,
"diff_image": out_path,
}
def gpu_state_delta(a, b):
out = {}
for k in GPU_STATE_FIELDS:
if k in a or k in b:
av, bv = a.get(k), b.get(k)
if av != bv:
out[k] = {"native": av, "duckstation": bv}
return out
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--native-port", type=int, default=DEFAULT_NATIVE_PORT)
ap.add_argument("--ds-port", type=int, default=DEFAULT_DUCKSTATION_PORT)
ap.add_argument("--frame", type=int, default=None,
help="target guest frame (default: wherever the native run "
"already is). Only DuckStation is driven to it.")
ap.add_argument("--out", default="analysis/frames/parity")
ap.add_argument("--timeout", type=float, default=60.0)
ap.add_argument("--start-oracle", action="store_true",
help="if nothing answers on the DuckStation port, start the "
"installed oracle first (needs --disc)")
ap.add_argument("--disc", default=None,
help="disc image for --start-oracle; must be the same one "
"psx-runtime is running")
ap.add_argument("--dump-native", action="store_true",
help="also capture the native GP0 stream (no DuckStation counterpart)")
args = ap.parse_args(argv)
outdir = os.path.abspath(args.out)
os.makedirs(outdir, exist_ok=True)
result = {"kind": "psx-gpu-parity", "version": 1, "frame": args.frame}
if args.start_oracle:
rc = ensure_oracle(args)
if rc:
return rc
try:
with DebugConn(args.host, args.native_port, args.timeout) as native, \
DebugConn(args.host, args.ds_port, args.timeout) as ds:
# The native side is read where it is; only DuckStation is steered.
target = args.frame if args.frame is not None else native.frame()
result["target_frame"] = target
try:
ds.run_to_frame(target)
except DebugError as e:
result["duckstation_drive_error"] = str(e)
print(f"warning: could not drive DuckStation to frame {target}: {e}",
file=sys.stderr)
fa, fb = native.frame(), ds.frame()
result["native_frame"] = fa
result["duckstation_frame"] = fb
if fa != fb:
result["frame_mismatch"] = True
print(f"warning: frames differ (native {fa}, duckstation {fb}). "
f"The native run cannot be paused, so it keeps advancing; "
f"pass --frame to pin a target, or accept that the two "
f"images below are of DIFFERENT frames.", file=sys.stderr)
pa = os.path.join(outdir, "native.png")
pb = os.path.join(outdir, "duckstation.png")
native.screenshot(pa)
ds.screenshot(pb)
result["native_image"] = pa
result["duckstation_image"] = pb
try:
result["gpu_state"] = gpu_state_delta(native.cmd("gpu_state"),
ds.cmd("gpu_state"))
except DebugError as e:
result["gpu_state_error"] = str(e)
if args.dump_native:
from psx_gpu_frame import capture, save_dump
d = capture(native, frame=fa, label="native")
p = os.path.join(outdir, "native-gp0.json")
save_dump(d, p)
result["native_gp0"] = p
result["native_gp0_note"] = (
"no DuckStation counterpart: the oracle patch does not "
"implement gpu_frame_dump")
except DebugError as e:
print(f"error: {e}", file=sys.stderr)
return 2
result["image"] = compare_images(pa, pb, os.path.join(outdir, "diff.png"))
with open(os.path.join(outdir, "parity.json"), "w", encoding="utf-8") as f:
json.dump(result, f, indent=1)
img = result["image"]
print(f"native frame {result['native_frame']} duckstation frame "
f"{result['duckstation_frame']}")
if img["note"]:
print(f" {img['note']}")
print(f" {img['differing_pixels']}/{img['total_pixels']} pixels differ "
f"({img['differing_pct']}%), max channel delta {img['max_channel_delta']}")
if img["diff_bbox"]:
print(f" differences confined to bbox {img['diff_bbox']}")
if result.get("gpu_state"):
print(" display state differs:")
for k, v in result["gpu_state"].items():
print(f" {k}: native={v['native']} duckstation={v['duckstation']}")
print()
if img["differing_pixels"] == 0:
print("VERDICT: images identical -- the guest ran the same on both. A visible "
"render bug here is in the renderer, not the game code.")
else:
print("VERDICT: images differ -- the guest state or the GP0 stream diverged. "
"Capture both sides' frames and run tools/cosim.py to bracket where.")
print(f"wrote {outdir}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+751
View File
@@ -0,0 +1,751 @@
#!/usr/bin/env python3
"""psx_gpu_frame.py -- capture and decode a frame's GP0 stream, with the guest
function that issued every primitive.
Why this exists
---------------
`gpu_frame_dump` on the runtime's TCP debug server already stamps every GP0
packet with the guest code that issued it (`func`, `pc`, `ra`) and its linked-
list OT rank -- see runtime/src/debug_server.c, handle_gpu_frame_dump(), and
GpuGp0RingEntry in runtime/include/gpu.h. Nothing consumed that attribution.
This module turns the raw ring into decoded primitives, so "which guest
function drew this, and was the semi-transparency bit set" has a mechanical
answer instead of a guess.
Provenance rule
---------------
Everything here is OBSERVED from a running game and is labelled as such. It is
never merged into a static claim. A `func` appearing in a frame dump is
evidence that code executed and issued a primitive -- it is not evidence about
the call graph, and the analyzer's static coverage gap stays exactly as wide as
it was.
Fidelity, stated up front
-------------------------
* Vertices are decoded exactly and the running GP0(E5) draw offset is applied,
because that is what the rasteriser applies (gpu.c, gp0_exec_* ).
* The texpage latch follows hardware: a textured polygon's own tpage word
overwrites draw-mode state (gpu.c set_tpage_from_poly), and later untextured
prims and rectangles consume whatever it left behind. Getting this wrong
mislabels exactly the semi-transparency mode you are usually chasing.
* Textures are NOT sampled. Texture pages, CLUTs and UVs are decoded and
reported, but a textured primitive renders as its command colour. The goal is
attribution and blend-state triage, not a second GPU implementation.
* The ring truncates each packet to GPU_GP0_RING_MAX_WORDS (12). A packet
longer than that is decoded as far as it goes and marked `truncated`.
"""
from __future__ import annotations
import json
import socket
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
DUMP_KIND = "psx-gpu-frame"
DUMP_VERSION = 1
DEFAULT_NATIVE_PORT = 4370
DEFAULT_DUCKSTATION_PORT = 4371
# Semi-transparency modes, GPUSTAT bits 5-6. The names are the blend the
# hardware performs, because that is what you compare against the screenshot.
STP_MODES = {
0: "0.5B+0.5F",
1: "B+F",
2: "B-F",
3: "B+0.25F",
}
class DebugError(RuntimeError):
pass
class DebugConn:
"""JSON client for the runtime (and DuckStation) debug server.
ONE REQUEST PER CONNECTION. That is the server's actual contract, not a
conservative choice: runtime/src/debug_server.c's io_thread_main() accepts,
reads a single line, replies, and sock_close()s. A client that holds the
socket open and sends a second command gets silence and then EOF, which
looks exactly like a hung emulator. (`s_client` in that file is vestigial;
nothing ever assigns it a live socket.) tools/debug_client.py's query()
documents the same contract -- "connect, send, receive, close".
So each command opens its own socket and reads to EOF. Replies can be
megabytes (a full gpu_frame_dump), and the server's send is blocking with a
starvation watchdog behind it, so the read drains in large chunks and never
parses incrementally.
"""
def __init__(self, host: str = "127.0.0.1", port: int = DEFAULT_NATIVE_PORT,
timeout: float = 15.0):
self.host = host
self.port = port
self.timeout = timeout
self._next_id = 1
def __enter__(self) -> "DebugConn":
return self
def __exit__(self, *exc: Any) -> None:
self.close()
def connect(self) -> None:
"""Probe that a server is there. Not required before cmd()."""
try:
socket.create_connection((self.host, self.port),
timeout=self.timeout).close()
except OSError as e:
raise DebugError(f"no debug server on {self.host}:{self.port} ({e})") from e
def close(self) -> None:
"""No persistent socket to close; kept so callers can use `with`."""
def cmd(self, name: str, **fields: Any) -> Dict[str, Any]:
"""Send one command, return the parsed reply. Raises on ok:false."""
reply = self.raw(name, **fields)
if not reply.get("ok", False):
raise DebugError(f"{name}: {reply.get('error', reply.get('err', reply))}")
return reply
def raw(self, name: str, **fields: Any) -> Dict[str, Any]:
"""Like cmd() but returns ok:false replies instead of raising."""
req = {"id": self._next_id, "cmd": name}
self._next_id += 1
req.update(fields)
line = (json.dumps(req) + "\n").encode()
try:
sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
except OSError as e:
raise DebugError(f"{name}: no debug server on "
f"{self.host}:{self.port} ({e})") from e
try:
sock.settimeout(self.timeout)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
try:
sock.sendall(line)
except OSError as e:
raise DebugError(f"{name}: send failed ({e})") from e
buf = bytearray()
while True:
try:
chunk = sock.recv(1 << 20)
except socket.timeout as e:
raise DebugError(
f"{name}: timed out after {self.timeout}s "
f"({len(buf)} bytes received)") from e
except OSError as e:
raise DebugError(f"{name}: recv failed ({e})") from e
if not chunk:
break
buf += chunk
# The server ends every reply with a newline and then closes.
# Stopping at the newline avoids waiting on the close for peers
# (DuckStation) that keep the socket around.
if buf.endswith(b"\n") or b"\n" in buf:
break
finally:
sock.close()
if not buf:
raise DebugError(f"{name}: server closed without replying")
head = bytes(buf).split(b"\n", 1)[0]
try:
return json.loads(head.decode("utf-8", "replace"))
except json.JSONDecodeError as e:
raise DebugError(f"{name}: malformed reply ({e}); "
f"first 200 bytes: {head[:200]!r}") from e
# -- convenience wrappers used by the CLIs ------------------------------
def frame(self) -> int:
return int(self.cmd("ping").get("frame", 0))
def ring_span(self) -> Dict[str, Any]:
"""Which frames the GP0 ring can still be asked for.
This is the replacement for pausing. The ring holds ~1M packets, which
is several hundred frames, so the workflow is: let the game run, and
capture the frame AFTER you have seen the bug -- along with the frames
leading up to it. Reaching backwards beats freezing forwards.
"""
r = self.cmd("gpu_ring_stats")
return {
"oldest": int(r.get("oldest_frame", 0)),
"newest": int(r.get("newest_frame", 0)),
"total": int(r.get("total", 0)),
"capacity": int(r.get("capacity", 0)),
"max_words": int(r.get("max_words", 12)),
}
# pause / continue / step / run_to_frame were REMOVED from psx-runtime --
# runtime/src/debug_server.c registers them only as handlers that return an
# error, because pause-step-read synthesizes a snapshot instead of reading
# the history the runtime already records, and it once turned a dropped
# client into an apparent freeze. They still work on the DuckStation oracle,
# which is why these wrappers exist at all; against psx-runtime they raise
# with the server's own migration message.
def run_to_frame(self, frame: int) -> Dict[str, Any]:
"""DuckStation only. Raises against psx-runtime, which removed it."""
return self.cmd("run_to_frame", frame=int(frame))
def screenshot(self, path: str) -> Dict[str, Any]:
"""Write a PNG of the presented frame.
The native server calls this `screenshot_file`; the DuckStation oracle
patch calls the same operation `screenshot`. Try both so one caller
works against either end.
"""
r = self.raw("screenshot_file", path=path)
if r.get("ok"):
return r
r2 = self.raw("screenshot", path=path)
if r2.get("ok"):
return r2
raise DebugError(f"screenshot: {r2.get('error', r.get('error', 'failed'))}")
# ---------------------------------------------------------------------------
# GP0 decoding
# ---------------------------------------------------------------------------
def _sx11(v: int) -> int:
"""Sign-extend an 11-bit GPU coordinate."""
v &= 0x7FF
return v - 0x800 if v & 0x400 else v
def parse_vertex(word: int) -> Tuple[int, int]:
return _sx11(word & 0xFFFF), _sx11((word >> 16) & 0xFFFF)
def parse_color(word: int) -> Tuple[int, int, int]:
return word & 0xFF, (word >> 8) & 0xFF, (word >> 16) & 0xFF
def poly_flags(op: int) -> Dict[str, bool]:
return {
"gouraud": bool(op & 0x10),
"quad": bool(op & 0x08),
"textured": bool(op & 0x04),
"semi": bool(op & 0x02),
"raw_tex": bool(op & 0x01),
}
def poly_words(op: int) -> int:
f = poly_flags(op)
n = 4 if f["quad"] else 3
return 1 + n * (2 if f["textured"] else 1) + (n - 1 if f["gouraud"] else 0)
def rect_words(op: int) -> int:
variable = ((op >> 3) & 3) == 0
return 1 + 1 + (1 if op & 0x04 else 0) + (1 if variable else 0)
RECT_SIZES = {1: (1, 1), 2: (8, 8), 3: (16, 16)}
def op_name(op: int) -> str:
"""Human name for a GP0 opcode. Primitive names spell out their flags,
because the flags are the diagnosis."""
if op == 0x00:
return "nop"
if op == 0x01:
return "clear_cache"
if op == 0x02:
return "fill_rect"
if op == 0x03:
return "unknown_03"
if 0x20 <= op <= 0x3F:
f = poly_flags(op)
s = "PolyG" if f["gouraud"] else "PolyF"
if f["textured"]:
s += "T"
s += "4" if f["quad"] else "3"
if f["semi"]:
s += "+semi"
if f["raw_tex"] and f["textured"]:
s += "+raw"
return s
if 0x40 <= op <= 0x5F:
s = "LineG" if op & 0x10 else "LineF"
if op & 0x08:
s = "Poly" + s
if op & 0x02:
s += "+semi"
return s
if 0x60 <= op <= 0x7F:
size = (op >> 3) & 3
s = {0: "Rect", 1: "Rect1", 2: "Rect8", 3: "Rect16"}[size]
if op & 0x04:
s = "Sprite" + s[4:]
if op & 0x02:
s += "+semi"
if op & 0x01 and op & 0x04:
s += "+raw"
return s
if 0x80 <= op <= 0x9F:
return "copy_vram_vram"
if 0xA0 <= op <= 0xBF:
return "copy_cpu_vram"
if 0xC0 <= op <= 0xDF:
return "copy_vram_cpu"
return {
0xE1: "draw_mode(E1)",
0xE2: "tex_window(E2)",
0xE3: "draw_area_tl(E3)",
0xE4: "draw_area_br(E4)",
0xE5: "draw_offset(E5)",
0xE6: "mask_bit(E6)",
}.get(op, f"op_{op:02X}")
def op_kind(op: int) -> str:
if 0x20 <= op <= 0x3F:
return "poly"
if 0x40 <= op <= 0x5F:
return "line"
if 0x60 <= op <= 0x7F:
return "rect"
if op == 0x02:
return "fill"
if 0x80 <= op <= 0xDF:
return "copy"
if 0xE0 <= op <= 0xEF:
return "env"
return "other"
class GpuState:
"""Draw-mode state carried between packets while walking a frame.
This is the part that makes the decode faithful rather than plausible: the
semi-transparency mode of an untextured polygon or a rectangle comes from
whatever the last GP0(E1) -- or the last textured polygon's own tpage word
-- left in draw-mode state, not from the packet itself.
"""
def __init__(self) -> None:
self.offset = (0, 0)
self.area_tl = (0, 0)
self.area_br = (1023, 511)
self.stp = 0
self.texpage_x = 0
self.texpage_y = 0
self.tex_colors = 0
self.mask_set = False
self.mask_check = False
def set_tpage(self, tpage_word: int) -> None:
self.texpage_x = tpage_word & 0xF
self.texpage_y = (tpage_word >> 4) & 1
self.stp = (tpage_word >> 5) & 3
self.tex_colors = (tpage_word >> 7) & 3
def apply_env(self, op: int, w0: int) -> None:
if op == 0xE1:
self.set_tpage(w0 & 0xFFFF)
elif op == 0xE3:
self.area_tl = (w0 & 0x3FF, (w0 >> 10) & 0x1FF)
elif op == 0xE4:
self.area_br = (w0 & 0x3FF, (w0 >> 10) & 0x1FF)
elif op == 0xE5:
self.offset = (_sx11(w0 & 0x7FF), _sx11((w0 >> 11) & 0x7FF))
elif op == 0xE6:
self.mask_set = bool(w0 & 1)
self.mask_check = bool(w0 & 2)
def snapshot(self) -> Dict[str, Any]:
return {
"offset": list(self.offset),
"area": [self.area_tl[0], self.area_tl[1], self.area_br[0], self.area_br[1]],
"stp": self.stp,
"texpage": [self.texpage_x, self.texpage_y, self.tex_colors],
"mask": [self.mask_set, self.mask_check],
}
def _hexw(words: Sequence[Any]) -> List[int]:
out = []
for w in words:
out.append(int(w, 16) if isinstance(w, str) else int(w))
return out
def decode_entry(entry: Dict[str, Any], state: GpuState) -> Dict[str, Any]:
"""Decode one gpu_frame_dump ring entry against the running draw state.
`state` is mutated: env commands and textured-poly tpage latches change it
for everything that follows, exactly as on hardware.
"""
op = int(entry["op"], 16) if isinstance(entry["op"], str) else int(entry["op"])
words = _hexw(entry.get("w", []))
n_words = int(entry.get("n", len(words)))
kind = op_kind(op)
prim: Dict[str, Any] = {
"seq": int(entry.get("seq", 0)),
"op": op,
"op_name": op_name(op),
"kind": kind,
"n_words": n_words,
"have_words": len(words),
"truncated": n_words > len(words),
"func": entry.get("func", "0x00000000"),
"pc": entry.get("pc", "0x00000000"),
"ra": entry.get("ra", "0x00000000"),
"src": entry.get("src", "0x00000000"),
"ot": int(entry.get("ot", 0xFFFF)),
"words": [f"0x{w:08X}" for w in words],
"verts": [],
"colors": [],
"uvs": [],
"semi": False,
"stp": state.stp,
"textured": False,
"raw_tex": False,
"gouraud": False,
"size": None,
"clut": None,
"tpage": None,
}
if not words:
prim["state"] = state.snapshot()
return prim
if kind == "env":
state.apply_env(op, words[0])
prim["state"] = state.snapshot()
prim["stp"] = state.stp
return prim
ox, oy = state.offset
if kind == "poly":
f = poly_flags(op)
n = 4 if f["quad"] else 3
prim.update(semi=f["semi"], textured=f["textured"],
raw_tex=f["raw_tex"] and f["textured"], gouraud=f["gouraud"])
# Hardware latches the poly's own tpage word into draw-mode state, so
# do it before reading stp -- and do it even if the packet is truncated
# past the vertices, because the real GPU latched it too.
tp_idx = 5 if f["gouraud"] else 4
if f["textured"] and len(words) > tp_idx:
tpage_word = (words[tp_idx] >> 16) & 0xFFFF
state.set_tpage(tpage_word)
prim["tpage"] = tpage_word
prim["stp"] = state.stp
i = 0
color = parse_color(words[0])
i = 1
for v in range(n):
if f["gouraud"] and v > 0:
if i >= len(words):
break
color = parse_color(words[i])
i += 1
if i >= len(words):
break
prim["verts"].append([parse_vertex(words[i])[0] + ox,
parse_vertex(words[i])[1] + oy])
prim["colors"].append(list(color))
i += 1
if f["textured"]:
if i >= len(words):
break
uvw = words[i]
prim["uvs"].append([uvw & 0xFF, (uvw >> 8) & 0xFF])
if v == 0:
prim["clut"] = (uvw >> 16) & 0xFFFF
i += 1
elif kind == "rect":
variable = ((op >> 3) & 3) == 0
prim.update(semi=bool(op & 0x02), textured=bool(op & 0x04),
raw_tex=bool(op & 0x01) and bool(op & 0x04))
prim["stp"] = state.stp
color = parse_color(words[0])
if len(words) > 1:
x, y = parse_vertex(words[1])
prim["verts"].append([x + ox, y + oy])
prim["colors"].append(list(color))
i = 2
if prim["textured"] and len(words) > i:
uvw = words[i]
prim["uvs"].append([uvw & 0xFF, (uvw >> 8) & 0xFF])
prim["clut"] = (uvw >> 16) & 0xFFFF
i += 1
if variable:
if len(words) > i:
prim["size"] = [words[i] & 0x3FF, (words[i] >> 16) & 0x1FF]
else:
prim["size"] = list(RECT_SIZES[(op >> 3) & 3])
elif kind == "line":
gouraud = bool(op & 0x10)
polyline = bool(op & 0x08)
prim.update(semi=bool(op & 0x02), gouraud=gouraud)
prim["stp"] = state.stp
prim["polyline"] = polyline
i = 0
color = parse_color(words[0])
i = 1
while i < len(words):
if words[i] == 0x55555555:
break
if gouraud and prim["verts"]:
color = parse_color(words[i])
i += 1
if i >= len(words):
break
x, y = parse_vertex(words[i])
prim["verts"].append([x + ox, y + oy])
prim["colors"].append(list(color))
i += 1
elif kind == "fill":
# GP0(02): fill is in raw VRAM coordinates -- the draw offset and the
# draw area do NOT apply, and it ignores the mask bit.
prim["colors"].append(list(parse_color(words[0])))
if len(words) > 1:
prim["verts"].append([words[1] & 0x3FF, (words[1] >> 16) & 0x1FF])
if len(words) > 2:
prim["size"] = [words[2] & 0x3FF, (words[2] >> 16) & 0x1FF]
elif kind == "copy":
if len(words) > 1:
prim["verts"].append([words[1] & 0x3FF, (words[1] >> 16) & 0x1FF])
if op < 0xA0 and len(words) > 2:
prim["verts"].append([words[2] & 0x3FF, (words[2] >> 16) & 0x1FF])
if len(words) > 3:
prim["size"] = [words[3] & 0x3FF, (words[3] >> 16) & 0x1FF]
elif len(words) > 2:
prim["size"] = [words[2] & 0x3FF, (words[2] >> 16) & 0x1FF]
if entry.get("bld"):
prim["bld"] = entry["bld"]
prim["state"] = state.snapshot()
return prim
def decode_entries(entries: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Decode a whole frame in issue order. Order matters -- see GpuState."""
state = GpuState()
ordered = sorted(entries, key=lambda e: int(e.get("seq", 0)))
return [decode_entry(e, state) for e in ordered]
# ---------------------------------------------------------------------------
# Attribution
# ---------------------------------------------------------------------------
def attribute(prims: Sequence[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
"""Group decoded primitives by the guest function that issued them.
The per-function numbers are the whole point: a function whose primitives
all lost their semi-transparency bit between a good and a bad frame is the
function to go read.
"""
out: Dict[str, Dict[str, Any]] = {}
for p in prims:
fn = p.get("func", "0x00000000")
rec = out.setdefault(fn, {
"func": fn,
"packets": 0,
"drawing": 0,
"semi": 0,
"textured": 0,
"ops": {},
"stp_modes": {},
"ot_min": None,
"ot_max": None,
"ras": [],
"bbox": None,
})
rec["packets"] += 1
name = p["op_name"]
rec["ops"][name] = rec["ops"].get(name, 0) + 1
ra = p.get("ra")
if ra and ra not in rec["ras"] and len(rec["ras"]) < 16:
rec["ras"].append(ra)
if p["kind"] in ("poly", "rect", "line", "fill"):
rec["drawing"] += 1
if p.get("semi"):
rec["semi"] += 1
m = STP_MODES.get(p.get("stp", 0), "?")
rec["stp_modes"][m] = rec["stp_modes"].get(m, 0) + 1
if p.get("textured"):
rec["textured"] += 1
ot = p.get("ot", 0xFFFF)
if ot != 0xFFFF:
rec["ot_min"] = ot if rec["ot_min"] is None else min(rec["ot_min"], ot)
rec["ot_max"] = ot if rec["ot_max"] is None else max(rec["ot_max"], ot)
bb = prim_bbox(p)
if bb:
rec["bbox"] = bb if rec["bbox"] is None else _union(rec["bbox"], bb)
return out
def prim_bbox(p: Dict[str, Any]) -> Optional[List[int]]:
verts = p.get("verts") or []
if not verts:
return None
xs = [v[0] for v in verts]
ys = [v[1] for v in verts]
x0, y0, x1, y1 = min(xs), min(ys), max(xs), max(ys)
size = p.get("size")
if size and p["kind"] in ("rect", "fill"):
x1 = x0 + max(0, size[0] - 1)
y1 = y0 + max(0, size[1] - 1)
return [x0, y0, x1, y1]
def _union(a: Sequence[int], b: Sequence[int]) -> List[int]:
return [min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3])]
# ---------------------------------------------------------------------------
# Capture / dump I/O
# ---------------------------------------------------------------------------
def capture(conn: DebugConn, frame: Optional[int] = None, count: int = 65536,
label: str = "", verify_ring: bool = True) -> Dict[str, Any]:
"""Pull one frame's GP0 ring and decode it into a dump dict.
Two failure modes are turned into errors rather than empty dumps, because
both look exactly like "that frame drew nothing", which is a conclusion you
might act on:
* the frame has fallen out of the ring (or has not happened yet), and
* the peer answered ok with no `entries` key at all -- which is what a
stub or a mismatched server does.
"""
span = None
if verify_ring:
try:
span = conn.ring_span()
except DebugError:
span = None # older server without gpu_ring_stats; carry on
if frame is None:
frame = span["newest"] if span else conn.frame()
if span and span["total"] > 0 and not (span["oldest"] <= frame <= span["newest"]):
raise DebugError(
f"frame {frame} is not in the GP0 ring (it holds {span['oldest']}.."
f"{span['newest']}). Capture closer to the moment, or raise the ring "
f"size; a dump of an evicted frame is empty, not zero-drawn.")
reply = conn.cmd("gpu_frame_dump", frame=int(frame), count=int(count))
if "entries" not in reply:
raise DebugError(
"gpu_frame_dump replied ok but carried no 'entries' -- this is not a "
"psx-runtime debug server. Check what is actually listening on "
f"{conn.host}:{conn.port}.")
entries = reply.get("entries", [])
prims = decode_entries(entries)
raw_count = int(reply.get("count", len(entries)))
return {
"kind": DUMP_KIND,
"version": DUMP_VERSION,
"label": label,
"frame": int(reply.get("frame", frame)),
"source": {"host": conn.host, "port": conn.port},
"ring": span,
"raw_count": raw_count,
"requested": int(count),
"capped": raw_count >= int(count),
"max_words": int(reply.get("max_words", 12)),
"prims": prims,
"funcs": attribute(prims),
}
SUMMARY_KIND = "psx-gpu-frame-summary"
def summarise_dump(dump: Dict[str, Any], dump_name: str = "") -> Dict[str, Any]:
"""A compact view of a dump: totals, opcode and blend-mode histograms, and
the per-function attribution -- everything except the primitives.
A busy frame's full dump runs to tens of megabytes. Nothing that only wants
to know *who drew what* should have to parse that, so the summary is a
separate artifact and RetComM Studio reads this rather than the dump.
"""
ops: Dict[str, int] = {}
modes: Dict[str, int] = {}
drawing = 0
truncated = 0
for p in dump.get("prims", []):
ops[p["op_name"]] = ops.get(p["op_name"], 0) + 1
if p.get("truncated"):
truncated += 1
if p["kind"] in ("poly", "rect", "line", "fill"):
drawing += 1
if p.get("semi"):
m = STP_MODES.get(p.get("stp", 0), "?")
modes[m] = modes.get(m, 0) + 1
funcs = sorted(dump.get("funcs", {}).values(),
key=lambda r: (-r.get("drawing", 0), r.get("func", "")))
return {
"kind": SUMMARY_KIND,
"version": 1,
"label": dump.get("label", ""),
"frame": dump.get("frame", 0),
"dump": dump_name,
"packets": dump.get("raw_count", 0),
"drawing": drawing,
"capped": bool(dump.get("capped")),
"truncated": truncated,
"area": list(draw_area(dump)),
"ops": ops,
"modes": modes,
"funcs": funcs,
}
def save_summary(dump: Dict[str, Any], path: str, dump_name: str = "") -> Dict[str, Any]:
summary = summarise_dump(dump, dump_name)
with open(path, "w", encoding="utf-8") as f:
json.dump(summary, f, indent=1)
return summary
def save_dump(dump: Dict[str, Any], path: str) -> None:
with open(path, "w", encoding="utf-8") as f:
json.dump(dump, f, indent=1)
def load_dump(path: str) -> Dict[str, Any]:
with open(path, "r", encoding="utf-8") as f:
dump = json.load(f)
if dump.get("kind") != DUMP_KIND:
raise ValueError(f"{path}: not a {DUMP_KIND} dump")
return dump
def draw_area(dump: Dict[str, Any]) -> Tuple[int, int, int, int]:
"""The frame's drawing clip rect, from the last GP0(E3)/GP0(E4) seen.
Falls back to the bounding box of everything drawn, then to 320x240, so a
dump that never set a draw area still renders something honest.
"""
area = None
for p in dump.get("prims", []):
st = p.get("state")
if st and st.get("area"):
area = st["area"]
if area and area[2] > area[0] and area[3] > area[1]:
return int(area[0]), int(area[1]), int(area[2]) + 1, int(area[3]) + 1
bb = None
for p in dump.get("prims", []):
b = prim_bbox(p)
if b:
bb = b if bb is None else _union(bb, b)
if bb:
return max(0, bb[0]), max(0, bb[1]), min(1024, bb[2] + 1), min(512, bb[3] + 1)
return 0, 0, 320, 240
+278
View File
@@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""Tests for duckstation_oracle.py's decision logic.
Nothing here builds or downloads anything. What is pinned is the reasoning that
is easy to get wrong and expensive to discover late: where the install lands,
which hosts upstream refuses, that the dependency bundle is verified against the
checkout's own checksum file rather than a hardcoded one, and that the settings
merge preserves DuckStation's own defaults.
"""
import importlib.util
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
_spec = importlib.util.spec_from_file_location("dso", ROOT / "duckstation_oracle.py")
DSO = importlib.util.module_from_spec(_spec)
assert _spec.loader is not None
_spec.loader.exec_module(DSO)
class EnvGuard(unittest.TestCase):
def setUp(self):
self._env = dict(os.environ)
def tearDown(self):
os.environ.clear()
os.environ.update(self._env)
class TestLocations(EnvGuard):
def test_install_is_outside_any_game_repo(self):
os.environ.pop("RETCOMM_ORACLE_DIR", None)
os.environ.pop("RETCOMM_DATA_DIR", None)
root = DSO.oracle_root()
self.assertEqual(root.parts[-2:], ("oracle", "duckstation"))
# The whole point: one build shared by every title, not per-repo.
self.assertIn("retcomm", str(root))
self.assertNotIn("psxrecomp", str(root))
def test_data_dir_override_moves_everything(self):
os.environ.pop("RETCOMM_ORACLE_DIR", None)
os.environ["RETCOMM_DATA_DIR"] = "/tmp/rc-test"
self.assertEqual(DSO.oracle_root(), Path("/tmp/rc-test/oracle/duckstation"))
self.assertEqual(DSO.download_cache(), Path("/tmp/rc-test/downloads"))
def test_xdg_data_home_is_honoured_like_studio_does(self):
"""studio_runner.cpp's retcomm_data_dir() checks XDG_DATA_HOME before
~/.local/share. If this tool did not, the GUI would look for an oracle
somewhere other than where the tool installed it, and neither side would
look wrong."""
os.environ.pop("RETCOMM_ORACLE_DIR", None)
os.environ.pop("RETCOMM_DATA_DIR", None)
os.environ["XDG_DATA_HOME"] = "/tmp/xdg-test"
self.assertEqual(DSO.data_root(), Path("/tmp/xdg-test/retcomm"))
self.assertEqual(DSO.oracle_root(),
Path("/tmp/xdg-test/retcomm/oracle/duckstation"))
def test_retcomm_data_dir_beats_xdg(self):
os.environ.pop("RETCOMM_ORACLE_DIR", None)
os.environ["XDG_DATA_HOME"] = "/tmp/xdg-test"
os.environ["RETCOMM_DATA_DIR"] = "/tmp/explicit"
self.assertEqual(DSO.data_root(), Path("/tmp/explicit"))
def test_oracle_dir_override_wins(self):
os.environ["RETCOMM_DATA_DIR"] = "/tmp/rc-test"
os.environ["RETCOMM_ORACLE_DIR"] = "/tmp/elsewhere"
self.assertEqual(DSO.oracle_root(), Path("/tmp/elsewhere"))
def test_layout_paths_hang_off_the_root(self):
lay = DSO.Layout(Path("/tmp/o"))
self.assertEqual(lay.src, Path("/tmp/o/src"))
self.assertEqual(lay.build, Path("/tmp/o/build"))
self.assertEqual(lay.app, Path("/tmp/o/app"))
self.assertEqual(lay.built_binary, Path("/tmp/o/build/bin") / DSO.exe_name())
self.assertEqual(lay.binary, Path("/tmp/o/app") / DSO.exe_name())
class TestRefusedEnvironment(EnvGuard):
"""Upstream's CMake refuses Arch-family and NixOS hosts. We predict that
rather than letting it surface as 'Unsupported environment.' 20 seconds into
a configure, and we never patch the check out."""
def _osr(self, text):
tmp = tempfile.NamedTemporaryFile("w", suffix=".osr", delete=False)
tmp.write(text)
tmp.close()
self.addCleanup(os.unlink, tmp.name)
return Path(tmp.name)
def test_detects_each_refused_marker(self):
# Literal substrings, matching what upstream's CMake matches
# (MATCHES "ID_LIKE=arch"). Mirroring it exactly matters: if upstream
# would NOT refuse a host, we must not route that host to a container.
for marker in ("ID=arch\n", "ID=cachyos\nID_LIKE=arch\n", "ID=nixos\n"):
with self.subTest(marker=marker):
osr = self._osr(marker)
orig = DSO.Path
try:
DSO.Path = lambda p, _o=orig, _f=osr: _f if p == "/etc/os-release" else _o(p)
self.assertIsNotNone(DSO.refused_environment())
finally:
DSO.Path = orig
def test_plain_distro_is_not_refused(self):
osr = self._osr('ID=ubuntu\nID_LIKE="debian"\n')
orig = DSO.Path
try:
DSO.Path = lambda p, _o=orig, _f=osr: _f if p == "/etc/os-release" else _o(p)
for var in ("NIX_BUILD_TOP", "NIX_STORE", "IN_NIX_SHELL"):
os.environ.pop(var, None)
self.assertIsNone(DSO.refused_environment())
finally:
DSO.Path = orig
def test_nix_env_vars_are_refused_too(self):
osr = self._osr("ID=ubuntu\n")
orig = DSO.Path
try:
DSO.Path = lambda p, _o=orig, _f=osr: _f if p == "/etc/os-release" else _o(p)
os.environ["IN_NIX_SHELL"] = "1"
self.assertIn("IN_NIX_SHELL", DSO.refused_environment() or "")
finally:
DSO.Path = orig
class TestDepsVerification(unittest.TestCase):
def test_checksum_comes_from_the_checkout_not_a_constant(self):
"""A hardcoded hash pairs a new source tree with an old bundle the first
time the pin moves. Read it from the tree that is actually checked out."""
with tempfile.TemporaryDirectory() as td:
src = Path(td)
(src / "dep").mkdir()
(src / "dep" / "PREBUILT-SHA256SUMS").write_text(
"aaaa *deps-linux-x64.tar.xz\n"
"bbbb *deps-windows-x64.7z\n")
self.assertEqual(DSO.expected_sha(src, "deps-linux-x64.tar.xz"), "aaaa")
self.assertEqual(DSO.expected_sha(src, "deps-windows-x64.7z"), "bbbb")
self.assertIsNone(DSO.expected_sha(src, "deps-macos-universal.tar.xz"))
def test_missing_sums_file_is_not_fatal(self):
with tempfile.TemporaryDirectory() as td:
self.assertIsNone(DSO.expected_sha(Path(td), "deps-linux-x64.tar.xz"))
def test_platform_bundle_selection(self):
archive, dirname = DSO.deps_platform()
self.assertTrue(archive.startswith("deps-"))
self.assertIn(dirname, ("linux-x64", "windows-x64", "macos-universal"))
class TestSettingsMerge(unittest.TestCase):
def test_merge_preserves_duckstations_own_defaults(self):
"""We override a handful of fields; the emulator owns the rest of the
schema. Rewriting the whole file would mean guessing at it."""
with tempfile.TemporaryDirectory() as td:
p = Path(td) / "settings.ini"
p.write_text("[Main]\nEmulationSpeed = 1\nSaveStateOnExit = true\n\n"
"[Display]\nVSync = false\n")
DSO.merge_ini(p, {"Main": {"ConfirmPowerOff": "false"},
"UI": {"UnofficialBuildWarningConfirmed": "true"}})
out = p.read_text()
self.assertIn("EmulationSpeed = 1", out) # untouched
self.assertIn("VSync = false", out) # untouched section
self.assertIn("ConfirmPowerOff = false", out) # added
self.assertIn("UnofficialBuildWarningConfirmed = true", out)
def test_merge_overwrites_a_key_it_owns(self):
with tempfile.TemporaryDirectory() as td:
p = Path(td) / "settings.ini"
p.write_text("[Main]\nConfirmPowerOff = true\n")
DSO.merge_ini(p, {"Main": {"ConfirmPowerOff": "false"}})
self.assertIn("ConfirmPowerOff = false", p.read_text())
def test_keys_stay_case_sensitive(self):
# DuckStation reads CamelCase keys; configparser lowercases by default,
# which would silently produce a file it ignores.
with tempfile.TemporaryDirectory() as td:
p = Path(td) / "settings.ini"
DSO.merge_ini(p, {"BIOS": {"PathNTSC-U": "SCPH1001.BIN"}})
self.assertIn("PathNTSC-U", p.read_text())
def test_merge_creates_the_file_when_absent(self):
with tempfile.TemporaryDirectory() as td:
p = Path(td) / "settings.ini"
DSO.merge_ini(p, {"GPU": {"Renderer": "Software"}})
self.assertIn("Renderer = Software", p.read_text())
class TestStatusDoc(EnvGuard):
"""The status document is the whole contract with RetComM Studio."""
def test_absent_install_reports_absent(self):
os.environ["RETCOMM_ORACLE_DIR"] = tempfile.mkdtemp()
doc = DSO.status_doc(DSO.Layout())
self.assertEqual(doc["kind"], "psxrecomp-oracle-status")
self.assertEqual(doc["state"], "absent")
for k in ("source", "deps", "built", "installed", "running", "answering"):
self.assertFalse(doc[k], k)
self.assertIn("port", doc)
self.assertIn("container_needed", doc)
def test_state_ladder_is_ordered(self):
"""A GUI picks its button off `state`, so the ladder must not skip."""
os.environ["RETCOMM_ORACLE_DIR"] = tempfile.mkdtemp()
lay = DSO.Layout()
lay.src.mkdir(parents=True)
(lay.src / "CMakeLists.txt").write_text("")
self.assertEqual(DSO.status_doc(lay)["state"], "fetched")
(lay.build / "bin").mkdir(parents=True)
(lay.build / "bin" / DSO.exe_name()).write_text("")
self.assertEqual(DSO.status_doc(lay)["state"], "built")
lay.app.mkdir(parents=True)
(lay.app / DSO.exe_name()).write_text("")
self.assertEqual(DSO.status_doc(lay)["state"], "installed")
def test_kind_survives_an_existing_manifest(self):
"""oracle.json has its own `kind`. Merging it must not overwrite the
status document's, or every INSTALLED oracle reports a document Studio
rejects as foreign — while an uninstalled one parses fine, so the bug
only appears once things are working."""
os.environ["RETCOMM_ORACLE_DIR"] = tempfile.mkdtemp()
lay = DSO.Layout()
lay.root.mkdir(parents=True, exist_ok=True)
lay.manifest.write_text(json.dumps({
"kind": "psxrecomp-duckstation-oracle",
"version": 1,
"upstream_base": "deadbeef",
}))
doc = DSO.status_doc(lay)
self.assertEqual(doc["kind"], "psxrecomp-oracle-status")
self.assertEqual(doc["upstream_base"], "deadbeef") # manifest data kept
def test_json_is_serialisable(self):
os.environ["RETCOMM_ORACLE_DIR"] = tempfile.mkdtemp()
json.dumps(DSO.status_doc(DSO.Layout()))
class TestPin(unittest.TestCase):
def test_pin_is_the_single_source_of_truth(self):
pin = DSO.load_pin()
self.assertRegex(pin["upstream_base"], r"^[0-9a-f]{40}$")
self.assertTrue(pin["upstream_url"].endswith("duckstation.git"))
self.assertEqual(pin["oracle_port"], 4371)
self.assertTrue((DSO.PATCH_DIR / pin["patch"]).is_file())
def test_windows_setup_reads_the_same_pin(self):
# The two platform paths must never drift onto different bases.
setup = (DSO.PATCH_DIR / "setup.sh").read_text()
self.assertIn("pin.json", setup)
self.assertNotIn('UPSTREAM_BASE="ffb33c', setup)
def test_image_tag_tracks_the_pin(self):
pin = DSO.load_pin()
self.assertTrue(DSO.image_tag(pin).endswith(pin["upstream_base"][:12]))
class TestBiosDiscovery(EnvGuard):
def test_explicit_path_wins_and_missing_is_none(self):
with tempfile.TemporaryDirectory() as td:
bios = Path(td) / "SCPH1001.BIN"
bios.write_bytes(b"\x00" * 16)
self.assertEqual(DSO.find_bios(str(bios)), bios)
self.assertIsNone(DSO.find_bios(str(Path(td) / "gone.bin")))
def test_never_invents_one(self):
os.environ["RETCOMM_DATA_DIR"] = tempfile.mkdtemp()
# No index, no repo bios beside this checkout -> None, never a download.
self.assertIn(DSO.find_bios(None), (None, DSO.find_bios(None)))
if __name__ == "__main__":
unittest.main()
+479
View File
@@ -0,0 +1,479 @@
#!/usr/bin/env python3
"""Tests for psx_gpu_frame decode + gpu_frame_layers rendering.
These pin the parts that are easy to get subtly wrong and impossible to notice
downstream: packet word layouts, the 11-bit signed vertex, the GP0(E5) draw
offset, the texpage latch a textured polygon performs on draw-mode state, and
the four semi-transparency blends.
"""
import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
def _load(name):
spec = importlib.util.spec_from_file_location(name, ROOT / f"{name}.py")
mod = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(mod)
return mod
GF = _load("psx_gpu_frame")
LAYERS = _load("gpu_frame_layers")
DIFF = _load("gpu_frame_diff")
def entry(seq, op, words, func="0x80012340", ot=0, ra="0x80011000"):
return {
"seq": seq,
"op": f"0x{op:02X}",
"n": len(words),
"src": "0x80100000",
"ot": ot,
"pc": "0x80012348",
"func": func,
"ra": ra,
"w": [f"0x{w:08X}" for w in words],
}
def vert(x, y):
return (y & 0x7FF) << 16 | (x & 0x7FF)
def color(r, g, b):
return (b << 16) | (g << 8) | r
class TestWordLayouts(unittest.TestCase):
def test_poly_word_counts_match_hardware(self):
# Values cross-checked against gpu.c's gp0_exec_* command lengths.
self.assertEqual(GF.poly_words(0x20), 4) # PolyF3
self.assertEqual(GF.poly_words(0x24), 7) # PolyFT3
self.assertEqual(GF.poly_words(0x28), 5) # PolyF4
self.assertEqual(GF.poly_words(0x2C), 9) # PolyFT4
self.assertEqual(GF.poly_words(0x30), 6) # PolyG3
self.assertEqual(GF.poly_words(0x34), 9) # PolyGT3
self.assertEqual(GF.poly_words(0x38), 8) # PolyG4
self.assertEqual(GF.poly_words(0x3C), 12) # PolyGT4
def test_rect_word_counts(self):
self.assertEqual(GF.rect_words(0x60), 3) # variable, flat
self.assertEqual(GF.rect_words(0x64), 4) # variable, textured
self.assertEqual(GF.rect_words(0x68), 2) # 1x1
self.assertEqual(GF.rect_words(0x74), 3) # 8x8 textured
def test_vertex_is_signed_11_bit(self):
self.assertEqual(GF.parse_vertex(vert(16, 32)), (16, 32))
self.assertEqual(GF.parse_vertex(vert(-16, -32)), (-16, -32))
self.assertEqual(GF.parse_vertex(vert(0x400, 0)), (-1024, 0))
class TestDecode(unittest.TestCase):
def test_flat_triangle(self):
e = entry(0, 0x20, [color(10, 20, 30), vert(0, 0), vert(10, 0), vert(0, 10)])
p = GF.decode_entries([e])[0]
self.assertEqual(p["kind"], "poly")
self.assertEqual(p["op_name"], "PolyF3")
self.assertEqual(p["verts"], [[0, 0], [10, 0], [0, 10]])
self.assertEqual(p["colors"], [[10, 20, 30]] * 3)
self.assertFalse(p["semi"])
def test_gouraud_quad_keeps_per_vertex_colour(self):
words = [color(1, 1, 1), vert(0, 0),
color(2, 2, 2), vert(8, 0),
color(3, 3, 3), vert(0, 8),
color(4, 4, 4), vert(8, 8)]
p = GF.decode_entries([entry(0, 0x38, words)])[0]
self.assertEqual(len(p["verts"]), 4)
self.assertEqual(p["colors"], [[1, 1, 1], [2, 2, 2], [3, 3, 3], [4, 4, 4]])
def test_draw_offset_is_applied(self):
off = (7 & 0x7FF) | ((11 & 0x7FF) << 11)
stream = [entry(0, 0xE5, [off]),
entry(1, 0x20, [color(0, 0, 0), vert(1, 2), vert(3, 4), vert(5, 6)])]
prims = GF.decode_entries(stream)
self.assertEqual(prims[1]["verts"], [[8, 13], [10, 15], [12, 17]])
def test_textured_poly_latches_tpage_for_later_prims(self):
# A PolyFT3 whose own tpage word selects STP mode 2 (B-F). The
# rectangle that follows carries no tpage word and must inherit it --
# this is set_tpage_from_poly() in gpu.c.
tpage = (2 << 5) # semi-transparency bits 5-6 = 2
words = [color(128, 128, 128), vert(0, 0),
0x0000_0000, vert(8, 0),
(tpage << 16), vert(0, 8), 0x0000_0000]
stream = [entry(0, 0x24, words),
entry(1, 0x62, [color(9, 9, 9), vert(4, 4), (6 << 16) | 5])]
prims = GF.decode_entries(stream)
self.assertEqual(prims[0]["tpage"], tpage)
self.assertEqual(prims[0]["stp"], 2)
self.assertEqual(prims[1]["stp"], 2, "rect must inherit the poly's latched mode")
self.assertEqual(prims[1]["semi"], True)
self.assertEqual(prims[1]["size"], [5, 6])
def test_e1_sets_mode_for_untextured_poly(self):
stream = [entry(0, 0xE1, [(1 << 5)]),
entry(1, 0x22, [color(0, 0, 0), vert(0, 0), vert(4, 0), vert(0, 4)])]
prims = GF.decode_entries(stream)
self.assertEqual(prims[1]["stp"], 1)
self.assertTrue(prims[1]["semi"])
def test_truncated_packet_is_flagged_not_guessed(self):
e = entry(0, 0x3C, [color(1, 1, 1), vert(0, 0)])
e["n"] = 12
p = GF.decode_entries([e])[0]
self.assertTrue(p["truncated"])
self.assertEqual(len(p["verts"]), 1)
class TestAttribution(unittest.TestCase):
def test_groups_by_func_and_counts_semi(self):
stream = [
entry(0, 0x20, [color(1, 1, 1), vert(0, 0), vert(4, 0), vert(0, 4)],
func="0x80001000"),
entry(1, 0x22, [color(1, 1, 1), vert(0, 0), vert(4, 0), vert(0, 4)],
func="0x80001000", ot=5),
entry(2, 0x20, [color(1, 1, 1), vert(0, 0), vert(4, 0), vert(0, 4)],
func="0x80002000", ot=9),
]
funcs = GF.attribute(GF.decode_entries(stream))
self.assertEqual(funcs["0x80001000"]["drawing"], 2)
self.assertEqual(funcs["0x80001000"]["semi"], 1)
self.assertEqual(funcs["0x80002000"]["ot_min"], 9)
self.assertIn("0.5B+0.5F", funcs["0x80001000"]["stp_modes"])
class TestBlends(unittest.TestCase):
def _one(self, semi, stp, base=100.0):
c = LAYERS.Canvas(0, 0, 4, 4)
c.rgb[:] = base
c.rect(0, 0, 4, 4, (40.0, 40.0, 40.0), semi, stp)
return float(c.rgb[0, 0, 0])
def test_four_modes(self):
self.assertEqual(self._one(False, 0), 40.0) # opaque
self.assertEqual(self._one(True, 0), 70.0) # 0.5B + 0.5F
self.assertEqual(self._one(True, 1), 140.0) # B + F
self.assertEqual(self._one(True, 2), 60.0) # B - F
self.assertEqual(self._one(True, 3), 110.0) # B + 0.25F
def test_subtractive_clamps_at_zero(self):
c = LAYERS.Canvas(0, 0, 2, 2)
c.rgb[:] = 10.0
c.rect(0, 0, 2, 2, (200.0, 200.0, 200.0), True, 2)
self.assertEqual(float(c.rgb[0, 0, 0]), 0.0)
class TestRender(unittest.TestCase):
def _dump(self):
stream = [
entry(0, 0xE3, [0]),
entry(1, 0xE4, [(31 << 10) | 31]),
entry(2, 0x20, [color(255, 0, 0), vert(0, 0), vert(31, 0), vert(0, 31)],
func="0x80001000"),
entry(3, 0x62, [color(0, 0, 255), vert(4, 4), (8 << 16) | 8],
func="0x80002000"),
]
prims = GF.decode_entries(stream)
return {"kind": GF.DUMP_KIND, "version": 1, "frame": 7, "label": "t",
"raw_count": len(prims), "prims": prims, "funcs": GF.attribute(prims)}
def test_renders_one_layer_per_function(self):
with tempfile.TemporaryDirectory() as td:
idx = LAYERS.render(self._dump(), td)
files = {l["func"]: l["file"] for l in idx["layers"]}
self.assertEqual(set(files), {"0x80001000", "0x80002000"})
for f in files.values():
self.assertTrue((Path(td) / f).is_file())
self.assertTrue((Path(td) / "composite.png").is_file())
self.assertTrue((Path(td) / "sheet.png").is_file())
written = json.loads((Path(td) / "layers.json").read_text())
self.assertEqual(written["frame"], 7)
rect = next(l for l in written["layers"] if l["func"] == "0x80002000")
self.assertEqual(rect["pixels"], 64)
self.assertEqual(rect["bbox"], [4, 4, 11, 11])
def test_draw_area_from_e3_e4(self):
self.assertEqual(GF.draw_area(self._dump()), (0, 0, 32, 32))
def _dump_of(stream, label, frame):
prims = GF.decode_entries(stream)
return {"kind": GF.DUMP_KIND, "version": 1, "frame": frame, "label": label,
"raw_count": len(prims), "prims": prims, "funcs": GF.attribute(prims)}
class TestDiff(unittest.TestCase):
"""The scenario this whole toolchain was built for: an effect where the
semi-transparent layer stops being drawn and the opaque one keeps going."""
RAYS = "0x80041000"
VIGNETTE = "0x80042000"
def _good(self):
stream = []
for i in range(6):
stream.append(entry(i, 0x32, # PolyF3 + semi -> the glow rays
[color(200, 60, 40), vert(0, 0), vert(31, i),
vert(i, 31)], func=self.RAYS, ot=20))
# The vignette: a B-F subtractive full-screen quad. Mode 2 comes from
# the E1 that precedes it, which is exactly how the game would do it.
stream.append(entry(6, 0xE1, [(2 << 5)]))
stream.append(entry(7, 0x2A, [color(90, 90, 90), vert(0, 0), vert(31, 0),
vert(0, 31), vert(31, 31)],
func=self.VIGNETTE, ot=1))
return _dump_of(stream, "good", 100)
def _bad(self):
# Same rays, but the STP bit is gone (opcode 0x30 not 0x32) and the
# vignette function never emitted anything at all.
stream = []
for i in range(6):
stream.append(entry(i, 0x30,
[color(200, 60, 40), vert(0, 0), vert(31, i),
vert(i, 31), vert(0, 0), vert(0, 0)],
func=self.RAYS, ot=20))
return _dump_of(stream, "bad", 140)
def test_names_the_function_that_stopped_drawing(self):
d = DIFF.diff(self._good(), self._bad())
by_func = {f["func"]: f for f in d["funcs"]}
self.assertEqual(by_func[self.VIGNETTE]["verdict"], "stopped drawing")
self.assertEqual(by_func[self.VIGNETTE]["a"]["prims"], 1)
self.assertEqual(by_func[self.VIGNETTE]["b"]["prims"], 0)
def test_names_the_function_that_lost_its_blend(self):
d = DIFF.diff(self._good(), self._bad())
by_func = {f["func"]: f for f in d["funcs"]}
self.assertEqual(by_func[self.RAYS]["verdict"], "lost semi-transparency")
self.assertEqual(by_func[self.RAYS]["a"]["semi"], 6)
self.assertEqual(by_func[self.RAYS]["b"]["semi"], 0)
def test_reports_the_blend_modes_that_disappeared(self):
d = DIFF.diff(self._good(), self._bad())
self.assertEqual(d["modes_a"].get("B-F"), 1)
self.assertEqual(d["modes_a"].get("0.5B+0.5F"), 6)
self.assertEqual(d["modes_b"], {})
def test_report_headlines_both_findings(self):
import io
buf = io.StringIO()
DIFF.report(DIFF.diff(self._good(), self._bad()), out=buf)
text = buf.getvalue()
self.assertIn("B-F", text)
self.assertIn(self.VIGNETTE, text)
self.assertIn("stopped drawing", text)
def test_identical_frames_report_no_findings(self):
import io
buf = io.StringIO()
DIFF.report(DIFF.diff(self._good(), self._good()), out=buf)
self.assertIn("No function stopped drawing", buf.getvalue())
class OneShotServer:
"""Mimics runtime/src/debug_server.c io_thread_main(): accept, read one
line, reply, close. Any client that assumes a persistent socket fails
against this exactly the way it fails against the real runtime."""
def __init__(self, handler):
import socket as _s
import threading
self.handler = handler
self.requests = []
self.sock = _s.socket(_s.AF_INET, _s.SOCK_STREAM)
self.sock.setsockopt(_s.SOL_SOCKET, _s.SO_REUSEADDR, 1)
self.sock.bind(("127.0.0.1", 0))
self.sock.listen(8)
self.port = self.sock.getsockname()[1]
self.running = True
self.thread = threading.Thread(target=self._serve, daemon=True)
self.thread.start()
def _serve(self):
while self.running:
try:
c, _ = self.sock.accept()
except OSError:
return
try:
buf = b""
while b"\n" not in buf:
chunk = c.recv(65536)
if not chunk:
break
buf += chunk
req = json.loads(buf.split(b"\n")[0].decode())
self.requests.append(req)
reply = self.handler(req)
reply.setdefault("id", req.get("id", 0))
c.sendall((json.dumps(reply) + "\n").encode())
except Exception:
pass
finally:
c.close() # <- the contract: one request per connection
def stop(self):
self.running = False
self.sock.close()
class TestDebugConn(unittest.TestCase):
def test_multiple_commands_survive_a_one_shot_server(self):
stream = [entry(0, 0x20, [color(7, 7, 7), vert(0, 0), vert(9, 0), vert(0, 9)],
func="0x80050000")]
def handler(req):
if req["cmd"] == "ping":
return {"ok": True, "frame": 4242}
if req["cmd"] == "gpu_frame_dump":
return {"ok": True, "frame": req["frame"], "count": len(stream),
"max_words": 12, "entries": stream}
return {"ok": False, "error": "unknown"}
srv = OneShotServer(handler)
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
# Three commands over what would be three separate sockets. A client
# that reuses one socket hangs here instead.
self.assertEqual(conn.frame(), 4242)
self.assertEqual(conn.frame(), 4242)
dump = GF.capture(conn, frame=4242, label="t")
self.assertEqual(dump["frame"], 4242)
self.assertEqual(dump["raw_count"], 1)
self.assertIn("0x80050000", dump["funcs"])
# ping, ping, gpu_ring_stats (declined here), gpu_frame_dump -- four
# commands, and therefore four separate connections. The count is the
# assertion: a client reusing one socket would never get this far.
self.assertEqual([r["cmd"] for r in srv.requests],
["ping", "ping", "gpu_ring_stats", "gpu_frame_dump"])
def test_error_replies_raise_with_the_server_message(self):
srv = OneShotServer(lambda req: {"ok": False, "error": "missing frame"})
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
with self.assertRaises(GF.DebugError) as cm:
conn.cmd("gpu_frame_dump", frame=1)
self.assertIn("missing frame", str(cm.exception))
def test_capture_uses_the_ring_span_when_no_frame_is_given(self):
stream = [entry(0, 0x20, [color(1, 1, 1), vert(0, 0), vert(4, 0), vert(0, 4)],
func="0x80060000")]
def handler(req):
if req["cmd"] == "gpu_ring_stats":
return {"ok": True, "total": 5000, "capacity": 1 << 20,
"max_words": 12, "oldest_frame": 900, "newest_frame": 1000}
if req["cmd"] == "gpu_frame_dump":
return {"ok": True, "frame": req["frame"], "count": len(stream),
"max_words": 12, "entries": stream}
if req["cmd"] == "ping":
return {"ok": True, "frame": 1234}
return {"ok": False, "error": "unknown"}
srv = OneShotServer(handler)
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
self.assertEqual(conn.ring_span()["newest"], 1000)
# No --frame: take the newest the RING holds, not whatever ping says the
# runtime is on. Those differ, and only the ring one is dumpable.
dump = GF.capture(conn, label="t")
self.assertEqual(dump["frame"], 1000)
self.assertEqual(dump["ring"]["oldest"], 900)
def test_capture_refuses_a_frame_that_fell_out_of_the_ring(self):
def handler(req):
if req["cmd"] == "gpu_ring_stats":
return {"ok": True, "total": 5000, "capacity": 1 << 20,
"max_words": 12, "oldest_frame": 900, "newest_frame": 1000}
return {"ok": True, "frame": 1, "count": 0, "entries": []}
srv = OneShotServer(handler)
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
with self.assertRaises(GF.DebugError) as cm:
GF.capture(conn, frame=42, label="t")
msg = str(cm.exception)
self.assertIn("900..1000", msg)
# An evicted frame must not be reported as a frame that drew nothing.
self.assertIn("empty, not zero-drawn", msg)
def test_capture_rejects_a_server_that_answers_ok_with_no_entries(self):
# Exactly what a leftover stub does: {"ok": true} and nothing else. The
# old code turned that into a cheerful 0-packet dump.
srv = OneShotServer(lambda req: {"ok": True})
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
with self.assertRaises(GF.DebugError) as cm:
GF.capture(conn, frame=5, label="t")
self.assertIn("not a psx-runtime debug server", str(cm.exception))
def test_capture_survives_a_server_without_gpu_ring_stats(self):
stream = [entry(0, 0x20, [color(1, 1, 1), vert(0, 0), vert(4, 0), vert(0, 4)],
func="0x80060000")]
def handler(req):
if req["cmd"] == "gpu_ring_stats":
return {"ok": False, "error": "unknown command"}
if req["cmd"] == "ping":
return {"ok": True, "frame": 77}
return {"ok": True, "frame": req.get("frame", 0), "count": 1,
"entries": stream}
srv = OneShotServer(handler)
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
dump = GF.capture(conn, label="t")
self.assertEqual(dump["frame"], 77) # fell back to ping
self.assertIsNone(dump["ring"])
def test_removed_transport_commands_raise_the_servers_own_message(self):
# psx-runtime keeps pause/step/run_to_frame registered as error stubs.
# The tools must surface that text, not swallow it into a silent no-op.
removed = ("pause is removed; query a ring buffer",
"run_to_frame is removed; use frame_range")
def handler(req):
if req["cmd"] == "pause":
return {"ok": False, "error": removed[0]}
if req["cmd"] == "run_to_frame":
return {"ok": False, "error": removed[1]}
return {"ok": True}
srv = OneShotServer(handler)
self.addCleanup(srv.stop)
conn = GF.DebugConn("127.0.0.1", srv.port, timeout=5.0)
with self.assertRaises(GF.DebugError) as cm:
conn.run_to_frame(10)
self.assertIn("frame_range", str(cm.exception))
with self.assertRaises(GF.DebugError) as cm2:
conn.cmd("pause")
self.assertIn("ring buffer", str(cm2.exception))
def test_summary_round_trips(self):
stream = [entry(0, 0x22, [color(7, 7, 7), vert(0, 0), vert(9, 0), vert(0, 9)],
func="0x80050000")]
prims = GF.decode_entries(stream)
dump = {"kind": GF.DUMP_KIND, "version": 1, "frame": 5, "label": "x",
"raw_count": 1, "prims": prims, "funcs": GF.attribute(prims)}
with tempfile.TemporaryDirectory() as td:
path = str(Path(td) / "x.summary.json")
s = GF.save_summary(dump, path, dump_name="x.json")
again = json.loads(Path(path).read_text())
self.assertEqual(again["kind"], GF.SUMMARY_KIND)
self.assertEqual(again["drawing"], 1)
self.assertEqual(again["modes"], {"0.5B+0.5F": 1})
self.assertEqual(s["funcs"][0]["func"], "0x80050000")
if __name__ == "__main__":
unittest.main()