SirStone
17c99f542f
The classic captures were OPEN-LOOP: replayed DrussGT never dodged OUR
bullets. These come from real TR battles through the working Java bridge, so
the recording contains genuine reactions to ModularBot's live fire. The
open-loop caveat is gone (perfect-information remains).
PRIMARY RESULT - the boss beats us badly. DrussGT 1447 - ModularBot 300 over
15 rounds, ModularBot winning only round 5 (DrussGT died at tick 1893). Rounds
are long, not truncated: mean 1335 ticks, ModularBot got off 1134 shots.
ModularBot 1134 shots / 60 hits = 5.3% real hit rate
DrussGT 1400 shots / 169 hits = 12.1% real hit rate
So DrussGT's gun is ~2.3x more accurate than our entire rack, on top of far
better movement. That is the number to move.
Also captured: shield-on variant (DrussGT 939-287, 9/10 - ModularBot takes
round 1 to the known shield warm-up), and vs SpinBot 1175-0, Crazy 1080-1,
Corners 1659-0. 20,026 + 12,629 + 10,824 + 11,507 + 2,575 ticks.
Movement statistics match the classic set within ~0.04 on the perpendicular
and radial fractions, so this is the same wave surfer in TR physics:
TR vs modularbot: perp 0.967, radial 0.001, 52.8% at full speed,
reversing 46.3%, median range 464 px.
CLOSED LOOP PROVEN, not asserted. ModularBot's fire is a heat-limited near
metronome (median interval 14 ticks), which gives a usable exogenous clock:
- event-locked |delta heading| oscillates 0.96 -> 2.69 deg about a 1.47 deg
mean with the fire period, almost every lag outside the 95% band of a
400-iteration phase-shuffled null;
- cross-correlation of |delta heading| against the fire impulse peaks at
r = +0.111, lag 12 ticks, permutation p = 0.005 (null peak mean +0.016);
- OWN-FIRE CONTROL is flat, so the oscillation is enemy-driven rather than
an internal cadence;
- range response is weak (~3 px over 30 ticks, near noise) and is therefore
NOT claimed, and per-bullet dodging is not claimed either because the
bullet detector (shield) is off.
CAVEATS: still perfect-information (observer gives true positions every tick,
unlike the live bot's stale between-scan WorldState) so these remain optimistic
vs live play; and they are open-loop AT REPLAY TIME - 'closed_loop' describes
the capture, not a later replay. TR conversion residual is ~1.5 deg mean
because the TR server moves along the pre-turn heading, vs 0.000 deg for the
classic captures.
Adds analyze_closed_loop.py (PSTH event-locking, phase-shuffle permutation
null, cross-correlation, own-fire control) and per-round result sidecars.
357 lines
14 KiB
Python
357 lines
14 KiB
Python
#!/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()
|