SirRoboGarage/common_libs/tests/measure_cornering_ram.py

#!/usr/bin/env python3
"""OFFLINE MEASUREMENT ONLY - no live battles, no bot rebuild.

Measurements for the "cornered enemy is predictable -> ram + better guns"
hypothesis, over the committed jsonl fixtures in tools/fixtures/.

  M1  real hit attribution (events sidecar) bucketed by enemy wall distance
  M2  movement predictability, wall-adjacent vs open (1-tick and H-tick)
  M3  ram / body-contact opportunity base rate, episodes, energy loss

Run:
  python3 common_libs/tests/measure_cornering_ram.py
"""
import json, math, os, statistics
from collections import defaultdict, Counter

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
FIX = os.path.join(ROOT, "tools", "fixtures")
META = os.path.join(FIX, "drussgt_meta")
EVENTS = os.path.join(ROOT, "tools", "robocode_shim", "evidence",
                      "tr_drussgt_vs_modularbot.events.json")

BOT_RADIUS = 18.0
CONTACT = 2 * BOT_RADIUS          # 36 px centre-to-centre
RAM_DAMAGE = 0.6                  # Robocode/Tank Royale per-collision energy loss to each bot

WALL_BUCKETS = [("<=30", 0, 30), ("30-60", 30, 60), ("60-120", 60, 120),
                ("120-250", 120, 250), (">250", 250, 1e9)]
ADJ = ("<=30", "30-60")
OPEN = ("60-120", "120-250", ">250")


def load_fixture(path):
    states, meta = [], {}
    for line in open(path):
        line = line.strip()
        if not line:
            continue
        d = json.loads(line)
        if "meta" in d:
            meta = d["meta"]
            continue
        if "end" in d:
            continue
        states.append(d)
    return meta, states


def load_rounds(path, n):
    p = os.path.join(META, os.path.basename(path) + ".rounds.json")
    if not os.path.exists(p):
        return [{"round": 1, "startTick": 0, "count": n}]
    return json.load(open(p))["rounds"]


def wall_dist(x, y, w=800.0, h=600.0):
    return min(x, y, w - x, h - y)


def bucket_name(d):
    for name, lo, hi in WALL_BUCKETS:
        if lo < d <= hi:
            return name
    return "<=30"


def z2prop(h1, n1, h2, n2):
    if n1 == 0 or n2 == 0:
        return 0.0
    p1, p2 = h1 / n1, h2 / n2
    p = (h1 + h2) / (n1 + n2)
    se = math.sqrt(p * (1 - p) * (1 / n1 + 1 / n2))
    return (p1 - p2) / se if se > 0 else 0.0


def ang_diff(a, b):
    d = (a - b) % 360.0
    return min(d, 360.0 - d)


def norm180(a):
    a = (a + 180.0) % 360.0 - 180.0
    return a


# ────────────────────────────────────────────────────────────── M1 real ──
def m1_real():
    fxpath = os.path.join(FIX, "tr_drussgt_vs_modularbot.jsonl")
    meta, states = load_fixture(fxpath)
    rounds = load_rounds(fxpath, len(states))
    rstarts = {r["round"]: r["startTick"] for r in rounds}
    events = [json.loads(l) for l in open(EVENTS)]
    fires = [e for e in events if e["type"] == "fire"]
    hits = {(e["round"], e["bullet"]) for e in events if e["type"] == "hit"}

    shots, hitc = defaultdict(int), defaultdict(int)
    perround = defaultdict(lambda: defaultdict(lambda: [0, 0]))
    st = ht = 0
    for e in fires:
        if e["owner"] != 2:      # 2 = ModularBot (s*), 1 = DrussGT (e*)
            continue
        g = rstarts[e["round"]] + e["tick"] - 2
        s = states[g]
        b = bucket_name(wall_dist(s["ex"], s["ey"]))
        ish = (e["round"], e["bullet"]) in hits
        shots[b] += 1
        hitc[b] += ish
        st += 1
        ht += ish
        perround[e["round"]][b][0] += ish
        perround[e["round"]][b][1] += 1

    print("== M1a REAL pooled hit attribution (ModularBot shots, all guns) ==")
    print(f"   {'bucket':>8} {'shots':>6} {'hits':>5} {'rate':>7}")
    for name, _, _ in WALL_BUCKETS:
        s, h = shots[name], hitc[name]
        print(f"   {name:>8} {s:>6} {h:>5} {h/s*100 if s else 0:>6.2f}%")
    print(f"   {'TOTAL':>8} {st:>6} {ht:>5} {ht/st*100:>6.2f}%")
    a = sum(shots[n] for n in ADJ); ah = sum(hitc[n] for n in ADJ)
    o = sum(shots[n] for n in OPEN); oh = sum(hitc[n] for n in OPEN)
    print(f"   ADJACENT(<=60) {ah}/{a}={ah/a*100:.2f}%   OPEN(>60) {oh}/{o}={oh/o*100:.2f}%"
          f"   z={z2prop(ah,a,oh,o):.3f}")
    ar = [sum(perround[r][b][0] for b in ADJ) / max(1, sum(perround[r][b][1] for b in ADJ)) * 100
          for r in range(1, 16) if sum(perround[r][b][1] for b in ADJ)]
    orr = [sum(perround[r][b][0] for b in OPEN) / max(1, sum(perround[r][b][1] for b in OPEN)) * 100
           for r in range(1, 16) if sum(perround[r][b][1] for b in OPEN)]
    print(f"   per-round ADJ rate range {min(ar):.1f}-{max(ar):.1f}% (mean {statistics.mean(ar):.2f})")
    print(f"   per-round OPEN rate range {min(orr):.1f}-{max(orr):.1f}% (mean {statistics.mean(orr):.2f})")
    print("   OVERLAP: per-round ranges fully overlap -> not separated by repo convention")
    print()


