lgbot: movement recorder (lgbot_record) and tools/lgbot_rec.py to compare player vs bot movement

Co-Authored-By: Claude Opus 5.5 <[email protected]>

tools/lgbot_rec.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+"""Compare how you and lgbot move, from a recording (set lgbot_record 1, play, set lgbot_record 0).
+  python3 tools/lgbot_rec.py [home/lgbot-rec.csv]
+Only fight time counts: both alive, within 1000 units and about the same floor. Sideways/toward are
+relative to the opponent (you vs the bot, the bot vs the nearest human). Prints numbers only.
+"""
+import math, sys
+
+FIGHT_DIST, FIGHT_DZ = 1000, 250
+SIDE_DEADBAND = 60     # u/s sideways before it counts as strafing to one side
+GAP_S = 0.2            # a hole in the recording longer than this breaks a run
+STOP_SPEED = 100
+JUMP_VZ = 150
+
+
+def parse(lines):
+    """-> list of sessions, each {t: {who: row}}; row = (alive, hp, x, y, z, vx, vy, vz, state)."""
+    sessions, cur = [], None
+    for line in lines:
+        line = line.strip()
+        if not line:
+            continue
+        if line.startswith("#"):
+            cur = {}
+            sessions.append(cur)
+            continue
+        f = line.split(",")
+        if len(f) < 11:
+            continue
+        if cur is None:
+            cur = {}
+            sessions.append(cur)
+        try:
+            t = float(f[0])
+            row = (int(f[2]), int(f[3])) + tuple(float(v) for v in f[4:10]) + (f[10],)
+        except ValueError:
+            continue
+        cur.setdefault(t, {})[f[1]] = row
+    return sessions
+
+
+def fight_frames(session):
+    """-> {"human": [...], "bot": [...]}, frames (t, x, y, z, vx, vy, vz, ox, oy, state) in time order."""
+    out = {"human": [], "bot": []}
+    for t in sorted(session):
+        rows = session[t]
+        bot = rows.get("bot")
+        if not bot or not bot[0]:
+            continue
+        best = None
+        for who, r in rows.items():
+            if who == "bot" or not r[0]:
+                continue
+            d = math.hypot(r[2] - bot[2], r[3] - bot[3])
+            if best is None or d < best[0]:
+                best = (d, r)
+        if best is None or best[0] > FIGHT_DIST or abs(best[1][4] - bot[4]) > FIGHT_DZ:
+            continue
+        h = best[1]
+        out["human"].append((t,) + h[2:8] + (bot[2], bot[3], ""))
+        out["bot"].append((t,) + bot[2:8] + (h[2], h[3], bot[8]))
+    return out
+
+
+def pct(vals, q):
+    if not vals:
+        return None
+    v = sorted(vals)
+    return v[min(len(v) - 1, int(q / 100.0 * len(v)))]
+
+
+def analyze(frames):
+    """Movement stats for one player's fight frames (see fight_frames)."""
+    n = len(frames)
+    res = {"seconds": 0.0, "frames": n}
+    if n < 2:
+        return res
+    ground_speed, air_speed, dist, side_share = [], [], [], []
+    holds, side, side_since, prev = [], 0, None, None
+    air = stops = jumps = steer = 0
+    for fr in frames:
+        t, x, y, z, vx, vy, vz, ox, oy, state = fr
+        dx, dy = ox - x, oy - y
+        d = math.hypot(dx, dy) or 1.0
+        ux, uy = dx / d, dy / d
+        lat = -uy * vx + ux * vy
+        rad = ux * vx + uy * vy
+        speed = math.hypot(vx, vy)
+        dist.append(d)
+        on_ground = abs(vz) < 1
+        (ground_speed if on_ground else air_speed).append(speed)
+        if not on_ground:
+            air += 1
+        if speed < STOP_SPEED:
+            stops += 1
+        if abs(lat) + abs(rad) > 1:
+            side_share.append(abs(lat) / (abs(lat) + abs(rad)))
+        if state == "steer":
+            steer += 1
+        gap = prev is not None and not 0 <= t - prev[0] <= GAP_S
+        if prev is not None and not gap and abs(prev[6]) < 1 and vz > JUMP_VZ:
+            jumps += 1
+        if gap:
+            side, side_since = 0, None
+        s = 1 if lat > SIDE_DEADBAND else -1 if lat < -SIDE_DEADBAND else 0
+        if s and s != side:
+            if side and side_since is not None:
+                holds.append(int((t - side_since) * 1000))
+            side, side_since = s, t
+        prev = fr
+    secs = sum(min(max(b[0] - a[0], 0), GAP_S) for a, b in zip(frames, frames[1:]))
+    res.update({
+        "seconds": secs,
+        "hold_ms": (pct(holds, 10), pct(holds, 50), pct(holds, 90)),
+        "reversals_per_s": len(holds) / secs if secs else 0,
+        "ground_speed": (pct(ground_speed, 50), pct(ground_speed, 90)),
+        "air_speed": (pct(air_speed, 50), pct(air_speed, 90)),
+        "air_share": air / n,
+        "jumps_per_s": jumps / secs if secs else 0,
+        "sideways_share": sum(side_share) / len(side_share) if side_share else 0,
+        "stopped_share": stops / n,
+        "dist": (pct(dist, 10), pct(dist, 50), pct(dist, 90)),
+        "steer_share": steer / n,
+    })
+    return res
+
+
+def fmt(v):
+    if v is None:
+        return "-"
+    if isinstance(v, tuple):
+        return " / ".join(fmt(x) for x in v)
+    if isinstance(v, float):
+        return "%.2f" % v if abs(v) < 10 else "%.0f" % v
+    return str(v)
+
+
+ROWS = [("seconds", "fight time s"), ("hold_ms", "strafe hold ms p10/p50/p90"),
+        ("reversals_per_s", "direction changes /s"), ("ground_speed", "ground speed p50/p90"),
+        ("air_speed", "air speed p50/p90"), ("air_share", "time in air"), ("jumps_per_s", "jumps /s"),
+        ("sideways_share", "sideways (vs toward/away)"), ("stopped_share", "time nearly stopped"),
+        ("dist", "distance p10/p50/p90"), ("steer_share", "plugin steering (bot)")]
+
+
+def report(human, bot):
+    lines = ["%-28s %-22s %s" % ("", "you", "lgbot")]
+    for k, label in ROWS:
+        lines.append("%-28s %-22s %s" % (label, fmt(human.get(k)), fmt(bot.get(k))))
+    return "\n".join(lines)
+
+
+def main(argv):
+    path = argv[1] if len(argv) > 1 else "home/lgbot-rec.csv"
+    with open(path) as f:
+        sessions = parse(f)
+    both = {"human": [], "bot": []}
+    for s in sessions:
+        fr = fight_frames(s)
+        for k in both:
+            both[k].extend(fr[k])  # t restarts at 0 each session: the gap rule keeps them apart
+    print("%d session(s)" % len(sessions))
+    print(report(analyze(both["human"]), analyze(both["bot"])))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

