ql-lg-server/tools/lgbot_rec.py

#!/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 nearest opponent (human or bot), so human-vs-human recordings (lgbot_record 2) work
too: "humans" is every human together. 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):
    """-> {who: [...]} ("bot", "h1", ...), frames (t, x, y, z, vx, vy, vz, ox, oy, state) in time order.
    Each alive player is measured against the nearest alive player in fight range (human or bot)."""
    out = {}
    for t in sorted(session):
        alive = [(who, r) for who, r in session[t].items() if r[0]]
        for who, me in alive:
            best = None
            for other, r in alive:
                if other == who or abs(r[4] - me[4]) > FIGHT_DZ:
                    continue
                d = math.hypot(r[2] - me[2], r[3] - me[3])
                if d <= FIGHT_DIST and (best is None or d < best[0]):
                    best = (d, r)
            if best is not None:
                out.setdefault(who, []).append(
                    (t,) + me[2:8] + (best[1][2], best[1][3], me[8]))
    return out


def by_kind(per_player):
    """{who: frames} -> {"human": frames, "bot": frames}: one player after another, so time jumps
    back at each join and the gap rule keeps players (and sessions) apart."""
    out = {"human": [], "bot": []}
    for who in sorted(per_player):
        out["bot" if who == "bot" else "human"].extend(per_player[who])
    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" % ("", "humans", "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)
    per_player = {}
    for s in sessions:
        for who, frames in fight_frames(s).items():
            per_player.setdefault(who, []).extend(frames)  # t restarts each session: a gap
    both = by_kind(per_player)
    print("%d session(s)" % len(sessions))
    print(report(analyze(both["human"]), analyze(both["bot"])))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))