#!/usr/bin/env python3 """Build the single self-updating SAC training dashboard. Pure-stdlib SVG output (matplotlib not available on this box). Generates ONE file: docs/campaign_dashboard.svg - four panels, current (v2) run only: 1. test-match win % vs opponents (campaign_v4_stdout.log eval lines) 2. critic_loss / |actor_loss| / alpha (training_metrics.jsonl; losses log left axis, alpha linear right axis) 3. throughput, games/hour buckets (training_metrics.jsonl 'epoch' deltas; 1 metrics row == one 10-game chunk, counts match stdout chunk markers) 4. max score per eval cycle (campaign_v4_stdout.log eval blocks; eval_log.jsonl only ever holds the LATEST cycle, so history comes from the stdout log) plus an embedded JS snippet that reloads the page every 60 s when the SVG is opened as a top-level document in Chrome. Usage: python3 tools/plot_progress.py [campaign_log] [metrics_jsonl] [outdir] All args optional; defaults relative to the SAC_LSTM_Bot/ root (parent of tools/). python3 tools/plot_progress.py --selftest # tiny built-in sanity check """ import json import math import os import re import sys import tempfile from datetime import datetime from pathlib import Path import xml.etree.ElementTree as ET ROOT = Path(__file__).resolve().parent.parent EVAL_RE = re.compile(r">>> \[eval\] win rate: (\d+)/(\d+) \(([\d.]+)%\) vs (\S+)") EVAL_BLOCK_RE = re.compile(r">>> \[eval\] \d+ deterministic rounds vs (\S+)") ROUND_RE = re.compile(r"Round \d+/\d+\D+ticks:\d+ score:(\d+) win:") TREND_WINDOW = 10 # rolling mean shown as the thick trend line (panel 1) RATE_BUCKET = 20 # metric intervals per throughput bucket (~200 games) GAMES_PER_ROW = 10 # one training_metrics.jsonl row per 10-round chunk COLORS = {"Corners": "#d62728", "Crazy": "#1f77b4", "Target": "#2ca02c", "RamFire": "#ff7f0e", "SacTwin": "#9467bd"} CRITIC_C, ACTOR_C = "#1f77b4", "#ff7f0e" ALPHA_C = "#9467bd" # alpha line on the losses panel RELOAD_JS = ('') W, H = 1400, 1720 TITLE_H, GUIDE_H = 80, 80 # rows: (header_y, panel_top_y, panel_bottom_y, x_left, x_right) # panel_top sits low enough to leave room under header+sub for the # per-panel how-to-read guide (3 italic lines, see panel_guide) C1_L, C1_R = 70, 697 C2_L, C2_R = 747, 1375 ROWS = { 1: (100, 184, 784, C1_L, C1_R), 2: (100, 184, 784, C2_L, C2_R), 3: (940, 1024, 1494, C1_L, C1_R), 4: (940, 1024, 1494, C2_L, C2_R), } PANEL_TITLES = [ "Test matches - win % vs opponents", "Training losses (log) & alpha (linear)", "Throughput - games per hour", "Max score per eval cycle", ] # per-panel how-to-read notes: (what it shows, axes, which direction is better) GUIDES = { 1: ("How often the bot wins against each opponent in test battles.", "X = training progress (match number); Y = win rate, 0-100%.", "higher is better."), 2: ("How well the brain is learning: losses should fall; alpha sets explore/exploit.", "X = training progress; left Y (log) = losses; right Y (linear) = alpha.", "lower losses = better; alpha falls over time as the bot gets confident."), 3: ("How many games per hour the bot trains (its learning speed).", "X = training progress (10-game chunks); Y = games per hour.", "higher = faster learning; a steady line beats a spiky one."), 4: ("Best single-round score the bot managed in each test cycle.", "X = eval cycle number; Y = best score achieved.", "higher is better; a rising trend means the bot is improving."), } GUIDE_LINES = [ "This file reloads itself in Chrome every sixty seconds.", "Regenerate anytime with tools/watch_dashboard.sh or the python command.", ] # ---------- parsing ---------- def parse_eval_series(path): """Return {opponent: [win% per eval, in file order]}.""" series = {} if not path.is_file(): print(f"[skip] eval log not found: {path}") return series for line in path.read_text(errors="replace").splitlines(): m = EVAL_RE.search(line) if not m: continue try: pct = float(m.group(3)) except ValueError: continue series.setdefault(m.group(4), []).append(pct) return series def parse_max_scores(path): """Return {opponent: [best single-round score per eval cycle, in file order]}. eval_log.jsonl is atomically overwritten every cycle (sac_train.sh mv), so per-cycle history only exists in the stdout log: each eval prints a '>>> [eval] N deterministic rounds vs X' header, then Round/score lines, closed by the '[eval] win rate' (or crashed / no results) line. Training rounds share the Round-line format, so they are ignored unless inside a block. """ out, cur, best = {}, None, None if not path.is_file(): print(f"[skip] campaign log not found: {path}") return out for line in path.read_text(errors="replace").splitlines(): m = EVAL_BLOCK_RE.search(line) if m: cur, best = m.group(1), None continue if cur is None: continue if "[eval]" in line: # win-rate / crashed / no-results closes the block if best is not None: out.setdefault(cur, []).append(best) cur, best = None, None continue m = ROUND_RE.search(line) if m: v = int(m.group(1)) best = v if best is None else max(best, v) return out def parse_metrics(path): """Return list of metric dicts, skipping malformed lines.""" rows = [] if not path.is_file(): print(f"[skip] metrics file not found: {path}") return rows for line in path.read_text(errors="replace").splitlines(): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def metric_col(rows, key, positive=False): """[(index, value)] for float-parseable rows; abs() applied; optional >0 filter.""" out = [] for i, r in enumerate(rows): try: v = abs(float(r[key])) except (KeyError, TypeError, ValueError): continue if positive and v <= 0: continue out.append((i, v)) return out def rolling(vals, w=TREND_WINDOW): out, s = [], 0.0 for i, v in enumerate(vals): s += v if i >= w: s -= vals[i - w] out.append(s / min(i + 1, w)) return out def bucket_means(vals, n): """Split vals into <=n contiguous buckets of near-equal size; per-bucket mean.""" if not vals: return [] n = min(n, len(vals)) k, rem = divmod(len(vals), n) out, i = [], 0 for b in range(n): size = k + (1 if b < rem else 0) out.append(sum(vals[i:i + size]) / size) i += size return out # ---------- tiny SVG helpers ---------- def esc(s): return str(s).replace("&", "&").replace("<", "<").replace(">", ">") def write_svg(path, text): """Validate the finished SVG, then atomically swap it into place. Readers never see partial output; an invalid render aborts without touching the previous good file.""" try: ET.fromstring(text) except ET.ParseError as e: print(f"[error] {path.name}: generated SVG invalid, keeping old file ({e})") return False tmp = path.with_name(path.name + ".tmp") tmp.write_text(text) os.replace(tmp, path) return True def polyline(pts, color, width=1.5, dash=None, opacity=1.0): if len(pts) < 2: return "" d = f' stroke-dasharray="{dash}"' if dash else "" p = " ".join(f"{x:.1f},{y:.1f}" for x, y in pts) return (f'\n') def dots(pts, color, r=2, opacity=0.25): return "".join(f'\n' for x, y in pts) def hgrid(x0, x1, ys, color="#dddddd"): return "".join(f'\n' for y in ys) def axis(x0, y0, x1, y1, xt, yt, xlabel, ylabel, ylog=False): """Draw axes + ticks + labels. xt/yt are (value, px) tick lists.""" s = (f'\n' f'\n') for v, px in xt: s += (f'\n' f'' f"{esc(v)}\n") for v, py in yt: s += (f'\n' f'' f"{esc(v)}\n") s += (f'' f"{esc(xlabel)}\n" f'{esc(ylabel)}\n') return s def ticks_linear(vmin, vmax, p0, p1, n=5, fmt="{:.0f}"): return [(fmt.format(vmin + (vmax - vmin) * i / (n - 1)), p0 + (p1 - p0) * i / (n - 1)) for i in range(n)] def ticks_log(vmin, vmax, p0, p1, n=5): vals = [10 ** (vmin + (vmax - vmin) * i / (n - 1)) for i in range(n)] return [("{:.3g}".format(v), p0 + (p1 - p0) * i / (n - 1)) for i, v in enumerate(vals)] def legend(items, x, y): """items: [(color, label)]""" s = "" for i, (color, label) in enumerate(items): yy = y + i * 18 s += (f'\n' f'{esc(label)}\n') return s def map_fn(p0, p1, vmin, vmax, log=False): def f(v): t = (math.log10(v) - vmin) / (vmax - vmin) if log else (v - vmin) / (vmax - vmin) return p0 + max(0.0, min(1.0, t)) * (p1 - p0) return f def guide_block(w, h, lines): """Plain-English "How to read" band at the bottom of the canvas.""" y = H - GUIDE_H s = f'\n' s += (f'' f"How to read\n" f'Regenerate anytime: python3 tools/plot_progress.py\n') for i, ln in enumerate(lines): s += f'{esc(ln)}\n' return s def header(x, y, title, sub=None): s = (f'' f"{esc(title)}\n") if sub: s += (f'' f"{esc(sub)}\n") return s def panel_guide(x0, y, guide): """Italic how-to-read note between a panel's header and its plot area.""" what, axes, better = guide s = "" for i, txt in enumerate((f"What: {what}", f"Axes: {axes}", f"Better: {better}")): s += (f'{esc(txt)}\n') return s # ---------- panels ---------- def panel_test_matches(series, geo): _, pt, pb, x0, x1 = geo s = "" if not series: s += f'' \ "no eval lines found\n" return xmax = max(max(len(v) for v in series.values()), 2) xm, ym = map_fn(x0, x1, 1, xmax), map_fn(pb, pt, 0, 100) s += hgrid(x0, x1, [ym(v) for v in range(0, 101, 20)]) step = max(1, (xmax // 8 // 10) * 10) xt = [(str(v), xm(v)) for v in range(step, xmax + 1, step)] or [("1", xm(1))] s += axis(x0, pb, x1, pt, xt, ticks_linear(0, 100, pb, pt, n=6), "test match number (each opponent)", "win rate (%)") items = [] for name in ("Corners", "Crazy", "Target"): vals = series.get(name, []) if not vals: continue c = COLORS[name] s += dots([(xm(i + 1), ym(v)) for i, v in enumerate(vals)], c) s += polyline([(xm(i + 1), ym(v)) for i, v in enumerate(rolling(vals))], c, 3.5) items.append((c, f"{name} - {len(vals)} evals")) s += legend(items, x0 + 12, pb + 52) return s def panel_max_score(maxes, geo): _, pt, pb, x0, x1 = geo s = "" if not maxes: s += f'' \ "no eval lines found\n" return ncyc = max(len(v) for v in maxes.values()) xmax = max(ncyc, 2) hi = max(1, max(v for vals in maxes.values() for v in vals)) * 1.1 xm, ym = map_fn(x0, x1, 1, xmax), map_fn(pb, pt, 0, hi) s += hgrid(x0, x1, [ym(hi * k / 4) for k in range(5)]) step = max(1, xmax // 8) xt = [(str(v), xm(v)) for v in range(step, xmax + 1, step)] or [("1", xm(1))] s += axis(x0, pb, x1, pt, xt, ticks_linear(0, hi, pb, pt, n=5), "eval cycle number", "best single-round score") items, per_opp = [], {} for name in ("Corners", "Crazy", "Target"): vals = maxes.get(name, []) if not vals: continue c = COLORS[name] pts = [(xm(i + 1), ym(v)) for i, v in enumerate(vals)] per_opp[name] = vals s += dots(pts, c) s += polyline(pts, c, 2.5) items.append((c, f"{name} - {len(vals)} evals")) if len(items) >= 2: # combined best across opponents, cycle-aligned comb = [max(vals[i] for vals in per_opp.values() if i < len(vals)) for i in range(ncyc)] s += polyline([(xm(i + 1), ym(v)) for i, v in enumerate(comb)], "#555555", 2.5, dash="6 4") items.append(("#555555", "combined max")) s += legend(items, x0 + 12, pb + 52) return s def panel_losses(rows, geo): _, pt, pb, x0, x1 = geo s = "" critic = metric_col(rows, "critic_loss", positive=True) actor = metric_col(rows, "actor_loss") # abs() applied; sign dropped actor = [(i, v) for i, v in actor if v > 0] alpha = metric_col(rows, "alpha") if not (critic or actor or alpha): s += f'' \ "no valid loss points\n" return x1 -= 46 # room on the right for the twin alpha axis labels n = len(rows) xm = lambda i: x0 + (x1 - x0) * i / max(n - 1, 1) base = critic + actor or alpha # log-domain source (losses in practice) lo = math.floor(math.log10(min(v for _, v in base))) hi = math.ceil(math.log10(max(v for _, v in base))) if lo == hi: hi = lo + 1 ym = map_fn(pb, pt, lo, hi, log=True) s += hgrid(x0, x1, [ym(10 ** e) for e in range(lo, hi + 1)]) s += axis(x0, pb, x1, pt, ticks_linear(1, n, x0, x1, n=5), ticks_log(lo, hi, pb, pt), "metric line number", "loss (log)") s += polyline([(xm(i), ym(v)) for i, v in critic], CRITIC_C, 1.8) s += polyline([(xm(i), ym(v)) for i, v in actor], ACTOR_C, 1.8) if alpha: # twin axis: alpha on its own linear scale, purple like the line ahi = max(1.0, max(v for _, v in alpha)) yma = map_fn(pb, pt, 0, ahi) s += hgrid(x0, x1, [yma(ahi * k / 4) for k in range(5)], "#e9dcf5") mid = (pt + pb) // 2 s += f'\n' for v, py in ticks_linear(0, ahi, pb, pt, n=5, fmt="{:.3g}"): s += (f'\n' f'{esc(v)}\n') s += (f'alpha\n') s += polyline([(xm(i), yma(v)) for i, v in alpha], ALPHA_C, 1.8) items = [(CRITIC_C, "critic_loss"), (ACTOR_C, "|actor_loss|")] if alpha: items.append((ALPHA_C, "alpha")) s += legend(items, x0 + 12, pb + 52) return s def panel_throughput(rows, geo): _, pt, pb, x0, x1 = geo s = "" eps = [] for r in rows: try: eps.append(float(r["epoch"])) except (KeyError, TypeError, ValueError): continue rates = [] # games/hour per inter-row interval for a, b in zip(eps, eps[1:]): dt = b - a if dt > 0: rates.append(3600.0 * GAMES_PER_ROW / dt) if not rates: s += f'' \ "no usable epoch timestamps\n" return bm = bucket_means(rates, RATE_BUCKET) xm = map_fn(x0, x1, 1, len(rates)) ymax = max(max(rates), max(bm)) * 1.1 ym = map_fn(pb, pt, 0, ymax) s += hgrid(x0, x1, [ym(ymax * k / 4) for k in range(5)]) step = max(1, len(rates) // 10) xt = [(str(v), xm(v)) for v in range(step, len(rates) + 1, step)] s += axis(x0, pb, x1, pt, xt, ticks_linear(0, ymax, pb, pt, n=5, fmt="{:.0f}"), f"chunk interval ({GAMES_PER_ROW}-game chunks)", "games / hour") s += dots([(xm(i + 1), ym(v)) for i, v in enumerate(rates)], "#7f7f7f", r=1.6) if len(bm) >= 2: ctr = [xm(round((i + 0.5) * len(rates) / len(bm))) for i in range(len(bm))] s += polyline(list(zip(ctr, map(ym, bm))), "#2ca02c", 3.5) return s # ---------- assembly ---------- def build_dashboard(campaign, metrics_f, out): series = parse_eval_series(campaign) print("[info] evals parsed: " + (", ".join(f"{k}={len(v)}" for k, v in sorted(series.items())) or "(none)")) maxes = parse_max_scores(campaign) print("[info] eval max-score cycles parsed: " + (", ".join(f"{k}={len(v)}" for k, v in sorted(maxes.items())) or "(none)")) rows = parse_metrics(metrics_f) print(f"[info] metric rows parsed: {len(rows)}") s = (f'\n' f'\n' f'SAC-LSTM campaign dashboard - live run (current only)\n' f'' f'generated {datetime.now():%Y-%m-%d %H:%M:%S} - auto-reloads every 60 s ' f'(open this file in Chrome)\n') drawers = [ (ROWS[1], PANEL_TITLES[0], GUIDES[1], "raw dots = single test matches, thick = rolling-mean-%d" % TREND_WINDOW, lambda: panel_test_matches(series, ROWS[1])), (ROWS[2], PANEL_TITLES[1], GUIDES[2], "training_metrics.jsonl - big early spikes are normal", lambda: panel_losses(rows, ROWS[2])), (ROWS[3], PANEL_TITLES[2], GUIDES[3], "method: training_metrics.jsonl 'epoch' deltas; 1 row = one 10-game chunk", lambda: panel_throughput(rows, ROWS[3])), (ROWS[4], PANEL_TITLES[3], GUIDES[4], "campaign_v4_stdout.log - max of the 10 deterministic round scores per eval", lambda: panel_max_score(maxes, ROWS[4])), ] for geo, title, guide, sub, drawer in drawers: s += header(geo[3], geo[0], title, sub) s += panel_guide(geo[3], geo[0] + 33, guide) s += drawer() or "" # panels bare-return None on their no-data path s += guide_block(W, H, GUIDE_LINES) s += RELOAD_JS + "\n" s += "\n" return write_svg(out, s) # ---------- selftest ---------- def selftest(): with tempfile.TemporaryDirectory() as td: td = Path(td) (td / "log").write_text( ">>> [eval] 10 deterministic rounds vs Corners\n" "Round 1/10 - ticks:100 score:3 win:true\n" "garbage line\n" "Round 2/10 - ticks:100 score:7 win:false\n" ">>> [eval] win rate: 3/10 (30%) vs Corners\n" ">>> [eval] 10 deterministic rounds vs Crazy\n" "Round 1/10 - ticks:100 score:70 win:false\n" ">>> [eval] win rate: 7/10 (70%) vs Crazy\n" ">>> [eval] broken\n" "Round 9/9 - ticks:1 score:999 win:false\n" ">>> [eval] 10 deterministic rounds vs Corners\n" "Round 1/10 - ticks:100 score:50 win:true\n" ">>> [eval] win rate: 5/10 (50%) vs Corners\n" ">>> [eval] 10 deterministic rounds vs Crazy\n" "Round 1/10 - ticks:100 score:40 win:true\n" ">>> [eval] win rate: 4/10 (40%) vs Crazy\n") ser = parse_eval_series(td / "log") assert ser == {"Corners": [30.0, 50.0], "Crazy": [70.0, 40.0]}, ser mxs = parse_max_scores(td / "log") # stray Round 999 after the unclosed '[eval] broken' line is ignored; # per-cycle max of the Round scores above assert mxs == {"Corners": [7, 50], "Crazy": [70, 40]}, mxs assert rolling([10] * 25, 20)[-1] == 10.0 assert rolling([1, 2, 3], 20) == [1.0, 1.5, 2.0] assert len(bucket_means(list(range(1287)), RATE_BUCKET)) == RATE_BUCKET bm = bucket_means([0, 10], RATE_BUCKET) assert bm == [0.0, 10.0], bm # fewer points than buckets -> no empty buckets (td / "m.jsonl").write_text( '{"epoch": 1000.0, "critic_loss": 10, "actor_loss": -2, "alpha": 0.5}\n' "not json\n" '{"epoch": 1060.0, "critic_loss": 100, "actor_loss": -4, "alpha": 0.25}\n' '{"epoch": 1090.0, "critic_loss": 50, "actor_loss": 3, "alpha": 0.2}\n') rows = parse_metrics(td / "m.jsonl") assert len(rows) == 3 and rows[1]["critic_loss"] == 100 assert metric_col(rows, "actor_loss") == [(0, 2.0), (1, 4.0), (2, 3.0)] dash = td / "dash.svg" assert build_dashboard(td / "log", td / "m.jsonl", dash) text = dash.read_text() ET.fromstring(text) # whole doc must parse -> closing tag present assert RELOAD_JS in text, "auto-reload script missing" for t in PANEL_TITLES: assert esc(t) in text, f"panel title missing: {t}" assert text.count(PANEL_TITLES[0]) == 1 assert "How to read" in text, "reading guide missing" assert 'width="1400"' in text and 'height="1720"' in text # every panel carries its own What/Axes/Better how-to-read note assert text.count("What:") == len(PANEL_TITLES), text.count("What:") for g in GUIDES.values(): assert esc(g[0]) in text, f"panel guide missing: {g[0]}" # circles: 4 eval dots (panel 1) + 2 throughput rate dots (panel 3) # + 2 cycles x 2 opponents max-score dots (panel 4) assert text.count("alpha<") == 2, text.count(">alpha<") assert text.count('fill="#9467bd"') == 6, text.count('fill="#9467bd"') assert "combined max" in text, "max-score combined line missing" # orientation guard: a known rising series (10% -> 90%) rendered through # the FULL build path must plot upward (smaller SVG y) and forward in # time (larger x). Fails loudly if axis mapping is ever inverted again. ori_log = td / "ori.log" ori_log.write_text( ">>> [eval] win rate: 1/10 (10%) vs Corners\n" ">>> [eval] win rate: 9/10 (90%) vs Corners\n") ori_dash = td / "ori.svg" assert build_dashboard(ori_log, td / "m.jsonl", ori_dash) m = re.search(r'90 but ink moved down ({ya} -> {yb})" assert xb > xa, f"x-axis reversed: newer eval plotted left ({xa} -> {xb})" print("selftest OK") def main(): if "--selftest" in sys.argv: selftest() return args = [a for a in sys.argv[1:] if not a.startswith("-")] campaign = Path(args[0]) if len(args) > 0 else ROOT / "campaign_v4_stdout.log" metrics = Path(args[1]) if len(args) > 1 else ROOT / "training_metrics.jsonl" outdir = Path(args[2]) if len(args) > 2 else ROOT / "docs" outdir.mkdir(parents=True, exist_ok=True) try: ok = build_dashboard(campaign, metrics, outdir / "campaign_dashboard.svg") except Exception as e: print(f"[error] dashboard build failed: {e}") ok = False if ok: print(f"[done] dashboard written to {outdir / 'campaign_dashboard.svg'}") else: print("[error] dashboard NOT updated - check paths above") sys.exit(1) if __name__ == "__main__": main()