Robot_Arm/ml/eval/acceptance.py

"""验收汇总（方案 §9.2 + §10.1 R2/R7/R8/R10）：把各阶段产物里的判据集中核对，输出 ml/runs/acceptance.json。
  python -m ml.eval.acceptance --ckpt ml/runs/current.pt --samples ml/data/samples --out ml/runs/acceptance.json
自己跑的一项：held-out（val 划分）全部任务闭环仿真（开仿真安全层）——可达任务到达率 ≥ 98%、D 类停靠距离 ≤ 老师 + 5 mm、起步失败 = 0。
其余项读已有文件（缺文件记为 missing，不算通过）：
  ml/runs/policy_bench.json   部署形态：各集合穿模帧 = 0；near 30/30（pos、pose）；route 全部到达；nc1000 0 失败；刹停率 < 2%；R8 恢复 ≥ 95%、活锁 = 0
  ml/data/r2.json             R2 近邻标签 std < 10% 步长
  ml/runs/bench_*/log.jsonl   R10 吞吐 ≥ 5000 样本/s（≈ 60000 步 ≤ 3.5 h）
  ml/runs/m5_gate_*.json      M5 头门禁
  ml/runs/mine/r*/gate.json   R7 最后一轮左右比例 / 行程门禁
  ml/runs/r4.json             R4 浏览器推理 batch=8 P99 < 15 ms（r4 探针）
  ml/runs/r9_browser.json     R9 查看器整次规划（推理 + 前瞻 + 安全层）P99 < 45 ms、饥饿节拍 < 1%（训练好的模型，bench 阶段测）
  webapp/models/*.meta.json   被评测的模型不能是 --random 探针
2026-10-18 新增（审查：§9 验收表在流水线里没有执行点）。
2026-10-18 第五轮：加 R4、随机探针模型拒绝（N-5）、R8 出题不足额判不通过（N-4）；R2 改为参考项（N-2：阈值与池大小耦合、未校准）。
2026-10-18 第六轮：加 R9（审查 B-1：纯推理口径会放过整周期丢步的配置）与安全层两把尺对照（B-2：同一批题浏览器 / Node 差 > 3× 标红）。"""
import argparse, glob, json, os, re
import torch


def jload(p):
    try: return json.load(open(p, encoding='utf-8'))
    except Exception: return None