# ────────────────────────────────────────────────────────────── M2 ──
def motion_dir(s):
    return s["eh"] if s["es"] >= 0 else s["eh"] + 180.0


def nearest_wall_dir(s, w=800.0, h=600.0):
    x, y = s["ex"], s["ey"]
    dl, dr, db, dt = x, w - x, y, h - y
    m = min(dl, dr, db, dt)
    if m == dl:
        return 0.0
    if m == dr:
        return 180.0
    if m == db:
        return 90.0
    return 270.0


def centre_dir(s, w=800.0, h=600.0):
    return math.degrees(math.atan2(h / 2 - s["ey"], w / 2 - s["ex"])) % 360.0


def m2_predictability():
    files = [f for f in sorted(os.listdir(FIX)) if f.endswith(".jsonl") and "drussgt" in f]
    tol = 30.0
    HOR = [1, 5, 10, 15]

    def newacc():
        return dict(n=0, keep=0, turn=0, turn_n=0, rev=0, rev_n=0,
                    absdh=0.0, absdh_n=0,
                    away=0, centre=0, vh_err=defaultdict(list), vh_n=defaultdict(int))

    acc = defaultdict(newacc)
    per_file_adj = defaultdict(lambda: defaultdict(lambda: [0, 0]))   # fixture -> bucket -> keep

    for fn in files:
        path = os.path.join(FIX, fn)
        _, states = load_fixture(path)
        rounds = load_rounds(path, len(states))
        starts = set(r["startTick"] for r in rounds)
        for i in range(1, len(states)):
            if i in starts:
                continue
            s0, s1 = states[i - 1], states[i]
            b = bucket_name(wall_dist(s0["ex"], s0["ey"]))
            a = acc[b]
            dh = norm180(s1["eh"] - s0["eh"])
            a["absdh"] += abs(dh); a["absdh_n"] += 1
            if abs(dh) >= 1.0 and i + 1 < len(states) and (i + 1) not in starts:
                dh2 = norm180(states[i + 1]["eh"] - s1["eh"])
                if abs(dh2) >= 1.0:
                    a["turn_n"] += 1
                    a["turn"] += (dh > 0) == (dh2 > 0)
            # 1-tick direction prediction
            if i + 1 < len(states) and (i + 1) not in starts:
                dx = states[i + 1]["ex"] - s1["ex"]
                dy = states[i + 1]["ey"] - s1["ey"]
                if math.hypot(dx, dy) >= 0.5:
                    actual = math.degrees(math.atan2(dy, dx)) % 360.0
                    a["n"] += 1
                    a["keep"] += ang_diff(actual, motion_dir(s0)) <= tol
                    a["away"] += ang_diff(actual, nearest_wall_dir(s0)) <= tol
                    a["centre"] += ang_diff(actual, centre_dir(s0)) <= tol
                    per_file_adj[fn][b][1] += 1
                    per_file_adj[fn][b][0] += ang_diff(actual, motion_dir(s0)) <= tol
            # velocity-sign reversals (both ticks have a definite sign)
            if s0["es"] != 0.0 and s1["es"] != 0.0:
                a["rev_n"] += 1
                a["rev"] += (s0["es"] > 0) != (s1["es"] > 0)
            # H-tick constant-velocity endpoint error
            for H in HOR[1:]:
                j = i + H
                if j >= len(states) or j in starts:
                    continue
                vx = s0["es"] * math.cos(math.radians(s0["eh"]))
                vy = s0["es"] * math.sin(math.radians(s0["eh"]))
                px, py = s0["ex"] + H * vx, s0["ey"] + H * vy
                err = math.hypot(px - states[j]["ex"], py - states[j]["ey"])
                travel = max(1e-6, H * abs(s0["es"]))
                a["vh_err"][H].append(err / travel)
                a["vh_n"][H] += 1

    print("== M2 predictability of DrussGT motion, by wall distance at prediction time ==")
    print(f"   tuned predictors: keep-direction (1 tick, {tol:.0f} deg tol, chance {2*tol/360*100:.1f}%),")
    print("   turn-persistence (sign of heading change), reversal rate, H-tick constant-velocity error")
    print(f"   {'bucket':>8} {'ticks':>7} {'keep':>7} {'turn':>7} {'rev':>7} {'|dh|':>6} "
          + " ".join(f"{'cv'+str(H):>7}" for H in HOR[1:]))
    for name, _, _ in WALL_BUCKETS:
        a = acc[name]
        if a["n"] == 0:
            continue
        cells = []
        for H in HOR[1:]:
            v = a["vh_err"][H]
            cells.append(f"{statistics.median(v)*100:>6.1f}%" if v else "     -")
        print(f"   {name:>8} {a['n']:>7} {a['keep']/a['n']*100:>6.1f}% "
              f"{(a['turn']/a['turn_n']*100 if a['turn_n'] else 0):>6.1f}% "
              f"{(a['rev']/a['rev_n']*100 if a['rev_n'] else 0):>6.1f}% "
              f"{a['absdh']/a['absdh_n']:>5.2f}d " + " ".join(cells))
    # binary
    def binc(keys):
        t = newacc()
        for k in keys:
            for kk, vv in acc[k].items():
                if kk == "vh_err":
                    for H, lst in vv.items():
                        t["vh_err"][H] += lst
                elif kk == "vh_n":
                    for H, nv in vv.items():
                        t["vh_n"][H] += nv
                else:
                    t[kk] += vv
        return t
    adj, opn = binc(ADJ), binc(OPEN)
    base = adj["n"] + opn["n"]
    print()
    print(f"   ADJACENT(<=60) = {adj['n']}/{base} ticks ({adj['n']/base*100:.1f}% of moving ticks)")
    print(f"   {'metric':>26} {'adjacent':>10} {'open':>10} {'delta':>9}")
    for key, lab in (("keep", "keep-direction 1t"), ("away", "away-from-wall 1t"),
                     ("centre", "toward-centre 1t")):
        ra, ro = adj[key] / adj["n"] * 100, opn[key] / opn["n"] * 100
        print(f"   {lab:>26} {ra:>9.1f}% {ro:>9.1f}% {ra-ro:>+8.1f}pp")
    ra = adj["turn"] / adj["turn_n"] * 100 if adj["turn_n"] else 0
    ro = opn["turn"] / opn["turn_n"] * 100 if opn["turn_n"] else 0
    print(f"   {'turn-persistence':>26} {ra:>9.1f}% {ro:>9.1f}% {ra-ro:>+8.1f}pp")
    ra = adj["rev"] / adj["rev_n"] * 100 if adj["rev_n"] else 0
    ro = opn["rev"] / opn["rev_n"] * 100 if opn["rev_n"] else 0
    print(f"   {'velocity-reversal-rate':>26} {ra:>9.1f}% {ro:>9.1f}% {ra-ro:>+8.1f}pp")
    for H in HOR[1:]:
        ma = statistics.median(adj["vh_err"][H]); mo = statistics.median(opn["vh_err"][H])
        print(f"   {'cv-error H=%d (median %%tr)'%H:>26} {ma*100:>9.1f}% {mo*100:>9.1f}% {(ma-mo)*100:>+8.1f}pp")

    # per-fixture consistency (paired sign test on keep-direction)
    print()
    print("   per-fixture keep-direction (1t) accuracy, adjacent vs open:")
    wins = 0; nf = 0
    for fn in files:
        f = per_file_adj[fn]
        na = sum(f[k][1] for k in ADJ); ha = sum(f[k][0] for k in ADJ)
        no = sum(f[k][1] for k in OPEN); ho = sum(f[k][0] for k in OPEN)
        if na == 0 or no == 0:
            continue
        nf += 1
        ra, ro = ha / na * 100, ho / no * 100
        wins += ra > ro
        print(f"     {fn:<38} adj {ra:5.1f}%  open {ro:5.1f}%  {'adj higher' if ra>ro else 'open higher'}")
    print(f"   fixtures where adjacent > open: {wins}/{nf}")


