Yaya48
c230971f07
Preserve formatted NAND boot and Wii Remote pairing, add DMA/SRAM conformance tests and a Wii comparison probe. Keep P1 profiling tools and native update A/B evidence; reject the unhelpful targeted SRAM shortcut.
103 lines
5.8 KiB
Python
103 lines
5.8 KiB
Python
"""Check a local P1 trace against the probe's known work; never qualify a Wii run."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from analyze_p1 import analyze, read_relevant_events
|
|
|
|
|
|
def sha256(path):
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest().upper()
|
|
|
|
|
|
def qualify(directory, events_path=None, xperf_summary_path=None):
|
|
manifest = json.loads((directory / "manifest.json").read_text(encoding="utf-8-sig"))
|
|
decoder = json.loads((directory / "decoder-summary.json").read_text(encoding="utf-8-sig"))
|
|
source = events_path or directory / "events.jsonl"
|
|
report = analyze(read_relevant_events(source, manifest["ProbePID"]),
|
|
manifest["ProbePID"])
|
|
log = (directory / "probe.stdout.log").read_text(encoding="utf-8-sig")
|
|
events = read_relevant_events(source, manifest["ProbePID"])
|
|
xperf_path = xperf_summary_path or directory / "xperf-summary.json"
|
|
xperf = json.loads(xperf_path.read_text(encoding="utf-8-sig")) if xperf_path.exists() else {}
|
|
counters = {(c["tid"], c["counter"]): c["value"]
|
|
for c in report["last_counters_by_thread"]}
|
|
# The local probe has one worker thread. Values are cumulative and exact.
|
|
worker = next(e["tid"] for e in events if e["native"] and e["event"] == "Marker"
|
|
and e["data"].get("Name") == "ProbeBegin")
|
|
checks = {
|
|
"probe_exit_success": manifest.get("ProbeExitCode") == 0,
|
|
"probe_self_checks": "P1Probe exact=OK" in log and "journal-pages=128 nat-arp-replies=32" in log,
|
|
"trace_requested_and_enabled": "trace-requested=1 trace-enabled=1" in log,
|
|
"probe_hash_unchanged": sha256(Path(manifest["Probe"])) == manifest["ProbeSHA256"],
|
|
"probe_pdb_present_and_unchanged": bool(manifest.get("ProbePDBSHA256")) and
|
|
sha256(Path(manifest["ProbePDB"])) == manifest["ProbePDBSHA256"],
|
|
"etl_decoded_without_loss": decoder.get("status") == 0
|
|
and decoder.get("events_lost") == 0 and decoder.get("buffers_lost") == 0
|
|
and decoder.get("native_decode_errors") == 0,
|
|
"qpc_frequency_agrees": decoder.get("qpc_frequency") == report["qpc_frequency"],
|
|
"arm_jit_mapping": any(e["native"] and e["event"] == "CodeMap"
|
|
and e["data"].get("Domain") == "ARM"
|
|
and e["data"].get("Kind") == "thumb" for e in events),
|
|
"nested_scope_consistency": report["scope_samples"] >= 128 and
|
|
report["accounting_errors"] == 0,
|
|
"no_counter_or_snapshot_regression": not report["counter_regressions"] and
|
|
not report["snapshot_regressions"],
|
|
"reset_distinct_generation": len(report["core_generations"]) >= 2,
|
|
"journal_exact": counters.get((worker, 0)) == 128 and
|
|
counters.get((worker, 1)) == 128 * (12 + 4 + 0x840 + 20) and
|
|
counters.get((worker, 2)) == 0,
|
|
"local_nat_exact": counters.get((worker, 3)) == 32 and
|
|
counters.get((worker, 5)) == 32 and
|
|
counters.get((worker, 7)) == 0 and
|
|
counters.get((worker, 8)) == 32 and
|
|
counters.get((worker, 10)) == 0 and
|
|
report["usb_queue_events"] >= 64,
|
|
}
|
|
system_checks = {
|
|
"arm_cpu_sample": report["cpu_leaf_samples"].get("ARM", 0) > 0,
|
|
"known_sleep_classified_waiting": bool(report["known_sleep"] and
|
|
report["known_sleep"]["wait_evidence_present"]),
|
|
"xperf_clock_and_target_events": xperf.get("anchors", 0) >= 2 and
|
|
xperf.get("max_anchor_error_qpc", float("inf")) <=
|
|
report["qpc_frequency"] / 100_000 and
|
|
xperf.get("xperf_invalid_target_events") == 0,
|
|
}
|
|
return {
|
|
"schema": 1, "local_provider_passed": all(checks.values()),
|
|
"full_collector_passed": all(checks.values()) and all(system_checks.values()),
|
|
"native_update_ready": False, "provider_checks": checks,
|
|
"system_checks": system_checks,
|
|
"remaining": ["Qualify real PPC JIT workload and native C++ stacks",
|
|
"Quantify full collector overhead on alternating fixed-work runs",
|
|
"Validate CPU domain coverage and critical path on a native trace"],
|
|
"analysis": report,
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("capture", type=Path)
|
|
parser.add_argument("--events", type=Path,
|
|
help="Combined native and xperf JSONL; defaults to capture/events.jsonl")
|
|
parser.add_argument("--xperf-summary", type=Path,
|
|
help="Summary from xperf_to_p1.py for the same combined events")
|
|
parser.add_argument("--output", type=Path,
|
|
help="New qualification JSON path; defaults to capture/qualification.json")
|
|
args = parser.parse_args()
|
|
result = qualify(args.capture, args.events, args.xperf_summary)
|
|
output = args.output or args.capture / "qualification.json"
|
|
with output.open("x", encoding="utf-8") as target:
|
|
json.dump(result, target, indent=2)
|
|
print(json.dumps({"local_provider_passed": result["local_provider_passed"],
|
|
"full_collector_passed": result["full_collector_passed"],
|
|
"provider_checks": result["provider_checks"],
|
|
"system_checks": result["system_checks"], "output": str(output)}, indent=2))
|
|
raise SystemExit(0 if result["local_provider_passed"] else 1)
|