完成 MuJoCo 环境下自动化数据采集系统构建

AGENTS.md

@@ -0,0 +1,84 @@
+# AKA-Sim2Real
+
+基于前视视角模拟器的数据采集与 ACT (Action Chunking Transformer) 模型训练系统。
+
+## 🛠️ 技术栈
+
+- **后端**: FastAPI, Socket.IO, PyTorch
+- **前端**: React, TypeScript, Socket.IO Client
+- **文档**: mdbook
+
+## 🚫 核心红线
+
+- **Python 环境**: 使用 `pip`，确保 `requirements.txt` 依赖完整，引入新依赖更新`requirements.txt`
+- **模型层**: `policies/models/act/` 为纯模型定义，禁止在此目录外修改模型结构
+- **API 调用**: 通过显式 runtime 接口调用推理，禁止直接依赖隐式模块状态
+- **测试**: 修改 ACT 相关代码后必须运行 `python3 backend/run_act_checks.py`
+
+## ⚙️ 常用命令
+
+```bash
+# 启动后端
+cd backend && python main.py
+
+# 启动前端
+cd ui && npm run dev
+
+# 运行测试
+coverage erase && python3 -m pytest tests/ --cov=backend --cov-report=html
+
+# 打开报告网页
+open htmlcov/index.html
+```
+
+## 🏗️ 架构与规范
+
+### 目录结构
+
+```
+├── backend/                 # Python 后端 (FastAPI + Socket.IO)
+│   ├── api/                # REST API 路由 (domains/inference, training, episode)
+│   ├── sio_handlers/       # Socket.IO 事件处理 (sim/real 命名空间)
+│   └── services/
+│       ├── inference/      # checkpoint.py, preprocess.py, execution.py, runtime.py
+│       ├── training/       # 训练编排
+│       └── episode/        # 数据采集/导出
+├── policies/models/act/    # ACT 模型 (modeling_act, configuration_act, defaults, train_act)
+├── ui/                    # React 前端
+├── tests/act/             # ACT 测试
+└── docs/                  # mdbook 文档
+```
+
+### 代码规范
+
+## 🧹 代码整洁规范 (Clean Code Principles)
+
+在编写任何代码时，必须严格遵守以下 Clean Code 原则：
+
+### 1. 命名即文档
+- **意图清晰**: 变量、函数、类的命名必须揭示其意图。避免使用魔术数字和模糊缩写（如 `d` 代表 `data`，`val` 代表 `value`）。
+- **可搜索性**: 避免使用魔法数字。如果必须使用，请定义为常量（如 `const DAYS_IN_WEEK = 7`）。
+- **类型命名**: 类名和对象名必须是名词（`Customer`, `AccountStatus`），函数名必须是动词（`getUser`, `calculateTotal`）。
+
+### 2. 函数设计
+- **单一职责**: 一个函数只做一件事。如果一个函数超过 **20 行**（或屏幕一屏），请尝试拆分。
+- **无副作用**: 尽量避免函数产生副作用（如修改全局变量、隐式修改传入的对象）。如果必须修改，请在命名中体现（如 `updateUser` vs `getUser`）。
+- **命令与查询分离**: 函数要么执行操作（Command），要么返回数据（Query），不要同时做这两件事。
+
+### 3. 注释与文档
+- **代码自解释**: 优先通过重构代码（如提取函数、优化变量名）来消除对注释的需求，而不是写注释来解释烂代码。
+- **禁止废话**: 不要写“获取用户 ID”这种显而易见的注释。
+- **解释“为什么”**: 只有在解释复杂的业务逻辑、权衡取舍或解决特定 Bug 时才写注释（解释“为什么这么做”，而不是“做了什么”）。
+- **TODO 规范**: 遗留代码必须标记 `// TODO: [描述] - [作者/日期]`。
+
+### 4. 错误处理
+- **使用异常而非返回码**: 不要通过返回 `null` 或 `-1` 来表示错误。使用 `try/catch` 或 Result/Either 模式。
+- **不要返回/传递 null**: 除非是特定语言特性（如 Optional），否则不要返回 `null`。返回空列表 `[]` 或空对象。
+- **精确捕获**: 不要捕获通用的 `Exception`，只捕获你明确知道如何处理的特定错误类型。
+
+### 5. 结构与格式
+- **垂直格式**: 相关的代码行应该放在一起（变量声明靠近使用处）。
+- **依赖倒置**: 高层模块不应依赖低层模块，二者都应依赖抽象（接口）。
+- **DRY**: 杜绝重复代码。如果你发现自己在复制粘贴代码，请立即提取为公共函数或组件。
+
+---

backend/api/domains/episode/routes.py

