{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 11 把 checkpoint 部署到 MuJoCo 闭环\n\n        这一节把 ACT、SmolVLA 或 pi_0 checkpoint 放回 `SimpleEnv2`，让策略按 20 Hz 读取双相机图像和 6 维关节状态，并输出 7 维动作。评估会保存视频和 JSONL，同时报告旧几何成功与严格 `physical_success`。\n\n        这是仿真闭环推理，不是把模型导出成服务。模型每次动作后都会重新读取环境观测；如果只在数据集上预测 GT 动作，那属于 open-loop replay，不能替代这里的成功率。\n"
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "TOPIC_ROOT = $NOTEBOOK_TOPIC_ROOT\nPROJECT_ROOT = $PROJECT_ROOT\nDATA_ROOT = $DATA_ROOT\nOUTPUT_ROOT = $OUTPUT_ROOT\nMODEL_ROOT = $MODEL_ROOT\n"
        }
      ],
      "source": "from pathlib import Path\nimport json\nimport os\nimport shutil\nimport subprocess\nimport sys\n\n\ndef find_topic_root():\n    cwd = Path.cwd().resolve()\n    for candidate in [cwd, *cwd.parents]:\n        if (candidate / \"assets\" / \"metrics_snapshot.json\").exists():\n            return candidate\n    raise RuntimeError(\"请从 AMD ROCm 专题目录或 notebooks 子目录启动 Jupyter。\")\n\n\nTOPIC_ROOT = find_topic_root()\nASSET_DIR = TOPIC_ROOT / \"assets\"\nNOTEBOOK_DIR = TOPIC_ROOT / \"notebooks\"\nPROJECT_ROOT = Path(os.environ.get(\"PROJECT_ROOT\", \"/path/to/every-embodied/mujoco_pnp\"))\nDATA_ROOT = Path(os.environ.get(\"DATA_ROOT\", \"/path/to/datasets/every_embodied\"))\nOUTPUT_ROOT = Path(os.environ.get(\"OUTPUT_ROOT\", TOPIC_ROOT / \"outputs\"))\nMODEL_ROOT = Path(os.environ.get(\"MODEL_ROOT\", PROJECT_ROOT / \"ckpt\"))\n\nprint(\"TOPIC_ROOT =\", TOPIC_ROOT)\nprint(\"PROJECT_ROOT =\", PROJECT_ROOT)\nprint(\"DATA_ROOT =\", DATA_ROOT)\nprint(\"OUTPUT_ROOT =\", OUTPUT_ROOT)\nprint(\"MODEL_ROOT =\", MODEL_ROOT)\n"
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {},
      "outputs": [],
      "source": "try:\n    from IPython.display import Image, Markdown, display\nexcept Exception:\n    class Markdown(str):\n        pass\n\n    def display(obj):\n        print(obj)\n\n    def Image(filename=None, width=None):\n        return f\"[image] {filename}\"\n\n\ndef show_asset(filename, width=960):\n    path = ASSET_DIR / filename\n    if path.exists():\n        display(Image(filename=str(path), width=width))\n    else:\n        print(f\"缺少素材：{path}\")\n\n\ndef md_table(headers, rows):\n    lines = [\"| \" + \" | \".join(headers) + \" |\", \"| \" + \" | \".join([\"---\"] * len(headers)) + \" |\"]\n    for row in rows:\n        lines.append(\"| \" + \" | \".join(str(x) for x in row) + \" |\")\n    display(Markdown(\"\\n\".join(lines)))\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 零训练成功预览：先看正确行为\n\n这一格不加载模型、不启动 MuJoCo，也不会消耗训练额度。它直接显示仓库内置的四视角严格成功回放，先确认接近、夹取、抬升、搬运、释放和稳定放置的正确顺序。视频来自固定环境上的 `pi0 + visual/history learned head`，不是 raw pi0，也不能代表随机位置泛化。\n"
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "当前是纯 Python 执行器；在 Jupyter 中运行本单元格即可嵌入播放视频。\n严格成功预览： $NOTEBOOK_TOPIC_ROOT/assets/pnp_four_view_strict_success.mp4\n"
        }
      ],
      "source": "SUCCESS_PREVIEW = ASSET_DIR / \"pnp_four_view_strict_success.mp4\"\nif not SUCCESS_PREVIEW.exists():\n    raise FileNotFoundError(f\"缺少成功预览视频：{SUCCESS_PREVIEW}\")\n\ntry:\n    from IPython.display import Video\nexcept ImportError:\n    print(\"当前是纯 Python 执行器；在 Jupyter 中运行本单元格即可嵌入播放视频。\")\nelse:\n    display(Video(filename=str(SUCCESS_PREVIEW), embed=True, html_attributes=\"controls muted\"))\n\nprint(\"严格成功预览：\", SUCCESS_PREVIEW)\n"
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {},
      "outputs": [],
      "source": "import shlex\n\ntry:\n    import yaml\nexcept ImportError as exc:\n    raise RuntimeError(\"当前环境缺少 PyYAML，请先执行 pip install pyyaml。\") from exc\n\n\ndef require_project_layout():\n    required = [\n        PROJECT_ROOT / \"train_model.py\",\n        PROJECT_ROOT / \"env_config.py\",\n        PROJECT_ROOT / \"asset\" / \"example_scene_y2.xml\",\n        PROJECT_ROOT / \"mujoco_env\" / \"y_env2.py\",\n    ]\n    missing = [path for path in required if not path.exists()]\n    if missing:\n        raise FileNotFoundError(\n            \"PROJECT_ROOT 不是 04mujoco 教程目录，缺少：\\n\"\n            + \"\\n\".join(str(path) for path in missing)\n        )\n    return True\n\n\ndef dataset_report(dataset_root):\n    dataset_root = Path(dataset_root)\n    info_path = dataset_root / \"meta\" / \"info.json\"\n    tasks_path = dataset_root / \"meta\" / \"tasks.jsonl\"\n    if not info_path.exists():\n        raise FileNotFoundError(f\"找不到 LeRobot 元数据：{info_path}\")\n    info = json.loads(info_path.read_text(encoding=\"utf-8\"))\n    tasks = []\n    if tasks_path.exists():\n        tasks = [\n            json.loads(line)[\"task\"]\n            for line in tasks_path.read_text(encoding=\"utf-8\").splitlines()\n            if line.strip()\n        ]\n    features = info.get(\"features\", {})\n    rows = [\n        (\"repo_id\", info.get(\"repo_id\", \"\")),\n        (\"episodes\", info.get(\"total_episodes\", 0)),\n        (\"frames\", info.get(\"total_frames\", 0)),\n        (\"fps\", info.get(\"fps\", \"\")),\n        (\"state shape\", features.get(\"observation.state\", {}).get(\"shape\")),\n        (\"action shape\", features.get(\"action\", {}).get(\"shape\")),\n        (\"tasks\", \" / \".join(tasks)),\n    ]\n    md_table([\"数据项\", \"读取结果\"], rows)\n    return info, tasks\n\n\ndef make_training_config(\n    policy_type,\n    dataset_repo_id,\n    dataset_root,\n    output_dir,\n    steps,\n    batch_size,\n    chunk_size,\n    n_action_steps,\n    save_freq,\n    seed=42,\n):\n    return {\n        \"dataset\": {\"repo_id\": dataset_repo_id, \"root\": str(Path(dataset_root))},\n        \"policy\": {\n            \"type\": policy_type,\n            \"chunk_size\": int(chunk_size),\n            \"n_action_steps\": int(n_action_steps),\n            \"device\": \"cuda\",\n        },\n        \"save_checkpoint\": True,\n        \"output_dir\": str(Path(output_dir)),\n        \"batch_size\": int(batch_size),\n        \"job_name\": Path(output_dir).name,\n        \"resume\": False,\n        \"seed\": int(seed),\n        \"num_workers\": 4,\n        \"steps\": int(steps),\n        \"eval_freq\": 0,\n        \"log_freq\": max(1, min(50, int(steps))),\n        \"save_freq\": int(save_freq),\n        \"use_policy_training_preset\": True,\n        \"wandb\": {\n            \"enable\": False,\n            \"project\": f\"every_embodied_{policy_type}\",\n            \"entity\": None,\n            \"disable_artifact\": True,\n        },\n    }\n\n\ndef write_training_config(name, config):\n    config_dir = OUTPUT_ROOT / \"configs\"\n    config_dir.mkdir(parents=True, exist_ok=True)\n    path = config_dir / f\"{name}.yaml\"\n    path.write_text(\n        yaml.safe_dump(config, allow_unicode=True, sort_keys=False),\n        encoding=\"utf-8\",\n    )\n    print(\"已写出配置：\", path)\n    return path\n\n\ndef training_command(config_path):\n    return [\n        sys.executable,\n        str(PROJECT_ROOT / \"train_model.py\"),\n        \"--config_path\",\n        str(Path(config_path)),\n    ]\n\n\ndef run_training(config_path, enabled=False):\n    require_project_layout()\n    command = training_command(config_path)\n    print(\"$\", shlex.join(command))\n    if not enabled:\n        print(\"当前只预览命令。确认配置后，把对应 RUN_* 开关改为 True。\")\n        return None\n    env = os.environ.copy()\n    env[\"PYTHONPATH\"] = f\"{PROJECT_ROOT}:{env.get('PYTHONPATH', '')}\"\n    return subprocess.run(command, cwd=PROJECT_ROOT, env=env, check=True)\n\n\ndef show_rocm_resources():\n    command = [\"rocm-smi\", \"--showuse\", \"--showmemuse\", \"--showtemp\"]\n    print(\"$\", shlex.join(command))\n    if shutil.which(command[0]) is None:\n        print(\"未找到 rocm-smi。确认当前机器是否安装 ROCm。\")\n        return\n    result = subprocess.run(command, check=False, capture_output=True, text=True)\n    if result.stdout:\n        print(result.stdout.rstrip())\n    if result.stderr:\n        print(result.stderr.rstrip(), file=sys.stderr)\n"
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {},
      "outputs": [],
      "source": "import random\nimport time\n\nimport numpy as np\nfrom PIL import Image\n\ntry:\n    import imageio.v2 as imageio\nexcept ImportError:\n    imageio = None\n\n\ndef find_pretrained_model(run_dir):\n    run_dir = Path(run_dir)\n    last = run_dir / \"checkpoints\" / \"last\" / \"pretrained_model\"\n    if last.exists():\n        return last\n    candidates = sorted((run_dir / \"checkpoints\").glob(\"*/pretrained_model\"))\n    if not candidates:\n        raise FileNotFoundError(f\"没有在 {run_dir} 下找到 pretrained_model\")\n    return candidates[-1]\n\n\ndef image_tensor(array, size=(256, 256)):\n    import torch\n\n    image = Image.fromarray(array).convert(\"RGB\").resize(size)\n    value = np.asarray(image, dtype=np.float32) / 255.0\n    return torch.from_numpy(value).permute(2, 0, 1).contiguous()\n\n\ndef load_policy(policy_type, policy_path, dataset_repo_id, dataset_root, device=\"cuda\"):\n    from lerobot.common.datasets.lerobot_dataset import LeRobotDatasetMetadata\n\n    if policy_type == \"act\":\n        from lerobot.common.policies.act.modeling_act import ACTPolicy as PolicyClass\n    elif policy_type == \"smolvla\":\n        from lerobot.common.policies.smolvla.modeling_smolvla import SmolVLAPolicy as PolicyClass\n    elif policy_type == \"pi0\":\n        from lerobot.common.policies.pi0.modeling_pi0 import PI0Policy as PolicyClass\n    else:\n        raise ValueError(f\"不支持的 policy_type：{policy_type}\")\n\n    metadata = LeRobotDatasetMetadata(dataset_repo_id, root=dataset_root)\n    policy = PolicyClass.from_pretrained(str(policy_path), dataset_stats=metadata.stats)\n    policy.to(device)\n    policy.eval()\n    return policy\n\n\ndef strict_snapshot(\n    env,\n    initial_target_z,\n    initial_plate_pos,\n    max_target_lift,\n    max_lifted_run,\n    stable_place_steps,\n):\n    target_pos = np.asarray(env.env.get_p_body(env.obj_target), dtype=np.float64)\n    plate_pos = np.asarray(env.env.get_p_body(\"body_obj_plate_11\"), dtype=np.float64)\n    target_R = np.asarray(env.env.get_R_body(env.obj_target), dtype=np.float64)\n    xy_dist = float(np.linalg.norm(target_pos[:2] - plate_pos[:2]))\n    upright_cos = float(target_R[2, 2])\n    gripper_open = bool(float(env.env.get_qpos_joint(\"rh_r1\")[0]) < 0.1)\n    tcp_high = bool(float(env.env.get_p_body(\"tcp_link\")[2]) > 0.9)\n    legacy_success = bool(env.check_success())\n    plate_xy_displacement = float(np.linalg.norm(plate_pos[:2] - initial_plate_pos[:2]))\n    placement_candidate = bool(\n        legacy_success\n        and max_target_lift >= 0.03\n        and max_lifted_run >= 3\n        and upright_cos >= 0.7\n        and abs(float(target_pos[2] - plate_pos[2])) < 0.08\n        and plate_xy_displacement < 0.05\n        and gripper_open\n        and tcp_high\n    )\n    stable_place_steps = stable_place_steps + 1 if placement_candidate else 0\n    physical_success = bool(placement_candidate and stable_place_steps >= 5)\n    return {\n        \"legacy_success\": legacy_success,\n        \"physical_success\": physical_success,\n        \"xy_dist\": xy_dist,\n        \"target_z\": float(target_pos[2]),\n        \"plate_z\": float(plate_pos[2]),\n        \"max_target_lift\": float(max_target_lift),\n        \"max_lifted_run\": int(max_lifted_run),\n        \"upright_cos\": upright_cos,\n        \"plate_xy_displacement\": plate_xy_displacement,\n        \"placement_candidate\": placement_candidate,\n        \"stable_place_steps\": int(stable_place_steps),\n        \"gripper_open\": gripper_open,\n        \"tcp_high\": tcp_high,\n    }, stable_place_steps\n\n\ndef run_closed_loop(\n    policy,\n    policy_type,\n    instruction,\n    seeds,\n    output_dir,\n    device=\"cuda\",\n    max_action_steps=300,\n    render=True,\n):\n    from mujoco_env.y_env2 import SimpleEnv2\n    import torch\n\n    output_dir = Path(output_dir)\n    output_dir.mkdir(parents=True, exist_ok=True)\n    xml_path = PROJECT_ROOT / \"asset\" / \"example_scene_y2.xml\"\n    results = []\n\n    for seed in seeds:\n        np.random.seed(seed)\n        random.seed(seed)\n        env = SimpleEnv2(str(xml_path), action_type=\"joint_angle\", state_type=\"joint_angle\", seed=seed)\n        env.set_instruction(instruction)\n        policy.reset()\n\n        initial_target_z = float(env.env.get_p_body(env.obj_target)[2])\n        initial_plate_pos = np.asarray(\n            env.env.get_p_body(\"body_obj_plate_11\"), dtype=np.float64\n        ).copy()\n        max_target_lift = 0.0\n        lifted_run = 0\n        max_lifted_run = 0\n        stable_place_steps = 0\n        frames = []\n        started = time.perf_counter()\n        final = None\n\n        action_step = 0\n        while action_step < max_action_steps and env.env.is_viewer_alive():\n            env.step_env()\n            if not env.env.loop_every(HZ=20):\n                continue\n\n            state = env.get_joint_state()[:6]\n            agent_image, wrist_image = env.grab_image()\n            observation = {\n                \"observation.state\": torch.as_tensor(state, dtype=torch.float32, device=device).unsqueeze(0),\n                \"observation.image\": image_tensor(agent_image).to(device).unsqueeze(0),\n                \"observation.wrist_image\": image_tensor(wrist_image).to(device).unsqueeze(0),\n                \"task\": [instruction],\n            }\n            with torch.inference_mode():\n                action = policy.select_action(observation)[0, :7].detach().cpu().numpy()\n            action[6] = np.clip(action[6], 0.0, 1.0)\n            env.step(action.astype(np.float32))\n\n            target_z = float(env.env.get_p_body(env.obj_target)[2])\n            lift = target_z - initial_target_z\n            max_target_lift = max(max_target_lift, lift)\n            lifted_run = lifted_run + 1 if lift >= 0.03 else 0\n            max_lifted_run = max(max_lifted_run, lifted_run)\n            final, stable_place_steps = strict_snapshot(\n                env,\n                initial_target_z,\n                initial_plate_pos,\n                max_target_lift,\n                max_lifted_run,\n                stable_place_steps,\n            )\n\n            if action_step % 2 == 0:\n                frames.append(np.asarray(agent_image))\n            if render:\n                env.render()\n            action_step += 1\n            if final[\"physical_success\"]:\n                break\n\n        video_path = output_dir / f\"{policy_type}_seed{seed}.mp4\"\n        video_saved = False\n        if frames and imageio is not None:\n            try:\n                imageio.mimsave(video_path, frames, fps=10, quality=8)\n                video_saved = True\n            except Exception as exc:\n                print(\"视频保存失败：\", repr(exc))\n        elif frames:\n            print(\"未安装 imageio，跳过视频保存。可执行 pip install imageio imageio-ffmpeg 后重跑。\")\n        row = {\n            \"policy_type\": policy_type,\n            \"seed\": int(seed),\n            \"instruction\": instruction,\n            \"action_steps\": int(action_step),\n            \"elapsed_s\": round(time.perf_counter() - started, 3),\n            \"video\": str(video_path) if video_saved else None,\n            **(final or {}),\n        }\n        print(json.dumps(row, ensure_ascii=False))\n        results.append(row)\n        env.env.close_viewer()\n\n    result_path = output_dir / \"results.jsonl\"\n    result_path.write_text(\n        \"\".join(json.dumps(row, ensure_ascii=False) + \"\\n\" for row in results),\n        encoding=\"utf-8\",\n    )\n    physical = sum(row.get(\"physical_success\", False) for row in results)\n    print(f\"physical_success = {physical}/{len(results)}\")\n    print(\"结果文件：\", result_path)\n    return results\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Checkpoint 1：选择模型、数据统计和评估 seed\n"
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "policy type = act\nmodel run = $MODEL_ROOT/act_rocm_full\ndataset = $TRAIN_DATA_ROOT\ntask = Place the blue mug on the plate.\nseeds = [1000, 1001, 1002, 1003]\n"
        }
      ],
      "source": "POLICY_TYPE = os.environ.get(\"POLICY_TYPE\", \"act\")  # act / smolvla / pi0\nDATASET_REPO_ID = os.environ.get(\"DATASET_REPO_ID\", \"datawhale_eai_pnp_language\")\nEVAL_DATA_ROOT = Path(\n    os.environ.get(\"EVAL_DATA_ROOT\", DATA_ROOT / \"omy_pnp_language\")\n).expanduser()\nMODEL_RUN_DIR = Path(\n    os.environ.get(\"MODEL_RUN_DIR\", MODEL_ROOT / f\"{POLICY_TYPE}_rocm_full\")\n).expanduser()\nTASK_TEXT = os.environ.get(\"TASK_TEXT\", \"Place the blue mug on the plate.\")\nEVAL_SEEDS = [int(value) for value in os.environ.get(\"EVAL_SEEDS\", \"1000,1001,1002,1003\").split(\",\")]\nMAX_ACTION_STEPS = int(os.environ.get(\"MAX_ACTION_STEPS\", \"300\"))\nRUN_CLOSED_LOOP = False\n\nprint(\"policy type =\", POLICY_TYPE)\nprint(\"model run =\", MODEL_RUN_DIR)\nprint(\"dataset =\", EVAL_DATA_ROOT)\nprint(\"task =\", TASK_TEXT)\nprint(\"seeds =\", EVAL_SEEDS)\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "先用 4 个固定 seed 做 smoke，但 4 条不具有统计代表性。模型通过小面板后，再扩到至少 20–30 个 held-out seeds，并把红杯、蓝杯分别统计。评估时不要读取 target/plate 坐标、数据集 phase 或 GT 动作作为策略输入。\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Checkpoint 2：检查数据与 checkpoint\n"
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "| 数据项 | 读取结果 |\n| --- | --- |\n| repo_id |  |\n| episodes | 20 |\n| frames | 13891 |\n| fps | 20 |\n| state shape | [6] |\n| action shape | [7] |\n| tasks | Pick up the red mug and place it on the white plate. |\n没有在 $MODEL_ROOT/act_rocm_full 下找到 pretrained_model\n"
        }
      ],
      "source": "require_project_layout()\nif (EVAL_DATA_ROOT / \"meta\" / \"info.json\").exists():\n    dataset_report(EVAL_DATA_ROOT)\nelse:\n    print(\"缺少评估数据统计：\", EVAL_DATA_ROOT)\n\ntry:\n    POLICY_PATH = Path(os.environ.get(\"POLICY_PATH\", \"\"))\n    if not str(POLICY_PATH) or str(POLICY_PATH) == \".\":\n        POLICY_PATH = find_pretrained_model(MODEL_RUN_DIR)\n    print(\"policy path =\", POLICY_PATH)\nexcept FileNotFoundError as exc:\n    POLICY_PATH = None\n    print(exc)\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Checkpoint 3：加载策略并执行闭环\n"
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "闭环评估默认关闭。确认 DISPLAY、checkpoint、数据统计和 seed 后，将 RUN_CLOSED_LOOP 改为 True。\n"
        }
      ],
      "source": "if RUN_CLOSED_LOOP:\n    if POLICY_PATH is None:\n        raise FileNotFoundError(\"请先完成训练，或通过 POLICY_PATH 指定 pretrained_model。\")\n    if POLICY_TYPE == \"pi0\" and not os.environ.get(\"HF_TOKEN\"):\n        print(\"本地 checkpoint 通常可离线加载；若仍需读取 gated 配置，请先私密设置 HF_TOKEN。\")\n    show_rocm_resources()\n    policy = load_policy(\n        policy_type=POLICY_TYPE,\n        policy_path=POLICY_PATH,\n        dataset_repo_id=DATASET_REPO_ID,\n        dataset_root=EVAL_DATA_ROOT,\n        device=\"cuda\",\n    )\n    results = run_closed_loop(\n        policy=policy,\n        policy_type=POLICY_TYPE,\n        instruction=TASK_TEXT,\n        seeds=EVAL_SEEDS,\n        output_dir=OUTPUT_ROOT / f\"{POLICY_TYPE}_closed_loop\",\n        device=\"cuda\",\n        max_action_steps=MAX_ACTION_STEPS,\n        render=True,\n    )\nelse:\n    print(\"闭环评估默认关闭。确认 DISPLAY、checkpoint、数据统计和 seed 后，将 RUN_CLOSED_LOOP 改为 True。\")\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Checkpoint 4：读取成功率和失败类型\n"
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "尚未生成闭环结果： $OUTPUT_ROOT/act_closed_loop/results.jsonl\n"
        }
      ],
      "source": "result_path = OUTPUT_ROOT / f\"{POLICY_TYPE}_closed_loop\" / \"results.jsonl\"\nif result_path.exists():\n    rows = [json.loads(line) for line in result_path.read_text(encoding=\"utf-8\").splitlines() if line.strip()]\n    physical = sum(row.get(\"physical_success\", False) for row in rows)\n    legacy = sum(row.get(\"legacy_success\", False) for row in rows)\n    md_table(\n        [\"口径\", \"结果\"],\n        [\n            (\"legacy_success\", f\"{legacy}/{len(rows)}\"),\n            (\"physical_success\", f\"{physical}/{len(rows)}\"),\n        ],\n    )\n    for row in rows:\n        if not row.get(\"physical_success\"):\n            print(\n                \"失败 seed\",\n                row[\"seed\"],\n                \"xy=\", round(row.get(\"xy_dist\", float(\"nan\")), 4),\n                \"lift=\", round(row.get(\"max_target_lift\", float(\"nan\")), 4),\n                \"upright=\", round(row.get(\"upright_cos\", float(\"nan\")), 4),\n            )\nelse:\n    print(\"尚未生成闭环结果：\", result_path)\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "如果 `legacy_success > physical_success`，优先查看失败视频，通常是推杯、未真正夹住、运输中掉落或只满足终态几何条件。下一步回到 02–06 的诊断 Notebook，把失败定位到 approach、contact、lift、transport、release 中的具体阶段。\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Checkpoint 5：严格判定与四视角实测结果\n"
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "| 严格条件 | 阈值 |\n| --- | --- |\n| 真实抬升 | >= 0.03 m，连续 3 步 |\n| 终态直立 | >= 0.7 |\n| 杯盘高度差 | <= 0.08 m |\n| 盘子位移 | <= 0.05 m |\n| 稳定放置 | 连续 5 步 |\n"
        }
      ],
      "source": "strict_results = json.loads(\n    (ASSET_DIR / \"pi0_strict_input_results.json\").read_text(encoding=\"utf-8\")\n)\npredicate = strict_results[\"strict_predicate\"]\nmd_table(\n    [\"严格条件\", \"阈值\"],\n    [\n        (\"真实抬升\", f'>= {predicate[\"min_lift_m\"]:.2f} m，连续 {predicate[\"min_lift_steps\"]} 步'),\n        (\"终态直立\", f'>= {predicate[\"min_upright_cos\"]:.1f}'),\n        (\"杯盘高度差\", f'<= {predicate[\"max_plate_z_gap_m\"]:.2f} m'),\n        (\"盘子位移\", f'<= {predicate[\"max_plate_xy_displacement_m\"]:.2f} m'),\n        (\"稳定放置\", f'连续 {predicate[\"stable_place_steps\"]} 步'),\n    ],\n)\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "![pi0 strict-input 闭环结果](../assets/pi0_strict_input_progress.png)\n\n            <video controls width=\"960\" src=\"../assets/pnp_four_view_strict_success.mp4\"></video>\n\n            视频对应环境 seed 0、策略采样 seed 3 的严格成功回放。注意它属于 `pi0 + visual/history learned head`，不是 raw pi0；模型输入只有双相机图像、语言、robot proprio 和历史执行动作，没有读取杯子坐标、盘子坐标、GT phase 或 oracle action。\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "随机环境 gate 只有 `1/4`，所以本章没有把固定场景 `6/12` 写成空间泛化成功。下一步应先用修正后的环境 seed 重新采集多位置示教，再扩大到至少 20–30 个真正随机环境评估。\n"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3 (ipykernel)",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3"
    },
    "amd_rocm_tutorial_execution": {
      "executor": "code/execute_tutorial_notebooks.py",
      "python": "3.12.3",
      "code_cells_executed": 10,
      "errors": 0,
      "long_tasks_enabled": false
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}