# ────────────────────────────────────────────────────────────── M3 ──
def near_episodes(flags, start, end):
    eps = []
    j = start
    while j < end:
        if flags[j]:
            k = j
            while k < end and flags[k]:
                k += 1
            eps.append((j, k))
            j = k
        else:
            j += 1
    return eps


def m3_ram():
    files = [f for f in sorted(os.listdir(FIX)) if f.endswith(".jsonl") and "drussgt" in f]
    print("== M3 ram / body-contact opportunity base rate ==")
    print(f"   strict contact = centre distance <= {CONTACT:.0f}px (2 x {BOT_RADIUS:.0f}px radius);")
    print(f"   ram collision signature = both bots lose ~{RAM_DAMAGE} energy on the same tick")
    print(f"   {'fixture':<36} {'ticks':>6} {'min':>6} {'<=36':>5} {'<=50':>5} {'<=60':>5} "
          f"{'<=80':>5} {'eps60':>5} {'maxep':>5} {'ram':>4} {'dE/ram':>8}")
    primary = None
    for fn in files:
        path = os.path.join(FIX, fn)
        _, states = load_fixture(path)
        rounds = load_rounds(path, len(states))
        ivals = [(r["startTick"], r["startTick"] + r["count"]) for r in rounds]
        dist = [math.hypot(s["ex"] - s["sx"], s["ey"] - s["sy"]) for s in states]
        for name, thr in (("<=36", CONTACT), ("<=50", 50), ("<=60", 60), ("<=80", 80)):
            pass
        cnt = {t: sum(1 for d in dist if d <= t) for t in (36, 50, 60, 80)}
        # near-contact episodes at <=60 inside rounds
        eps = []
        for a, b in ivals:
            flags = [dist[i] <= 60 for i in range(len(states))]
            eps += near_episodes(flags, a, b)
        # ram collisions: distance <=60 and both energies drop by ~equal amount
        rams = []
        for a, b in ivals:
            for i in range(max(a, 1), b):
                de = states[i]["ee"] - states[i - 1]["ee"]
                ds = states[i]["se"] - states[i - 1]["se"]
                if (dist[i] <= 60 and de <= -0.4 and ds <= -0.4
                        and abs(de - ds) <= 0.15):
                    rams.append((i, de, ds))
        dE = sum((de + ds) / 2 for _, de, ds in rams)
        dEcol = f"{dE:>+8.2f}"
        print(f"   {fn:<36} {len(states):>6} {min(dist):>6.1f} {cnt[36]:>5} {cnt[50]:>5} "
              f"{cnt[60]:>5} {cnt[80]:>5} {len(eps):>5} "
              f"{(max(k-j for j,k in eps) if eps else 0):>5} {len(rams):>4} "
              f"{dEcol}")
        if "tr_drussgt_vs_modularbot.jsonl" == fn:
            primary = (min(dist), cnt, len(eps), len(rams))
    print()
    print("   PRIMARY fixture tr_drussgt_vs_modularbot (ModularBot vs the real DrussGT):")
    if primary:
        mind, cnt, neps, nram = primary
        print(f"     closest approach = {mind:.1f}px (never within {CONTACT:.0f}px);"
              f" ticks <=50/60/80 = {cnt[50]}/{cnt[60]}/{cnt[80]};"
              f" near-contact episodes = {neps}; ram collisions = {nram}")
    print()


def main():
    m1_real()
    m2_predictability()
    m3_ram()


if __name__ == "__main__":
    main()