tools/test_lgbot_plugin.py

@@ -58,6 +58,9 @@ def load(world):
         def get_cvar(self, name):
             return world.cvars.get(name)
 
+        def set_cvar(self, name, value):
+            world.cvars[name] = value
+
         def add_hook(self, event, fn):
             pass
 
@@ -139,6 +142,33 @@ class Tick(unittest.TestCase):
         # per frame: at most the acceleration step (plus friction compensation), never a 640 u/s flip
         self.assertLess(biggest, self.mod.ACCEL * 0.025 * comp + 0.15 * 320 * comp + 1)
 
+    def test_recorder_writes_rows_without_names(self):
+        import tempfile
+        path = os.path.join(tempfile.mkdtemp(), "rec.csv")
+        self.mod.REC_FILE = path
+        self.w.cvars["lgbot_record"] = "1"
+        self.run_frames(80)
+        self.w.cvars["lgbot_record"] = "0"
+        self.w.now += 1.1                         # cvars are reread once a second
+        self.run_frames(2)
+        with open(path) as f:
+            text = f.read()
+        rows = [l.split(",") for l in text.splitlines() if not l.startswith("#")]
+        self.assertGreater(len(rows), 100)
+        self.assertEqual({row[1] for row in rows}, {"bot", "h1"})
+        self.assertIn("steer", {row[10] for row in rows})
+        self.assertNotIn(str(HUMAN_ID), text)
+        self.assertIsNone(self.plugin.rec)
+
+    def test_recorder_stops_itself(self):
+        import tempfile
+        self.mod.REC_FILE = os.path.join(tempfile.mkdtemp(), "rec.csv")
+        self.mod.REC_MAX_S = 1
+        self.w.cvars["lgbot_record"] = "1"
+        self.run_frames(80)
+        self.assertEqual(self.w.cvars["lgbot_record"], "0")
+        self.assertIsNone(self.plugin.rec)
+
     def test_no_target_means_no_steering(self):
         self.human.team = "spectator"
         self.run_frames(40)