@@ -2,6 +2,7 @@
 AKA-Sim 后端 - Episode/数据采集域 API
 """
 
+import json
 import logging
 from pathlib import Path
 
@@ -27,10 +28,13 @@ async def list_dataset_dirs(user_id: str):
 
     datasets = []
     for item in user_dataset_path.iterdir():
-        if item.is_dir() and (item / "data").exists() or (item / "meta").exists():
-            datasets.append(item.name)
+        if item.is_dir() and ((item / "data").exists() or (item / "meta").exists()):
+            info_path = item / "meta" / "info.json"
+            sort_time = info_path.stat().st_mtime if info_path.exists() else item.stat().st_mtime
+            datasets.append((item.name, sort_time))
 
-    return {"datasets": sorted(datasets)}
+    datasets.sort(key=lambda entry: entry[1], reverse=True)
+    return {"datasets": [name for name, _ in datasets]}
 
 
 @router.get("/models")
@@ -43,15 +47,52 @@ async def list_models(user_id: str, dataset_name: str = "default"):
     if not train_path.exists():
         return {"models": []}
 
+    if dataset_name:
+        dataset_model_path = train_path / dataset_name
+        has_model = (dataset_model_path / "model.pt").exists() or (dataset_model_path / "final_model.pt").exists()
+        if dataset_model_path.is_dir() and has_model:
+            return {"models": [dataset_name]}
+        return {"models": []}
+
     # 返回子文件夹名称
     models = []
     for item in train_path.iterdir():
-        if item.is_dir():
+        has_model = (item / "model.pt").exists() or (item / "final_model.pt").exists()
+        if item.is_dir() and has_model:
             models.append(item.name)
 
     return {"models": sorted(models)}
 
 
+@router.get("/info")
+async def get_dataset_info(user_id: str, dataset_name: str):
+    """读取数据集元信息。"""
+    if not user_id:
+        raise HTTPException(status_code=400, detail="user_id is required")
+    if not dataset_name or ".." in dataset_name or "/" in dataset_name or "\\" in dataset_name:
+        raise HTTPException(status_code=400, detail="invalid dataset_name")
+
+    project_root = Path(__file__).resolve().parents[4]
+    info_path = project_root / "output" / "dataset" / user_id / dataset_name / "meta" / "info.json"
+    if not info_path.exists():
+        return {
+            "dataset_name": dataset_name,
+            "total_frames": 0,
+            "total_episodes": 0,
+            "exists": False,
+        }
+
+    with open(info_path, "r") as f:
+        info = json.load(f)
+
+    return {
+        "dataset_name": dataset_name,
+        "total_frames": int(info.get("total_frames", 0) or 0),
+        "total_episodes": int(info.get("total_episodes", 0) or 0),
+        "exists": True,
+    }
+
+
 @router.post("/collect")
 async def collect_image(payload: CollectImagePayload):
     """将前端直接采集到的图像写入当前 episode。"""

backend/api/domains/training/routes.py

@@ -22,6 +22,13 @@ def set_sio_server(sio):
     _sio_server = sio
 
 
+async def _run_training_task(**kwargs):
+    try:
+        await training.train_model(_sio_server, **kwargs)
+    except Exception:
+        logger.exception("后台训练任务异常退出")
+
+
 @router.post("")
 async def start_training(request: TrainRequest, user_id: str = Query(...)):
     """启动训练。"""
@@ -59,21 +66,25 @@ async def start_training(request: TrainRequest, user_id: str = Query(...)):
         if request.resume_from:
             resume_path = str(project_root / request.resume_from)
 
-        logger.info(f"收到开始训练请求: user={user_id}, epochs={request.epochs}, batch_size={request.batch_size}, lr={request.lr}, resume_from={resume_path}")
-
-        asyncio.create_task(
-            training.train_model(
-                _sio_server,
-                user_id=user_id,
-                data_dir=str(data_path),
-                output_dir=str(output_path),
-                epochs=request.epochs,
-                batch_size=request.batch_size,
-                lr=request.lr,
-                resume_from=resume_path,
-            )
+        logger.info(
+            "收到开始训练请求: user=%s, epochs=%s, batch_size=%s, lr=%s, resume_from=%s",
+            user_id,
+            request.epochs,
+            request.batch_size,
+            request.lr,
+            resume_path,
         )
 
+        asyncio.create_task(_run_training_task(
+            user_id=user_id,
+            data_dir=str(data_path),
+            output_dir=str(output_path),
+            epochs=request.epochs,
+            batch_size=request.batch_size,
+            lr=request.lr,
+            resume_from=resume_path,
+        ))
+
         resume_msg = f"，从模型继续: {request.resume_from}" if request.resume_from else ""
         return {
             "success": True,

backend/config.py

@@ -10,7 +10,7 @@ class Config:
 
     # 服务器配置
     HOST = os.getenv("HOST", "0.0.0.0")
-    PORT = int(os.getenv("PORT", "8000"))
+    PORT = int(os.getenv("PORT", "8001"))
 
     # 模型配置
     MODEL_PATH = os.getenv("MODEL_PATH", None)

backend/run_act_checks.py

@@ -6,7 +6,7 @@
 
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
-PYTEST_CMD = ["python3", "-m", "pytest", "tests/act", "-q"]
+PYTEST_CMD = [sys.executable, "-m", "pytest", "tests/act", "-q"]
 
 
 def main() -> int:

backend/services/episode/exporter.py

@@ -176,17 +176,23 @@ def _build_stats_entry(self, key: str, values: np.ndarray) -> Dict[str, Any]:
         batch_count = int(values.shape[0])
         batch_min = values.min(axis=0).astype(np.float64)
         batch_max = values.max(axis=0).astype(np.float64)
+        batch_sum = values.sum(axis=0).astype(np.float64)
+        batch_sumsq = np.square(values).sum(axis=0).astype(np.float64)
 
         existing = self._stats_accumulators.get(key)
         if existing is None:
             accum = {
                 "count": batch_count,
+                "sum": batch_sum,
+                "sumsq": batch_sumsq,
                 "min": batch_min,
                 "max": batch_max,
             }
         else:
             accum = {
                 "count": existing["count"] + batch_count,
+                "sum": existing.get("sum", np.zeros_like(batch_sum)) + batch_sum,
+                "sumsq": existing.get("sumsq", np.zeros_like(batch_sumsq)) + batch_sumsq,
                 "min": np.minimum(existing["min"], batch_min),
                 "max": np.maximum(existing["max"], batch_max),
             }
@@ -205,15 +211,52 @@ def _build_stats_entry(self, key: str, values: np.ndarray) -> Dict[str, Any]:
         all_values = self._stats_accumulators[values_key]
         q01 = np.percentile(all_values, 1, axis=0).astype(np.float64)
         q99 = np.percentile(all_values, 99, axis=0).astype(np.float64)
+        mean = accum["sum"] / max(accum["count"], 1)
+        variance = accum["sumsq"] / max(accum["count"], 1) - np.square(mean)
+        std = np.sqrt(np.maximum(variance, 0.0))
 
         return {
             "count": int(accum["count"]),
+            "sum": accum["sum"].tolist(),
+            "sumsq": accum["sumsq"].tolist(),
             "min": accum["min"].tolist(),
             "max": accum["max"].tolist(),
+            "mean": mean.tolist(),
+            "std": std.tolist(),
             "q01": q01.tolist(),
             "q99": q99.tolist(),
         }
 
+    def _read_saved_numeric_columns(self) -> tuple[np.ndarray, np.ndarray] | None:
+        """从已写入 parquet 的全量数据重建状态和动作数组。"""
+        if not self.data_dir.exists():
+            return None
+
+        states = []
+        actions = []
+        for parquet_file in sorted(self.data_dir.glob("chunk-*/file-*.parquet")):
+            df = pd.read_parquet(parquet_file, columns=["observation.state", "action"])
+            states.extend(np.asarray(value, dtype=np.float32) for value in df["observation.state"])
+            actions.extend(np.asarray(value, dtype=np.float32) for value in df["action"])
+
+        if not states or not actions:
+            return None
+
+        return np.stack(states), np.stack(actions)
+
+    def _rebuild_stats_from_saved_data(self):
+        """确保 stats.json 反映整个数据集，而不是最近一次导出的 episode。"""
+        saved_data = self._read_saved_numeric_columns()
+        if saved_data is None:
+            return
+
+        states_array, actions_array = saved_data
+        self._stats_accumulators = {}
+        self._stats = {
+            "observation.state": self._build_stats_entry("observation.state", states_array),
+            "action": self._build_stats_entry("action", actions_array),
+        }
+
     def _ensure_dirs(self):
         """确保目录结构存在"""
         self.data_dir.mkdir(parents=True, exist_ok=True)
@@ -415,6 +458,7 @@ def _save_all_metadata(self):
         """保存所有元数据文件"""
         # 保存 info.json
         self._ensure_dirs()
+        self._rebuild_stats_from_saved_data()
         info = self.info
         with open(self.meta_dir / "info.json", "w") as f:
             json.dump(info, f, indent=2)

backend/services/training/dataset.py

@@ -72,4 +72,4 @@ def __getitem__(self, idx):
                 "state": state,
             },
             "action": action,
-        }
+        }