SirRoboGarage/common_libs/tests/measure_ramfloor_energy

#!/usr/bin/env python3
"""j160 open-loop energy measurement for the FIRING FLOOR / EXHAUSTION RAM.

NO battle, NO server, NO counterfactual replay. This reads the ALREADY RECORDED
closed-loop captures under /tmp and reports, per tick:

  * how often SELF energy sits below a floor candidate,
  * whether the owner's "both low, nobody firing" situation actually occurs,
  * who crosses a low-energy line FIRST (self or the enemy),
  * how often the enemy is low while we are healthy -- the opportunity the
    exhaustion trigger (TR_RAM_ENEMY_ENERGY) would act on.

Deliberately produces NO "damage if we had not fired" number: the offline
harness scored 0/6 on closed-loop questions (docs/offline_harness_trust.md),
so that class of number is worthless here.

Usage:  python3 common_libs/tests/measure_ramfloor_energy [glob-dir]
"""
import json, os, glob, statistics, sys, array

ROOTS = sys.argv[1:] or ["/tmp"]

def recordings():
    out = []
    for root in ROOTS:
        for f in glob.glob(os.path.join(root, "**", "*.jsonl"), recursive=True):
            if f.endswith(".events.jsonl"): continue
            try:
                with open(f) as fh: first = fh.readline()
            except OSError: continue
            if '"closed_loop":true' not in first.replace(" ", ""): continue
            out.append(f)
    return sorted(out)

def split_rounds(path):
    """Yield per-round [(self, enemy)] from a recording, using its round map."""
    rf = path.replace(".jsonl", ".jsonl.rounds.json")
    bounds = []
    if os.path.exists(rf):
        try:
            for r in json.load(open(rf))["rounds"]:
                bounds.append((r["startTick"], r["startTick"] + r["count"]))
        except Exception: bounds = []
    rows = []
    with open(path) as fh:
        for line in fh:
            if '"tick"' not in line: continue
            try: d = json.loads(line)
            except ValueError: continue
            if "se" in d and "ee" in d: rows.append((d["tick"], d["se"], d["ee"]))
    if not rows: return []
    if not bounds: bounds = [(rows[0][0], rows[-1][0] + 1)]
    rounds = []
    for s, e in bounds:
        r = [(se, ee) for t, se, ee in rows if s <= t < e]
        if not r: continue
        # trim the trailing both-disabled tail: a dead bot sits at ~0 forever
        last = max(i for i, (a, b) in enumerate(r) if a > 0 and b > 0)
        rounds.append(r[:last + 1])
    return rounds

def main():
    files = recordings()
    rounds = []
    for f in files: rounds += split_rounds(f)
    if not rounds:
        print("no closed-loop recordings found"); return
    N = sum(len(r) for r in rounds)
    se, ee = array.array("d"), array.array("d")
    for r in rounds:
        for a, b in r: se.append(a); ee.append(b)
    print(f"recordings={len(files)}  rounds={len(rounds)}  ticks={N}\n")

    THR = [5, 10, 15, 20, 25]
    print("=== A) SELF energy below a floor candidate (share of ticks) ===")
    print(f"{'floor':>5} {'pct':>7} {'rounds hit':>10} {'med run':>8} {'p90 run':>8} {'max run':>8}")
    for t in THR:
        tot, hit, lens = 0, 0, []
        for r in rounds:
            cur, got = 0, False
            for a, _ in r:
                if a <= t: cur += 1; tot += 1; got = True
                elif cur: lens.append(cur); cur = 0
            if cur: lens.append(cur)
            hit += 1 if got else 0
        lens.sort()
        print(f"{t:>5} {100*tot/N:>6.2f}% {hit:>10} "
              f"{statistics.median(lens) if lens else 0:>8.0f} "
              f"{lens[int(.9*len(lens))] if lens else 0:>8} "
              f"{lens[-1] if lens else 0:>8}")

    print("\n=== B) the owner's \"both low, nobody firing\" situation ===")
    for t in THR:
        both = sum(1 for a, b in zip(se, ee) if a <= t and b <= t)
        sonly = sum(1 for a, b in zip(se, ee) if a <= t < b)
        eonly = sum(1 for a, b in zip(se, ee) if b <= t < a)
        print(f"  both<={t:>2}: {100*both/N:6.3f}%   self-only {100*sonly/N:6.2f}%"
              f"   enemy-only {100*eonly/N:6.2f}%")

    print("\n=== C) who crosses a low-energy line FIRST (per round) ===")
    for t in [10, 15, 20, 25]:
        s = e = n = 0
        for r in rounds:
            fs = next((i for i, x in enumerate(r) if x[0] <= t), None)
            fe = next((i for i, x in enumerate(r) if x[1] <= t), None)
            if fs is None and fe is None: n += 1
            elif fs is None or (fe is not None and fs < fe): s += 1
            else: e += 1
        m = len(rounds)
        print(f"  t={t:>2}: self-first {s:>5} ({100*s/m:5.1f}%)  "
              f"enemy-first {e:>5} ({100*e/m:5.1f}%)  neither {n:>4} ({100*n/m:4.1f}%)")

    print("\n=== D) \"the enemy can no longer fire\" (server rejects energy <= power) ===")
    for p in (0.4, 1.0, 1.95, 3.0):
        c = sum(1 for b in ee if b <= p)
        c2 = sum(1 for a, b in zip(se, ee) if b <= p and a > 20)
        print(f"  enemy <= {p:>4}: {100*c/N:6.3f}%   and self>20: {100*c2/N:6.3f}%")

    print("\n=== E) exhaustion-trigger OPPORTUNITY: enemy low while we are healthy ===")
    for t in [10, 20, 30]:
        row = "  ".join(f"self>{fl}: {100*sum(1 for a,b in zip(se,ee) if b<=t and a>fl)/N:6.2f}%"
                        for fl in (0, 20, 25))
        print(f"  enemy<={t:>2}  {row}")

    print("\n=== F) the ALREADY-SHIPPED finisher (enemy<20 & self>enemy & dist<300) ===")
    c = 0
    # dist needs the raw file; recompute over the whole trimmed corpus
    for f in files:
        with open(f) as fh:
            for line in fh:
                if '"se"' not in line: continue
                try: d = json.loads(line)
                except ValueError: continue
                if "se" not in d: continue
                if d["ee"] < 20 and d["se"] > d["ee"] and \
                   ((d["ex"]-d["sx"])**2 + (d["ey"]-d["sy"])**2) ** .5 < 300:
                    c += 1
    print(f"  {c} ticks ({100*c/N:.4f}% of the trimmed corpus)")

if __name__ == "__main__":
    main()