#!/usr/bin/env python3 """Closed-loop reactivity analysis for the Tank-Royale bridge fixtures. Given a fixture JSONL (per-tick DrussGT + adversary state, produced by ``TrBattleCapture``) and the matching raw-events JSONL (bullet fire / hit / death events), this measures whether **DrussGT's movement responds to the adversary's live bullets** at capture time -- the property the classic-replay fixtures cannot have. Why the raw events: the fixture alone cannot distinguish an enemy *fire* from enemy *damage* (both are energy drops), and it has no bullet positions. The events sidecar gives the exact fire tick, muzzle position, direction and power of every shot, so we can event-lock DrussGT's response and compute predicted bullet-arrival times. Outputs ------- * fire cadence, hits / misses / bullet-bullet, per-round death victim (who won the round) and DrussGT hits taken; * event-locked PSTH (peri-stimulus time histogram) of DrussGT's response (``|dHeading|``, reversal rate, radial-velocity fraction) as a function of ticks since an adversary fire, with a permutation null band (fire ticks circularly shifted inside each round); * cross-correlation of the fire impulse train with the response, peak lag. Note: fixture roles are fixed by ``TrBattleCapture``: e* = DrussGT (subject), s* = adversary. The muzzle-position match is used only to *validate* that owner id mapping independently. Usage: python3 tools/robocode_shim/analyze_closed_loop.py FIXTURE.jsonl EVENTS.jsonl """ import json import math import os import random import statistics as st import sys def wrap180(a): return ((a + 180.0) % 360.0) - 180.0 def load_fixture(path): rows = [] for line in open(path): line = line.strip() if not line: continue o = json.loads(line) if "tick" in o: rows.append(o) rounds = [] for cand in (path + ".rounds.json", os.path.join(os.path.dirname(path), "drussgt_meta", os.path.basename(path) + ".rounds.json")): try: rounds = json.load(open(cand))["rounds"] break except Exception: rounds = [] return rows, rounds def load_events(path): return [json.loads(l) for l in open(path) if l.strip()] def build_rounds(rows, rounds): if not rounds: out, cur = [], [rows[0]] for r in rows[1:]: if math.hypot(r["ex"] - cur[-1]["ex"], r["ey"] - cur[-1]["ey"]) > 20: out.append(cur) cur = [] cur.append(r) out.append(cur) return [{"rows": c, "start": c[0]["tick"], "round": i + 1} for i, c in enumerate(out)] by_tick = {r["tick"]: r for r in rows} out = [] for rn in rounds: s = rn["startTick"] out.append({"round": rn["round"], "start": s, "rows": [by_tick[t] for t in range(s, s + rn["count"]) if t in by_tick]}) return out def response_series(rlist): res = [] for rd in rlist: R = rd["rows"] n = len(R) hc = [float("nan")] * n rev = [float("nan")] * n rad = [float("nan")] * n for i in range(1, n): hc[i] = abs(wrap180(R[i]["eh"] - R[i - 1]["eh"])) v0, v1 = R[i - 1]["es"], R[i]["es"] rev[i] = 1.0 if (abs(v0) > 0.5 and v0 * v1 < 0) else 0.0 theta = math.degrees(math.atan2(R[i]["sy"] - R[i]["ey"], R[i]["sx"] - R[i]["ex"])) mu = R[i]["eh"] if v1 >= 0 else R[i]["eh"] + 180.0 rad[i] = abs(math.cos(math.radians(mu - theta))) res.append({"round": rd["round"], "start": rd["start"], "n": n, "hc": hc, "rev": rev, "rad": rad}) return res def main(): if len(sys.argv) < 3: print(__doc__) sys.exit(2) fixture, events_path = sys.argv[1], sys.argv[2] rows, rounds = load_fixture(fixture) events = load_events(events_path) rlist = build_rounds(rows, rounds) arrays = response_series(rlist) n_by_round = {a["round"]: a["n"] for a in arrays} # ---- validate owner -> role mapping from muzzle positions ---- idx = {r["tick"]: i for i, r in enumerate(rows)} start_of = {rn["round"]: rn["startTick"] for rn in rounds} role_by_owner = {} for e in events: if e["type"] != "fire" or "x" not in e: continue start = start_of.get(e["round"]) if start is None: continue gi = start + (e["tick"] - 1) if gi not in idx: continue r = rows[idx[gi]] de = math.hypot(r["ex"] - e["x"], r["ey"] - e["y"]) ds = math.hypot(r["sx"] - e["x"], r["sy"] - e["y"]) role_by_owner.setdefault(e["owner"], []).append("e" if de <= ds else "s") owner_role = {o: max(set(v), key=v.count) for o, v in role_by_owner.items()} subj_owner = next((o for o, ro in owner_role.items() if ro == "e"), None) adv_owner = next((o for o, ro in owner_role.items() if ro == "s"), None) print(f"fixture : {fixture}") print(f"events : {events_path}") print(f"ticks : {len(rows)} rounds: {len(rlist)}") for o, v in role_by_owner.items(): print(f"owner {o}: muzzle-position role votes e*={v.count('e')} s*={v.count('s')} " f"-> {owner_role[o]}*") print(f"subject (DrussGT, e*) = owner {subj_owner}; adversary (s*) = owner {adv_owner}") fires_adv = [e for e in events if e["type"] == "fire" and e["owner"] == adv_owner] fires_subj = [e for e in events if e["type"] == "fire" and e["owner"] == subj_owner] hits_adv = [e for e in events if e["type"] == "hit" and e["owner"] == adv_owner] hits_subj = [e for e in events if e["type"] == "hit" and e["owner"] == subj_owner] bhb = [e for e in events if e["type"] == "hitbullet"] print(f"fires : adversary={len(fires_adv)} subject(DrussGT)={len(fires_subj)}") print(f"hits : adversary->DrussGT={len(hits_adv)} DrussGT->adversary={len(hits_subj)}" f" bullet-bullet={len(bhb)}") if fires_adv: print(f"adversary hit rate = {len(hits_adv)/len(fires_adv):.3f}") fire_ticks_by_round = {rd["round"]: [] for rd in rlist} for e in fires_adv: fire_ticks_by_round[e["round"]].append(e["tick"]) print("\n=== per-round (ticks, adversary fires, DrussGT hits taken, round death) ===") per_round_hits = {} for e in hits_adv: per_round_hits[e["round"]] = per_round_hits.get(e["round"], 0) + 1 death_by_round = {} for e in events: if e["type"] == "death": death_by_round.setdefault(e["round"], []).append(e["victim"]) for rd in rlist: rn = rd["round"] dth = death_by_round.get(rn) if dth is None: winner = "no death (timeout?)" else: # victimId == adversary owner -> adversary died -> DrussGT won winner = (f"DrussGT (adversary victimId={dth[0]})" if dth[0] == adv_owner else f"adversary (DrussGT victimId={dth[0]})") print(f" round {rn:2d}: ticks={n_by_round[rn]:5d} advFires={len(fire_ticks_by_round[rn]):4d} " f"DrussGT hits taken={per_round_hits.get(rn,0):2d} winner={winner}") deltas = [] for rn, fts in fire_ticks_by_round.items(): fts = sorted(fts) deltas += [b - a for a, b in zip(fts, fts[1:])] if deltas: s = sorted(deltas) print(f"\nadversary fire interval (ticks): median={st.median(deltas):.1f} " f"mean={st.mean(deltas):.1f} p10={s[len(s)//10]} p90={s[len(s)*9//10]} " f"(min={s[0]}, max={s[-1]})") # ---- event-locked PSTH around adversary fire tick ---- tau_lo, tau_hi = -5, 80 fire_by_round = {rd["round"]: [rd["start"] + (t - 1) for t in sorted(fire_ticks_by_round[rd["round"]])] for rd in rlist} def psth_for(key, fires_map): acc = {t: [] for t in range(tau_lo, tau_hi + 1)} for arr in arrays: lo, n = arr["start"], arr["n"] for f in fires_map[arr["round"]]: for tau in range(tau_lo, tau_hi + 1): gi = f + tau if lo <= gi < lo + n: v = arr[key][gi - lo] if v == v: acc[tau].append(v) return {t: (st.mean(acc[t]) if acc[t] else float("nan")) for t in acc} random.seed(1234) for key, label in [("hc", "|dHeading| deg"), ("rev", "reversal rate"), ("rad", "|cos| radial frac")]: real = psth_for(key, fire_by_round) nulls = {t: [] for t in range(tau_lo, tau_hi + 1)} for _ in range(150): shifted = {} for arr in arrays: lo, n = arr["start"], arr["n"] base = fire_by_round[arr["round"]] k = random.randint(0, max(1, n - 1)) shifted[arr["round"]] = [lo + ((x - lo + k) % n) for x in base] acc = psth_for(key, shifted) for t in nulls: if acc[t] == acc[t]: nulls[t].append(acc[t]) print(f"\n=== event-locked PSTH: {label} vs ticks-since-adversary-fire ===") print(" tau : real null95-low null95-high flag") for t in range(tau_lo, tau_hi + 1): band = sorted(nulls[t]) lo_b = band[int(0.025 * len(band))] if band else float("nan") hi_b = band[int(0.975 * len(band))] if band else float("nan") flag = "" if real[t] == real[t] and band and (real[t] < lo_b or real[t] > hi_b): flag = "OUTSIDE-NULL" if t <= 45: print(f" {t:4d}: {real[t]:8.4f} {lo_b:9.4f} {hi_b:9.4f} {flag}") # ---- control: is DrussGT's turning also locked to its OWN fire cadence? ---- own_fire_by_round = {rd["round"]: [] for rd in rlist} _start = {rd["round"]: rd["start"] for rd in rlist} for e in fires_subj: own_fire_by_round[e["round"]].append(_start[e["round"]] + (e["tick"] - 1)) for rn in own_fire_by_round: own_fire_by_round[rn].sort() print("\n=== CONTROL PSTH (|dHeading|) locked to DrussGT's OWN fire times ===") print(" tau : real null95-low null95-high flag") real_own = psth_for("hc", own_fire_by_round) for t in range(0, 41): band = [] for _ in range(80): shifted = {} for arr in arrays: lo, n = arr["start"], arr["n"] k = random.randint(0, max(1, n - 1)) shifted[arr["round"]] = [lo + ((x - lo + k) % n) for x in own_fire_by_round[arr["round"]]] band.append(psth_for("hc", shifted)[t]) band = sorted(v for v in band if v == v) lo_b = band[int(0.025 * len(band))] if band else float("nan") hi_b = band[int(0.975 * len(band))] if band else float("nan") flag = "OUTSIDE-NULL" if (real_own[t] == real_own[t] and band and (real_own[t] < lo_b or real_own[t] > hi_b)) else "" print(f" {t:4d}: {real_own[t]:8.4f} {lo_b:9.4f} {hi_b:9.4f} {flag}") # ---- range-keeping vs ticks since adversary fire ---- print("\n=== range response: mean range (px) vs ticks-since-adversary-fire ===") for t in range(0, 41, 5): vals = [] for rd in rlist: R = rd["rows"] lo = rd["start"] for f in fire_by_round[rd["round"]]: gi = f + t if lo <= gi < lo + len(R): r = R[gi - lo] vals.append(math.hypot(r["ex"] - r["sx"], r["ey"] - r["sy"])) print(f" tau={t:3d}: mean range = {st.mean(vals):7.1f} (n={len(vals)})") # ---- cross-correlation: fire impulse train vs response ---- print("\n=== cross-correlation of adversary-fire impulse with response ===") # Sparse computation: fires are rare (k ~ 1000 out of ~20000 ticks), so the # only permutation-dependent term is sum_xy at each lag. for key, label in [("hc", "|dHeading|"), ("rev", "reversal")]: resp = [arr[key] for arr in arrays] fire_rel = [] for arr in arrays: lo, n = arr["start"], arr["n"] fire_rel.append([gi - lo for gi in fire_by_round[arr["round"]] if lo <= gi < lo + n]) denom = {} for lag in range(0, 41): npar = 0 sy = sy2 = 0.0 for r in resp: n = len(r) start = 1 if lag == 0 else 0 # r[0] is NaN (round start) for j in range(start + lag, n): v = r[j] if v == v: npar += 1 sy += v sy2 += v * v denom[lag] = (npar, sy, sy2) def sparse_corr(fr_lists): out = {} for lag in range(0, 41): npar, sy, sy2 = denom[lag] if npar < 10: out[lag] = float("nan") continue k = 0 sxy = 0.0 for r, fl in zip(resp, fr_lists): n = len(r) start = 1 if lag == 0 else 0 for fi in fl: if start <= fi < n - lag: v = r[fi + lag] if v == v: k += 1 sxy += v mux = k / npar muy = sy / npar cov = sxy / npar - mux * muy varx = k / npar - mux * mux vary = sy2 / npar - muy * muy out[lag] = (cov / math.sqrt(varx * vary) if varx > 0 and vary > 0 else float("nan")) return out real = sparse_corr(fire_rel) best = max((v, k) for k, v in real.items() if v == v) print(f" {label}: peak r={best[0]:+.4f} at lag={best[1]} ticks; " f"r(0)={real[0]:+.4f} r(10)={real[10]:+.4f} r(20)={real[20]:+.4f} " f"r(30)={real[30]:+.4f}") null_peaks = [] for _ in range(200): shifted = [] for arr, fl in zip(arrays, fire_rel): n = arr["n"] kk = random.randint(0, max(1, n - 1)) shifted.append([(x + kk) % n for x in fl]) null_peaks.append(max(v for v in sparse_corr(shifted).values() if v == v)) p = (1 + sum(1 for v in null_peaks if v >= best[0])) / (1 + len(null_peaks)) print(f" permutation p(peak >= real peak) = {p:.4f} " f"(null peak mean={st.mean(null_peaks):+.4f}, max={max(null_peaks):+.4f})") if __name__ == "__main__": main()