add docs+code
880dff9 verified about 2 months ago
57.5 kB
#!/usr/bin/env python3
"""bg50 四科判决（只跑 ARoPE 臂）：复现旧报告《世界RoPE·四科判决》的口径。
考场：data/eval/eybx_bg_50/{01..50}_*.png（50 个未见背景，中央已合成玩家），文本条件 NULL_PROMPT
（训练时场景 dropout 用的串；这些背景不属于 18 个训练场景，不配场景从句，--scene 可人为覆盖）。
四科（度量全在 CPU 上只读 mp4，位移一律用 verify_sidecar.measure_shift：SIFT+RANSAC 仿射场在玩家位置取值，
返回的是"背景位移"= −frame_offset_px）：
  1. 增益线性  gain_<bg>_x<mult>  10 个背景 × {0.5, 0.75, 1.0, 1.25, 1.5}× 直线右走；帧 0→80 实测位移 vs 指令位移过原点回归，报斜率 / R²。
  2. 折返闭环  back_<bg>          50 个背景，1.0× 步速右走 40 帧再左走 40 帧回原点（frame_offset[80] = 0）；
                                   终帧 vs 首帧的 PSNR / SSIM / NCC（明暗对齐后、中心 160×160 以外），回归残差（终帧相对首帧的实测位置误差 px），
                                   位置补偿后 NCC；守门：帧 0→40 的实走距离。
  3. 运动平滑  straight_<bg>      50 个背景，1.0× 直线右走；逐帧水平速度（相邻帧 SIFT，失败退化为隔 2 帧）：倒退帧数、速度 std/mean、归一 jerk、平均速度/承诺。
  4. 方向服从  同一批 straight    帧 0→80 累计位移（measure_displacement 的 sift 链式）与指令的角度误差 / 幅度比 / 失控数。
用法：
    python -m actionrope.eval_bg50 --ckpt outputs/arope_5k/step-5000.safetensors --run step5000 --stages gen,measure --gpus 0-7
    python -m actionrope.eval_bg50 --run step5000 --stages measure            # 只重算度量
    python -m actionrope.eval_bg50 --selfcheck                                 # 合成数据自检（不需要 GPU）
产物 outputs/eval_bg50/<run>/：plan.jsonl、videos/<id>.mp4 + <id>.json、metrics.jsonl、summary.json、report.md、figures.png、selfcheck.json。
"""
from __future__ import annotations

import argparse
import json
import math
import os
import subprocess
import sys
import time
from concurrent.futures import ProcessPoolExecutor

import numpy as np

os.environ.setdefault("DIFFSYNTH_SKIP_DOWNLOAD", "True")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
from actionrope import geometry as G  # noqa: E402

PY = os.path.join(ROOT, ".venv", "bin", "python")
EVAL_DIR = os.path.join(ROOT, "data", "eval", "eybx_bg_50")
OUT_ROOT = os.path.join(ROOT, "outputs", "eval_bg50")

GAIN_MULTS = (0.5, 0.75, 1.0, 1.25, 1.5)
GAIN_N_BG = 10
BACK_MULT = 1.0   # 折返也用 1 倍速（人定：全部科目统一 1×）
STRAIGHT_MULT = 1.0
BACK_TURN = 40                # 帧 1..40 向右、41..80 向左
CENTER_BOX = 160              # 折返相似度剔除的中心角色框（px，正方形）
ANGLE_FAIL_DEG = 15.0
RATIO_FAIL = 0.5

# 旧报告《世界RoPE·四科判决》里世界 RoPE@5k 的数字（任务给出的部分；其余未给出记 None）
OLD_REPORT = {
    "gain_slope": 1.06, "gain_r2": 0.95,
    "back_psnr_med": None, "back_ssim_med": None, "back_ncc_med": None,
    "back_resid_med": None, "back_ncc_comp_med": None,
    "smooth_backward_frames_total": 6,
    "smooth_speed_cv_med": None, "smooth_jerk_med": None, "smooth_speed_ratio_med": None,
    "dir_angle_med": None, "dir_angle_p90": None, "dir_ratio_med": None, "dir_fail": None,
}


# ------------------------------------------------------------------ 素材 / 计划 ----
def list_backgrounds() -> list[dict]:
    """manifest.jsonl 里 status=ok 的背景，按文件名的两位序号排序。"""
    rows = []
    with open(os.path.join(EVAL_DIR, "manifest.jsonl"), encoding="utf-8") as fh:
        for line in fh:
            if line.strip():
                rows.append(json.loads(line))
    rows = [r for r in rows if r.get("status") == "ok"]
    for r in rows:
        r["idx"] = int(r["file"][:2])
        r["path"] = os.path.join(EVAL_DIR, r["file"])
    rows.sort(key=lambda r: r["idx"])
    return rows


def parse_bgs(spec: str | None, rows: list[dict]) -> list[dict]:
    if not spec:
        return rows
    want = set()
    for tok in spec.split(","):
        tok = tok.strip()
        if "-" in tok:
            a, b = tok.split("-")
            want.update(range(int(a), int(b) + 1))
        elif tok:
            want.add(int(tok))
    return [r for r in rows if r["idx"] in want]


def parse_gpus(spec: str) -> list[int]:
    out = []
    for tok in spec.split(","):
        tok = tok.strip()
        if "-" in tok:
            a, b = tok.split("-")
            out += list(range(int(a), int(b) + 1))
        elif tok:
            out.append(int(tok))
    return out


def straight_offsets(vx: float) -> np.ndarray:
    """直线右走：frame_offset[i] = (i/16 · vx, 0)，玩家向右 ⇒ 背景左移。"""
    off = np.zeros((G.FRAMES, 2), np.float64)
    off[:, 0] = np.arange(G.FRAMES) / G.FPS * vx
    return off


def back_offsets(vx: float, turn: int = BACK_TURN) -> np.ndarray:
    """折返：帧 1..turn−1 每帧 +vx/16，帧 turn / turn+1 各 +½ / −½（2 帧线性掉头），帧 turn+2..80 每帧 −vx/16。
    正负步数严格相等 ⇒ frame_offset[80] = 0（不靠 cell 数凑）。"""
    vel = np.zeros(G.FRAMES, np.float64)
    vel[1:turn] = vx
    vel[turn] = 0.5 * vx
    vel[turn + 1] = -0.5 * vx
    vel[turn + 2:] = -vx
    off = np.zeros((G.FRAMES, 2), np.float64)
    off[:, 0] = np.cumsum(vel) / G.FPS
    assert abs(off[-1, 0]) < 1e-9, off[-1, 0]
    return off


def build_plan(rows: list[dict], walk_x: float, seed: int) -> list[dict]:
    plan = []
    for r in rows[:GAIN_N_BG]:
        for m in GAIN_MULTS:
            off = straight_offsets(walk_x * m)
            plan.append(dict(id=f"gain_{r['idx']:02d}_x{m:.2f}", exam="gain", bg=r["idx"], bg_file=r["file"],
                             mult=m, instruction=f"right x{m:.2f} straight", frame_offset_px=off.tolist(),
                             expected_bg_shift_80=(-off[80]).tolist(), seed=seed))
    for r in rows:
        off = back_offsets(walk_x * BACK_MULT)
        plan.append(dict(id=f"back_{r['idx']:02d}", exam="back", bg=r["idx"], bg_file=r["file"], mult=BACK_MULT,
                         instruction=f"right x{BACK_MULT} 40 frames then left x{BACK_MULT} 40 frames",
                         frame_offset_px=off.tolist(), expected_bg_shift_80=(-off[80]).tolist(),
                         expected_bg_shift_turn=(-off[BACK_TURN]).tolist(), turn_frame=BACK_TURN, seed=seed))
    for r in rows:
        off = straight_offsets(walk_x * STRAIGHT_MULT)
        plan.append(dict(id=f"straight_{r['idx']:02d}", exam="straight", bg=r["idx"], bg_file=r["file"],
                         mult=STRAIGHT_MULT, instruction=f"right x{STRAIGHT_MULT:.2f} straight",
                         frame_offset_px=off.tolist(), expected_bg_shift_80=(-off[80]).tolist(), seed=seed))
    return plan


def video_done(vdir: str, vid: str) -> bool:
    """mp4 与 json 都在，且 json 记录的字节数与 mp4 实际一致 ⇒ 视为完整，可跳过。"""
    mp4, js = os.path.join(vdir, vid + ".mp4"), os.path.join(vdir, vid + ".json")
    if not (os.path.isfile(mp4) and os.path.isfile(js)):
        return False
    try:
        with open(js, encoding="utf-8") as fh:
            info = json.load(fh)
        return bool(info.get("ok")) and info.get("mp4_bytes") == os.path.getsize(mp4) and info.get("frames") == G.FRAMES
    except (json.JSONDecodeError, OSError):
        return False


