SirRoboGarage/common_libs/tests/measure_fire_ghost_lag.py

#!/usr/bin/env python3
"""Ghost-vs-observer probe (j147) — is the 1-tick aura lag a DECISION lag?

For every enemy fire in the recorded event sidecar we locate the ghost our
mover spawned for it (the `[firediag] SPAWN` line) and compare

  * WHEN  the ghost was spawned  (its tick) vs WHEN the enemy fired, and
  * WHERE  the ghost was drawn   vs WHERE the true bullet is at that instant.

Timeline anchoring is done in ABSOLUTE arena coordinates: the bot logs its own
position at the spawn tick, and the capture row carrying that position pins
`captureRow = botTick + k` (a per-round constant = the server's delivery
offset).  With k known,

    trueBullet(row) = fireOrigin + (row - fireRow) * v      (v = 20 - 3*power)
    ghost(row)      = spawnGhostPos + (row - spawnRow) * v

so both the time lag and the pixel displacement are directly measurable.
"""
import json, math, re, sys, collections

SPAWN_RE = re.compile(
    r"SPAWN tick=(\d+) sx=([-\d.eE+]+) sy=([-\d.eE+]+) gx=([-\d.eE+]+) "
    r"gy=([-\d.eE+]+) p=([-\d.eE+]+) eta=([-\d.eE+]+)")


def load(arm_dir):
    rows = [json.loads(l) for l in open(f"{arm_dir}/run1.jsonl") if '"tick"' in l]
    ev = [json.loads(l) for l in open(f"{arm_dir}/run1.events.jsonl") if l.strip()]
    rounds = json.load(open(f"{arm_dir}/run1.jsonl.rounds.json"))["rounds"]
    spawns = []
    for line in open(f"{arm_dir}/run1.bot.stdout.log"):
        m = SPAWN_RE.search(line)
        if m:
            g = [float(x) for x in m.groups()]
            spawns.append(dict(tick=int(g[0]), sx=g[1], sy=g[2], gx=g[3], gy=g[4],
                               p=g[5], eta=g[6]))
    return rows, ev, rounds, spawns


def split_rounds(spawns):
    """The bot's tick restarts every round -> segment on a tick decrease."""
    out, cur = [], []
    for s in spawns:
        if cur and s["tick"] <= cur[-1]["tick"]:
            out.append(cur); cur = []
        cur.append(s)
    if cur:
        out.append(cur)
    return out


