| |
| """ |
| 绘图脚本 —— 用 PIL 手绘学习曲线 / 损失曲线 / 评估分布 / 基线对比 |
| ================================================================= |
| （工程环境里没有 matplotlib，所以用 PIL 直接画图，零额外依赖） |
| |
| 用法:: |
| |
| python plot_curves.py --run runs/main_2.5m --baseline runs/baseline_50k --out artifacts |
| |
| 产物（artifacts/ 下）： |
| learning_curve.png 训练回报 + 评估回报随步数变化（含 200 分"解决"线） |
| losses.png policy loss / value loss / entropy / KL |
| eval_progress.png 每次评估的均值 ± 标准差 |
| eval_histogram.png 100 局评估的回报分布（对比随机策略） |
| compare_baseline.png 课件 5 万步命令 vs 正式训练的对比 |
| """ |
|
|
| import argparse |
| import csv |
| import json |
| import os |
|
|
| import numpy as np |
| from PIL import Image, ImageDraw, ImageFont |
|
|
| C = { |
| "blue": (86, 156, 255), "green": (80, 200, 120), "orange": (255, 165, 60), |
| "red": (240, 90, 90), "purple": (170, 120, 240), "gray": (150, 150, 160), |
| "cyan": (90, 210, 220), "yellow": (240, 210, 80), |
| } |
| BG = (18, 20, 25) |
| PANEL_BG = (30, 33, 40) |
| GRID = (55, 60, 70) |
| TXT = (235, 235, 240) |
| TXT_DIM = (190, 190, 195) |
|
|
|
|
| |
| |
| |
| def _font(size): |
| for name in ("arial.ttf", "segoeui.ttf", "msyh.ttc"): |
| try: |
| return ImageFont.truetype(name, size) |
| except Exception: |
| continue |
| return ImageFont.load_default() |
|
|
|
|
| def _rot_text(img, xy, text, font, fill): |
| """把文字旋转 90° 画到 img 上（用于 y 轴标题）。""" |
| tmp = Image.new("RGBA", (240, 30), (0, 0, 0, 0)) |
| ImageDraw.Draw(tmp).text((0, 0), text, font=font, fill=fill) |
| tmp = tmp.crop(tmp.getbbox() or (0, 0, 1, 1)).rotate(90, expand=True) |
| img.paste(tmp, (int(xy[0]), int(xy[1])), tmp) |
|
|
|
|
| def _nan_safe(vals): |
| return [v if np.isfinite(v) else 0.0 for v in vals] |
|
|
|
|
| def draw_panel(img, box, series, title, xlabel, ylabel, |
| hline=None, legend=True, xticklabels=None, ylim=None): |
| """在 box=(x0,y0,x1,y1) 里画坐标系。 |
| |
| series 元素：{"name","xs","ys","color","kind":"line"/"scatter"/"bar","width"} |
| xticklabels: 给定则 x 轴用分类标签（xs 为 0..n-1）。 |
| ylim: (ymin, ymax) 强制 y 轴范围（超出范围的点不画，避免极端离群点压扁曲线）。 |
| """ |
| d = ImageDraw.Draw(img) |
| x0, y0, x1, y1 = box |
| pad_l, pad_r, pad_t, pad_b = 78, 18, 48, 48 |
| px0, py0 = x0 + pad_l, y0 + pad_t |
| px1, py1 = x1 - pad_r, y1 - pad_b |
| d.rectangle([px0, py0, px1, py1], fill=PANEL_BG) |
|
|
| xs_all, ys_all = [], [] |
| for s in series: |
| xs_all.extend(s["xs"]) |
| ys_all.extend(s["ys"]) |
| if hline is not None: |
| ys_all.append(hline[0]) |
| if not xs_all: |
| return |
| xmin, xmax = float(min(xs_all)), float(max(xs_all)) |
| if ylim is not None: |
| ymin, ymax = float(ylim[0]), float(ylim[1]) |
| else: |
| ymin, ymax = float(min(ys_all)), float(max(ys_all)) |
| pad_y = (ymax - ymin) * 0.10 |
| ymin, ymax = ymin - pad_y, ymax + pad_y |
| if xmax == xmin: |
| xmax = xmin + 1 |
| if ymax == ymin: |
| ymax = ymin + 1 |
|
|
| def px(xv): |
| return px0 + (xv - xmin) / (xmax - xmin) * (px1 - px0) |
|
|
| def py(yv): |
| return py1 - (yv - ymin) / (ymax - ymin) * (py1 - py0) |
|
|
| |
| for i in range(5): |
| yv = ymin + (ymax - ymin) * i / 4 |
| y = py(yv) |
| d.line([px0, y, px1, y], fill=GRID, width=1) |
| label = f"{yv:,.0f}" if abs(yv) >= 1 else f"{yv:.2f}" |
| d.text((x0 + 30, y - 8), label, font=_font(13), fill=TXT_DIM) |
| if xticklabels is None: |
| xv = xmin + (xmax - xmin) * i / 4 |
| x = px(xv) |
| d.line([x, py0, x, py1], fill=GRID, width=1) |
| lab = f"{xv/1e6:.1f}M" if abs(xv) >= 1e6 else ( |
| f"{xv/1e3:.0f}k" if abs(xv) >= 1e3 else ( |
| f"{xv:.0f}" if abs(xv) >= 10 else f"{xv:.1f}")) |
| d.text((x - 14, py1 + 10), lab, font=_font(13), fill=TXT_DIM) |
|
|
| |
| if xticklabels is not None: |
| n = len(xticklabels) |
| width = (px1 - px0) / max(1, n) |
| for i, lab in enumerate(xticklabels): |
| x = px0 + width * (i + 0.5) |
| d.text((x - d.textlength(lab, font=_font(13)) / 2, py1 + 10), |
| lab, font=_font(13), fill=TXT_DIM) |
|
|
| |
| if hline is not None: |
| yv, label = hline[0], hline[1] |
| color = hline[2] if len(hline) > 2 else C["red"] |
| y = py(yv) |
| for xx in range(int(px0), int(px1), 14): |
| d.line([xx, y, min(xx + 7, px1), y], fill=color, width=2) |
| d.text((px1 - d.textlength(label, font=_font(13)) - 6, y - 19), |
| label, font=_font(13), fill=color) |
|
|
| |
| for s in series: |
| color = C.get(s.get("color", "blue"), C["blue"]) |
| kind, width = s.get("kind", "line"), s.get("width", 2) |
| raw = [(xv, yv) for xv, yv in zip(s["xs"], s["ys"]) if ymin <= yv <= ymax] |
| pts = [(px(xv), py(yv)) for xv, yv in raw] |
| if kind == "bar": |
| if xticklabels is not None: |
| slot = (px1 - px0) / len(xticklabels) |
| bw = slot * 0.5 |
| centers = [px0 + slot * (i + 0.5) for i in range(len(s["ys"]))] |
| else: |
| bw = max(4, (px1 - px0) / max(1, len(pts)) * 0.7) |
| centers = [x for x, _ in pts] |
| for cx, (_, cy) in zip(centers, pts): |
| d.rectangle([cx - bw / 2, min(cy, py1), cx + bw / 2, max(cy, py1)], fill=color) |
| elif kind == "scatter": |
| for (cx, cy) in pts: |
| d.ellipse([cx - 3, cy - 3, cx + 3, cy + 3], fill=color) |
| else: |
| if len(pts) > 1: |
| d.line(pts, fill=color, width=width, joint="curve") |
|
|
| |
| d.text((x0 + pad_l, y0 + 10), title, font=_font(19), fill=TXT) |
| d.text((px0 + (px1 - px0) / 2 - d.textlength(xlabel, font=_font(13)) / 2, y1 - 20), |
| xlabel, font=_font(13), fill=TXT_DIM) |
| _rot_text(img, (x0 + 3, (py0 + py1) / 2 - 60), ylabel, _font(13), TXT_DIM) |
|
|
| |
| if legend: |
| entries = [(s["name"], s.get("color", "blue")) for s in series] |
| lh = 22 * len(entries) + 10 |
| lw = 16 + max(d.textlength(n, font=_font(13)) for n, _ in entries) + 34 |
| d.rectangle([px0 + 8, py0 + 8, px0 + 8 + lw, py0 + 8 + lh], fill=(24, 27, 33)) |
| ly = py0 + 14 |
| for name, color in entries: |
| d.line([px0 + 16, ly + 7, px0 + 38, ly + 7], fill=C[color], width=3) |
| d.text((px0 + 46, ly), name, font=_font(13), fill=TXT) |
| ly += 22 |
|
|
|
|
| def moving_average(xs, ys, window): |
| if len(ys) <= window: |
| return xs, ys |
| ys = _nan_safe(ys) |
| out = [float(np.mean(ys[max(0, i - window + 1):i + 1])) for i in range(len(ys))] |
| return xs, out |
|
|
|
|
| |
| |
| |
| def read_csv_rows(path): |
| if not os.path.exists(path): |
| return [] |
| with open(path, "r", encoding="utf-8-sig") as f: |
| return list(csv.DictReader(f)) |
|
|
|
|
| def load_run(run_dir): |
| train = read_csv_rows(os.path.join(run_dir, "train_log.csv")) |
| evals = read_csv_rows(os.path.join(run_dir, "eval_log.csv")) |
| summary = {} |
| sp = os.path.join(run_dir, "summary.json") |
| if os.path.exists(sp): |
| with open(sp, "r", encoding="utf-8") as f: |
| summary = json.load(f) |
| return train, evals, summary |
|
|
|
|
| |
| |
| |
| def plot_learning_curve(run_dir, out_path, title): |
| train, evals, _ = load_run(run_dir) |
| if not train: |
| print(f"[跳过] 缺 {run_dir}/train_log.csv") |
| return |
| steps = [int(r["global_step"]) for r in train] |
| rets = [float(r["mean_return"]) for r in train] |
| series = [] |
| if any(np.isfinite(rets)): |
| series.append({"name": "train return (raw)", "xs": steps, "ys": rets, |
| "color": "gray", "kind": "line", "width": 1}) |
| mx, my = moving_average(steps, rets, max(5, len(steps) // 60)) |
| series.append({"name": "train return (moving avg)", "xs": mx, "ys": my, |
| "color": "blue", "kind": "line", "width": 3}) |
| if evals: |
| ex = [int(r["global_step"]) for r in evals] |
| ey = [float(r["eval_mean"]) for r in evals] |
| series.append({"name": "eval (argmax policy)", "xs": ex, "ys": ey, |
| "color": "green", "kind": "line", "width": 3}) |
| series.append({"name": "eval points", "xs": ex, "ys": ey, |
| "color": "green", "kind": "scatter"}) |
| W, H = 1150, 580 |
| img = Image.new("RGB", (W, H), BG) |
| draw_panel(img, (10, 10, W - 10, H - 10), series, title, |
| "environment steps", "episode return", |
| hline=(200.0, "solved threshold = 200", C["red"]), |
| ylim=(-320, 330)) |
| d = ImageDraw.Draw(img) |
| d.text((100, 44), |
| "note: the first 2 argmax evals (-1983 / -142) are off-scale — early policy not yet deterministic, see README", |
| font=_font(12), fill=(150, 150, 160)) |
| img.save(out_path) |
| print(f"[图] {out_path}") |
|
|
|
|
| |
| |
| |
| def plot_losses(run_dir, out_path): |
| train, _, _ = load_run(run_dir) |
| if not train: |
| return |
| steps = [int(r["global_step"]) for r in train] |
| W, H = 1150, 800 |
| img = Image.new("RGB", (W, H), BG) |
| panels = [("policy_loss", "1) policy loss (PPO-Clip)", "orange"), |
| ("value_loss", "2) value loss", "purple"), |
| ("entropy", "3) policy entropy", "cyan"), |
| ("approx_kl", "4) approximate KL", "green")] |
| for i, (key, title, color) in enumerate(panels): |
| ys = [float(r[key]) for r in train] |
| r, c = divmod(i, 2) |
| box = (10 + c * (W // 2), 10 + r * (H // 2), 10 + (c + 1) * (W // 2) - 10, |
| 10 + (r + 1) * (H // 2) - 10) |
| mx, my = moving_average(steps, ys, max(3, len(ys) // 80)) |
| draw_panel(img, box, |
| [{"name": "raw", "xs": steps, "ys": _nan_safe(ys), "color": "gray", "width": 1}, |
| {"name": "moving avg", "xs": mx, "ys": my, "color": color, "width": 2}], |
| title, "environment steps", key) |
| img.save(out_path) |
| print(f"[图] {out_path}") |
|
|
|
|
| |
| |
| |
| def plot_eval_progress(run_dir, out_path): |
| _, evals, _ = load_run(run_dir) |
| if not evals: |
| return |
| steps = [int(r["global_step"]) for r in evals] |
| means = [float(r["eval_mean"]) for r in evals] |
| stds = [float(r["eval_std"]) for r in evals] |
| W, H = 1150, 540 |
| img = Image.new("RGB", (W, H), BG) |
| d = ImageDraw.Draw(img) |
| ylo, yhi = -260.0, 340.0 |
| draw_panel(img, (10, 10, W - 10, H - 10), |
| [{"name": "eval mean", "xs": steps, "ys": means, "color": "green", "width": 3}], |
| "evaluation during training (argmax policy)", "environment steps", "eval return", |
| hline=(200.0, "solved threshold = 200", C["red"]), ylim=(ylo, yhi)) |
| |
| px0, py0, px1, py1 = 10 + 78, 10 + 48, W - 10 - 18, H - 10 - 48 |
| xmin, xmax = min(steps), max(steps) |
| for xv, m, s in zip(steps, means, stds): |
| x = px0 + (xv - xmin) / max(1, xmax - xmin) * (px1 - px0) |
| hi_, lo_ = m + s, m - s |
| y_hi = py1 - (min(max(hi_, ylo), yhi) - ylo) / (yhi - ylo) * (py1 - py0) |
| y_lo = py1 - (min(max(lo_, ylo), yhi) - ylo) / (yhi - ylo) * (py1 - py0) |
| d.line([x, y_hi, x, y_lo], fill=C["blue"], width=2) |
| d.line([x - 4, y_hi, x + 4, y_hi], fill=C["blue"], width=2) |
| d.line([x - 4, y_lo, x + 4, y_lo], fill=C["blue"], width=2) |
| img.save(out_path) |
| print(f"[图] {out_path}") |
|
|
|
|
| |
| |
| |
| def plot_eval_histogram(eval_json, out_path): |
| if not os.path.exists(eval_json): |
| return |
| with open(eval_json, "r", encoding="utf-8") as f: |
| data = json.load(f) |
| returns = np.asarray(data["ppo"]["returns"], dtype=float) |
| rand = data.get("random_baseline") |
| lo, hi = float(returns.min()), float(returns.max()) |
| pad = max(5.0, (hi - lo) * 0.05) |
| bins = np.linspace(lo - pad, hi + pad, 15) |
| hist, edges = np.histogram(returns, bins=bins) |
| centers = (edges[:-1] + edges[1:]) / 2 |
| W, H = 1150, 540 |
| img = Image.new("RGB", (W, H), BG) |
| draw_panel(img, (10, 10, W - 10, H - 10), |
| [{"name": f"PPO agent ({len(returns)} episodes, argmax)", |
| "xs": centers.tolist(), "ys": hist.tolist(), "color": "blue", "kind": "bar"}], |
| "evaluation return distribution (argmax policy)", "episode return", |
| "number of episodes", |
| hline=(200.0, "solved threshold = 200", C["red"]), |
| ylim=(0, max(1, int(hist.max()) * 1.25))) |
| if rand: |
| d = ImageDraw.Draw(img) |
| txt = (f"random policy baseline: mean {rand['mean']:.0f} ± {rand['std']:.0f}, " |
| f"landing rate {rand['landing_rate']*100:.0f}% (far to the left, off-scale)") |
| d.text((100, 84), txt, font=_font(13), fill=C["orange"]) |
| img.save(out_path) |
| print(f"[图] {out_path}") |
|
|
|
|
| |
| |
| |
| def plot_compare(run_dir, baseline_dir, out_path): |
| _, evals, summary = load_run(run_dir) |
| _, base_evals, base_summary = load_run(baseline_dir) |
| if not evals: |
| return |
| items = [] |
| if base_evals: |
| items.append((f"course command\n{int(base_summary.get('total_timesteps', 0))/1000:.0f}k steps", |
| float(base_evals[-1]["eval_mean"]))) |
| items.append((f"this run\n{summary.get('total_timesteps', 0)/1e6:.1f}M steps", |
| float(evals[-1]["eval_mean"]))) |
|
|
| W, H = 980, 460 |
| img = Image.new("RGB", (W, H), BG) |
| d = ImageDraw.Draw(img) |
| d.text((24, 16), "course command baseline vs. full PPO training", font=_font(19), fill=TXT) |
| left, right, top = 180, W - 150, 80 |
| vmin = min(v for _, v in items + [("", 0.0)]) |
| vmax = max(v for _, v in items + [("", 220.0)]) |
| span = vmax - vmin |
|
|
| def vx(v): |
| return left + (v - vmin) / span * (right - left) |
|
|
| rowh = 78 |
| for i, (label, value) in enumerate(items): |
| y = top + i * rowh |
| d.text((20, y + 10), label, font=_font(14), fill=TXT_DIM) |
| color = C["gray"] if i == 0 and len(items) > 1 else C["green"] |
| x0, x1 = sorted([vx(0), vx(value)]) |
| d.rectangle([x0, y, x1, y + 40], fill=color) |
| tx = max(vx(value), vx(0.0)) + 8 |
| d.text((tx, y + 10), f"{value:+.1f}", font=_font(16), fill=TXT) |
| |
| for val, color, lab in [(0.0, C["gray"], "0"), (200.0, C["red"], "solved = 200")]: |
| x = vx(val) |
| d.line([x, top - 12, x, top + len(items) * rowh], fill=color, width=2) |
| d.text((x - 20, top - 34), lab, font=_font(13), fill=color) |
| img.save(out_path) |
| print(f"[图] {out_path}") |
|
|
|
|
| def main(): |
| p = argparse.ArgumentParser(description="画学习曲线与评估图") |
| p.add_argument("--run", type=str, default="runs/main_2.5m") |
| p.add_argument("--baseline", type=str, default="runs/baseline_50k") |
| p.add_argument("--out", type=str, default="artifacts") |
| p.add_argument("--title", type=str, default="PPO on LunarLander — implemented from scratch") |
| p.add_argument("--eval-json", type=str, default="") |
| args = p.parse_args() |
| os.makedirs(args.out, exist_ok=True) |
|
|
| plot_learning_curve(args.run, os.path.join(args.out, "learning_curve.png"), args.title) |
| plot_losses(args.run, os.path.join(args.out, "losses.png")) |
| plot_eval_progress(args.run, os.path.join(args.out, "eval_progress.png")) |
| plot_compare(args.run, args.baseline, os.path.join(args.out, "compare_baseline.png")) |
| plot_eval_histogram(args.eval_json or os.path.join(args.out, "eval_results.json"), |
| os.path.join(args.out, "eval_histogram.png")) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