def main():
    from ml.common.kin_torch import Kin
    from ml.model.policy import build
    from ml.data import load_tasks
    from ml.sim.closed_loop import evaluate
    ap = argparse.ArgumentParser()
    ap.add_argument('--ckpt', required=True); ap.add_argument('--samples', default='ml/data/samples')
    ap.add_argument('--out', default='ml/runs/acceptance.json')
    ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu')
    a = ap.parse_args()
    items = []
    def add(name, value, ok, rule):
        items.append({'item': name, 'value': value, 'pass': (None if ok is None else bool(ok)), 'rule': rule})

    # 1) held-out 闭环（仿真安全层）
    ck = torch.load(a.ckpt, map_location=a.device)
    model = build(ck['cfg']).to(a.device); model.load_state_dict(ck['model']); model.eval()
    kin = Kin(device=a.device)
    val = [t for t in load_tasks([a.samples]) if t['split'] == 'val']
    reach = [t for t in val if t.get('reached') and t['cat'] != 'D']
    dock = [t for t in val if t['cat'] == 'D']
    if reach:
        r, _ = evaluate(model, kin, reach, a.device, safety=True)
        add('held-out 可达任务到达率', r['success'], r['success'] >= 0.98, '≥ 0.98（仿真安全层）')
        add('起步失败率（R2）', r['start_fail'], r['start_fail'] <= 0.002, '≤ 0.2%（约 2500 题容许 ≤5 题，0 题判据过脆）')
    else:
        add('held-out 可达任务到达率', None, False, 'val 中无可达任务')
    if dock:
        _, pers = evaluate(model, kin, dock, a.device, safety=True)
        err = torch.cat([p['err'] for _, p in pers])
        ref = torch.tensor([t.get('err') if t.get('err') is not None else float('inf') for t in dock])
        rate = (err <= ref + 5).float().mean().item()
        add('D 类停靠距离 ≤ 老师 + 5 mm 的比例', rate, rate >= 0.98, '≥ 0.98')

    # 2) 部署形态
    mm = jload('webapp/models/traj_policy.meta.json')
    add('部署模型是训练产物（非随机探针）', None if mm is None else not mm.get('random_init'), mm is not None and not mm.get('random_init'), 'meta.random_init 不为真')
    r4 = jload('ml/runs/r4.json')
    add('R4 浏览器推理 batch=8 P99（ms）', (r4 or {}).get('deploy', {}).get('p99') if r4 else 'missing', bool(r4 and r4.get('pass')), '< 15（随机初始化同结构模型实测）')
    r9 = jload('ml/runs/r9_browser.json'); rr = (r9 or {}).get('result') or {}
    add('R9 查看器稳态规划 P99（ms）/ 运动中饥饿占比 / 认证超时占比', [rr.get('planWarmP99'), rr.get('starve_rate'), rr.get('cert_time_rate')] if r9 else 'missing', bool(r9 and r9.get('pass')),
        '< 45 ms、< 1%、< 5%（训练好的模型、30 道贴柱题，ml.eval.r4_probe --viewer；冷启动 P99 / 刹停率 / tickGapMax 见 r9_browser.json）')
    pb = jload('ml/runs/policy_bench.json')
    if pb is None:
        add('部署形态评测', 'missing', False, '先跑 bash ml/pipeline.sh bench')
    else:
        sets = {k: v for k, v in pb.items() if isinstance(v, dict) and 'n' in v}
        cf = sum(v['collisionFrames'] for v in sets.values())
        add('部署形态穿模帧（全部集合）', cf, cf == 0, '= 0')
        for k in ('near_column_pos', 'near_column_pose', 'route40', 'near_column_1000'):
            if k in sets: add(f'{k} 通过', f"{sets[k]['ok']}/{sets[k]['n']}", sets[k]['ok'] == sets[k]['n'], '全部通过')
            else: add(f'{k} 通过', 'missing', False, '全部通过')
        plans = sum(v.get('plans', 0) for v in sets.values()); br = sum(v.get('brakes', 0) for v in sets.values())
        add('刹停发生率（刹停 / 推理）', br / plans if plans else None, (br / plans < 0.02) if plans else False, '< 0.02')
        r8 = sets.get('r8_adversarial')
        if r8 and r8.get('requested') and r8['n'] < r8['requested']:
            add('R8 出题数', f"{r8['n']}/{r8['requested']}", False, '出满要求题数（不足则恢复率分母缩水）')
        if r8:
            add('R8 对抗集 3 s 内恢复率', r8['recoveryRate'], r8['recoveryRate'] >= 0.95, '≥ 0.95')
            add('R8 活锁', r8['livelock'], r8['livelock'] == 0, '= 0')
        else: add('R8 对抗集', 'missing', False, 'bench 需含 r8')
        add('推理延迟 P99（Node CPU，参考）', max((v.get('inferMsP99') or 0) for v in sets.values()), None, '浏览器以 r9_browser.json 为准')
        nc = sets.get('near_column_pos') or {}
        bs, ns = rr.get('safetyP99'), nc.get('safetyMsP99')
        ratio = (bs / ns) if (bs is not None and ns) else None
        add('安全层 P99 浏览器 / Node（同一批贴柱题）', [bs, ns, None if ratio is None else round(ratio, 2)], None if ratio is None else ratio <= 3,
            '≤ 3×；超了看 certQueriesP99 / certMemoHitRate / certFailWhy 定位')

    # 3) 数据 / 训练 / 后训练门禁
    r2 = jload('ml/data/r2.json')
    add('R2 近邻标签 std（步长比例）', [r2.get('label_std_mean'), r2.get('label_std_norm')] if r2 else 'missing', None, '参考：< 0.10；按近邻距离归一后看 label_std_norm（见 check_r2）')
    sps = None
    for f in sorted(glob.glob('ml/runs/bench_*/log.jsonl')):
        recs = [json.loads(l) for l in open(f) if '"samples_per_s"' in l]
        if recs: sps = max(sps or 0, recs[-1]['samples_per_s'])
    add('R10 训练吞吐（样本/s）', sps, sps is not None and sps >= 5000, '≥ 5000（≈ 60000 步 ≤ 3.5 h）')
    m5 = [jload(f) for f in sorted(glob.glob('ml/runs/m5_gate_*.json'))]
    m5 = [m for m in m5 if m]
    add('M5 头门禁', [{'head': m['head'], 'collision_rate': m['policy_only']['collision_rate']} for m in m5] or 'missing',
        any(m['pass'] for m in m5), '纯策略贴立柱穿柱率 ≤ 0.02')
    rnum = lambda p: int((re.findall(r'[\\/]r(\d+)[\\/]gate\.json$', p) or ['0'])[0])
    gates = sorted(glob.glob('ml/runs/mine/r*/gate.json'), key=rnum)
    if gates:
        gj = jload(gates[-1])
        add('R7 左右比例变化（pp）/ 行程比', [gj.get('R7_left_share_shift_pp'), gj.get('R7_travel_median_ratio')], gj.get('R7_ok'), '≤ 5 pp 且 ≤ 1.05')
    else: add('R7 门禁', 'missing', False, '先跑 mine')

    hard = [i for i in items if i['pass'] is not None]
    out = {'ckpt': a.ckpt, 'all_pass': all(i['pass'] for i in hard), 'failed': [i['item'] for i in hard if not i['pass']], 'items': items}
    os.makedirs(os.path.dirname(a.out) or '.', exist_ok=True)
    json.dump(out, open(a.out, 'w', encoding='utf-8'), indent=1, ensure_ascii=False)
    for i in items:
        mark = {True: '通过', False: '未过', None: '参考'}[i['pass']]
        print(f"[{mark}] {i['item']}: {i['value']}  （{i['rule']}）")
    print('全部通过' if out['all_pass'] else '未通过：' + '、'.join(out['failed']))


if __name__ == '__main__':
    main()