def analyse(arm_dir, verbose=True):
    rows, ev, rounds, spawns = load(arm_dir)
    pos = {r["tick"]: r for r in rows}
    starts = {r["round"]: r["startTick"] for r in rounds}
    counts = {r["round"]: r["count"] for r in rounds}

    # who is the enemy, and which capture row does a fire turn correspond to
    votes = collections.Counter()
    for e in ev:
        if e["type"] != "fire":
            continue
        gt = starts[e["round"]] + e["tick"]
        for off in (-3, -2, -1, 0, 1):
            d = pos.get(gt + off)
            if not d:
                continue
            de = math.hypot(d["ex"] - e["x"], d["ey"] - e["y"])
            ds = math.hypot(d["sx"] - e["x"], d["sy"] - e["y"])
            votes[(e["owner"], "e" if de < ds else "s")] += 1
            break
    enemy_owner = max(votes.items(), key=lambda kv: kv[1])[0][0]
    fires = [e for e in ev if e["type"] == "fire" and e["owner"] == enemy_owner]
    by_round = collections.defaultdict(list)
    for e in fires:
        by_round[e["round"]].append(e)

    recs, unmatched = [], 0
    chunks = split_rounds(spawns)
    # which capture row holds a given position (the bot's own, for the anchor)
    where = collections.defaultdict(list)
    for r in rows:
        where[(round(r["sx"], 3), round(r["sy"], 3))].append(r["tick"])
    for chunk in chunks:
        for s in chunk:
            # TIMELINE, no guessing.  MEASURED LIVE (`[firediag] EV hit` lines vs
            # the capture's event sidecar, exact matches): the sidecar's per-round
            # `tick` IS the server getTurn and the bot runs with
            # `getTurn = bot.tick + 1` (j134), so a ghost logged at bot tick `t`
            # was placed during server turn `t + 1` of its round.  The round is
            # found from the bot's OWN logged position (unique per round).
            hits = where.get((round(s["sx"], 3), round(s["sy"], 3)), ())
            rnd = None
            for hrow in hits:
                for r_ in rounds:
                    if r_["startTick"] <= hrow < r_["startTick"] + r_["count"]:
                        rnd = r_["round"]
                        break
                if rnd is not None:
                    break
            if rnd is None:
                unmatched += 1
                continue
            sturn = s["tick"] + 1
            cands = []
            for e in by_round[rnd]:
                if abs(e["tick"] - sturn) > 2 or abs(e["power"] - s["p"]) > 1e-6:
                    continue
                oerr = math.hypot(s["gx"] - e["x"], s["gy"] - e["y"])
                if oerr < 32.0:                    # the scanned enemy IS the shooter
                    cands.append((abs(e["tick"] - sturn), oerr, e, e["tick"]))
            if not cands:
                unmatched += 1
                continue
            cands.sort()
            _, oerr, e, eturn = cands[0]
            sp = 20.0 - 3.0 * e["power"]
            th = math.radians(e["dir"])
            vx, vy = sp * math.cos(th), sp * math.sin(th)
            d = sturn - eturn                       # +ve = ghost spawned LATE
            # a bullet takes its FIRST step during the turn it is fired, so at
            # the start of server turn `eturn + n` the true bullet sits at
            # origin + n * v.
            tx, ty = e["x"] + vx * d, e["y"] + vy * d
            # arrival deadline: the mover's own eta (logged) vs the true remaining
            # flight time to the TRUE bullet from the same reference point.
            etaTrue = math.hypot(s["sx"] - tx, s["sy"] - ty) / sp
            recs.append(dict(round=rnd, tick=s["tick"], lag=d, power=e["power"],
                             speed=sp, origin_err=oerr, eta=s["eta"],
                             err=math.hypot(s["gx"] - tx, s["gy"] - ty),
                             lagerr=abs(s["eta"] - etaTrue)))

    if verbose:
        print(f"\n=== {arm_dir}")
        print(f"enemy owner id={enemy_owner}  spawns={len(spawns)}  matched={len(recs)}"
              f"  unmatched={unmatched}")
        if not recs:
            return None, []
        print("DETECTION LAG  (capture row of the spawn - capture row of the fire),"
              " +ve = detected LATE:")
        for lag, n in sorted(collections.Counter(r["lag"] for r in recs).items()):
            print(f"   lag={lag:+d} ticks : {n:4d}  ({100.0*n/len(recs):5.1f}%)")
        errs = sorted(r["err"] for r in recs)
        n = len(errs)
        print(f"GHOST-vs-TRUTH displacement px: mean={sum(errs)/n:.2f} median={errs[n//2]:.2f}"
              f" p90={errs[int(0.9*n)]:.2f} max={errs[-1]:.2f}")
        oe = sorted(r["origin_err"] for r in recs)
        print(f"  of which ghost ORIGIN vs fire origin (the scanned enemy position):"
              f" mean={sum(oe)/n:.2f} median={oe[n//2]:.2f} max={oe[-1]:.2f}")
        # the pure time part: lag * speed
        pure = sorted(abs(r["lag"]) * r["speed"] for r in recs)
        print(f"TIME part only (|lag| * speed): mean={sum(pure)/n:.2f} "
              f"median={pure[n//2]:.2f} max={pure[-1]:.2f}")
        le = sorted(r["lagerr"] for r in recs)
        print(f"ARRIVAL-DEADLINE error (mover's eta - the true remaining flight),"
              f" ticks: mean={sum(le)/n:.3f} median={le[n//2]:.3f}"
              f" p90={le[int(0.9*n)]:.3f} max={le[-1]:.3f}")
    return None, recs


if __name__ == "__main__":
    for d in sys.argv[1:]:
        analyse(d)