retroDE_ps2/sim/golden/trace_compare.py

#!/usr/bin/env python3
"""Compare two traces in the retroDE_ps2 common envelope.

Implements the diff semantics specified in sim/golden/trace_compare_spec.md:

  - parse both files, skipping blank lines and comments starting with '#',
  - enforce exactly 8 columns per non-comment line,
  - validate cycle monotonicity within each input,
  - filter to (subsystem, event) pair (default: EE IFETCH),
  - require non-empty filtered sets on both sides,
  - require equal filtered lengths,
  - compare by record order, not by cycle,
  - the fields that must match exactly: subsystem, event, arg0..arg3, flags,
  - cycle is informational only.

Exit codes:
  0  pass
  1  semantic mismatch (length, field, or empty filtered set)
  2  usage or missing-file error
  3  malformed input / parse failure
"""
import argparse
import sys
from dataclasses import dataclass

EXIT_PASS = 0
EXIT_MISMATCH = 1
EXIT_USAGE = 2
EXIT_MALFORMED = 3


@dataclass
class Record:
    line_no: int
    cycle: int
    subsystem: str
    event: str
    arg0: int
    arg1: int
    arg2: int
    arg3: int
    flags: int


def _parse_int(s: str) -> int:
    s = s.strip()
    if s.startswith(("0x", "0X")):
        return int(s, 16)
    return int(s, 10)


def _parse_flags(s: str) -> int:
    s = s.strip()
    if s == "-":
        return 0
    if s.startswith(("0x", "0X")):
        return int(s, 16)
    raise ValueError(f"unrecognized flags token: {s!r}")


def _fail_malformed(path: str, line_no: int, reason: str) -> None:
    print(f"ERROR: {path}:{line_no}: {reason}", file=sys.stderr)
    sys.exit(EXIT_MALFORMED)


def parse_trace(path: str) -> list[Record]:
    try:
        f = open(path, "r")
    except OSError as e:
        print(f"ERROR: cannot open {path}: {e}", file=sys.stderr)
        sys.exit(EXIT_USAGE)

    records: list[Record] = []
    with f:
        for line_no, raw in enumerate(f, 1):
            stripped = raw.strip()
            if not stripped or stripped.startswith("#"):
                continue
            parts = stripped.split()
            if len(parts) != 8:
                _fail_malformed(path, line_no,
                                f"expected 8 columns, got {len(parts)}")
            try:
                cycle = _parse_int(parts[0])
                arg0 = _parse_int(parts[3])
                arg1 = _parse_int(parts[4])
                arg2 = _parse_int(parts[5])
                arg3 = _parse_int(parts[6])
                flags = _parse_flags(parts[7])
            except ValueError as e:
                _fail_malformed(path, line_no, str(e))
            records.append(Record(
                line_no=line_no, cycle=cycle,
                subsystem=parts[1], event=parts[2],
                arg0=arg0, arg1=arg1, arg2=arg2, arg3=arg3, flags=flags,
            ))

    # Monotonic cycle check on the raw (unfiltered) stream.
    for i in range(1, len(records)):
        if records[i].cycle < records[i - 1].cycle:
            _fail_malformed(
                path, records[i].line_no,
                f"non-monotonic cycle: {records[i].cycle} "
                f"< previous {records[i - 1].cycle}"
            )

    return records


def _filter(recs: list[Record], subsystem: str, event: str) -> list[Record]:
    return [r for r in recs if r.subsystem == subsystem and r.event == event]


def _fmt_record(r: Record, subsystem: str, event: str) -> str:
    return (f"{subsystem} {event} "
            f"arg0=0x{r.arg0:016x} arg1=0x{r.arg1:016x} "
            f"arg2=0x{r.arg2:016x} arg3=0x{r.arg3:016x} "
            f"flags=0x{r.flags:08x}")


def main() -> int:
    ap = argparse.ArgumentParser(description="retroDE_ps2 trace comparator")
    ap.add_argument("rtl", help="RTL trace file")
    ap.add_argument("golden", help="golden trace file")
    ap.add_argument("--subsystem", default="EE",
                    help="filter subsystem (default: EE)")
    ap.add_argument("--event", default="IFETCH",
                    help="filter event (default: IFETCH)")
    args = ap.parse_args()

    rtl_all = parse_trace(args.rtl)
    golden_all = parse_trace(args.golden)

    rtl = _filter(rtl_all, args.subsystem, args.event)
    golden = _filter(golden_all, args.subsystem, args.event)

    if not rtl:
        print(f"FAIL: {args.rtl}: no {args.subsystem}/{args.event} records "
              f"after filter", file=sys.stderr)
        return EXIT_MISMATCH
    if not golden:
        print(f"FAIL: {args.golden}: no {args.subsystem}/{args.event} records "
              f"after filter", file=sys.stderr)
        return EXIT_MISMATCH

    if len(rtl) != len(golden):
        longer = "rtl" if len(rtl) > len(golden) else "golden"
        first_missing = min(len(rtl), len(golden))
        print(f"FAIL: length mismatch — rtl={len(rtl)} golden={len(golden)}",
              file=sys.stderr)
        print(f"  {longer} has more records; "
              f"first missing filtered index = {first_missing}",
              file=sys.stderr)
        return EXIT_MISMATCH

    must_match = ("subsystem", "event", "arg0", "arg1", "arg2", "arg3", "flags")
    for i, (r, g) in enumerate(zip(rtl, golden)):
        diffs = [f for f in must_match if getattr(r, f) != getattr(g, f)]
        if diffs:
            print(f"FAIL: mismatch at filtered record {i}", file=sys.stderr)
            print(f"  rtl:    {_fmt_record(r, args.subsystem, args.event)}",
                  file=sys.stderr)
            print(f"  golden: {_fmt_record(g, args.subsystem, args.event)}",
                  file=sys.stderr)
            print(f"  diff:   {', '.join(diffs)}", file=sys.stderr)
            return EXIT_MISMATCH

    print(f"PASS: matched {len(rtl)} {args.subsystem}/{args.event} records "
          f"(cycle ignored, order-based compare)")
    return EXIT_PASS


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