# ------------------------------------------------------------------ 生成（GPU 子进程）----
def gen_worker(run_dir: str, ckpt: str, rank: int, world: int, steps: int, shift: float, scene: str | None,
               text_table: str | None):
    """一个子进程只加载一次模型，跑 plan 里 idx % world == rank 的条目。"""
    import torch
    from actionrope import infer as I
    from actionrope.arope import loss_weight_map, MASK_EMBEDDING_NAME
    from actionrope.prompts import NULL_PROMPT, strip_action
    from diffsynth.utils.data import save_video

    plan = [json.loads(l) for l in open(os.path.join(run_dir, "plan.jsonl"), encoding="utf-8") if l.strip()]
    mine = [p for i, p in enumerate(plan) if i % world == rank]
    vdir = os.path.join(run_dir, "videos")
    os.makedirs(vdir, exist_ok=True)
    todo = [p for p in mine if not video_done(vdir, p["id"])]
    log = lambda s: print(f"[gen rank{rank}] {s}", flush=True)
    log(f"{len(mine)} 条，其中 {len(todo)} 条待生成")
    if not todo:
        return
    t0 = time.time()
    # 臂按 ckpt 自动识别（baseline/SPEC.md）：arope / linear / xattn / adaln 文本剥动作词；plain / prompt 动作走文本
    pipe, dit, arm = I.load_models(ckpt, None, "cuda")
    arm_name, _ = I.resolve_arm(ckpt, None)
    text_arm = arm_name in I.TEXT_ARMS
    has_mask = hasattr(dit, MASK_EMBEDDING_NAME)
    enc = I.TextEncoder(text_table or I.TEXT_TABLE["plain" if text_arm else "arope"], "cuda")
    per_cell = getattr(arm, "per_cell_context", False) and getattr(arm, "require_per_cell", False)
    if scene:
        prompts = [strip_action(p) for p in I.compose_cell_prompts(scene, None, [7] * G.CELLS)]
    else:
        prompts = [NULL_PROMPT] * G.CELLS
    context = context_ids = None
    if not text_arm:  # 动作走文本的臂每条视频的文本不同，在循环里按指令拼
        context, context_ids = I.build_context(enc, prompts, "cuda", per_cell=per_cell)
    log(f"模型加载 {time.time() - t0:.0f}s · arm={arm_name} mask_channel={has_mask} · 文本 {prompts[0]!r} · t5_fallback={enc.n_fallback}")

    z0_cache: dict[str, torch.Tensor] = {}
    for p in todo:
        t1 = time.time()
        info = dict(id=p["id"], exam=p["exam"], bg=p["bg"], bg_file=p["bg_file"], instruction=p["instruction"],
                    mult=p["mult"], frame_offset_px=p["frame_offset_px"], expected_bg_shift_80=p["expected_bg_shift_80"],
                    ckpt=ckpt, steps=steps, shift=shift, seed=p["seed"], prompt=prompts[0], mask_channel=has_mask)
        try:
            if p["bg_file"] not in z0_cache:
                z0_cache.clear()
                img = I.load_first_frame(os.path.join(EVAL_DIR, p["bg_file"]))
                z0_cache[p["bg_file"]] = (img, I.encode_first_frame(pipe, img))
            img, z0 = z0_cache[p["bg_file"]]
            frame_off = np.array(p["frame_offset_px"], np.float64)
            offset_px = G.frames_to_cells(frame_off)
            info["offset_px"] = offset_px.tolist()
            # 逐 cell 真实屏幕方向标签（与 infer.main 的 true_labels 同一算法）：cell 速度 px/s，一个 cell = 4 帧 = 0.25 s
            vel = np.zeros_like(offset_px)
            vel[1:] = (offset_px[1:] - offset_px[:-1]) / 0.25
            true_labels = [I.velocity_to_label(v, mirror=False) for v in vel]
            info["action_idx"] = true_labels
            info["arm"] = arm_name
            off_px_t = torch.tensor(offset_px, dtype=torch.float32).unsqueeze(0)
            if text_arm:
                # 动作走文本：无场景从句时用数据集的裸动作串（场景 dropout 用的那 9 条，表里有）
                cell_prompts = (I.compose_cell_prompts(scene, None, true_labels) if scene
                                else [I.BARE_TMPL.format(action=I.ACTIONS[l]) for l in true_labels])
                ctx, ctx_ids = I.build_context(enc, cell_prompts, "cuda", per_cell=per_cell)
                info["prompt"] = cell_prompts[1]
            else:
                ctx, ctx_ids = context, context_ids
            if arm_name == "arope":
                off_t = off_px_t.to("cuda")
                # 与 infer.main 一致：mask 通道输入 = loss_weight_map(...)==1（known），有 mask 通道才给
                mask_in = loss_weight_map(off_t).eq(1).float() if has_mask else None
            else:
                off_t, mask_in = None, None  # baseline 臂：普通 RoPE、无 mask，动作走 action_inputs
            action_inputs = None
            if arm is not None:
                action_inputs = {k: v.to("cuda") for k, v in I.make_action_inputs(off_px_t, torch.tensor([true_labels])).items()}
            ts = time.time()
            latents = I.sample(dit, z0, ctx, ctx_ids, off_t, mask_in, steps=steps, shift=shift,
                               cfg=1.0, seed=p["seed"], arm=arm, action_inputs=action_inputs, log=lambda s: None)
            torch.cuda.synchronize()
            info["sample_sec"] = time.time() - ts
            ts = time.time()
            frames = I.decode_video(pipe, latents)
            info["decode_sec"] = time.time() - ts
            mp4 = os.path.join(vdir, p["id"] + ".mp4")
            save_video(list(frames), mp4, fps=int(G.FPS), quality=9)
            info.update(ok=True, frames=int(frames.shape[0]), resolution=[int(frames.shape[2]), int(frames.shape[1])],
                        mp4_bytes=os.path.getsize(mp4), total_sec=time.time() - t1,
                        peak_mem_gb=torch.cuda.max_memory_allocated() / 1e9)
        except Exception as e:  # 单条失败不拖垮分片
            info.update(ok=False, error=repr(e), total_sec=time.time() - t1)
            log(f"{p['id']} FAILED: {e!r}")
        with open(os.path.join(vdir, p["id"] + ".json"), "w", encoding="utf-8") as fh:
            json.dump(info, fh, ensure_ascii=False, indent=1, default=float)
        log(f"{p['id']} · {info['total_sec']:.0f}s · ok={info['ok']}")
    log(f"分片完成 · {time.time() - t0:.0f}s")


def stage_gen(args, run_dir: str) -> dict:
    gpus = parse_gpus(args.gpus)
    plan = [json.loads(l) for l in open(os.path.join(run_dir, "plan.jsonl"), encoding="utf-8") if l.strip()]
    vdir = os.path.join(run_dir, "videos")
    os.makedirs(vdir, exist_ok=True)
    n_todo = sum(not video_done(vdir, p["id"]) for p in plan)
    print(f"[gen] {len(plan)} 条计划，{n_todo} 条待生成，{len(gpus)} 张卡 {gpus}", flush=True)
    t0 = time.time()
    procs, pids = [], []
    log_dir = os.path.join(run_dir, "logs")
    os.makedirs(log_dir, exist_ok=True)
    for rank, gpu in enumerate(gpus):
        env = dict(os.environ, CUDA_VISIBLE_DEVICES=str(gpu), PYTHONPATH=ROOT, DIFFSYNTH_SKIP_DOWNLOAD="True")
        cmd = [PY, "-m", "actionrope.eval_bg50", "--_worker", "--run", args.run, "--ckpt", args.ckpt,
               "--rank", str(rank), "--world", str(len(gpus)), "--steps", str(args.steps), "--shift", str(args.shift)]
        if args.scene:
            cmd += ["--scene", args.scene]
        if args.text_table:
            cmd += ["--text_table", args.text_table]
        fh = open(os.path.join(log_dir, f"gen_rank{rank}_gpu{gpu}.log"), "a")
        pr = subprocess.Popen(cmd, env=env, stdout=fh, stderr=subprocess.STDOUT, cwd=ROOT)
        procs.append((pr, fh))
        pids.append(pr.pid)
    with open(os.path.join(run_dir, "gen_pids.json"), "w") as fh:
        json.dump(dict(pids=pids, gpus=gpus, started=time.time()), fh)
    rc = []
    for pr, fh in procs:
        rc.append(pr.wait())
        fh.close()
    gen_sec = time.time() - t0
    n_fail = 0
    for p in plan:
        js = os.path.join(vdir, p["id"] + ".json")
        ok = False
        if os.path.isfile(js):
            with open(js, encoding="utf-8") as fh:
                ok = bool(json.load(fh).get("ok"))
        n_fail += not ok
    stat = dict(gen_wall_sec=gen_sec, n_plan=len(plan), n_generated_now=n_todo, n_failed=n_fail, worker_rc=rc, gpus=gpus)
    with open(os.path.join(run_dir, "gen_stat.json"), "w") as fh:
        json.dump(stat, fh, indent=1)
    print(f"[gen] 完成 · 墙钟 {gen_sec:.0f}s · 失败 {n_fail} · rc={rc}", flush=True)
    return stat