tools/test_lgbot_rec.py

@@ -0,0 +1,75 @@
+"""tools/lgbot_rec.py: parsing a recording and the movement stats."""
+import os, sys, unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import lgbot_rec as r  # noqa: E402
+
+
+def rec(seconds, human_fn, bot_fn, dt=0.025):
+    """Recording lines: human_fn/bot_fn(t) -> (x, y, z, vx, vy, vz)."""
+    lines = ["# start 0 map test"]
+    t = 0.0
+    while t < seconds:
+        for who, fn, state in (("h1", human_fn, ""), ("bot", bot_fn, "steer")):
+            x, y, z, vx, vy, vz = fn(t)
+            lines.append("%.3f,%s,1,100,%.0f,%.0f,%.0f,%.0f,%.0f,%.0f,%s" % (t, who, x, y, z, vx, vy, vz, state))
+        t += dt
+    return lines
+
+
+def adad(hold_s):
+    # sideways (y) at 320 u/s, switching every hold_s, standing 450 in front of the other on x
+    def fn(t):
+        d = 1 if int(t / hold_s) % 2 == 0 else -1
+        return (0, 0, 0, 0, 320 * d, 0)
+    return fn
+
+
+def still_at(x):
+    return lambda t: (x, 0, 0, 0, 0, 0)
+
+
+class Stats(unittest.TestCase):
+    def test_strafe_holds_and_sideways(self):
+        s = r.parse(rec(10, adad(0.5), still_at(450)))
+        st = r.analyze(r.fight_frames(s[0])["human"])
+        self.assertAlmostEqual(st["hold_ms"][1], 500, delta=30)
+        self.assertGreater(st["sideways_share"], 0.95)
+        self.assertEqual(st["air_share"], 0)
+        self.assertAlmostEqual(st["ground_speed"][1], 320, delta=1)
+        self.assertAlmostEqual(st["seconds"], 10, delta=0.1)
+
+    def test_bot_steer_share_and_stopped(self):
+        s = r.parse(rec(5, adad(0.5), still_at(450)))
+        st = r.analyze(r.fight_frames(s[0])["bot"])
+        self.assertEqual(st["steer_share"], 1)
+        self.assertEqual(st["stopped_share"], 1)
+
+    def test_jumps_counted(self):
+        def hop(t):  # one jump a second: 0.7 s in the air, then ground
+            ph = t % 1.0
+            return (0, 0, 0, 0, 320, 270 - 800 * ph if ph < 0.7 else 0)
+        st = r.analyze(r.fight_frames(r.parse(rec(10, hop, still_at(450)))[0])["human"])
+        self.assertAlmostEqual(st["jumps_per_s"], 1.0, delta=0.15)
+        self.assertAlmostEqual(st["air_share"], 0.7, delta=0.05)
+
+    def test_far_or_other_floor_is_not_fight_time(self):
+        far = r.fight_frames(r.parse(rec(2, adad(0.5), still_at(1500)))[0])
+        up = r.fight_frames(r.parse(rec(2, adad(0.5), lambda t: (450, 0, 400, 0, 0, 0)))[0])
+        self.assertEqual(far["human"], [])
+        self.assertEqual(up["human"], [])
+
+    def test_sessions_do_not_join(self):
+        lines = rec(2, adad(0.5), still_at(450)) + rec(2, adad(0.5), still_at(450))
+        sessions = r.parse(lines)
+        self.assertEqual(len(sessions), 2)
+        frames = r.fight_frames(sessions[0])["human"] + r.fight_frames(sessions[1])["human"]
+        self.assertAlmostEqual(r.analyze(frames)["seconds"], 4, delta=0.1)
+
+    def test_report_prints(self):
+        st = r.analyze(r.fight_frames(r.parse(rec(2, adad(0.5), still_at(450)))[0])["human"])
+        self.assertIn("strafe hold", r.report(st, r.analyze([])))
+
+
+if __name__ == "__main__":
+    unittest.main()