#!/usr/bin/env python3 """STATE-WINDOW GATE: does a temporal WINDOW of wave-relative states predict DrussGT's future lateral position better than a SINGLE state? This is the cheap veto test for the "feed SBC a temporal list of states" design (docs/state_window_gate.md). It is NOT a gun and makes no live-win claim. DATA (real live battles, never regenerated here) /tmp/tfil_ab2/out//runN.jsonl + .events.jsonl + .rounds.json 70 battles / 490 rounds / ~55k shots fired by ModularBot at the real unmodified DrussGT, recorded by tools/robocode_shim/run_bridge_battle.sh. In the capture rows `e*` is DrussGT (the subject) and `s*` is ModularBot (us); the per-shot geometry is re-derived by the validated instrument in common_libs/tests/analyze_drussgt_dodge_vs_power.py, which this file imports. STATE (a design artifact -- see docs/state_window_gate.md for the rationale) One state at absolute tick t, in the frame of the bullet fired at t0 along direction u = (cos dir, sin dir): lat = (D(t) - P0) x u lateral offset from the bullet line, px vlat = lat(t) - lat(t-1) lateral velocity, px/tick (crossing/returning) toa = (t0 - t) + karr ticks until the bullet reaches arrival room = ray distance from D(t) along sign(vlat)*n until the arena wall, px turn = wrap180(eh(t) - eh(t-1)) signed turn rate, deg/tick Each field is quantised into Q in {2,3,4} bins (the numerosity dial). A window is K consecutive states; K=1 is the single-state baseline. Two frames are tested, and BOTH compare a window against the single state at the SAME decision tick D (otherwise a longer window would win only because its decision tick is later): pre D = t0 (the fire tick); the window is the K pre-fire states ending at D. fly D = t0+K-1; the window is the first K states of the flight, and the baseline is the single state at D. Only shots with karr > 31 are used, so the whole window is strictly before the bullet's arrival. TARGET perp_arr = DrussGT's SIGNED perpendicular offset from the bullet line at the tick our bullet reaches its along-track plane (the miss offset that decides the hit). Quantised into 7 bins (edges +-120, +-60, +-18 px); the central +-18 px bin is the hit window. MODEL (deliberately dull: the question is about INFORMATION, not modelling) An interpolated (Jelinek-Mercer) suffix-backoff table over the quantised window: P = global; for j=1..K, P <- (count(suffix_j) + A*P)/(total+A). This is the direct analogue of the SBC "count coincidences" idea, it is order-sensitive, and it can never do much worse than the shorter context, so the sweep isolates information rather than overfitting. The context depth is capped at 12 (orders above that are never observed often enough to matter). A in {1,5,20} is swept as a robustness check (A=5 is primary). SPLIT BY BATTLE, never by tick. All rounds of a battle go to one side. 70 %/30 % battle split, repeated over 3 seeds; model hyper-parameters (bin edges) are derived from the TRAIN side only. CONTROLS 1. shuffle -- random permutation of the K states inside each window (the multiset is preserved, order destroyed). MANDATORY. 2. reverse -- deterministic reversal (preserves recurrence, reverses time). 3. majority -- no-window floor. 4. recurrence -- distinct windows, and how often a window repeats. Run: python3 common_libs/tests/state_window_gate.py --tfil /tmp/tfil_ab2/out \ --json common_libs/tests/fixtures/state_window_gate.json \ > common_libs/tests/fixtures/state_window_gate_report.txt """ from __future__ import annotations import argparse import bisect import collections import json import math import os import random import statistics import sys from array import array HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import analyze_drussgt_dodge_vs_power as dodge # noqa: E402 ARENA_W, ARENA_H = 800.0, 600.0 BOT_R = 18.0 MAXK_PRE = 48 # pre-fire history depth (all K <= 48 are feasible) MAXK_FLY = 32 # during-flight window depth (max real flight is 43) FIELDS = ("lat", "vlat", "toa", "room", "turn") D_CAP = 12 # model context depth cap (see docs; orders > cap never observed) TARGET_EDGES = [-120.0, -60.0, -18.0, 18.0, 60.0, 120.0] HIT_BIN = 3 # bin [ -18, 18 ) -> the bot radius K_PRE = (1, 4, 8, 16, 32, 48) K_FLY = (1, 4, 8, 16, 32) def wrap180(a): return ((a + 180.0) % 360.0) - 180.0 def room_to_wall(px, py, dx, dy): """Distance from (px,py) along unit (dx,dy) until leaving the arena (accounting for the 18 px bot radius).""" t = float("inf") for p, d, lo, hi in ((px, dx, BOT_R, ARENA_W - BOT_R), (py, dy, BOT_R, ARENA_H - BOT_R)): if abs(d) > 1e-9: cand = (hi - p) / d if d > 0 else (lo - p) / d if cand < t: t = cand if t == float("inf"): return 0.0 return max(0.0, t) def tbin(v): return bisect.bisect_right(TARGET_EDGES, v) def raw_state(run, t0, karr, P0, ux, uy, t, rnd): rs = run.start[rnd] re = rs + run.count[rnd] - 1 tc = min(max(t, rs), re) tp = max(tc - 1, rs) r = run.by_tick.get(tc) rp = run.by_tick.get(tp) if r is None or rp is None: return None lat = (r["ex"] - P0[0]) * uy - (r["ey"] - P0[1]) * ux latp = (rp["ex"] - P0[0]) * uy - (rp["ey"] - P0[1]) * ux vlat = lat - latp toa = (t0 - tc) + karr turn = wrap180(r["eh"] - rp["eh"]) sg = 1.0 if vlat >= 0 else -1.0 room = room_to_wall(r["ex"], r["ey"], -uy * sg, ux * sg) return (lat, vlat, toa, room, turn) def extract_run(run): out = [] for s in run.shots(): t0 = s["tick"] karr = s["flight"] if karr is None: continue th = math.radians(s["_dir"]) ux, uy = math.cos(th), math.sin(th) P0 = (s["_x"], s["_y"]) rnd = s["rnd"] pre = array("f") ok = True for off in range(-(MAXK_PRE - 1), 1): rs = raw_state(run, t0, karr, P0, ux, uy, t0 + off, rnd) if rs is None: ok = False break pre.extend(rs) if not ok: continue fly = [] for off in range(MAXK_FLY): rs = raw_state(run, t0, karr, P0, ux, uy, t0 + off, rnd) fly.append(None if rs is None else rs) out.append(dict( battle=os.path.basename(os.path.dirname(run.cap_path)) + "/" + os.path.basename(run.cap_path), rnd=rnd, t0=t0, karr=karr, perp_arr=s["perp_arr"], perp_fire=s["perp_fire"], range=s["range"], power=s["power"], hit=s["hit"], tbin=tbin(s["perp_arr"]), pre=pre, fly=fly, )) return out # ------------------------------------------------------------------ quantise def edges_for(vals, q): xs = sorted(vals) n = len(xs) return [xs[min(n - 1, int((qi / q) * n))] for qi in range(1, q)] def frame_edges(samples, frame, Q): """Quantile bin edges per field, derived from the TRAIN side only.""" edges = [] for f in range(5): vals = [] if frame == "pre": for s in samples: if s["_split"] != "train": continue a = s["pre"] for i in range(MAXK_PRE): vals.append(a[i * 5 + f]) else: for s in samples: if s["_split"] != "train": continue for st in s["fly"]: if st is not None: vals.append(st[f]) edges.append(edges_for(vals, Q)) return edges def apply_codes(samples, frame, edges, Q): res = [] for s in samples: if frame == "pre": a = s["pre"] n = MAXK_PRE codes = [0] * n for i in range(n): base = i * 5 c = 0 for f in range(5): c += bisect.bisect_right(edges[f], a[base + f]) * (Q ** f) codes[i] = c else: a = s["fly"] n = MAXK_FLY codes = [0] * n for i in range(n): if a[i] is None: codes[i] = -1 continue c = 0 for f in range(5): c += bisect.bisect_right(edges[f], a[i][f]) * (Q ** f) codes[i] = c res.append(codes) return res # ------------------------------------------------------------------ the model def order_window(w, mode, rng): if mode == "temporal": return w if mode == "reverse": return w[::-1] if mode == "shuffled": return [w[i] for i in rng.sample(range(len(w)), len(w))] raise ValueError(mode) def build_counts(windows, ys, mode, Dcap, rng): """cbyorder[j][context] = [ {target_bin: count}, total ]. The model is a standard interpolated (Jelinek-Mercer) suffix backoff: for a window w the prediction for the next-state target is P = global for j = 1..K: P = (count(suffix_j) + A*P) / (total(suffix_j) + A) so a longer context is only believed as far as the data supports it, and the model can never do much worse than the shorter one. This keeps the MODEL uninteresting, which is what the gate needs.""" cbyorder = [dict() for _ in range(Dcap + 1)] for w, y in zip(windows, ys): wo = order_window(w, mode, rng) ctx = () for j in range(1, min(Dcap, len(wo)) + 1): ctx = (wo[-j],) + ctx e = cbyorder[j].get(ctx) if e is None: e = [{}, 0] cbyorder[j][ctx] = e e[0][y] = e[0].get(y, 0) + 1 e[1] += 1 return cbyorder NBINS = len(TARGET_EDGES) + 1 def glob_vec(glob, nbins=NBINS): tot = sum(glob.values()) return [(glob.get(b, 0)) / tot for b in range(nbins)] def predict(w, cbyorder, P0, K, A): P = list(P0) for j in range(1, min(K, len(w), len(cbyorder) - 1) + 1): ctx = tuple(w[-j:]) e = cbyorder[j].get(ctx) if e is None: continue cnt, tot = e P = [(cnt.get(b, 0) + A * P[b]) / (tot + A) for b in range(len(P))] return P def evaluate(windows, ys, cbyorder, P0, mode, Kdepth, A, rng): acc = 0 n = 0 ll = 0.0 hitp = 0.0 for w, y in zip(windows, ys): wo = order_window(w, mode, rng) P = predict(wo, cbyorder, P0, Kdepth, A) best = max(range(len(P)), key=lambda b: (P[b], -b)) if best == y: acc += 1 ll += -math.log2(max(P[y], 1e-12)) hitp += P[HIT_BIN] n += 1 return {"n": n, "acc": acc / n, "logloss": ll / n, "implied_hit": hitp / n} def majority_floor(samples): train = [s for s in samples if s["_split"] == "train"] glob = collections.Counter(s["tbin"] for s in train) top = max(glob, key=lambda b: glob[b]) acc = ll = 0.0 n = 0 test_hit = 0 for s in samples: if s["_split"] != "test": continue n += 1 if s["tbin"] == top: acc += 1 ll += -math.log2(glob[s["tbin"]] / sum(glob.values())) test_hit += 1 if s["tbin"] == HIT_BIN else 0 return {"top": top, "n": n, "acc": acc / n, "logloss": ll / n, "emp_hit": test_hit / n, "dist": {str(k): v / sum(glob.values()) for k, v in sorted(glob.items())}} def recurrence(windows, K): """Distinct windows and repeat rate over the WHOLE corpus (no split).""" seen = collections.Counter() for w in windows: k = min(K, len(w)) seen[tuple(w[len(w) - k:])] += 1 total = len(windows) rep2 = sum(c for c in seen.values() if c >= 2) return {"distinct": len(seen), "total": total, "repeat_frac": rep2 / total, "mean_count": total / len(seen)} # ------------------------------------------------------------------ driver def split_battles(samples, seed): battles = sorted({s["battle"] for s in samples}) rng = random.Random(seed) rng.shuffle(battles) ntr = int(round(0.70 * len(battles))) train = set(battles[:ntr]) for s in samples: s["_split"] = "train" if s["battle"] in train else "test" def window_list(codes, frame, K): """The ordered window for one sample. pre frame: the full 48-state pre-fire history ENDING at the fire tick; the model's depth parameter selects how many of the most recent states it may use, so K=1 is the fire-tick state itself. fly frame: the first K states of the flight, i.e. ending at t0+K-1; the single-state baseline is then the state at that same tick. """ if frame == "pre": return list(codes) return list(codes[:K]) def run_frame(samples, frame, Ks, seeds, As): res = {"frame": frame, "seeds": {}} for seed in seeds: split_battles(samples, seed) tr = [i for i, s in enumerate(samples) if s["_split"] == "train"] te = [i for i, s in enumerate(samples) if s["_split"] == "test"] ytr = [samples[i]["tbin"] for i in tr] yte = [samples[i]["tbin"] for i in te] seed_res = {"majority": majority_floor(samples), "Q": {}} for Q in (2, 3, 4): edges = frame_edges(samples, frame, Q) codes = apply_codes(samples, frame, edges, Q) glob = {} for i in tr: glob[samples[i]["tbin"]] = glob.get(samples[i]["tbin"], 0) + 1 P0 = glob_vec(glob) qres = {} for mode in ("temporal", "shuffled", "reverse"): mres = {str(A): {} for A in As} cache = {} for K in Ks: wid = "full" if frame == "pre" else K if wid not in cache: wl = [window_list(codes[i], frame, K) for i in range(len(samples))] cby = build_counts([wl[i] for i in tr], ytr, mode, D_CAP, random.Random(1000 + seed)) cache[wid] = (cby, wl) cby, wl = cache[wid] wte = [wl[i] for i in te] kd = min(K, D_CAP) for A in As: rnge = random.Random(4000 + seed * 13 + K) mres[str(A)][str(K)] = { "window": evaluate(wte, yte, cby, P0, mode, kd, A, rnge)} qres[mode] = mres # single-state-at-the-same-decision-tick baseline (true recent state) sres = {str(A): {} for A in As} for K in Ks: wid = "full" if frame == "pre" else K wl = [window_list(codes[i], frame, K) for i in range(len(samples))] cby = build_counts([wl[i] for i in tr], ytr, "temporal", D_CAP, random.Random(1000 + seed)) wte = [wl[i] for i in te] for A in As: sres[str(A)][str(K)] = evaluate( wte, yte, cby, P0, "temporal", 1, A, random.Random(5000 + seed * 17 + K)) qres["single"] = sres qres["recurrence"] = {str(K): recurrence( [window_list(codes[i], frame, K) for i in range(len(samples))], K) for K in Ks} seed_res["Q"][str(Q)] = qres res["seeds"][str(seed)] = seed_res return res def mean_over(xs): return statistics.fmean(xs) if xs else float("nan") def cell(res, frame, Q, mode, A, K, key="window"): if key == "single": q = lambda sd: res[frame]["seeds"][sd]["Q"][str(Q)]["single"][str(A)][str(K)] else: q = lambda sd: res[frame]["seeds"][sd]["Q"][str(Q)][mode][str(A)][str(K)][key] a = [q(sd) for sd in res[frame]["seeds"]] return (mean_over([c["acc"] for c in a]), mean_over([c["logloss"] for c in a]), mean_over([c["implied_hit"] for c in a])) def render(res, out, A_primary): lines = [] def p(s=""): lines.append(s) print(s) for frame in res: Ks = K_PRE if frame == "pre" else K_FLY p("=" * 100) p("FRAME %s%s" % (frame, " (window ENDS at the fire tick, looks BACK)" if frame == "pre" else " (window STARTS at the fire tick, ends at t0+K-1)")) p("=" * 100) maj = [res[frame]["seeds"][sd]["majority"] for sd in res[frame]["seeds"]] p("MAJORITY / NO-WINDOW FLOOR (test acc %.4f, log-loss %.4f bits," " empirical hit %.4f)" % (mean_over([m["acc"] for m in maj]), mean_over([m["logloss"] for m in maj]), mean_over([m["emp_hit"] for m in maj]))) p(" target-bin edges %s px -> central +-18 px hit bin. At the median" % TARGET_EDGES) p(" fire range (487 px) the 36 px hit window subtends %.2f deg; at 450 px" % math.degrees(2 * math.atan(18.0 / 487.0))) p(" it is %.2f deg; at 100 px %.2f deg. (atan(18/range).)" % (math.degrees(2 * math.atan(18.0 / 450.0)), math.degrees(2 * math.atan(18.0 / 100.0)))) p(" target-bin distribution on train: %s" % ", ".join("%s:%.3f" % (k, v) for k, v in sorted(maj[0]["dist"].items()))) p() for Q in (2, 3, 4): nstates = Q ** 5 p("-" * 100) p("COARSENESS Q=%d -> %d distinct single states (%.1f bits)" % (Q, nstates, math.log2(nstates))) p(" %-4s | %-24s | %-24s | %-13s | %-13s" % ("K", "single@D (temporal)", "window (temporal)", "window=SHUFFLED", "window=REVERSE")) p(" %-4s | %-24s | %-24s | %-13s | %-13s" % ("", "logloss acc hitP", "logloss acc hitP", "logloss", "logloss")) for K in Ks: sc = cell(res, frame, Q, "temporal", A_primary, K, "single") tc = cell(res, frame, Q, "temporal", A_primary, K, "window") sh = cell(res, frame, Q, "shuffled", A_primary, K, "window") rv = cell(res, frame, Q, "reverse", A_primary, K, "window") p(" %-4d | %.4f %.4f %.4f | %.4f %.4f %.4f | %.4f | %.4f" % (K, sc[1], sc[0], sc[2], tc[1], tc[0], tc[2], sh[1], rv[1])) rec = res[frame]["seeds"][sorted(res[frame]["seeds"])[0]]["Q"][str(Q)]["recurrence"] p(" recurrence, distinct ordered window tuples over the whole corpus:") for K in Ks: r = rec[str(K)] p(" K=%-2d distinct=%-8d of %-6d (mean count %.2f, repeat_frac %.3f)" % (K, r["distinct"], r["total"], r["mean_count"], r["repeat_frac"])) p() p("=" * 100) p("HEADLINE (mean over %d battle-split seeds; A=%g)" % (len(res[frame]["seeds"]), A_primary)) p(" held-out log-loss. delta_window = window - single@D (NEGATIVE = the") p(" window beats the single state at the same decision tick)") for Q in (2, 3, 4): srow = [cell(res, frame, Q, "temporal", A_primary, K, "single")[1] for K in Ks] trow = [cell(res, frame, Q, "temporal", A_primary, K, "window")[1] for K in Ks] p(" Q=%d single@D " % Q + " ".join("K=%d %.4f" % (K, v) for K, v in zip(Ks, srow))) p(" window " + " ".join("K=%d %.4f(%+.4f)" % (K, t, t - s) for K, t, s in zip(Ks, trow, srow))) p(" shuffle control: window(shuffled) - window(temporal) (must be >>0)") for Q in (2, 3, 4): row = [cell(res, frame, Q, "shuffled", A_primary, K, "window")[1] - cell(res, frame, Q, "temporal", A_primary, K, "window")[1] for K in Ks] p(" Q=%d " % Q + " ".join("K=%d %+.4f" % (K, v) for K, v in zip(Ks, row))) p(" reverse control: window(reverse) - window(temporal)") for Q in (2, 3, 4): row = [cell(res, frame, Q, "reverse", A_primary, K, "window")[1] - cell(res, frame, Q, "temporal", A_primary, K, "window")[1] for K in Ks] p(" Q=%d " % Q + " ".join("K=%d %+.4f" % (K, v) for K, v in zip(Ks, row))) p(" robustness in the interpolation strength A (window temporal log-loss):") for Q in (2, 3, 4): for A in out.get("A_all", []): row = [cell(res, frame, Q, "temporal", A, K, "window")[1] for K in Ks] p(" Q=%d A=%-4g " % (Q, A) + " ".join("K=%d %.4f" % (K, v) for K, v in zip(Ks, row))) p("=" * 100) out["report"] = lines def main(): ap = argparse.ArgumentParser() ap.add_argument("--tfil", default="/tmp/tfil_ab2/out") ap.add_argument("--json", default=None) ap.add_argument("--limit-runs", type=int, default=0) ap.add_argument("--seed", type=int, default=1) ap.add_argument("--seeds", type=int, default=3) ap.add_argument("--frames", default="pre") ap.add_argument("--alphas", default="1,5,20") a = ap.parse_args() runs = dodge.discover_tfil(a.tfil) if a.limit_runs: runs = runs[:a.limit_runs] print("[gate] %d battles" % len(runs), file=sys.stderr) samples = [] for k, run in enumerate(runs): samples.extend(extract_run(run)) if (k + 1) % 10 == 0: print("[gate] %d/%d battles, %d shots" % (k + 1, len(runs), len(samples)), file=sys.stderr) print("[gate] %d samples" % len(samples), file=sys.stderr) hit = sum(1 for s in samples if s["tbin"] == HIT_BIN) print("[gate] empirical arrival hit bin (|perp|<18): %.4f" % (hit / len(samples)), file=sys.stderr) seeds = [a.seed + i for i in range(a.seeds)] As = [float(x) for x in a.alphas.split(",")] A_primary = 5.0 res = {} nfly = 0 for frame in a.frames.split(","): if frame == "pre": fs = samples else: kmin = max(K_FLY) + 1 # keep the whole window strictly before arrival fs = [s for s in samples if s["karr"] >= kmin] nfly = len(fs) print("[gate] frame %s: %d shots with karr >= %d" % (frame, len(fs), kmin), file=sys.stderr) res[frame] = run_frame(fs, frame, K_PRE if frame == "pre" else K_FLY, seeds, As) out = {"samples": len(samples), "battles": len(runs), "seeds": seeds, "fly_samples": nfly, "alphas": As, "A_primary": A_primary, "A_all": As, "D_cap": D_CAP, "target_edges": TARGET_EDGES, "hit_bin": HIT_BIN, "k_pre": list(K_PRE), "k_fly": list(K_FLY)} render(res, out, A_primary) out["raw"] = res if a.json: with open(a.json, "w") as f: json.dump(out, f, indent=1, default=str) return out if __name__ == "__main__": main()