# ------------------------------------------------------------------ 度量（CPU）----
def _gray(frames: np.ndarray) -> np.ndarray:
    import cv2
    return np.stack([cv2.cvtColor(f, cv2.COLOR_RGB2GRAY) for f in frames], 0)


def per_frame_velocity(gray: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
    """帧 i→i+1 的背景位移 (dx, dy) [80,2]；相邻帧 SIFT 测不出时退化为隔 2 帧（i→i+2 或 i−1→i+1）的一半。
    source[i]：1=相邻，2=隔 2 帧，0=测不出（NaN）。"""
    from actionrope.verify_sidecar import measure_shift
    n = gray.shape[0]
    v = np.full((n - 1, 2), np.nan)
    src = np.zeros(n - 1, int)
    cache: dict[tuple[int, int], dict | None] = {}

    def shift(a, b):
        if (a, b) not in cache:
            cache[(a, b)] = measure_shift(gray[a], gray[b])
        return cache[(a, b)]

    for i in range(n - 1):
        m = shift(i, i + 1)
        if m is not None:
            v[i] = (m["dx"], m["dy"]); src[i] = 1
            continue
        for a, b in ((i, i + 2), (i - 1, i + 1)):
            if 0 <= a and b < n:
                m = shift(a, b)
                if m is not None:
                    v[i] = (m["dx"] / 2, m["dy"] / 2); src[i] = 2
                    break
    return v, src


def cumulative_shift(gray: np.ndarray, k: int) -> dict:
    """帧 0→k 的累计背景位移：measure_displacement 的 sift（直接测得出用直接，否则链式）。"""
    from actionrope.infer import measure_displacement
    m = measure_displacement(gray, k=k)
    return dict(sift=m["sift"], source=m["sift_source"], direct=m["sift_direct"], chain=m["sift_chain"],
                chain_failed_at=m.get("sift_chain_failed_at"))


def _region_mask(h: int = G.H, w: int = G.W, box: int = CENTER_BOX) -> np.ndarray:
    m = np.ones((h, w), bool)
    cy, cx = h // 2, w // 2
    m[cy - box // 2: cy + box // 2, cx - box // 2: cx + box // 2] = False
    return m


def _match_gain(fk: np.ndarray, f0: np.ndarray, m: np.ndarray) -> np.ndarray:
    """把终帧逐通道做线性明暗对齐到首帧（区域内均值 / 标准差匹配），剔除整体明暗漂移。"""
    out = fk.astype(np.float64).copy()
    for c in range(fk.shape[2]):
        a, b = fk[..., c][m].astype(np.float64), f0[..., c][m].astype(np.float64)
        sa, sb = a.std(), b.std()
        g = sb / sa if sa > 1e-6 else 1.0
        out[..., c] = (fk[..., c].astype(np.float64) - a.mean()) * g + b.mean()
    return np.clip(out, 0, 255)


def psnr_region(a: np.ndarray, b: np.ndarray, m: np.ndarray) -> float:
    mse = np.mean((a[m].astype(np.float64) - b[m].astype(np.float64)) ** 2)
    return 99.0 if mse < 1e-12 else float(10 * np.log10(255.0 ** 2 / mse))


def ssim_region(a: np.ndarray, b: np.ndarray, m: np.ndarray) -> float:
    """灰度 SSIM（高斯窗 11×11 σ=1.5，Wang 2004 常数），只在 m 区域平均。"""
    import cv2
    a = a.astype(np.float64); b = b.astype(np.float64)
    C1, C2 = (0.01 * 255) ** 2, (0.03 * 255) ** 2
    blur = lambda x: cv2.GaussianBlur(x, (11, 11), 1.5)
    mu_a, mu_b = blur(a), blur(b)
    va = blur(a * a) - mu_a ** 2
    vb = blur(b * b) - mu_b ** 2
    cov = blur(a * b) - mu_a * mu_b
    s = ((2 * mu_a * mu_b + C1) * (2 * cov + C2)) / ((mu_a ** 2 + mu_b ** 2 + C1) * (va + vb + C2))
    return float(s[m].mean())


def ncc_region(a: np.ndarray, b: np.ndarray, m: np.ndarray) -> float:
    x = a[m].astype(np.float64); y = b[m].astype(np.float64)
    x -= x.mean(); y -= y.mean()
    d = np.sqrt((x * x).sum() * (y * y).sum())
    return float((x * y).sum() / d) if d > 1e-9 else 0.0


def translate(img: np.ndarray, t: tuple[float, float]) -> tuple[np.ndarray, np.ndarray]:
    """dst(p) = src(p + t)：把"背景位移了 t"的帧平移回去与帧 0 对齐；返回 (对齐图, 有效区掩码)。"""
    import cv2
    M = np.array([[1, 0, -t[0]], [0, 1, -t[1]]], np.float64)
    h, w = img.shape[:2]
    out = cv2.warpAffine(img, M, (w, h), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_CONSTANT, borderValue=0)
    valid = cv2.warpAffine(np.full((h, w), 255, np.uint8), M, (w, h), flags=cv2.INTER_NEAREST,
                           borderMode=cv2.BORDER_CONSTANT, borderValue=0) > 0
    return out, valid


def back_metrics(frames: np.ndarray, gray: np.ndarray, turn: int) -> dict:
    f0, fk = frames[0], frames[-1]
    m = _region_mask()
    fk_adj = _match_gain(fk, f0, m)
    g0, gk = gray[0].astype(np.float64), _gray(np.clip(fk_adj, 0, 255).astype(np.uint8)[None])[0].astype(np.float64)
    out = dict(psnr=psnr_region(fk_adj, f0, m), ssim=ssim_region(gk, g0, m), ncc=ncc_region(gk, g0, m))
    # 回归残差：终帧相对首帧的实测背景位置误差（SIFT 直接，测不出退链式）
    cs = cumulative_shift(gray, G.FRAMES - 1)
    out["resid_source"] = cs["source"]
    if cs["sift"] is None:
        out.update(resid_dx=None, resid_dy=None, resid_px=None, ncc_comp=None)
    else:
        dx, dy = cs["sift"]
        out.update(resid_dx=dx, resid_dy=dy, resid_px=float(math.hypot(dx, dy)))
        aligned, valid = translate(fk_adj, (dx, dy))
        ga = _gray(np.clip(aligned, 0, 255).astype(np.uint8)[None])[0].astype(np.float64)
        out["ncc_comp"] = ncc_region(ga, g0, m & valid)
        out["psnr_comp"] = psnr_region(aligned, f0, m & valid)
    # 守门：实走距离 = 帧 0→turn 的位移
    ct = cumulative_shift(gray, turn)
    out["walk_source"] = ct["source"]
    out["walk_dx"] = None if ct["sift"] is None else ct["sift"][0]
    out["walk_px"] = None if ct["sift"] is None else float(math.hypot(*ct["sift"]))
    return out


def smooth_metrics(gray: np.ndarray, commit_px_per_frame: float) -> dict:
    """直线右走：逐帧水平速度（背景位移，指令方向为负）。"""
    v, src = per_frame_velocity(gray)
    vx = v[:, 0]
    ok = np.isfinite(vx)
    n_ok = int(ok.sum())
    out = dict(n_frames_measured=n_ok, n_adjacent=int((src == 1).sum()), n_skip2=int((src == 2).sum()),
               n_unmeasured=int((src == 0).sum()), vx=[None if not np.isfinite(x) else float(x) for x in vx],
               vy=[None if not np.isfinite(x) else float(x) for x in v[:, 1]])
    if n_ok < 8:
        out.update(backward_frames=None, backward_frames_1px=None, backward_max_px=None, speed_cv=None,
                   speed_cv_cell=None, jerk_norm=None, speed_mean=None, speed_ratio=None)
        return out
    vv = vx[ok]
    # 指令右走 ⇒ 背景速度应为负；符号翻正 = 倒退帧
    out["backward_frames"] = int((vv > 0).sum())
    out["backward_frames_1px"] = int((vv > 1.0).sum())
    mean = float(vv.mean())
    out["speed_mean"] = mean
    out["speed_cv"] = float(vv.std() / abs(mean)) if abs(mean) > 1e-6 else None
    out["speed_ratio"] = float(mean / commit_px_per_frame)
    out["backward_max_px"] = float(vv.max()) if (vv > 0).any() else 0.0
    # 逐 cell（4 帧）平均后的速度 CV：因果 VAE 每 4 帧一个 latent cell，帧内节奏抖动在这一步被平均掉
    vs = np.where(ok, vx, np.nan)
    vc = np.nanmean(vs.reshape(-1, 4), 1)
    out["speed_cv_cell"] = float(np.nanstd(vc) / abs(np.nanmean(vc))) if np.isfinite(vc).sum() >= 4 else None
    # 归一 jerk：速度序列的二阶差分 RMS / |平均速度|（只在连续可测段上取差分）
    j = np.diff(vs, n=2)
    j = j[np.isfinite(j)]
    out["jerk_norm"] = float(np.sqrt((j ** 2).mean()) / abs(mean)) if j.size and abs(mean) > 1e-6 else None
    return out


def direction_metrics(gray: np.ndarray, expected: np.ndarray) -> dict:
    cs = cumulative_shift(gray, G.FRAMES - 1)
    out = dict(source=cs["source"], expected=expected.tolist())
    if cs["sift"] is None:
        out.update(measured=None, angle_deg=None, ratio=None, fail=True, fail_reason="unmeasured")
        return out
    d = np.array(cs["sift"], np.float64)
    ang = math.degrees(math.atan2(d[1], d[0]) - math.atan2(expected[1], expected[0]))
    ang = (ang + 180) % 360 - 180
    ratio = float(np.linalg.norm(d) / np.linalg.norm(expected))
    fail = abs(ang) > ANGLE_FAIL_DEG or ratio < RATIO_FAIL
    out.update(measured=d.tolist(), angle_deg=float(ang), ratio=ratio, fail=bool(fail),
               fail_reason=("angle" if abs(ang) > ANGLE_FAIL_DEG else "") + ("ratio" if ratio < RATIO_FAIL else ""))
    return out


def measure_one(item: dict, frames: np.ndarray | None = None) -> dict:
    """一条视频的全部度量。item 来自 plan（含 exam / frame_offset_px）；frames 不给就读 item['mp4']。"""
    from actionrope.infer import read_frames
    t0 = time.time()
    out = dict(id=item["id"], exam=item["exam"], bg=item["bg"], mult=item["mult"])
    try:
        if frames is None:
            frames = read_frames(item["mp4"])
        if frames.shape[0] != G.FRAMES:
            raise ValueError(f"帧数 {frames.shape[0]} != {G.FRAMES}")
        gray = _gray(frames)
        fo = np.array(item["frame_offset_px"], np.float64)
        expected = -fo[80]
        out["expected_bg_shift_80"] = expected.tolist()
        if item["exam"] == "gain":
            cs = cumulative_shift(gray, G.FRAMES - 1)
            out["measured_80"] = cs["sift"]
            out["source"] = cs["source"]
            out["commanded_dx"] = float(expected[0])
            out["measured_dx"] = None if cs["sift"] is None else float(cs["sift"][0])
        elif item["exam"] == "back":
            out.update(back_metrics(frames, gray, item.get("turn_frame", BACK_TURN)))
            out["expected_walk_dx"] = float(-fo[item.get("turn_frame", BACK_TURN), 0])
        elif item["exam"] == "straight":
            commit = float((-fo[80, 0]) / (G.FRAMES - 1))     # px/帧，负值
            out["commit_px_per_frame"] = commit
            out["smooth"] = smooth_metrics(gray, commit)
            out["direction"] = direction_metrics(gray, expected)
        out["ok"] = True
    except Exception as e:
        out.update(ok=False, error=repr(e))
    out["measure_sec"] = time.time() - t0
    return out


def _measure_job(item):
    return measure_one(item)


def _med(xs) -> float | None:
    xs = [x for x in xs if x is not None and np.isfinite(x)]
    return float(np.median(xs)) if xs else None


def _p90(xs) -> float | None:
    xs = [x for x in xs if x is not None and np.isfinite(x)]
    return float(np.percentile(xs, 90)) if xs else None


def fit_origin(x: np.ndarray, y: np.ndarray) -> dict:
    """过原点回归 y = k·x：k = Σxy/Σx²；R² = 1 − SS_res/SS_tot（SS_tot 绕均值）。"""
    k = float((x * y).sum() / (x * x).sum())
    res = y - k * x
    ss_tot = float(((y - y.mean()) ** 2).sum())
    r2 = float(1 - (res ** 2).sum() / ss_tot) if ss_tot > 0 else None
    return dict(slope=k, r2=r2, n=int(x.size), resid_rms=float(np.sqrt((res ** 2).mean())))


def summarize(metrics: list[dict], walk_x: float) -> dict:
    s: dict = {"n_measured": len(metrics), "n_measure_failed": sum(not m.get("ok") for m in metrics)}
    # 1. 增益线性：指令幅值 vs 实测幅值（都取沿指令方向的分量，指令右走 ⇒ 背景 dx<0，取负号后为正）
    g = [m for m in metrics if m["exam"] == "gain" and m.get("ok") and m.get("measured_dx") is not None]
    if g:
        x = np.array([-m["commanded_dx"] for m in g]); y = np.array([-m["measured_dx"] for m in g])
        fit = fit_origin(x, y)
        by_mult = {}
        for mult in GAIN_MULTS:
            ys = [-m["measured_dx"] / (-m["commanded_dx"]) for m in g if abs(m["mult"] - mult) < 1e-6]
            by_mult[f"{mult:.2f}"] = dict(n=len(ys), ratio_med=_med(ys))
        # 敏感性：逐个剔除一个背景重新拟合，记录影响最大的那个（生成的画面若带纵深视差，SIFT 链式在玩家位置取值会失稳）
        per_bg = {}
        for bg in sorted({m["bg"] for m in g}):
            keep = np.array([m["bg"] != bg for m in g])
            f = fit_origin(x[keep], y[keep])
            ratios = [-m["measured_dx"] / (-m["commanded_dx"]) for m in g if m["bg"] == bg]
            per_bg[bg] = dict(slope_wo=f["slope"], r2_wo=f["r2"], ratio_med=float(np.median(ratios)))
        worst = max(per_bg, key=lambda k: per_bg[k]["r2_wo"])
        s["gain"] = dict(slope=fit["slope"], r2=fit["r2"], n=fit["n"], resid_rms_px=fit["resid_rms"],
                         leave_one_bg_out={str(k): v for k, v in per_bg.items()},
                         worst_bg=int(worst), slope_wo_worst=per_bg[worst]["slope_wo"], r2_wo_worst=per_bg[worst]["r2_wo"],
                         n_unmeasured=sum(1 for m in metrics if m["exam"] == "gain" and m.get("measured_dx") is None),
                         ratio_by_mult=by_mult, points=[[float(a), float(b)] for a, b in zip(x, y)])
    # 2. 折返闭环
    b = [m for m in metrics if m["exam"] == "back" and m.get("ok")]
    if b:
        s["back"] = dict(n=len(b), psnr_med=_med([m["psnr"] for m in b]), psnr_p10=_p90([-m["psnr"] for m in b]) and -_p90([-m["psnr"] for m in b]),
                         ssim_med=_med([m["ssim"] for m in b]), ncc_med=_med([m["ncc"] for m in b]),
                         resid_px_med=_med([m["resid_px"] for m in b]), resid_px_p90=_p90([m["resid_px"] for m in b]),
                         resid_dx_med=_med([m["resid_dx"] for m in b]), resid_dy_med=_med([m["resid_dy"] for m in b]),
                         ncc_comp_med=_med([m.get("ncc_comp") for m in b]), psnr_comp_med=_med([m.get("psnr_comp") for m in b]),
                         walk_px_med=_med([m["walk_px"] for m in b]), walk_dx_med=_med([m["walk_dx"] for m in b]),
                         expected_walk_dx=b[0].get("expected_walk_dx"),
                         n_resid_unmeasured=sum(m["resid_px"] is None for m in b),
                         n_walk_unmeasured=sum(m["walk_px"] is None for m in b),
                         n_walk_short=sum(1 for m in b if m["walk_px"] is not None and m["walk_px"] < 0.5 * abs(b[0].get("expected_walk_dx", 1))),
                         psnr_all=[m["psnr"] for m in b])
    # 3. 运动平滑
    st = [m for m in metrics if m["exam"] == "straight" and m.get("ok")]
    if st:
        sm = [m["smooth"] for m in st]
        bf = [x["backward_frames"] for x in sm if x["backward_frames"] is not None]
        s["smooth"] = dict(n=len(sm), backward_frames_total=int(sum(bf)), backward_frames_1px_total=int(sum(x["backward_frames_1px"] for x in sm if x["backward_frames_1px"] is not None)),
                           n_videos_with_backward=int(sum(1 for v in bf if v > 0)), backward_frames_per_video=bf,
                           backward_max_px=max((x["backward_max_px"] or 0.0) for x in sm),
                           speed_cv_med=_med([x["speed_cv"] for x in sm]), speed_cv_p90=_p90([x["speed_cv"] for x in sm]),
                           speed_cv_cell_med=_med([x["speed_cv_cell"] for x in sm]),
                           vx_mean_by_frame=[float(v) for v in np.nanmean(np.array([[np.nan if v is None else v for v in x["vx"]] for x in sm]), 0)],
                           jerk_norm_med=_med([x["jerk_norm"] for x in sm]), jerk_norm_p90=_p90([x["jerk_norm"] for x in sm]),
                           speed_ratio_med=_med([x["speed_ratio"] for x in sm]), speed_ratio_mean=float(np.mean([x["speed_ratio"] for x in sm if x["speed_ratio"] is not None])),
                           speed_mean_px_per_frame_med=_med([x["speed_mean"] for x in sm]),
                           commit_px_per_frame=st[0]["commit_px_per_frame"],
                           frames_adjacent=int(sum(x["n_adjacent"] for x in sm)), frames_skip2=int(sum(x["n_skip2"] for x in sm)),
                           frames_unmeasured=int(sum(x["n_unmeasured"] for x in sm)))
        d = [m["direction"] for m in st]
        s["direction"] = dict(n=len(d), expected=d[0]["expected"], angle_abs_med=_med([abs(x["angle_deg"]) for x in d if x["angle_deg"] is not None]),
                              angle_abs_p90=_p90([abs(x["angle_deg"]) for x in d if x["angle_deg"] is not None]),
                              angle_signed_med=_med([x["angle_deg"] for x in d if x["angle_deg"] is not None]),
                              ratio_med=_med([x["ratio"] for x in d]), ratio_p10=(lambda v: None if v is None else -v)(_p90([-x["ratio"] for x in d if x["ratio"] is not None])),
                              n_fail=int(sum(x["fail"] for x in d)), n_unmeasured=int(sum(x["measured"] is None for x in d)),
                              fail_ids=[m["id"] for m in st if m["direction"]["fail"]],
                              dy_med=_med([x["measured"][1] for x in d if x["measured"] is not None]))
    s["old_report_world_rope_5k"] = OLD_REPORT
    s["walk_speed_px_s_x"] = walk_x
    return s


def stage_measure(args, run_dir: str, plan: list[dict]) -> dict:
    vdir = os.path.join(run_dir, "videos")
    items = []
    n_missing = 0
    for p in plan:
        if video_done(vdir, p["id"]):
            items.append(dict(p, mp4=os.path.join(vdir, p["id"] + ".mp4")))
        else:
            n_missing += 1
    print(f"[measure] {len(items)} 条视频，{n_missing} 条缺失/未完成 · {args.workers} 进程", flush=True)
    t0 = time.time()
    with ProcessPoolExecutor(args.workers) as ex:
        metrics = list(ex.map(_measure_job, items, chunksize=1))
    print(f"[measure] 度量完成 · {time.time() - t0:.0f}s", flush=True)
    with open(os.path.join(run_dir, "metrics.jsonl"), "w", encoding="utf-8") as fh:
        for m in metrics:
            fh.write(json.dumps(m, ensure_ascii=False, default=float) + "\n")
    return dict(metrics=metrics, measure_sec=time.time() - t0, n_missing=n_missing)


# ------------------------------------------------------------------ 合成自检 ----
def synth_video(img: np.ndarray, bg_shift: np.ndarray) -> np.ndarray:
    """按逐帧"背景位移"（帧 i 相对帧 0，四舍五入到整数 px）造 81 帧视频。
    背景位移 t ⇒ 帧 k 位置 s 的内容在帧 0 位于 s − t ⇒ 帧 k = 画布上从 x0 − tx 起的窗口。
    横向用 [镜像 | img | 镜像] 的 3W 宽画布滑窗（非周期，|位移| ≤ 832 px 不会像 np.roll 那样回绕成 t ∓ W），
    纵向位移小，直接 np.roll。"""
    h, w = img.shape[:2]
    canvas = np.concatenate([img[:, ::-1], img, img[:, ::-1]], 1)
    out = np.empty((bg_shift.shape[0],) + img.shape, np.uint8)
    for i, (tx, ty) in enumerate(bg_shift):
        x0 = w - int(round(tx))
        if not 0 <= x0 <= 2 * w:
            raise ValueError(f"合成横向位移 {tx} 超出画布")
        out[i] = np.roll(canvas[:, x0:x0 + w], int(round(ty)), axis=0)
    return out


def selfcheck(walk_x: float, bg_path: str) -> dict:
    """度量代码在合成平移视频上的自检：速度 / 位移 / 倒退帧 / 残差 / 补偿都应逐位可预期。"""
    from actionrope.infer import load_first_frame
    img = load_first_frame(bg_path)
    res = {"bg": os.path.basename(bg_path)}
    # A. 直线右走 1.0×：每帧背景 −4.33 px（整数取整后 −4/−5 交替），无倒退帧，0→80 = −346
    fo = straight_offsets(walk_x)
    fr = synth_video(img, -fo)
    m = measure_one(dict(id="synth_straight", exam="straight", bg=0, mult=1.0, frame_offset_px=fo.tolist()), fr)
    exp_v = np.diff(np.round(-fo[:, 0]))
    vx = np.array([np.nan if v is None else v for v in m["smooth"]["vx"]])
    res["straight"] = dict(expected_dx_80=float(round(-fo[80, 0])), measured_dx_80=m["direction"]["measured"][0],
                           measured_dy_80=m["direction"]["measured"][1], angle_deg=m["direction"]["angle_deg"],
                           ratio=m["direction"]["ratio"], backward_frames=m["smooth"]["backward_frames"],
                           expected_backward_frames=0, n_unmeasured=m["smooth"]["n_unmeasured"],
                           n_skip2=m["smooth"]["n_skip2"],
                           vel_max_abs_err_px=float(np.nanmax(np.abs(vx - exp_v))),
                           speed_ratio=m["smooth"]["speed_ratio"], speed_cv=m["smooth"]["speed_cv"],
                           expected_speed_cv=float(exp_v.std() / abs(exp_v.mean())), jerk_norm=m["smooth"]["jerk_norm"])
    # B. 直线右走但帧 20/21/22 各倒退 3 px（3 个倒退帧），再继续
    bg = -fo[:, 0].copy()
    bg = np.round(bg)
    bg[20:] += 3 * 3           # 从帧 20 起整体少走 9 px
    bg[20] = bg[19] + 3; bg[21] = bg[20] + 3; bg[22] = bg[21] + 3
    shift = np.stack([bg, np.zeros_like(bg)], 1)
    fr = synth_video(img, shift)
    m = measure_one(dict(id="synth_backward", exam="straight", bg=0, mult=1.0, frame_offset_px=fo.tolist()), fr)
    res["backward"] = dict(expected_backward_frames=3, backward_frames=m["smooth"]["backward_frames"],
                           backward_frames_1px=m["smooth"]["backward_frames_1px"],
                           expected_dx_80=float(bg[80]), measured_dx_80=m["direction"]["measured"][0],
                           vx_frames_19_23=[m["smooth"]["vx"][i] for i in range(19, 24)])
    # C. 折返：右 40 帧回 40 帧，终帧 = 首帧（roll 回绕）⇒ 残差 0、PSNR 99、NCC 1；实走距离 ≈ 256
    fo_b = back_offsets(walk_x * BACK_MULT)
    fr = synth_video(img, -fo_b)
    m = measure_one(dict(id="synth_back", exam="back", bg=0, mult=BACK_MULT, frame_offset_px=fo_b.tolist(), turn_frame=BACK_TURN), fr)
    res["back_exact"] = dict(psnr=m["psnr"], ssim=m["ssim"], ncc=m["ncc"], resid_px=m["resid_px"], ncc_comp=m["ncc_comp"],
                             walk_dx=m["walk_dx"], expected_walk_dx=m["expected_walk_dx"])
    # D. 折返但终帧多走了 (+12, −5) px（回归残差应 ≈ 13 px；补偿后 NCC 应回到 ≈1，补偿前明显低）
    shift = -fo_b.copy()
    shift[80] += (12, -5)
    fr = synth_video(img, shift)
    m = measure_one(dict(id="synth_back_resid", exam="back", bg=0, mult=BACK_MULT, frame_offset_px=fo_b.tolist(), turn_frame=BACK_TURN), fr)
    res["back_resid"] = dict(expected_resid=[12, -5], resid_dx=m["resid_dx"], resid_dy=m["resid_dy"], resid_px=m["resid_px"],
                             expected_resid_px=float(math.hypot(12, 5)), psnr=m["psnr"], ncc=m["ncc"], ncc_comp=m["ncc_comp"],
                             psnr_comp=m.get("psnr_comp"))
    # E. 折返 + 终帧整体变亮 20%（明暗漂移）⇒ 明暗对齐后 PSNR/NCC 仍应接近 exact
    fr = synth_video(img, -fo_b)
    fr[80] = np.clip(fr[80].astype(np.float64) * 1.2 + 10, 0, 255).astype(np.uint8)
    m = measure_one(dict(id="synth_back_bright", exam="back", bg=0, mult=BACK_MULT, frame_offset_px=fo_b.tolist(), turn_frame=BACK_TURN), fr)
    res["back_bright"] = dict(psnr=m["psnr"], ssim=m["ssim"], ncc=m["ncc"], resid_px=m["resid_px"])
    # F. 增益：5 档合成 ⇒ 斜率 1、R² 1
    pts = []
    for mult in GAIN_MULTS:
        fo_g = straight_offsets(walk_x * mult)
        fr = synth_video(img, -fo_g)
        m = measure_one(dict(id=f"synth_gain_{mult}", exam="gain", bg=0, mult=mult, frame_offset_px=fo_g.tolist()), fr)
        pts.append((-m["commanded_dx"], -m["measured_dx"]))
    x = np.array([p[0] for p in pts]); y = np.array([p[1] for p in pts])
    res["gain"] = dict(fit_origin(x, y), points=[[float(a), float(b)] for a, b in pts])
    # 判定
    checks = dict(
        straight_dx=abs(res["straight"]["measured_dx_80"] - res["straight"]["expected_dx_80"]) < 3,
        straight_backward=res["straight"]["backward_frames"] == 0,
        straight_vel=res["straight"]["vel_max_abs_err_px"] < 1.0,
        backward_count=res["backward"]["backward_frames"] == 3,
        back_exact=res["back_exact"]["resid_px"] < 1.5 and res["back_exact"]["ncc"] > 0.99,
        back_resid=abs(res["back_resid"]["resid_px"] - res["back_resid"]["expected_resid_px"]) < 1.5 and res["back_resid"]["ncc_comp"] > 0.98,
        back_bright=res["back_bright"]["ncc"] > 0.99 and res["back_bright"]["psnr"] > 35,
        gain=abs(res["gain"]["slope"] - 1) < 0.02 and res["gain"]["r2"] > 0.999,
    )
    res["checks"] = checks
    res["all_pass"] = all(checks.values())
    return res


# ------------------------------------------------------------------ 报告 / 图 ----
def make_figure(summary: dict, metrics: list[dict], path: str):
    import matplotlib
    matplotlib.use("Agg")
    import matplotlib.pyplot as plt
    fig, axes = plt.subplots(1, 3, figsize=(15, 4.4))
    ax = axes[0]
    g = summary.get("gain")
    if g:
        pts = np.array(g["points"])
        ax.scatter(pts[:, 0], pts[:, 1], s=18, alpha=0.7, label="videos")
        xs = np.linspace(0, pts[:, 0].max() * 1.05, 10)
        ax.plot(xs, g["slope"] * xs, "r-", label=f"fit: slope {g['slope']:.3f}, R² {g['r2']:.3f}")
        ax.plot(xs, xs, "k--", lw=0.8, label="identity")
        ax.set_xlabel("commanded |dx| (px, frame 0→80)"); ax.set_ylabel("measured |dx| (px)")
        ax.legend(fontsize=8)
    ax.set_title("1. gain linearity (0.5–1.5×, 10 bg)")
    ax = axes[1]
    b = summary.get("back")
    if b:
        ax.hist(b["psnr_all"], bins=20, color="tab:green", alpha=0.8)
        ax.axvline(b["psnr_med"], color="r", label=f"median {b['psnr_med']:.1f} dB")
        ax.set_xlabel("PSNR(frame 80, frame 0) dB, outside center 160²"); ax.legend(fontsize=8)
    ax.set_title("2. return-trip PSNR (50 bg)")
    ax = axes[2]
    sm = summary.get("smooth")
    if sm:
        bf = sm["backward_frames_per_video"]
        ax.hist(bf, bins=range(0, max(bf + [1]) + 2), color="tab:orange", alpha=0.8, align="left")
        ax.set_xlabel("backward frames per video"); ax.set_ylabel("videos")
        ax.set_title(f"3. backward frames (total {sm['backward_frames_total']} / 50 bg)")
    fig.tight_layout()
    fig.savefig(path, dpi=130)
    plt.close(fig)


def _f(v, nd=2):
    return "—" if v is None else (f"{v:.{nd}f}" if isinstance(v, float) else str(v))


def make_report(summary: dict, sc: dict | None, gen_stat: dict | None, run: str, ckpt: str | None) -> str:
    o = OLD_REPORT
    g, b, sm, d = summary.get("gain", {}), summary.get("back", {}), summary.get("smooth", {}), summary.get("direction", {})
    L = [f"# bg50 四科判决 · run `{run}`", "",
         f"ckpt `{ckpt}` · 文本条件 NULL_PROMPT · 50 步 shift 5 · bf16 · seed 固定 · 首帧 26:15 居中裁剪后缩放到 832×480。",
         f"横向步速 walk_speed_px_s[0] = {summary['walk_speed_px_s_x']:.2f} px/s（1.0×）；折返/直行只用横向指令，纵向为 0"
         "（纵向增益已知偏大 ~25%，见 README 已知问题 3）。位移量法：SIFT+RANSAC 仿射场在玩家位置取值（`verify_sidecar.measure_shift`），"
         "帧 0→k 直接测不出时每 16 帧链式累加（`infer.measure_displacement`）。", ""]
    if gen_stat:
        L += [f"生成：{gen_stat['n_plan']} 条计划，本次生成 {gen_stat['n_generated_now']} 条，墙钟 {gen_stat['gen_wall_sec']:.0f} s，"
              f"{len(gen_stat['gpus'])} 卡，失败 {gen_stat['n_failed']} 条。", ""]
    L += ["## 汇总（与旧报告世界 RoPE@5k 并列）", "",
          "| 科目 | 指标 | 本次 step-5000 | 旧报告 世界RoPE@5k |", "|---|---|---|---|",
          f"| 1 增益线性 | 斜率 | {_f(g.get('slope'), 3)} | {o['gain_slope']} |",
          f"| 1 增益线性 | R² | {_f(g.get('r2'), 3)} | {o['gain_r2']} |",
          f"| 1 增益线性 | 样本数 / 测不出 | {g.get('n')} / {g.get('n_unmeasured')} | — |",
          f"| 2 折返闭环 | PSNR 中位 (dB) | {_f(b.get('psnr_med'))} | {_f(o['back_psnr_med'])} |",
          f"| 2 折返闭环 | SSIM 中位 | {_f(b.get('ssim_med'), 3)} | {_f(o['back_ssim_med'])} |",
          f"| 2 折返闭环 | NCC 中位 | {_f(b.get('ncc_med'), 3)} | {_f(o['back_ncc_med'])} |",
          f"| 2 折返闭环 | 回归残差中位 (px) / p90 | {_f(b.get('resid_px_med'), 1)} / {_f(b.get('resid_px_p90'), 1)} | {_f(o['back_resid_med'])} |",
          f"| 2 折返闭环 | 补偿后 NCC 中位 | {_f(b.get('ncc_comp_med'), 3)} | {_f(o['back_ncc_comp_med'])} |",
          f"| 2 折返闭环 | 实走距离中位 (px, 帧 0→40) / 指令 | {_f(b.get('walk_px_med'), 1)} / {_f(abs(b.get('expected_walk_dx') or 0), 1)} | — |",
          f"| 3 运动平滑 | 倒退帧总数（50 景） | {sm.get('backward_frames_total')}（>1 px：{sm.get('backward_frames_1px_total')}；含倒退帧的视频 {sm.get('n_videos_with_backward')}） | {o['smooth_backward_frames_total']} |",
          f"| 3 运动平滑 | 归一速度波动 std/mean 中位 / p90 | {_f(sm.get('speed_cv_med'), 3)} / {_f(sm.get('speed_cv_p90'), 3)} | {_f(o['smooth_speed_cv_med'])} |",
          f"| 3 运动平滑 | 归一 jerk 中位 / p90 | {_f(sm.get('jerk_norm_med'), 3)} / {_f(sm.get('jerk_norm_p90'), 3)} | {_f(o['smooth_jerk_med'])} |",
          f"| 3 运动平滑 | 平均速度/承诺 中位 | {_f(sm.get('speed_ratio_med'), 3)}（承诺 {_f(sm.get('commit_px_per_frame'))} px/帧） | {_f(o['smooth_speed_ratio_med'])} |",
          f"| 4 方向服从 | \\|角度误差\\| 中位 / p90 (°) | {_f(d.get('angle_abs_med'))} / {_f(d.get('angle_abs_p90'))} | {_f(o['dir_angle_med'])} |",
          f"| 4 方向服从 | 幅度比中位 | {_f(d.get('ratio_med'), 3)} | {_f(o['dir_ratio_med'])} |",
          f"| 4 方向服从 | 失控数（\\|角\\|>15° 或 幅<0.5）/ 测不出 | {d.get('n_fail')} / {d.get('n_unmeasured')} | {_f(o['dir_fail'])} |",
          ""]
    rb = g.get("ratio_by_mult", {})
    L += ["### 第一科 增益线性", "",
          "各档实测/指令幅度比中位：" + "、".join(f"{k}× {_f(v['ratio_med'], 3)}（n={v['n']}）" for k, v in rb.items()) + "。",
          f"同一批 10 个背景、只改指令位移，实测帧 0→80 横向位移对指令位移过原点回归斜率 {_f(g.get('slope'), 3)}、R² {_f(g.get('r2'), 3)}"
          f"（旧报告 {o['gain_slope']} / {o['gain_r2']}）。",
          ("斜率 <1：1.5× 档幅度比明显掉到 " + _f(rb.get('1.50', {}).get('ratio_med'), 3) + "，高速档欠冲拉低了斜率" if (g.get('slope') or 1) < 1 else
           "斜率 >1：模型在未见背景上仍有系统性超调") + "，与 README 里训练场景 tidal_flats 首帧的三档斜率 ≈1.05 不同 —— 未见背景下增益略欠、离散更大；"
          f"R² 反映不同背景之间的离散（残差 RMS {_f(g.get('resid_rms_px'), 1)} px）。散点里明显偏低的一组点全来自背景 {(g.get('worst_bg') if g.get('worst_bg') is not None else -1):02d}"
          f"（幅度比中位 {_f(g.get('leave_one_bg_out', {}).get(str(g.get('worst_bg')), {}).get('ratio_med'), 3)}）：生成画面带纵深视差"
          "（bg 01 赛博城市：远景楼群带比地面带慢 ~15%），SIFT 仿射场在玩家位置取值时各段锁到不同层，16 帧链式给 −235 px 而逐帧速度累加给 −391 px（指令 −433）；"
          f"这是量法在视差场景下的失稳而非模型不动。剔除该背景后斜率 {_f(g.get('slope_wo_worst'), 3)}、R² {_f(g.get('r2_wo_worst'), 3)}（作敏感性参考，头条仍按旧报告口径全量报）。", "",
          "### 第二科 折返闭环", "",
          f"{BACK_MULT}× 步速右走 40 帧再左走 40 帧，frame_offset[80] = 0。守门：帧 0→40 实走距离中位 {_f(b.get('walk_px_med'), 1)} px"
          f"（指令 {_f(abs(b.get('expected_walk_dx') or 0), 1)} px，{b.get('n_walk_short')} 条 <50%，{b.get('n_walk_unmeasured')} 条测不出），确实走出去了。",
          f"终帧 vs 首帧（明暗线性对齐、剔除中心 160×160）：PSNR 中位 {_f(b.get('psnr_med'))} dB、SSIM {_f(b.get('ssim_med'), 3)}、NCC {_f(b.get('ncc_med'), 3)}；"
          f"回归残差中位 {_f(b.get('resid_px_med'), 1)} px（dx 中位 {_f(b.get('resid_dx_med'), 1)}、dy 中位 {_f(b.get('resid_dy_med'), 1)}），"
          f"按残差平移对齐后 NCC {_f(b.get('ncc_comp_med'), 3)}（PSNR {_f(b.get('psnr_comp_med'))} dB）。",
          "补偿后 NCC 与补偿前的差距是位置误差贡献；补偿后仍达不到 1 的部分是往返期间生成内容对首帧的遗忘 / 重绘。", "",
          "### 第三科 运动平滑", "",
          f"1.0× 直线右走 50 条，逐帧水平速度：相邻帧 SIFT 成功 {sm.get('frames_adjacent')} 帧、退化隔 2 帧 {sm.get('frames_skip2')}、测不出 {sm.get('frames_unmeasured')}。",
          f"倒退帧（指令右走而实测背景速度翻正）总数 {sm.get('backward_frames_total')}（旧报告 {o['smooth_backward_frames_total']}），>1 px 的 {sm.get('backward_frames_1px_total')}；"
          f"归一速度波动中位 {_f(sm.get('speed_cv_med'), 3)}、归一 jerk（速度二阶差分 RMS/|均速|）中位 {_f(sm.get('jerk_norm_med'), 3)}，"
          f"平均速度/承诺中位 {_f(sm.get('speed_ratio_med'), 3)}。",
          f"倒退帧全部是幅值 ≤ {_f(sm.get('backward_max_px'))} px 的停顿帧（>1 px 的为 0），集中在 cell 边界（帧 20→21、42→43），不是真正的往回走。"
          f"逐帧速度呈稳定的帧内节奏（各视频均值序列前 8 帧 {[round(v, 1) for v in sm.get('vx_mean_by_frame', [])[:8]]} px/帧，跨视频 std 仅 0.4–1.1 px），"
          f"来自因果 VAE 每 4 帧一个 latent cell 的时间压缩；按 cell（4 帧）平均后速度波动中位降到 {_f(sm.get('speed_cv_cell_med'), 3)}"
          "（合成整数平移的下限 0.108），即宏观速度平稳、微观有周期抖动；jerk 同理主要反映这个节奏，只宜横向比较。", "",
          "### 第四科 方向服从", "",
          f"同一批直线右走视频，帧 0→80 累计位移（链式 SIFT）与指令 (−{abs(d.get('expected', [0])[0]) if d.get('expected') else 0:.0f}, 0) 的角度误差 |角| 中位 {_f(d.get('angle_abs_med'))}°、p90 {_f(d.get('angle_abs_p90'))}°"
          f"（带符号中位 {_f(d.get('angle_signed_med'))}°，纵向漂移 dy 中位 {_f(d.get('dy_med'), 1)} px），幅度比中位 {_f(d.get('ratio_med'), 3)}，"
          f"失控 {d.get('n_fail')} 条（{', '.join(d.get('fail_ids', [])) or '无'}），测不出 {d.get('n_unmeasured')} 条。",
          "角度误差主要来自纵向漂移（指令纵向为 0），幅度比与第一科 1.0× 档一致。", ""]
    if sc:
        st, bw, be, br, bb, gg = sc["straight"], sc["backward"], sc["back_exact"], sc["back_resid"], sc["back_bright"], sc["gain"]
        L += ["## 度量代码合成数据自检", "",
              f"用 `{sc['bg']}` 的 832×480 首帧按指令 np.roll 平移造视频（整数 px），全部通过：**{sc['all_pass']}**（{sc['checks']}）。", "",
              "| 合成用例 | 预期 | 实测 |", "|---|---|---|",
              f"| 直线 1.0× 帧 0→80 dx | {st['expected_dx_80']:.0f} | {st['measured_dx_80']:.1f}（dy {st['measured_dy_80']:.1f}，角 {st['angle_deg']:.2f}°，幅比 {st['ratio']:.3f}） |",
              f"| 直线 1.0× 逐帧速度 | −4/−5 交替，倒退帧 0 | 最大逐帧误差 {st['vel_max_abs_err_px']:.2f} px，倒退帧 {st['backward_frames']}，测不出 {st['n_unmeasured']}，隔 2 帧 {st['n_skip2']} |",
              f"| 直线 1.0× 速度 std/mean | {st['expected_speed_cv']:.3f}（整数取整固有） | {st['speed_cv']:.3f}；jerk {st['jerk_norm']:.3f}；速度/承诺 {st['speed_ratio']:.3f} |",
              f"| 注入 3 个倒退帧（+3 px） | 倒退帧 3，dx80 {bw['expected_dx_80']:.0f} | 倒退帧 {bw['backward_frames']}（>1 px {bw['backward_frames_1px']}），dx80 {bw['measured_dx_80']:.1f}，帧 19–23 速度 {[round(v, 1) for v in bw['vx_frames_19_23']]} |",
              f"| 折返精确回原点 | 残差 0，PSNR 99，NCC 1，实走 {abs(be['expected_walk_dx']):.0f} | 残差 {be['resid_px']:.2f} px，PSNR {be['psnr']:.1f}，SSIM {be['ssim']:.3f}，NCC {be['ncc']:.4f}，补偿后 {be['ncc_comp']:.4f}，实走 dx {be['walk_dx']:.1f} |",
              f"| 折返终帧多走 (+12, −5) | 残差 13.0 px，补偿后 NCC ≈1 | 残差 ({br['resid_dx']:.1f}, {br['resid_dy']:.1f}) = {br['resid_px']:.1f} px，PSNR {br['psnr']:.1f}，NCC {br['ncc']:.3f} → 补偿后 {br['ncc_comp']:.4f}（PSNR {br['psnr_comp']:.1f}） |",
              f"| 折返 + 终帧亮 20% | 明暗对齐后 ≈ 精确用例 | PSNR {bb['psnr']:.1f}，SSIM {bb['ssim']:.3f}，NCC {bb['ncc']:.4f}，残差 {bb['resid_px']:.2f} |",
              f"| 增益 5 档合成 | 斜率 1，R² 1 | 斜率 {gg['slope']:.4f}，R² {gg['r2']:.5f} |", ""]
    L += ["## 文件", "", "`plan.jsonl`（每条视频的 id / 背景 / 指令 / frame_offset）、`videos/<id>.mp4|json`、`metrics.jsonl`、`summary.json`、`figures.png`、`selfcheck.json`。"]
    return "\n".join(L)


# ------------------------------------------------------------------ 入口 ----
def main(argv=None):
    ap = argparse.ArgumentParser(description="bg50 四科判决（ARoPE 臂）")
    ap.add_argument("--ckpt", default=os.path.join(ROOT, "outputs", "arope_5k", "step-5000.safetensors"))
    ap.add_argument("--run", default="step5000")
    ap.add_argument("--stages", default="gen,measure", help="gen,measure 的子集")
    ap.add_argument("--gpus", default="0-7")
    ap.add_argument("--steps", type=int, default=50)
    ap.add_argument("--shift", type=float, default=5.0)
    ap.add_argument("--bgs", default=None, help="背景序号，如 1-10,15；默认全部 50")
    ap.add_argument("--seed", type=int, default=0)
    ap.add_argument("--scene", default=None, help="人为覆盖文本条件（scene key 或 'in a ...' 从句）；默认 NULL_PROMPT")
    ap.add_argument("--text_table", default=None)
    ap.add_argument("--workers", type=int, default=48, help="度量阶段 CPU 进程数")
    ap.add_argument("--selfcheck", action="store_true", help="只跑合成数据自检")
    ap.add_argument("--no_selfcheck", action="store_true", help="measure 阶段跳过自检")
    ap.add_argument("--_worker", action="store_true", help=argparse.SUPPRESS)
    ap.add_argument("--rank", type=int, default=0, help=argparse.SUPPRESS)
    ap.add_argument("--world", type=int, default=1, help=argparse.SUPPRESS)
    args = ap.parse_args(argv)

    run_dir = os.path.join(OUT_ROOT, args.run)
    os.makedirs(run_dir, exist_ok=True)
    if args._worker:
        gen_worker(run_dir, args.ckpt, args.rank, args.world, args.steps, args.shift, args.scene, args.text_table)
        return

    import torch
    from actionrope.infer import SIDECAR
    sc_meta = torch.load(SIDECAR, map_location="cpu", weights_only=False)["meta"]
    walk_x = float(sc_meta["walk_speed_px_s"][0])
    rows = parse_bgs(args.bgs, list_backgrounds())

    if args.selfcheck:
        res = selfcheck(walk_x, rows[0]["path"])
        print(json.dumps(res, ensure_ascii=False, indent=1, default=float))
        with open(os.path.join(run_dir, "selfcheck.json"), "w", encoding="utf-8") as fh:
            json.dump(res, fh, ensure_ascii=False, indent=1, default=float)
        return res

    stages = [s.strip() for s in args.stages.split(",") if s.strip()]
    plan = build_plan(rows, walk_x, args.seed)
    plan_path = os.path.join(run_dir, "plan.jsonl")
    if os.path.isfile(plan_path):
        old = [json.loads(l) for l in open(plan_path, encoding="utf-8") if l.strip()]
        old_ids = {p["id"]: p for p in old}
        for p in plan:   # 计划必须与已有的一致，否则续跑会把不同指令的视频当成同一条
            if p["id"] in old_ids and not np.allclose(old_ids[p["id"]]["frame_offset_px"], p["frame_offset_px"]):
                raise RuntimeError(f"{plan_path} 里 {p['id']} 的指令与本次不同，换 --run")
    with open(plan_path, "w", encoding="utf-8") as fh:
        for p in plan:
            fh.write(json.dumps(p, ensure_ascii=False) + "\n")
    print(f"[plan] {len(plan)} 条（gain {sum(p['exam'] == 'gain' for p in plan)} / back {sum(p['exam'] == 'back' for p in plan)} / "
          f"straight {sum(p['exam'] == 'straight' for p in plan)}）· walk_x {walk_x:.2f} px/s · run_dir {run_dir}", flush=True)

    gen_stat = None
    if "gen" in stages:
        gen_stat = stage_gen(args, run_dir)
    elif os.path.isfile(os.path.join(run_dir, "gen_stat.json")):
        with open(os.path.join(run_dir, "gen_stat.json")) as fh:
            gen_stat = json.load(fh)

    if "measure" in stages:
        sc = None
        sc_path = os.path.join(run_dir, "selfcheck.json")
        if args.no_selfcheck and os.path.isfile(sc_path):
            with open(sc_path, encoding="utf-8") as fh:
                sc = json.load(fh)
            print(f"[selfcheck] 复用 {sc_path}（all_pass={sc['all_pass']}）", flush=True)
        elif not args.no_selfcheck:
            t0 = time.time()
            sc = selfcheck(walk_x, rows[0]["path"])
            sc["sec"] = time.time() - t0
            with open(sc_path, "w", encoding="utf-8") as fh:
                json.dump(sc, fh, ensure_ascii=False, indent=1, default=float)
            print(f"[selfcheck] all_pass={sc['all_pass']} {sc['checks']} · {sc['sec']:.0f}s", flush=True)
        mres = stage_measure(args, run_dir, plan)
        summary = summarize(mres["metrics"], walk_x)
        summary.update(run=args.run, ckpt=args.ckpt, n_plan=len(plan), n_videos_missing=mres["n_missing"],
                       measure_sec=mres["measure_sec"], gen_stat=gen_stat,
                       selfcheck_all_pass=None if sc is None else sc["all_pass"], seed=args.seed, steps=args.steps,
                       prompt=args.scene or "NULL_PROMPT")
        with open(os.path.join(run_dir, "summary.json"), "w", encoding="utf-8") as fh:
            json.dump(summary, fh, ensure_ascii=False, indent=1, default=float)
        make_figure(summary, mres["metrics"], os.path.join(run_dir, "figures.png"))
        rep = make_report(summary, sc, gen_stat, args.run, args.ckpt)
        with open(os.path.join(run_dir, "report.md"), "w", encoding="utf-8") as fh:
            fh.write(rep)
        print(rep)
        return summary


if __name__ == "__main__":
    main()