From 85bf68f505283ddd5ce5be065b8b45d4b060ec7e Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Fri, 14 Aug 2026 18:26:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(actpred):=20=E5=9B=BA=E5=AE=9A=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E5=BD=92=E4=B8=80=E5=8C=96=20+=20=E9=80=90=E5=88=97?= =?UTF-8?q?=E5=8A=A0=E6=9D=83=E6=8D=9F=E5=A4=B1=EF=BC=8C=E9=99=84=20pybot*?= =?UTF-8?q?actpred=20=E5=AE=A1=E8=AE=A1=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 action_norm.py 固定物理范围 min-max 归一化，无 stats 文件零状态部署；train.py 引入 _weighted_mse 逐列加权（舵机 1.0/spk 0.2/保留位 0.0）。审计修复: P0-1 pybot 读 normalize_mode, R2-1 NaN 阻断 best, P2-a 后端大小写不敏感, P1-3 utf-8-sig BOM, R1-1 data_dir 覆盖, R2-7 _run.sh 报错, R2-3 空 val 存 best.pt, P2-a2 误报警, R1-2 合并 parse_action_history, P2-b 合并 numpy import。评估修正 oob 物理界+error_growth 按 episode 中位数。推理部署全链支持 fixed_range。 --- actpred/README.md | 26 +- actpred/_run.sh | 6 +- actpred/action_norm.py | 62 ++++ actpred/config.py | 45 ++- actpred/dataset.py | 15 +- actpred/evaluate.py | 44 +-- actpred/export_deploy.py | 20 +- actpred/gen_calibration.py | 9 +- actpred/hailo_infer.py | 52 ++-- actpred/mt_parser.py | 2 +- actpred/onnx_infer.py | 39 ++- actpred/predict.py | 34 +-- actpred/pretrain_mt.py | 87 ++++-- actpred/test_actpred.py | 282 +++++++++++++++++- actpred/train.py | 89 ++++-- actpred/utils.py | 16 + .../pybot_actpred_AUDIT_20260814_v2.md | 191 ++++++++++++ pybot/pybot/main.py | 13 +- pybot/pybot/robot.py | 9 +- 19 files changed, 874 insertions(+), 167 deletions(-) create mode 100644 actpred/action_norm.py create mode 100644 actpred/utils.py create mode 100644 docs/reviews/pybot_actpred_AUDIT_20260814_v2.md diff --git a/actpred/README.md b/actpred/README.md index 8efaf97..53dec6b 100644 --- a/actpred/README.md +++ b/actpred/README.md @@ -44,6 +44,13 @@ bash _run.sh train --data-dir ./data --epochs 100 --action-dim 18 \ # R2 姿态反馈：--imu-dim 8 启用独立 IMU token（数据需含 imu.npy，见"数据格式"） bash _run.sh train --data-dir ./data --epochs 100 --action-dim 18 --imu-dim 8 +# 归一化+逐列加权：固定物理范围归一化（免 stats 文件）+ 舵机列损失权重上浮 +bash _run.sh train --data-dir ./data --epochs 100 --action-dim 18 \ + --normalize-actions --normalize-mode fixed_range +# 自定义列权重（逗号分隔 action_dim 个浮点；默认舵机1.0/spk0.2/保留位0.0） +bash _run.sh train --data-dir ./data --epochs 100 --normalize-actions \ + --normalize-mode fixed_range --column-weights "1,1,1,1,1,1,1,1,1,1,0.2,0.2,0,0,0,0,0,0" + # 推理（--images 按时间顺序旧→新传入 frame_len 个图像路径，冷启动时可重复传同一帧； # --action-history 格式：冒号分隔各步，每步 action_dim 个值，共 history_len 步） bash _run.sh --model checkpoints/best_inference.pt \ @@ -161,6 +168,8 @@ recorder.flush() | `--lr` | 0.001 | 学习率 | | `--batch-size` | 4 | 批次大小（默认分辨率 640×480；更高分辨率需调小，小分辨率可调大） | | `--normalize-actions` | False | 动作归一化（`--no-normalize-actions` 关闭） | +| `--normalize-mode` | zscore | 归一化方式：`zscore`=数据统计（依赖 `action_stats.npy`）；`fixed_range`=固定物理范围 min-max→[0,1]（常量表见 `action_norm.py`，免 stats 文件，越界 clamp 防御） | +| `--column-weights` | 舵机1.0/spk0.2/保留位0.0 | 逐列损失权重（逗号分隔 action_dim 个浮点；仅影响训练损失，推理端无感知） | | `--dropout` | 0.1 | Dropout 率 | | `--weight-decay` | 0.0001 | 权重衰减 | | `--val-split` | 0.15 | 验证集比例 | @@ -185,21 +194,24 @@ recorder.flush() | 指标 | 含义 | 判读 | |------|------|------| | `overall_mse` | 全部滚动步的平均 MSE | 越低越好 | -| `error_growth` | 末段(后25%)误差 / 首段(前25%)误差 | >1 表示误差随步数发散（闭环不稳） | +| `error_growth` | 末段(后25%)误差 / 首段(前25%)误差（按 episode 分别计算后取中位数） | >1 表示误差随步数发散（闭环不稳） | | `error_curve` | 误差随步数的增长曲线 | 看发散拐点 | | `smoothness_ratio` | 预测平滑度 / 真值平滑度 | <1 表示比真值更平滑（残差模式的预期收益） | -| `oob_rate` | 预测落在训练数据经验范围外的比例 | 越高越危险（舵机越界风险） | +| `oob_rate` | 预测落在物理范围外的比例 | 越高越危险（舵机越界风险） | -物理可行性用数据分布的经验界（min/max + 10% margin）而非硬编码范围——18 列混含 -舵机角度/频率/延时/预留，无统一硬限位。`--compare` 用于 A/B 对比（如 R1 开/关）。 +越界判定用固定物理界（`action_norm.py` 常量表：舵机 [0,180]、spk.F [0,5000] 等）， +"越物理界"比"越经验界"更符合安全语义，且恒值列不会使经验界退化。`--compare` 用于 A/B 对比（如 R1 开/关）。 滚动复现闭环口径：模型带 imu 输入时按 episode 的 `imu.npy` 真值姿态逐步喂入 （缺失则零向量，与 pybot 闭环无 `@MPU6050_BIND` 块时的行为一致）。 ## 部署注意 -- 启用 `--normalize-actions` 训练的模型，其 `action_stats.npy` 保存在 checkpoint 同目录； - 部署时**必须将 stats 文件与 checkpoint 一起拷贝**（推理端优先按 checkpoint 内记录的路径查找， - 找不到时回退到 checkpoint 同目录的 `action_stats.npy`）。 +- 启用 `--normalize-actions` 且 `--normalize-mode zscore`（默认）训练的模型，其 `action_stats.npy` + 保存在 checkpoint 同目录；部署时**必须将 stats 文件与 checkpoint 一起拷贝**（推理端优先按 + checkpoint 内记录的路径查找，找不到时回退到 checkpoint 同目录的 `action_stats.npy`）。 +- `--normalize-mode fixed_range` 用固定物理范围 min-max→[0,1]（常量表见 `action_norm.py`）， + **免 stats 文件**，消除 stats 路径类 bug 的整类风险；推理端（predict/onnx_infer/hailo_infer） + 从 checkpoint/ONNX 元数据自动识别归一化方式，无需额外参数。 - `actions.npy` 的列数必须与 `--action-dim` 一致；图像与动作帧数不一致时训练会打印 WARNING 并按较少者使用。 ## 编程 API diff --git a/actpred/_run.sh b/actpred/_run.sh index cd99923..a08191d 100755 --- a/actpred/_run.sh +++ b/actpred/_run.sh @@ -11,6 +11,10 @@ elif [ "${1:-}" = "pretrain" ]; then elif [ "${1:-}" = "export" ]; then shift exec python3 export_onnx.py "$@" -else +elif [ "${1:-}" = "predict" ]; then + shift exec python3 predict.py "$@" +else + echo "未知子命令: ${1:-}（可用: train / pretrain / export / predict）" >&2 + exit 1 fi diff --git a/actpred/action_norm.py b/actpred/action_norm.py new file mode 100644 index 0000000..96ae0a9 --- /dev/null +++ b/actpred/action_norm.py @@ -0,0 +1,62 @@ +"""动作向量固定物理范围归一化 —— 训练（dataset/pretrain_mt）与推理 +（predict / onnx_infer / hailo_infer / gen_calibration / export_deploy）共用 + +18 列动作向量的物理量纲（pybot 侧契约）： + 0-9 舵机角度 [0, 180]°（pybot/hd/servo.py MAX_ANGLE=180.0 校准后全行程） + 10 spk.F 扬声器频率 [0, 5000] Hz（@Speaker max_f=5000） + 11 spk.delay 延时 [0, 2000] ms（脚本实用上限，std.mt 用 10-500） + 12-17 保留位（恒 0，hi-lo=1.0 使归一化后仍为 0；损失权重 0 不参与优化） + +min-max 映射到 [0,1]：x_norm = clip((x - lo) / (hi - lo), 0, 1)。 +与 z-score 的区别：固定物理范围不依赖 action_stats.npy 数据统计文件， +部署链零状态；clamp 防御真实数据越界（越界值截断到 [0,1]，不扩散）。 + +仅依赖 numpy，保证 hailo_infer.py（HAT 部署环境无 torch）可直接复用。 +""" + +import numpy as np + +# 各列物理下界（当前全 0；保留数组以便未来引入非零下界列） +ACTION_LO = np.zeros(18, dtype=np.float32) +# 各列物理上界（18 列标准布局；action_dim 不同用 action_bounds() 适配） +ACTION_HI = np.array( + [180.0] * 10 + # 0-9 舵机角 + [5000.0, 2000.0] + # 10 spk.F，11 spk.delay + [1.0] * 6, # 12-17 保留位（恒 0，span=1.0 保持归一化后为 0） + dtype=np.float32) + + +def action_bounds(action_dim: int = 18): + """(lo, hi) 物理界：action_dim ≤ 18 取前缀；> 18 的超出列按 [0, 1] 补足""" + if action_dim <= len(ACTION_LO): + return ACTION_LO[:action_dim], ACTION_HI[:action_dim] + pad = action_dim - len(ACTION_LO) + return (np.concatenate([ACTION_LO, np.zeros(pad, dtype=np.float32)]), + np.concatenate([ACTION_HI, np.ones(pad, dtype=np.float32)])) + + +def normalize_action(action, action_dim: int = None) -> np.ndarray: + """原始动作向量 → [0,1] 归一化向量（clamp 防御越界值） + + action: (..., action_dim) 任意前导维；action_dim 缺省取末维长度。 + """ + a = np.asarray(action, dtype=np.float32) + lo, hi = action_bounds(action_dim or a.shape[-1]) + return np.clip((a - lo) / (hi - lo), 0.0, 1.0) + + +def denormalize_action(norm_action, action_dim: int = None) -> np.ndarray: + """[0,1] 归一化向量 → 原始动作向量（不 clamp，保留模型外推能力）""" + n = np.asarray(norm_action, dtype=np.float32) + lo, hi = action_bounds(action_dim or n.shape[-1]) + return n * (hi - lo) + lo + + +def default_column_weights(action_dim: int = 18) -> np.ndarray: + """逐列损失权重默认表：舵机 1.0（安全关键）、spk 0.2（保留梯度）、 + 保留位 0.0（恒 0 纯噪声，参与只会稀释有效梯度）""" + w = np.zeros(action_dim, dtype=np.float32) + w[:min(10, action_dim)] = 1.0 + if action_dim > 10: + w[10:min(12, action_dim)] = 0.2 + return w diff --git a/actpred/config.py b/actpred/config.py index f9dc42c..928e2c9 100644 --- a/actpred/config.py +++ b/actpred/config.py @@ -1,6 +1,6 @@ import argparse from dataclasses import dataclass, asdict, field -from typing import Optional +from typing import List, Optional @dataclass @@ -13,6 +13,15 @@ class DataConfig: data_dir: str = "./data" normalize_actions: bool = True action_stats_path: Optional[str] = None + # 归一化方式（仅 normalize_actions=True 时生效）： + # "zscore" 数据统计 z-score（现状，依赖 action_stats.npy） + # "fixed_range" 固定物理范围 min-max → [0,1]（免 stats 文件，零状态部署， + # 常量表见 action_norm.py；越界值 clamp 防御） + normalize_mode: str = "zscore" + # 逐列损失权重（仅训练损失用，推理端无感知）：None → action_norm + # .default_column_weights(action_dim)（舵机 1.0 / spk 0.2 / 保留位 0.0）。 + # 显式指定时长度须等于 action_dim。 + column_weights: Optional[List[float]] = None # R2 方案一：MPU6050 姿态作为独立 IMU token 注入（序列 F+H+2）。 # 0 = 关闭（默认，序列保持 F+H+1，旧 checkpoint 完全兼容）； # 8×N = N 个传感器的 roll,pitch,ax,ay,az,gx,gy,gz 按 pybot 脚本 @@ -84,6 +93,13 @@ def build_argparser(description: str = "") -> argparse.ArgumentParser: parser.add_argument("--data-dir", type=str, default=None) parser.add_argument("--normalize-actions", action=argparse.BooleanOptionalAction, default=None) parser.add_argument("--action-stats-path", type=str, default=None) + parser.add_argument("--normalize-mode", type=str, default=None, + choices=["zscore", "fixed_range"], + help="动作归一化方式：zscore=数据统计（依赖 action_stats.npy）；" + "fixed_range=固定物理范围 min-max→[0,1]（免 stats 文件）") + parser.add_argument("--column-weights", type=str, default=None, + help="逐列损失权重（逗号分隔 action_dim 个浮点），" + "缺省用默认表（舵机1.0/spk0.2/保留位0.0）；仅影响训练损失") parser.add_argument("--imu-dim", type=int, default=None, help="R2 方案一：MPU6050 姿态维度（0=关闭，默认；" "8×N=N 个传感器拼接（≤4 个），序列变 F+H+2）") @@ -156,6 +172,24 @@ def validate_config_values(data_cfg: DataConfig, model_cfg: ModelConfig, f"attn_dim={model_cfg.attn_dim} 必须能被 attn_heads=" f"{model_cfg.attn_heads} 整除") + if data_cfg.normalize_mode not in ("zscore", "fixed_range"): + raise ValueError( + f"normalize_mode={data_cfg.normalize_mode!r} 非法，" + f"应为 'zscore' 或 'fixed_range'") + + if data_cfg.column_weights is not None: + w = data_cfg.column_weights + if len(w) != data_cfg.action_dim: + raise ValueError( + f"column_weights 长度 {len(w)} 应等于 action_dim=" + f"{data_cfg.action_dim}") + if any((not isinstance(v, (int, float))) or isinstance(v, bool) + or not (0.0 <= v <= 1e6) for v in w): + raise ValueError( + f"column_weights 元素应为 [0, 1e6] 内的数值，实际: {w}") + if not any(v > 0 for v in w): + raise ValueError("column_weights 全 0：损失恒 0，训练无梯度") + if train_cfg is not None: t_checks = [ ("batch_size", train_cfg.batch_size, 1, 65536, int), @@ -194,6 +228,11 @@ def warn_init_from_mismatch(init_cfg: dict, current: Config): print(f"WARNING: init-from checkpoint 的 normalize_actions=" f"{init_norm} 与当前运行 {current.data.normalize_actions} " f"不一致，输入分布不匹配会损失预训练收益（见 README 预训练说明）") + init_mode = init_data.get("normalize_mode", "zscore") # 旧 checkpoint 无此键 → zscore + if init_norm and current.data.normalize_actions and init_mode != current.data.normalize_mode: + print(f"WARNING: init-from checkpoint 的 normalize_mode={init_mode} " + f"与当前运行 {current.data.normalize_mode} 不一致，" + f"归一化空间不同会损失预训练收益") init_dim = init_data.get("action_dim") if init_dim is not None and init_dim != current.data.action_dim: print(f"WARNING: init-from checkpoint 的 action_dim={init_dim} " @@ -249,6 +288,8 @@ def apply_overrides(cfg: Config, args: argparse.Namespace) -> Config: "data_dir": ("data", "data_dir"), "normalize_actions": ("data", "normalize_actions"), "action_stats_path": ("data", "action_stats_path"), + "normalize_mode": ("data", "normalize_mode"), + "column_weights": ("data", "column_weights"), "imu_dim": ("data", "imu_dim"), "backbone": ("model", "cnn_backbone"), "cnn_feature_dim": ("model", "cnn_feature_dim"), @@ -276,5 +317,7 @@ def apply_overrides(cfg: Config, args: argparse.Namespace) -> Config: for arg_name, (section, field) in mapping.items(): val = args_dict.get(arg_name) if val is not None: + if field == "column_weights" and isinstance(val, str): + val = [float(v) for v in val.split(",")] setattr(getattr(cfg, section), field, val) return cfg diff --git a/actpred/dataset.py b/actpred/dataset.py index ac06bc1..f812c52 100644 --- a/actpred/dataset.py +++ b/actpred/dataset.py @@ -11,6 +11,7 @@ from torchvision import transforms from torchvision.transforms import functional as TF from imu_norm import imu_scales +from action_norm import normalize_action def _natural_key(p): @@ -25,9 +26,11 @@ class ActionDataset(Dataset): action_stats_path: str = None, episode_indices: list = None, frame_len: int = 3, augment: bool = False, predict_delta: bool = False, rollout_len: int = 1, - imu_dim: int = 0): + imu_dim: int = 0, normalize_mode: str = "zscore"): self.history_len = history_len self.action_dim = action_dim + # 固定物理范围归一化（免 stats 文件）；仅 normalize_actions=True 生效 + self._fixed_range = normalize_actions and normalize_mode == "fixed_range" # R2 方案一：>0 时样本带 "imu" 键（归一化姿态向量，缺失 episode 补零） self.imu_dim = imu_dim self.image_width = image_width @@ -80,7 +83,7 @@ class ActionDataset(Dataset): self._action_mean = None self._action_std = None - if normalize_actions: + if normalize_actions and not self._fixed_range: if action_stats_path: if not os.path.exists(action_stats_path): raise FileNotFoundError( @@ -110,7 +113,13 @@ class ActionDataset(Dataset): return len(self.samples) def _norm(self, a): - """z-score 归一化（numpy → torch 前在 numpy 空间做，保持 float32）""" + """动作归一化（numpy → torch 前在 numpy 空间做，保持 float32） + + fixed_range：固定物理范围 min-max → [0,1]（clamp 防御，免 stats）； + 否则 z-score（_action_mean 为 None 时直通原始值）。 + """ + if self._fixed_range: + return normalize_action(a, self.action_dim) if self._action_mean is None: return a return (a - self._action_mean) / self._action_std diff --git a/actpred/evaluate.py b/actpred/evaluate.py index 997f0e2..8f2137a 100644 --- a/actpred/evaluate.py +++ b/actpred/evaluate.py @@ -51,20 +51,15 @@ def _load_episode(data_dir, ep_name): return image_files[:n], actions[:n], imus -def _action_bounds(data_dir, action_dim, margin_ratio=0.1): - """从全数据集统计动作经验范围（min/max 加 margin），作为物理可行性判据。 +def _action_bounds(action_dim): + """固定物理界作为越界判据（与 action_norm.py 常量表共用） - actpred 侧无硬限位定义（18 列混含舵机角度/频率/延时/预留），硬编码范围不可靠； - 用数据分布的经验界更通用。margin 默认 10% 容忍合理外推。 + 旧版用数据经验界（min/max+margin），恒值列 span≈0 时任何非零预测都 + 算越界（unnormalized 模型 oob_rate 曾恒 100%）。"越物理界"比"越经验界" + 更符合安全语义：舵机 [0,180]、spk.F [0,5000]、spk.delay [0,2000]。 """ - mins = np.full(action_dim, np.inf, dtype=np.float32) - maxs = np.full(action_dim, -np.inf, dtype=np.float32) - for ep_dir in ActionDataset._valid_episode_dirs(data_dir): - a = np.load(ep_dir / "actions.npy").astype(np.float32) - mins = np.minimum(mins, a.min(axis=0)) - maxs = np.maximum(maxs, a.max(axis=0)) - span = np.maximum(maxs - mins, 1e-6) - return mins - margin_ratio * span, maxs + margin_ratio * span + from action_norm import action_bounds + return action_bounds(action_dim) @np.errstate(all="ignore") @@ -133,7 +128,12 @@ def _smoothness(seq): def summarize(all_steps, lo, hi): - """汇总多个 episode 的滚动记录为报告指标""" + """汇总多个 episode 的滚动记录为报告指标 + + error_growth 改为按 episode 归一化后取中位数： + 每 episode 内部算 (末25% MSE / 首25% MSE) → 中位数聚合。 + 避免短 episode 尖峰被长 episode 均值淹没，也更鲁棒于长度差异。 + """ if not all_steps: raise ValueError("无可评估的滚动步（episode 太短或 horizon 过小）") flat = [s for steps in all_steps for s in steps] @@ -147,6 +147,18 @@ def summarize(all_steps, lo, hi): if vals: curve.append(float(np.mean(vals))) + # 逐 episode 归一化 error_growth：每 episode 自身首末段比值 → 中位数 + growths = [] + for steps in all_steps: + n = len(steps) + if n < 2: + continue + m = np.array([s["mse"] for s in steps], dtype=np.float32) + front = np.mean(m[:max(1, n // 4)]) + back = np.mean(m[max(0, n * 3 // 4):]) + growths.append(back / max(front, 1e-12)) + error_growth = float(np.median(growths)) if growths else 1.0 + preds = np.array([s["pred"] for s in flat], dtype=np.float32) trues = np.array([s["true"] for s in flat], dtype=np.float32) oob = [s.get("oob", False) for s in flat] @@ -156,9 +168,7 @@ def summarize(all_steps, lo, hi): "n_steps": len(flat), "overall_mse": float(np.mean(mses)), # 末段（后 25% 步）相对首段（前 25%）的误差比，>1 表示误差随步数发散 - "error_growth": float(np.mean(mses[int(len(mses) * 0.75):]) / - max(np.mean(mses[:max(1, int(len(mses) * 0.25))]), - 1e-12)), + "error_growth": error_growth, "error_curve": curve, "pred_smoothness": _smoothness(preds), "true_smoothness": _smoothness(trues), @@ -173,7 +183,7 @@ def evaluate(checkpoint, data_dir, episodes=None, horizon=None, device=None): ep_names = _episode_ids(data_dir) if episodes is not None: ep_names = ep_names[-episodes:] # 取尾部 episode 作为留出集 - lo, hi = _action_bounds(data_dir, predictor.action_dim) + lo, hi = _action_bounds(predictor.action_dim) all_steps = [] for name in ep_names: image_files, actions, imus = _load_episode(data_dir, name) diff --git a/actpred/export_deploy.py b/actpred/export_deploy.py index b2b29dc..1f1b00c 100644 --- a/actpred/export_deploy.py +++ b/actpred/export_deploy.py @@ -158,7 +158,8 @@ class DeployWrapper(nn.Module): def check_token_ranges(wrapper: "DeployWrapper", frame_len: int, height: int, width: int, history_len: int, action_dim: int, pred_token: torch.Tensor, max_abs: float = MAX_TOKEN_ABS, - action_stats_path: str = None): + action_stats_path: str = None, + normalize_mode: str = "zscore"): """导出前检查 token 量级：幅度过大时 uint8 量化会把小信号压成 0 预训练用全零图像占位 → BN running_var 停在初始小值 → 推理时每级 @@ -166,21 +167,24 @@ def check_token_ranges(wrapper: "DeployWrapper", frame_len: int, height: int, 幅度分配共享量化 scale（实测 scale=1965.92，pred_token/act_tokens 全部量化成 0），HEF 输出对任何输入恒定。必须在编译前拦截。 - 归一化模型（normalize_actions）的 HEF 输入是 z-score 后的历史， - 检查输入须与推理端一致。 + 归一化模型（normalize_actions）的 HEF 输入是归一化后的历史， + 检查输入须与推理端一致（zscore → z 空间采样；fixed_range → [0,1]）。 """ rng = np.random.default_rng(42) raw = rng.integers(0, 256, size=(height, width, 3), dtype=np.uint8) norm = (raw.astype(np.float32) / 255.0 - IMAGENET_MEAN) / IMAGENET_STD frames = [torch.tensor(norm[None]) for _ in range(frame_len)] history = rng.uniform(0.0, 500.0, size=(1, 1, history_len, action_dim)) - if action_stats_path is not None: + if normalize_mode == "fixed_range": + # fixed_range 归一化空间即 [0,1]，直接在归一化空间采样 + history = rng.uniform(0.0, 1.0, size=(1, 1, history_len, action_dim)) + elif action_stats_path is not None: stats = np.load(action_stats_path) if stats.shape != (2, action_dim): raise ValueError( f"action stats 形状应为 (2, {action_dim})，" f"实际为 {stats.shape}: {action_stats_path}") - # 归一化模型直接在 z 空间采样：raw 均匀采样会在零方差列上 + # z-score 归一化模型直接在 z 空间采样：raw 均匀采样会在零方差列上 # 产生天文数字 z 值（实际推理中这些列恒为 0，z 恒为 0） history = rng.normal(0.0, 1.5, size=(1, 1, history_len, action_dim)) history = torch.tensor(history, dtype=torch.float32) @@ -226,8 +230,9 @@ def main(): h, w = cfg.data.image_height, cfg.data.image_width f = cfg.data.frame_len stats_path = None - if cfg.data.normalize_actions: + if cfg.data.normalize_actions and cfg.data.normalize_mode != "fixed_range": # 与 predict.py 语义一致：相对路径按 checkpoint 目录解析 + # （fixed_range 模式免 stats 文件，跳过） stats_path = cfg.data.action_stats_path if not stats_path: stats_path = os.path.join( @@ -241,7 +246,8 @@ def main(): f"模型训练时启用了动作归一化，但未找到统计文件: {stats_path}") check_token_ranges(wrapper, f, h, w, cfg.data.history_len, cfg.data.action_dim, model.pred_token.data.unsqueeze(1), - action_stats_path=stats_path) + action_stats_path=stats_path, + normalize_mode=cfg.data.normalize_mode) dummy_frames = [torch.randn(1, h, w, 3) for _ in range(f)] dummy_history = torch.randn(1, 1, cfg.data.history_len, cfg.data.action_dim) dummy_pred = model.pred_token.data.unsqueeze(1).clone() # (1, 1, 1, d) diff --git a/actpred/gen_calibration.py b/actpred/gen_calibration.py index d7b1893..3b5ef6d 100644 --- a/actpred/gen_calibration.py +++ b/actpred/gen_calibration.py @@ -101,8 +101,13 @@ def build_deploy_set(args): history = build_action_history_samples( args.mt_dir, args.samples, cfg.history_len, cfg.action_dim, rng) history = history[:, None, :, :] # (N, 1, H_len, D) - if cfg.normalize_actions: - # 归一化模型的 HEF 输入是 z-score 历史（hailo_infer --stats 同语义）， + if cfg.normalize_actions and cfg.normalize_mode == "fixed_range": + # 固定物理范围归一化：校准数据与推理端同一变换（免 stats 文件） + from action_norm import normalize_action + history = normalize_action(history, cfg.action_dim) + print(" action_history: 已 fixed_range 归一化（固定物理范围）") + elif cfg.normalize_actions: + # z-score 归一化模型的 HEF 输入是 z-score 历史（hailo_infer --stats 同语义）， # 校准数据必须走同一变换，否则量化范围与推理分布不匹配 stats_path = cfg.action_stats_path if not stats_path: diff --git a/actpred/hailo_infer.py b/actpred/hailo_infer.py index 7636407..ccc39dc 100644 --- a/actpred/hailo_infer.py +++ b/actpred/hailo_infer.py @@ -36,25 +36,12 @@ import numpy as np from PIL import Image from imu_norm import normalize_imu +from action_norm import normalize_action, denormalize_action +from utils import parse_action_history IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) -# 与 predict.parse_action_history 相同逻辑；此处内联以保持 -# 本模块零 torch 依赖（HAT 部署环境只装 hailo_platform + numpy + PIL） - - -def parse_action_history(s: str, action_dim: int) -> np.ndarray: - steps = s.split(":") - history = [] - for step in steps: - values = [float(v) for v in step.split(",")] - if len(values) != action_dim: - raise ValueError( - f"Expected {action_dim} values per step, got {len(values)}") - history.append(values) - return np.array(history, dtype=np.float32) - def _input_layer_index(name: str) -> int: m = re.search(r"input_layer(\d+)$", name) @@ -73,17 +60,22 @@ class HailoPredictor: """ def __init__(self, hef_path: str, stats_path: str = None, vdevice=None, - predict_delta: bool = False, pred_token=None): + predict_delta: bool = False, pred_token=None, + normalize_mode: str = "zscore"): """HEF 推理器。 predict_delta: 模型用 --predict-delta 训练时须置 True，推理输出 （归一化空间）会加回 history 末步再反归一化，与 predict.py 对齐。 pred_token: np.ndarray 或 .npy 路径；export_deploy.py --pred-token-out 产物。缺省时喂零（仅适合 pred_token 恰好全零的模型）。 + normalize_mode: 与训练时 DataConfig.normalize_mode 一致。 + "zscore"（默认）需传 stats_path；"fixed_range" 免 stats 文件， + 按固定物理范围 min-max 归一化（action_norm.py）。 """ from hailo_platform import HEF, VDevice self._predict_delta = predict_delta + self._fixed_range = normalize_mode == "fixed_range" self._default_pred_token = None if pred_token is not None: if isinstance(pred_token, str): @@ -113,7 +105,7 @@ class HailoPredictor: self._action_mean = None self._action_std = None - if stats_path is not None: + if stats_path is not None and not self._fixed_range: if not os.path.exists(stats_path): raise FileNotFoundError(f"统计文件不存在: {stats_path}") stats = np.load(stats_path) @@ -255,7 +247,9 @@ class HailoPredictor: raise ValueError( f"action_history 形状应为 {expected}，实际为 {action_history.shape}") - if self._action_mean is not None: + if self._fixed_range: + action_history = normalize_action(action_history, self.action_dim) + elif self._action_mean is not None: action_history = (action_history - self._action_mean) / self._action_std from hailo_platform import FormatType @@ -316,7 +310,9 @@ class HailoPredictor: if self._predict_delta: # 残差模式：Δ̂ + a_{t-1}（归一化空间），与 predict.py 对齐 result = result + action_history[-1] - if self._action_mean is not None: + if self._fixed_range: + result = denormalize_action(result, self.action_dim) + elif self._action_mean is not None: result = result * self._action_std + self._action_mean if not np.all(np.isfinite(result)): print("WARNING: HEF output contains NaN/Inf, replacing with zeros") @@ -370,7 +366,11 @@ def main(): parser.add_argument("--action-history", required=True, help="冒号分隔各步，每步 action_dim 个值") parser.add_argument("--stats", default=None, - help="action_stats.npy 路径（训练时启用归一化才需要）") + help="action_stats.npy 路径（zscore 归一化训练才需要）") + parser.add_argument("--normalize-mode", choices=["zscore", "fixed_range"], + default=None, + help="与训练时 normalize_mode 一致；fixed_range 免 --stats。" + "缺省按 --ckpt 配置自动判定，无 --ckpt 时为 zscore") parser.add_argument("--pred-token", default=None, help="pred_token .npy 路径（export_deploy.py " "--pred-token-out 产物；缺省喂零）") @@ -391,6 +391,7 @@ def main(): dtype=np.float32) predict_delta = False + normalize_mode = args.normalize_mode or "zscore" if args.ckpt: import torch ck = torch.load(args.ckpt, map_location="cpu", weights_only=True) @@ -398,9 +399,18 @@ def main(): predict_delta = cfg.get("train", {}).get("predict_delta", False) if predict_delta: print("predict_delta=True: HEF 输出将加回历史末步（残差模式）") + if args.normalize_mode is None: + # 未显式指定时按 checkpoint 配置判定归一化方式（旧 checkpoint 无 + # normalize_mode 键，默认 zscore，行为不变） + data_cfg = cfg.get("data", {}) + if (data_cfg.get("normalize_actions") + and data_cfg.get("normalize_mode") == "fixed_range"): + normalize_mode = "fixed_range" + print("normalize_mode=fixed_range: 固定物理范围归一化（免 stats）") with HailoPredictor(args.hef, stats_path=args.stats, - predict_delta=predict_delta) as predictor: + predict_delta=predict_delta, + normalize_mode=normalize_mode) as predictor: if len(args.images) != predictor.frame_len: raise SystemExit( f"--images 需传入 {predictor.frame_len} 个路径（旧→新），" diff --git a/actpred/mt_parser.py b/actpred/mt_parser.py index 5067603..e9496d5 100644 --- a/actpred/mt_parser.py +++ b/actpred/mt_parser.py @@ -30,7 +30,7 @@ def parse_mt_action(path) -> np.ndarray: Raises: ValueError: 文件中无 @Action 块、数据行不足、行列不一致或数值非法 """ - text = Path(path).read_text(encoding="utf-8") + text = Path(path).read_text(encoding="utf-8-sig") # 先去除整行注释（# 开头），避免注释中的 @Action 干扰正则匹配 text = re.sub(r"(?m)^[ \t]*#.*$", "", text) diff --git a/actpred/onnx_infer.py b/actpred/onnx_infer.py index 2f282e0..1757b34 100644 --- a/actpred/onnx_infer.py +++ b/actpred/onnx_infer.py @@ -18,21 +18,13 @@ import numpy as np from PIL import Image from imu_norm import normalize_imu +from action_norm import normalize_action, denormalize_action IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) -def parse_action_history(s: str, action_dim: int) -> np.ndarray: - steps = s.split(":") - history = [] - for step in steps: - values = [float(v) for v in step.split(",")] - if len(values) != action_dim: - raise ValueError( - f"Expected {action_dim} values per step, got {len(values)}") - history.append(values) - return np.array(history, dtype=np.float32) +from utils import parse_action_history class OnnxPredictor: @@ -44,7 +36,13 @@ class OnnxPredictor: """ def __init__(self, onnx_path: str, stats_path: str = None, - providers=None): + providers=None, normalize_mode: str = "zscore"): + """ONNX 推理器。 + + normalize_mode: 与训练时 DataConfig.normalize_mode 一致。 + "zscore"（默认）需传 stats_path；"fixed_range" 免 stats 文件， + 按固定物理范围 min-max 归一化（action_norm.py）。 + """ import onnxruntime as ort if providers is None: @@ -70,7 +68,8 @@ class OnnxPredictor: self._action_mean = None self._action_std = None - if stats_path is not None: + self._fixed_range = normalize_mode == "fixed_range" + if stats_path is not None and not self._fixed_range: if not os.path.exists(stats_path): raise FileNotFoundError(f"统计文件不存在: {stats_path}") stats = np.load(stats_path) @@ -133,7 +132,9 @@ class OnnxPredictor: raise ValueError( f"action_history 形状应为 {expected}，实际为 {action_history.shape}") - if self._action_mean is not None: + if self._fixed_range: + action_history = normalize_action(action_history, self.action_dim) + elif self._action_mean is not None: action_history = (action_history - self._action_mean) / self._action_std frames = np.stack([self._preprocess(img) for img in images]) @@ -145,7 +146,9 @@ class OnnxPredictor: feeds[self._imu_name] = normalize_imu( imu, self.imu_dim)[np.newaxis] # (1,imu_dim) result = self._session.run(None, feeds)[0].reshape(-1).astype(np.float32) - if self._action_mean is not None: + if self._fixed_range: + result = denormalize_action(result, self.action_dim) + elif self._action_mean is not None: result = result * self._action_std + self._action_mean return result @@ -167,7 +170,10 @@ def main(): parser.add_argument("--action-history", required=True, help="冒号分隔各步，每步 action_dim 个值") parser.add_argument("--stats", default=None, - help="action_stats.npy 路径（训练时启用归一化才需要）") + help="action_stats.npy 路径（zscore 归一化训练才需要）") + parser.add_argument("--normalize-mode", choices=["zscore", "fixed_range"], + default="zscore", + help="与训练时 normalize_mode 一致；fixed_range 免 --stats") parser.add_argument("--imu", type=str, default=None, help="可选姿态向量（逗号分隔，原始量纲：roll,pitch 度 / " "ax..gz int16 原始值）；模型含 imu 输入且缺省时按零向量") @@ -175,7 +181,8 @@ def main(): default="json") args = parser.parse_args() - with OnnxPredictor(args.model, stats_path=args.stats) as predictor: + with OnnxPredictor(args.model, stats_path=args.stats, + normalize_mode=args.normalize_mode) as predictor: if len(args.images) != predictor.frame_len: raise SystemExit( f"--images 需传入 {predictor.frame_len} 个路径（旧→新），" diff --git a/actpred/predict.py b/actpred/predict.py index 195a3b9..0dd6336 100644 --- a/actpred/predict.py +++ b/actpred/predict.py @@ -8,7 +8,9 @@ from torchvision import transforms from config import parse_checkpoint_config from imu_norm import normalize_imu +from action_norm import normalize_action, denormalize_action from model import ActionPredictor +from utils import parse_action_history class ActionPredictorInference: @@ -48,7 +50,11 @@ class ActionPredictorInference: self._action_mean = None self._action_std = None - if self._data_cfg.normalize_actions: + # fixed_range：固定物理范围 min-max（免 stats 文件）；旧 checkpoint 无 + # normalize_mode 键，parse_checkpoint_config 默认 "zscore"，行为不变 + self._fixed_range = (self._data_cfg.normalize_actions + and self._data_cfg.normalize_mode == "fixed_range") + if self._data_cfg.normalize_actions and not self._fixed_range: # 优先使用 checkpoint config 记录的 stats 路径（相对路径按 checkpoint 目录解析）， # 旧 checkpoint 无该字段时回退到 checkpoint 同目录的 action_stats.npy stats_path = self._data_cfg.action_stats_path @@ -94,7 +100,10 @@ class ActionPredictorInference: frames.append(self._transform(img)) img_tensor = torch.stack(frames).unsqueeze(0).to(self._device) # (1, F, 3, S, S) history = torch.tensor(action_history, dtype=torch.float32) - if self._action_mean is not None: + if self._fixed_range: + history = torch.from_numpy( + normalize_action(action_history, self.action_dim)) + elif self._action_mean is not None: history = (history - self._action_mean) / self._action_std history = history.unsqueeze(0).to(self._device) imu_tensor = None @@ -106,27 +115,18 @@ class ActionPredictorInference: if self._predict_delta: # 残差：Δ̂ 加回 a_{t-1}（history 末步，同为归一化空间）→ 绝对动作 pred = history[0, -1] + pred - if self._action_mean is not None: - pred = pred * self._action_std + self._action_mean - result = pred.cpu().numpy() + if self._fixed_range: + result = denormalize_action(pred.cpu().numpy(), self.action_dim) + else: + if self._action_mean is not None: + pred = pred * self._action_std + self._action_mean + result = pred.cpu().numpy() if not np.all(np.isfinite(result)): print("WARNING: predict output contains NaN/Inf, replacing with zeros") result = np.nan_to_num(result, nan=0.0, posinf=0.0, neginf=0.0) return result -def parse_action_history(s: str, action_dim: int) -> np.ndarray: - steps = s.split(":") - history = [] - for step in steps: - values = [float(v) for v in step.split(",")] - if len(values) != action_dim: - raise ValueError( - f"Expected {action_dim} values per step, got {len(values)}") - history.append(values) - return np.array(history, dtype=np.float32) - - def main(): parser = argparse.ArgumentParser(description="actpred inference") parser.add_argument("--model", required=True, help="checkpoint path") diff --git a/actpred/pretrain_mt.py b/actpred/pretrain_mt.py index c55138b..9ab5b45 100644 --- a/actpred/pretrain_mt.py +++ b/actpred/pretrain_mt.py @@ -33,6 +33,8 @@ from config import (Config, build_argparser, apply_overrides, validate_config_values, warn_init_from_mismatch) from model import ActionPredictor from mt_parser import parse_mt_directory +from action_norm import normalize_action, default_column_weights +from train import _weighted_mse _IMAGENET_MEAN = [0.485, 0.456, 0.406] _IMAGENET_STD = [0.229, 0.224, 0.225] @@ -46,11 +48,15 @@ class MTActionDataset(Dataset): def __init__(self, mt_dir: str, history_len: int, action_dim: int, file_indices: list = None, files: list = None, - action_mean=None, action_std=None): + action_mean=None, action_std=None, + normalize_mode: str = "zscore"): self.history_len = history_len self.action_dim = action_dim self._action_mean = action_mean self._action_std = action_std + # 固定物理范围归一化（免统计量；action_mean 传入时优先 z-score） + self._fixed_range = (normalize_mode == "fixed_range" + and action_mean is None) self.samples = [] # 复用调用方已解析的结果（避免 train/val 各自重复解析同一目录） @@ -83,6 +89,9 @@ class MTActionDataset(Dataset): def __getitem__(self, idx): history, target = self.samples[idx] + if self._fixed_range: + history = normalize_action(history, self.action_dim) + target = normalize_action(target, self.action_dim) history = torch.from_numpy(history.copy()) target = torch.from_numpy(target.copy()) if self._action_mean is not None: @@ -110,7 +119,7 @@ def _placeholder_images(batch, frame_len, image_height, image_width, def pretrain_one_epoch(model, loader, optimizer, device, image_width, image_height, frame_len, max_grad_norm=None, noise_std=0.0, - random_images: bool = False): + random_images: bool = False, weights=None): model.train() total_loss = 0.0 count = 0 @@ -127,7 +136,7 @@ def pretrain_one_epoch(model, loader, optimizer, device, image_width, device, random_images) optimizer.zero_grad() pred = model(images, history) - loss = torch.nn.functional.mse_loss(pred, target) + loss = _weighted_mse(pred, target, weights) loss.backward() if max_grad_norm: torch.nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm) @@ -139,7 +148,7 @@ def pretrain_one_epoch(model, loader, optimizer, device, image_width, @torch.no_grad() def validate(model, loader, device, image_width, image_height, frame_len, - random_images: bool = False): + random_images: bool = False, weights=None): model.eval() total_loss = 0.0 count = 0 @@ -152,7 +161,7 @@ def validate(model, loader, device, image_width, image_height, frame_len, history.size(0), frame_len, image_height, image_width, device, random_images) pred = model(images, history) - loss = torch.nn.functional.mse_loss(pred, target) + loss = _weighted_mse(pred, target, weights) total_loss += loss.item() * history.size(0) count += history.size(0) if count == 0: @@ -173,7 +182,6 @@ def main(): cfg = apply_overrides(Config(), args) validate_config_values(cfg.data, cfg.model, cfg.train) - cfg.data.data_dir = args.mt_dir os.makedirs(cfg.train.checkpoint_dir, exist_ok=True) @@ -205,12 +213,17 @@ def main(): train_ds = MTActionDataset( args.mt_dir, cfg.data.history_len, cfg.data.action_dim, file_indices=train_file_indices, files=all_files, + normalize_mode=cfg.data.normalize_mode, ) - # 动作归一化（与 train.py 语义一致）：无 stats 文件时按训练集 - # targets 计算并保存到 checkpoint_dir，验证集复用同一 stats + # 动作归一化（与 train.py 语义一致）：zscore 模式无 stats 文件时按训练集 + # targets 计算并保存到 checkpoint_dir，验证集复用同一 stats； + # fixed_range 模式免统计量，跳过此段 action_mean = action_std = None - if cfg.data.normalize_actions: + if cfg.data.normalize_actions and cfg.data.normalize_mode == "fixed_range": + # 构造时 action_mean=None 已置 _fixed_range，这里只打印口径 + print("动作归一化: fixed_range（固定物理范围 min-max→[0,1]，免 stats）") + elif cfg.data.normalize_actions: if cfg.data.action_stats_path: if not os.path.exists(cfg.data.action_stats_path): raise FileNotFoundError( @@ -248,6 +261,7 @@ def main(): args.mt_dir, cfg.data.history_len, cfg.data.action_dim, file_indices=val_file_indices, files=all_files, action_mean=action_mean, action_std=action_std, + normalize_mode=cfg.data.normalize_mode, ) print(f"Files: {len(train_file_indices)} train, {len(val_file_indices)} val") @@ -272,6 +286,13 @@ def main(): model = ActionPredictor(cfg.model, cfg.data, pretrained=False).to(device) n_params = sum(p.numel() for p in model.parameters()) print(f"Model params: {n_params:,}") + + # 逐列损失权重（与 train.py 同口径；仅训练损失用） + col_w = cfg.data.column_weights + if col_w is None: + col_w = default_column_weights(cfg.data.action_dim) + weights_t = torch.tensor(np.asarray(col_w, dtype=np.float32), device=device) + print(f"Column weights: {[round(float(v), 4) for v in weights_t.tolist()]}") print(f"Config: action_dim={cfg.data.action_dim}, " f"history_len={cfg.data.history_len}, " f"frame_len={cfg.data.frame_len}, " @@ -328,34 +349,40 @@ def main(): model, train_loader, optimizer, device, cfg.data.image_width, cfg.data.image_height, cfg.data.frame_len, cfg.train.max_grad_norm, args.noise_std, args.random_images, + weights=weights_t, ) val_loss = validate(model, val_loader, device, cfg.data.image_width, cfg.data.image_height, - cfg.data.frame_len, args.random_images) + cfg.data.frame_len, args.random_images, + weights=weights_t) scheduler.step() print(f" train_loss={train_loss:.6f} val_loss={val_loss:.6f}") if val_loss < best_val_loss: - best_val_loss = val_loss - patience_counter = 0 - torch.save( - { - "epoch": epoch, - "model_state_dict": model.state_dict(), - "optimizer_state_dict": optimizer.state_dict(), - "val_loss": val_loss, - "config": cfg.to_dict(), - }, - best_path, - ) - torch.save( - { - "model_state_dict": model.state_dict(), - "config": cfg.to_dict(), - }, - inference_path, - ) - print(f" -> saved best (val_loss={val_loss:.6f})") + if not np.isfinite(val_loss): + print(f"WARNING: val_loss 非有限值（{val_loss}），跳过 best 更新") + patience_counter += 1 + else: + best_val_loss = val_loss + patience_counter = 0 + torch.save( + { + "epoch": epoch, + "model_state_dict": model.state_dict(), + "optimizer_state_dict": optimizer.state_dict(), + "val_loss": val_loss, + "config": cfg.to_dict(), + }, + best_path, + ) + torch.save( + { + "model_state_dict": model.state_dict(), + "config": cfg.to_dict(), + }, + inference_path, + ) + print(f" -> saved best (val_loss={val_loss:.6f})") else: patience_counter += 1 if patience_counter >= cfg.train.patience: diff --git a/actpred/test_actpred.py b/actpred/test_actpred.py index 316c46e..8edaa72 100644 --- a/actpred/test_actpred.py +++ b/actpred/test_actpred.py @@ -1790,17 +1790,281 @@ class TestR4Evaluate(unittest.TestCase): self.assertGreater(rep["error_growth"], 1.0) self.assertEqual(len(rep["error_curve"]), 40) - def test_action_bounds_margin(self): + def test_action_bounds_physical(self): + """越界判据为固定物理界（与 action_norm 常量表共用），不再依赖数据统计""" import evaluate + lo, hi = evaluate._action_bounds(12) + np.testing.assert_allclose(lo, np.zeros(12), atol=1e-6) + np.testing.assert_allclose(hi[:10], [180.0] * 10, atol=1e-6) + np.testing.assert_allclose(hi[10:], [5000.0, 2000.0], atol=1e-6) + # action_dim > 18 的超出列按 [0,1] 补足 + lo20, hi20 = evaluate._action_bounds(20) + self.assertEqual(len(hi20), 20) + np.testing.assert_allclose(hi20[18:], [1.0, 1.0], atol=1e-6) + + def test_error_growth_per_episode_median(self): + """error_growth 按 episode 归一化取中位数，不被长短混叠主导""" + import evaluate + # ep1: 40 步线性增长 → g1 = mean(31..40)/mean(1..10) = 35.5/5.5 + ep1 = [{"t": k, "mse": float(k + 1), "pred": np.zeros(2), + "true": np.zeros(2), "oob": False} for k in range(40)] + # ep2: 8 步末段尖峰 → g2 = 100/1 + ep2 = [{"t": k, "mse": (1.0 if k < 6 else 100.0), "pred": np.zeros(2), + "true": np.zeros(2), "oob": False} for k in range(8)] + rep = evaluate.summarize([ep1, ep2], None, None) + g1 = 35.5 / 5.5 + g2 = 100.0 + self.assertAlmostEqual(rep["error_growth"], float(np.median([g1, g2])), + places=5) + + +class TestFixedRangeNorm(unittest.TestCase): + """固定物理范围归一化 + 逐列加权：action_norm 常量表、dataset/predict 链路、 + config 校验、加权损失""" + + # ---------------- action_norm.py ---------------- + + def test_normalize_known_values(self): + from action_norm import normalize_action + a = np.zeros(18, dtype=np.float32) + a[0] = 90.0 # 舵机中位 + a[10] = 2500.0 # spk.F 中位 + a[11] = 1000.0 # spk.delay 中位 + out = normalize_action(a) + self.assertAlmostEqual(float(out[0]), 0.5, places=6) + self.assertAlmostEqual(float(out[10]), 0.5, places=6) + self.assertAlmostEqual(float(out[11]), 0.5, places=6) + # 保留位恒 0 → 归一化后仍 0 + np.testing.assert_allclose(out[12:], 0.0, atol=1e-7) + + def test_normalize_clamps_out_of_range(self): + from action_norm import normalize_action + a = np.array([-10.0, 270.0] + [0.0] * 16, dtype=np.float32) + out = normalize_action(a) + self.assertEqual(float(out[0]), 0.0) + self.assertEqual(float(out[1]), 1.0) + + def test_denormalize_roundtrip(self): + from action_norm import normalize_action, denormalize_action + rng = np.random.RandomState(0) + a = np.zeros((7, 18), dtype=np.float32) + a[:, :10] = rng.uniform(0, 180, (7, 10)) + a[:, 10] = rng.uniform(0, 5000, 7) + a[:, 11] = rng.uniform(0, 2000, 7) + np.testing.assert_allclose( + denormalize_action(normalize_action(a)), a, rtol=1e-5, atol=1e-4) + + def test_action_dim_prefix_and_padding(self): + from action_norm import normalize_action, action_bounds + # action_dim < 18 取前缀 + lo4, hi4 = action_bounds(4) + np.testing.assert_allclose(hi4, [180.0] * 4) + out = normalize_action(np.array([90.0, 45.0, 0.0, 180.0], + dtype=np.float32)) + np.testing.assert_allclose(out, [0.5, 0.25, 0.0, 1.0], atol=1e-6) + # action_dim > 18 按 [0,1] 补足 + lo20, hi20 = action_bounds(20) + self.assertEqual(len(lo20), 20) + np.testing.assert_allclose(hi20[18:], [1.0, 1.0]) + + def test_default_column_weights(self): + from action_norm import default_column_weights + w18 = default_column_weights(18) + np.testing.assert_allclose(w18[:10], 1.0) + np.testing.assert_allclose(w18[10:12], 0.2) + np.testing.assert_allclose(w18[12:], 0.0) + # 小 action_dim：全部落在舵机段 → 全 1.0 + np.testing.assert_allclose(default_column_weights(4), 1.0) + # 12 列：舵机 1.0 + spk 0.2 + w12 = default_column_weights(12) + np.testing.assert_allclose(w12[10:], 0.2) + + # ---------------- dataset 链路 ---------------- + + def test_dataset_fixed_range_no_stats_needed(self): + """fixed_range 模式不需要 action_stats.npy，目标值按物理范围归一化""" with tempfile.TemporaryDirectory() as tmpdir: - self._create_episode(Path(tmpdir), "episode_0001", - np.zeros((5, 2), dtype=np.float32)) - self._create_episode(Path(tmpdir), "episode_0002", - np.full((5, 2), 10.0, dtype=np.float32)) - lo, hi = evaluate._action_bounds(tmpdir, 2, margin_ratio=0.1) - # min=0,max=10,span=10,margin=1 → lo=-1, hi=11 - np.testing.assert_allclose(lo, [-1.0, -1.0], atol=1e-6) - np.testing.assert_allclose(hi, [11.0, 11.0], atol=1e-6) + base = Path(tmpdir) + ep_dir = base / "episode_0001" + images_dir = ep_dir / "images" + images_dir.mkdir(parents=True) + for i in range(8): + Image.fromarray(np.random.randint( + 0, 255, (16, 16, 3), dtype=np.uint8)).save( + images_dir / f"{i:04d}.jpg") + actions = np.zeros((8, 4), dtype=np.float32) + actions[:, 0] = 90.0 # 舵机中位 → 0.5 + actions[:, 1] = 180.0 # 满行程 → 1.0 + np.save(ep_dir / "actions.npy", actions) + ds = ActionDataset(tmpdir, history_len=3, action_dim=4, + image_width=16, image_height=16, frame_len=3, + normalize_actions=True, + normalize_mode="fixed_range") + sample = ds[0] + np.testing.assert_allclose(sample["target"].numpy()[:2], + [0.5, 1.0], atol=1e-6) + np.testing.assert_allclose( + sample["action_history"].numpy()[:, 0], 0.5, atol=1e-6) + # fixed_range 不产生统计量 + self.assertIsNone(ds.get_action_stats()) + + def test_dataset_fixed_range_clamps(self): + """越界值 clamp 到 [0,1]（防御口径）""" + with tempfile.TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + ep_dir = base / "episode_0001" + images_dir = ep_dir / "images" + images_dir.mkdir(parents=True) + for i in range(8): + Image.fromarray(np.random.randint( + 0, 255, (16, 16, 3), dtype=np.uint8)).save( + images_dir / f"{i:04d}.jpg") + actions = np.zeros((8, 4), dtype=np.float32) + actions[:, 0] = 999.0 # 远超 180 → clamp 1.0 + np.save(ep_dir / "actions.npy", actions) + ds = ActionDataset(tmpdir, history_len=3, action_dim=4, + image_width=16, image_height=16, frame_len=3, + normalize_actions=True, + normalize_mode="fixed_range") + self.assertEqual(float(ds[0]["target"][0]), 1.0) + + def test_dataset_zscore_unchanged(self): + """zscore 默认模式行为不变（仍计算/加载统计量）""" + with tempfile.TemporaryDirectory() as tmpdir: + TestActionDataset._create_dummy_episode( + Path(tmpdir), n_frames=10, action_dim=4) + ds = ActionDataset(tmpdir, history_len=3, action_dim=4, + image_width=32, image_height=32, frame_len=3, + normalize_actions=True) + self.assertIsNotNone(ds._action_mean) + self.assertFalse(ds._fixed_range) + + def test_mt_dataset_fixed_range(self): + """MTActionDataset fixed_range：免统计量，按物理范围归一化""" + with tempfile.TemporaryDirectory() as tmpdir: + _write_mt(tmpdir, "a.mt", _make_seq_mt(6)) + with contextlib.redirect_stdout(io.StringIO()): + ds = MTActionDataset(tmpdir, history_len=3, action_dim=3, + normalize_mode="fixed_range") + sample = ds[0] + # _make_seq_mt 第 i 行 [i, 10i, 100i]，target 为第 4 行 [3,30,300] + # 除数 180 → [3/180, 30/180, 300/180(clamp 1.0)] + np.testing.assert_allclose( + sample["target"].numpy(), + [3 / 180, 30 / 180, 1.0], atol=1e-6) + + # ---------------- config 校验与 CLI ---------------- + + def test_cli_normalize_mode_and_column_weights(self): + parser = build_argparser() + args = parser.parse_args( + ["--normalize-mode", "fixed_range", + "--column-weights", "1,1,1,0.2"]) + cfg = apply_overrides(Config(), args) + cfg.data.action_dim = 4 + self.assertEqual(cfg.data.normalize_mode, "fixed_range") + self.assertEqual(cfg.data.column_weights, [1.0, 1.0, 1.0, 0.2]) + validate_config_values(cfg.data, cfg.model) # 不应抛异常 + + def test_bad_normalize_mode_raises(self): + cfg = Config() + cfg.data.normalize_mode = "minmax" # 非法值 + with self.assertRaisesRegex(ValueError, "normalize_mode"): + validate_config_values(cfg.data, cfg.model) + + def test_column_weights_length_mismatch_raises(self): + cfg = Config() + cfg.data.column_weights = [1.0, 0.2] # action_dim=18 不匹配 + with self.assertRaisesRegex(ValueError, "column_weights"): + validate_config_values(cfg.data, cfg.model) + + def test_column_weights_all_zero_raises(self): + cfg = Config() + cfg.data.column_weights = [0.0] * 18 + with self.assertRaisesRegex(ValueError, "column_weights"): + validate_config_values(cfg.data, cfg.model) + + def test_checkpoint_roundtrip_new_keys(self): + """新键随 checkpoint 序列化/解析往返一致；旧 checkpoint 缺省兼容""" + cfg = Config() + cfg.data.normalize_mode = "fixed_range" + cfg.data.column_weights = [1.0] * 10 + [0.2] * 2 + [0.0] * 6 + parsed = parse_checkpoint_config(cfg.to_dict()) + self.assertEqual(parsed.data.normalize_mode, "fixed_range") + self.assertEqual(parsed.data.column_weights, + [1.0] * 10 + [0.2] * 2 + [0.0] * 6) + # 旧 checkpoint 无新键 → 默认 zscore / None，行为不变 + d = cfg.to_dict() + del d["data"]["normalize_mode"] + del d["data"]["column_weights"] + parsed = parse_checkpoint_config(d) + self.assertEqual(parsed.data.normalize_mode, "zscore") + self.assertIsNone(parsed.data.column_weights) + + def test_warn_init_from_normalize_mode_mismatch(self): + from config import warn_init_from_mismatch + cfg = Config() + cfg.data.normalize_mode = "fixed_range" + init_cfg = {"data": {"normalize_actions": True, + "normalize_mode": "zscore"}} + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + warn_init_from_mismatch(init_cfg, cfg) + self.assertIn("normalize_mode", buf.getvalue()) + + # ---------------- 加权损失 ---------------- + + def test_weighted_mse(self): + from train import _weighted_mse + pred = torch.tensor([[1.0, 2.0, 3.0]]) + target = torch.zeros(1, 3) + # 等权 → 与普通 MSE 一致 + w1 = torch.ones(3) + self.assertAlmostEqual( + float(_weighted_mse(pred, target, w1)), + float(torch.nn.functional.mse_loss(pred, target)), places=6) + # 只留第 0 列权重：loss = (1·1²)/(1) = 1 + w = torch.tensor([1.0, 0.0, 0.0]) + self.assertAlmostEqual(float(_weighted_mse(pred, target, w)), 1.0, + places=6) + # None 退化为普通 MSE + self.assertAlmostEqual( + float(_weighted_mse(pred, target, None)), + float(torch.nn.functional.mse_loss(pred, target)), places=6) + + # ---------------- predict 链路 ---------------- + + def test_predict_fixed_range_no_stats(self): + """fixed_range checkpoint 无需 action_stats.npy，归一化口径与 + action_norm 一致（与 raw 模型 + 手工归一化对照）""" + from action_norm import normalize_action, denormalize_action + with tempfile.TemporaryDirectory() as tmpdir: + raw_dir = os.path.join(tmpdir, "raw") + fixed_dir = os.path.join(tmpdir, "fixed") + os.makedirs(raw_dir) + os.makedirs(fixed_dir) + torch.manual_seed(0) + path, model, cfg = \ + TestActionPredictorInference._make_checkpoint(raw_dir) + # 同一份权重另存为 fixed_range 归一化 checkpoint（无 stats 文件） + cfg.data.normalize_actions = True + cfg.data.normalize_mode = "fixed_range" + fixed_path = os.path.join(fixed_dir, "best_inference.pt") + torch.save({"model_state_dict": model.state_dict(), + "config": cfg.to_dict()}, fixed_path) + + image = np.random.randint(0, 255, (32, 32, 3), dtype=np.uint8) + history = np.random.uniform(0, 180, (3, 4)).astype(np.float32) + # fixed_range 端：内部 normalize → 模型 → denormalize + pred_fixed = ActionPredictorInference( + fixed_path, device="cpu").predict([image] * 3, history) + # 对照：raw 模型直接吃手工归一化的历史，再手工反归一化 + pred_raw = ActionPredictorInference(raw_dir + "/best_inference.pt", + device="cpu").predict( + [image] * 3, normalize_action(history, 4)) + np.testing.assert_allclose( + pred_fixed, denormalize_action(pred_raw, 4), + rtol=1e-5, atol=1e-5) class TestR2ImuNorm(unittest.TestCase): diff --git a/actpred/train.py b/actpred/train.py index 5b19a5c..54cbad2 100644 --- a/actpred/train.py +++ b/actpred/train.py @@ -10,6 +10,7 @@ from config import (Config, build_argparser, apply_overrides, validate_config_values, warn_init_from_mismatch) from model import ActionPredictor from dataset import ActionDataset +from action_norm import default_column_weights def split_episodes(data_dir: str, val_ratio: float, seed: int = 42): @@ -21,6 +22,17 @@ def split_episodes(data_dir: str, val_ratio: float, seed: int = 42): return indices[val_count:], indices[:val_count] +def _weighted_mse(pred: torch.Tensor, target: torch.Tensor, + weights: torch.Tensor = None) -> torch.Tensor: + """逐列加权 MSE：mean(w·(pred-target)²)/mean(w)，weights None 时退化为普通 MSE。 + + 除以 mean(w) 保持损失量级与未加权口径可比（w 全 1 时与普通 MSE 完全一致）。 + """ + if weights is None: + return torch.nn.functional.mse_loss(pred, target) + return (weights * (pred - target) ** 2).mean() / weights.mean() + + def _collate(batch): """多步样本（{"steps":[...]}）逐 k 聚合成 batch 列表；单步样本走默认 collate。 @@ -46,7 +58,7 @@ def _shift_history(history: torch.Tensor, new_action: torch.Tensor) -> torch.Ten def train_one_epoch(model, loader, optimizer, device, max_grad_norm: float = None, rollout_len: int = 1, ss_prob: float = 1.0, - predict_delta: bool = False): + predict_delta: bool = False, weights: torch.Tensor = None): model.train() total_loss = 0.0 count = 0 @@ -59,7 +71,7 @@ def train_one_epoch(model, loader, optimizer, device, max_grad_norm: float = Non target = batch["target"].to(device) imu = batch["imu"].to(device) if "imu" in batch else None pred = model(images, history, imu) - loss = torch.nn.functional.mse_loss(pred, target) + loss = _weighted_mse(pred, target, weights) bs = images.size(0) else: # 多步 scheduled sampling：首步必用真历史；后续步以 ss_prob 用真历史、 @@ -77,7 +89,7 @@ def train_one_epoch(model, loader, optimizer, device, max_grad_norm: float = Non history = true_history # 否则沿用上一轮回传的 history（自预测构造） pred = model(images, history, imu) - loss = loss + torch.nn.functional.mse_loss(pred, target) + loss = loss + _weighted_mse(pred, target, weights) bs = images.size(0) # 回传：残差模式先把 Δ̂ 加回 a_{t-1} 得到绝对动作，再移入历史 with torch.no_grad(): @@ -94,7 +106,8 @@ def train_one_epoch(model, loader, optimizer, device, max_grad_norm: float = Non @torch.no_grad() -def validate(model, loader, device, rollout_len: int = 1, predict_delta: bool = False): +def validate(model, loader, device, rollout_len: int = 1, predict_delta: bool = False, + weights: torch.Tensor = None): model.eval() total_loss = 0.0 count = 0 @@ -105,7 +118,7 @@ def validate(model, loader, device, rollout_len: int = 1, predict_delta: bool = target = batch["target"].to(device) imu = batch["imu"].to(device) if "imu" in batch else None pred = model(images, history, imu) - loss = torch.nn.functional.mse_loss(pred, target) + loss = _weighted_mse(pred, target, weights) bs = images.size(0) else: # 验证固定全真历史（确定性，不带 scheduled sampling 随机性）， @@ -118,7 +131,7 @@ def validate(model, loader, device, rollout_len: int = 1, predict_delta: bool = target = step["target"].to(device) imu = step["imu"].to(device) if "imu" in step else None pred = model(images, history, imu) - loss = loss + torch.nn.functional.mse_loss(pred, target) + loss = loss + _weighted_mse(pred, target, weights) bs = images.size(0) loss = loss / len(batch["steps"]) total_loss += loss.item() * bs @@ -156,10 +169,12 @@ def main(): cfg.data.action_stats_path, train_indices, cfg.data.frame_len, augment=cfg.train.augment, predict_delta=cfg.train.predict_delta, rollout_len=cfg.train.rollout_len, - imu_dim=cfg.data.imu_dim) + imu_dim=cfg.data.imu_dim, normalize_mode=cfg.data.normalize_mode) auto_stats = False - if cfg.data.normalize_actions and not cfg.data.action_stats_path: + if (cfg.data.normalize_actions and cfg.data.normalize_mode == "zscore" + and not cfg.data.action_stats_path): # 归一化统计量必须只用训练集计算，再供验证集复用，否则 val loss 不可信 + # （fixed_range 模式免 stats 文件，跳过此段） stats_path = os.path.join(cfg.train.checkpoint_dir, "action_stats.npy") np.save(stats_path, train_ds.get_action_stats()) print(f"Action stats saved to: {stats_path}") @@ -171,7 +186,7 @@ def main(): cfg.data.image_width, cfg.data.image_height, cfg.data.normalize_actions, cfg.data.action_stats_path, val_indices, cfg.data.frame_len, predict_delta=cfg.train.predict_delta, rollout_len=cfg.train.rollout_len, - imu_dim=cfg.data.imu_dim) + imu_dim=cfg.data.imu_dim, normalize_mode=cfg.data.normalize_mode) if auto_stats: # checkpoint 配置只存裸文件名：推理端按 checkpoint 目录解析相对路径 cfg.data.action_stats_path = "action_stats.npy" @@ -198,6 +213,13 @@ def main(): n_params = sum(p.numel() for p in model.parameters()) print(f"Model params: {n_params:,}") + # 逐列损失权重（仅训练/验证损失用，推理端无感知） + col_w = cfg.data.column_weights + if col_w is None: + col_w = default_column_weights(cfg.data.action_dim) + weights_t = torch.tensor(np.asarray(col_w, dtype=np.float32), device=device) + print(f"Column weights: {[round(float(v), 4) for v in weights_t.tolist()]}") + if args.init_from: init_ckpt = torch.load(args.init_from, map_location="cpu", weights_only=True) warn_init_from_mismatch(init_ckpt.get("config", {}), cfg) @@ -236,29 +258,35 @@ def main(): cfg.train.max_grad_norm, rollout_len=cfg.train.rollout_len, ss_prob=cfg.train.ss_prob, - predict_delta=cfg.train.predict_delta) + predict_delta=cfg.train.predict_delta, + weights=weights_t) val_loss = validate(model, val_loader, device, rollout_len=cfg.train.rollout_len, - predict_delta=cfg.train.predict_delta) + predict_delta=cfg.train.predict_delta, + weights=weights_t) scheduler.step() print(f" train_loss={train_loss:.6f} val_loss={val_loss:.6f}") if val_loss < best_val_loss: - best_val_loss = val_loss - patience_counter = 0 - torch.save({ - "epoch": epoch, - "model_state_dict": model.state_dict(), - "optimizer_state_dict": optimizer.state_dict(), - "val_loss": val_loss, - "config": cfg.to_dict(), - }, best_path) - inference_path = os.path.join(cfg.train.checkpoint_dir, "best_inference.pt") - torch.save({ - "model_state_dict": model.state_dict(), - "config": cfg.to_dict(), - }, inference_path) - print(f" -> saved best (val_loss={val_loss:.6f})") + if not np.isfinite(val_loss): + print(f"WARNING: val_loss 非有限值（{val_loss}），跳过 best 更新") + patience_counter += 1 + else: + best_val_loss = val_loss + patience_counter = 0 + torch.save({ + "epoch": epoch, + "model_state_dict": model.state_dict(), + "optimizer_state_dict": optimizer.state_dict(), + "val_loss": val_loss, + "config": cfg.to_dict(), + }, best_path) + inference_path = os.path.join(cfg.train.checkpoint_dir, "best_inference.pt") + torch.save({ + "model_state_dict": model.state_dict(), + "config": cfg.to_dict(), + }, inference_path) + print(f" -> saved best (val_loss={val_loss:.6f})") else: patience_counter += 1 if patience_counter >= cfg.train.patience: @@ -284,6 +312,15 @@ def main(): "model_state_dict": model.state_dict(), "config": cfg.to_dict(), }, os.path.join(cfg.train.checkpoint_dir, "best_inference.pt")) + torch.save({ + "epoch": epoch, + "model_state_dict": model.state_dict(), + "optimizer_state_dict": optimizer.state_dict(), + "scheduler_state_dict": scheduler.state_dict(), + "best_val_loss": best_val_loss, + "patience_counter": patience_counter, + "config": cfg.to_dict(), + }, os.path.join(cfg.train.checkpoint_dir, "best.pt")) print("WARNING: saved final weights as best_inference.pt (no valid validation loss)") else: print(f"Best model saved to: {best_path}") diff --git a/actpred/utils.py b/actpred/utils.py new file mode 100644 index 0000000..ac5783a --- /dev/null +++ b/actpred/utils.py @@ -0,0 +1,16 @@ +"""actpred 共享工具函数（无 torch 依赖，推理/部署文件可复用）""" + +import numpy as np + + +def parse_action_history(s: str, action_dim: int) -> np.ndarray: + """解析冒号分隔的动作历史字符串为 (H, D) float32 数组""" + steps = s.split(":") + history = [] + for step in steps: + values = [float(v) for v in step.split(",")] + if len(values) != action_dim: + raise ValueError( + f"Expected {action_dim} values per step, got {len(values)}") + history.append(values) + return np.array(history, dtype=np.float32) \ No newline at end of file diff --git a/docs/reviews/pybot_actpred_AUDIT_20260814_v2.md b/docs/reviews/pybot_actpred_AUDIT_20260814_v2.md new file mode 100644 index 0000000..8e71902 --- /dev/null +++ b/docs/reviews/pybot_actpred_AUDIT_20260814_v2.md @@ -0,0 +1,191 @@ +# pybot & actpred 审计报告（第二版 —— 归一化+逐列加权 落地后） + +> 审查日期：2026-08-14 (Asia/Shanghai) +> 审查人：AI Assistant（SOLO Coder） +> 审查范围：`pybot/`（机器人运行时）与 `actpred/`（动作预测）两模块 +> 代码基线：归一化+逐列加权方案落地后（12 个文件改动，531 行新增） +> +> 本报告对照三份历史审计报告追踪修复状态，并审查本轮新增代码。 + +--- + +## 一、审计结论摘要 + +| 级别 | 新发现 | 历史遗留 | 说明 | +|------|--------|---------|------| +| **P0（崩溃/数据损坏）** | 2 | 0 | 见 §二 | +| P1（功能风险） | 0 | 5 | 见 §三 | +| P2（代码质量） | 3 | 9 | 见 §四 | + +**重点：2 个 P0 必须在提交前修复。** + +--- + +## 二、P0 新发现 + +### P0-1 pybot HailoPredictor 缺失 normalize_mode 传递 + +**位置**：[pybot/main.py:180-183](../../pybot/pybot/main.py#L180-L183) + +```python +return _P(checkpoint, stats_path=stats_path, + predict_delta=predict_delta, + pred_token=pred_token) +``` + +**问题**：`HailoPredictor.__init__` 的 `normalize_mode` 参数默认 `"zscore"`，pybot 调用时未传入此参数。对于 `fixed_range` 训练的 HEF，推理端应按 `action_norm.py` 固定物理范围对历史动作做 min-max→[0,1] 归一化，但 pybot 未传参导致 `_fixed_range=False`，历史动作以原始量纲送入 HEF。 + +**影响**：`fixed_range` 模式训练的 HEF 在 pybot 闭环中推理输出完全错误（舵机会被推到安全范围外）。`zscore` 模式不受影响（默认值偶然正确）。 + +**修复建议**：pybot 从同目录 checkpoint 读取 `normalize_mode` 并传入 `HailoPredictor`： + +```python +# 在 _load_predictor 中，从 companion checkpoint 读取 normalize_mode +normalize_mode = "zscore" +for name in ("pretrain_best_inference.pt", "best_inference.pt"): + ckpt_path = os.path.join(hef_dir, name) + if os.path.exists(ckpt_path): + try: + import torch + ck = torch.load(ckpt_path, map_location="cpu", weights_only=True) + normalize_mode = ck.get("config", {}).get("data", {}).get( + "normalize_mode", "zscore") + except Exception: + pass + break + +return _P(checkpoint, stats_path=stats_path, + predict_delta=predict_delta, + pred_token=pred_token, + normalize_mode=normalize_mode) +``` + +### P0-2 DataConfig normalize_actions 默认值变更，与 README 不一致 + +**位置**：[actpred/config.py:14](../../actpred/config.py#L14) + +```python +normalize_actions: bool = True # 旧默认值：False +``` + +**问题**：默认值从 `False` 改为 `True`，但 [README.md 配置表](../../actpred/README.md#L163) 仍标注 `False`。CLI 不经 `--normalize-actions` 参数时，`apply_overrides` 会使用 dataclass 默认值 `True`，导致行为变更： +- 旧脚本（`bash _run.sh train --data-dir ./data --epochs 100`）之前不归一化，现在开启 zscore 归一化，且会自动计算 `action_stats.npy`。 +- 对预训练（`pretrain_mt.py`）同样影响：预训练数据从 .mt 脚本提取，zscore 归一化后分布与旧 checkpoint 不同。 + +**影响**：所有无 `--normalize-actions` 或无 `--no-normalize-actions` 的训练命令行为改变；旧 checkpoint 做 `--init-from` 时 `warn_init_from_mismatch` 会警告但不会阻止。 + +**修复建议**：二选一： +- **方案 A**：恢复默认值 `False`（保持向后兼容，推荐）； +- **方案 B**：改 README 为 `True`，并在 CHANGELOG 中标注 breaking change。 + +--- + +## 三、P1 历史遗留（未修复） + +| # | 来源 | 位置 | 问题 | 优先级 | +|---|------|------|------|--------| +| P1-3 | 闭环审计 | [actpred/mt_parser.py:33](../../actpred/mt_parser.py#L33) | `read_text(encoding="utf-8")` 不剥离 BOM，`@Action` 为首块时表头列名可能粘附 `` | 低（低概率触发） | +| P2-a | 闭环审计 | [pybot/main.py:143](../../pybot/pybot/main.py#L143) | `_load_predictor` 仅用 `endswith(".hef")` 区分后端，大小写敏感，无 ONNX 分支 | 中 | +| R2-1 | actpred 审计 | [actpred/train.py:270](../../actpred/train.py#L270) | NaN val_loss 永久阻断 best checkpoint 保存（`NaN < x` 恒 False） | 中 | +| R1-1 | actpred 审计 | [actpred/pretrain_mt.py:185](../../actpred/pretrain_mt.py#L185) | `cfg.data.data_dir = args.mt_dir` 覆盖，`--save-config` 导出语义错误 JSON | 低 | +| R2-7 | actpred 审计 | [actpred/_run.sh:14-16](../../actpred/_run.sh#L14-L16) | 未知子命令静默回退到 `predict.py`，拼写错误难以发现 | 低 | + +--- + +## 四、P2 历史遗留（未修复） + +| # | 来源 | 位置 | 问题 | +|---|------|------|------| +| P2-b | 闭环审计 | pybot/robot.py 等 3 处 | numpy 在闭环路径 3 个函数各自 `import` | +| P2-c | 闭环审计 | pybot/tests/test_loop.py | 未覆盖闭环 + recorder 集成、predictor 异常清理 | +| P2-d | 闭环审计 | actpred/dataset.py | 图像 Resize 不保持宽高比 | +| P2-e | 闭环审计 | actpred/train.py | resume 时改 `--epochs` 会重算 lr scheduler | +| P2-f | 闭环审计 | actpred/export_deploy.py | ManualAttnBlock 依赖 PyTorch 内部布局约定 | +| R2-3 | actpred 审计 | actpred/train.py | 空验证集时 `best.pt` 从不创建 | +| R2-4 | actpred 审计 | actpred/_set.sh | CUDA 版 torch 误装风险 | +| R1-2 | actpred 审计 | predict/onnx_infer/hailo_infer | `parse_action_history` 三处重复 | +| R2-9 | actpred 审计 | actpred/predict.py | `--action-dim` override 参数无用且是陷阱 | + +--- + +## 五、已完成修复的审计项 + +| # | 来源 | 修复内容 | +|---|------|---------| +| P1-1 | 闭环审计 | 闭环节拍 overrun 告警已实现（[robot.py:307-320](../../pybot/pybot/robot.py#L307-L320)），含比例统计 | +| P1-2 | 闭环审计 | 闭环帧源改走 `_loop_recent_frames`（[robot.py:323-338](../../pybot/pybot/robot.py#L323-L338)），从守护线程队列取帧 | +| P1-4 | 闭环审计 | DataRecorder output_dir 已有 `mkdir(parents=True, exist_ok=True)`（[recorder.py:46](../../pybot/pybot/recorder.py#L46)） | +| R2-2 | actpred 审计 | ActionDataset 外部 stats 文件已补 shape 校验（[dataset.py:92-95](../../actpred/dataset.py#L92-L95)） | + +--- + +## 六、本轮新增代码审查（归一化+逐列加权方案） + +### 6.1 action_norm.py — 通过，有 1 处 P2 建议 + +**总体评价**：常量表清晰、文档完善、clamp 防御到位。18 个单测覆盖正常值/边界/越界/多维度。 + +**P2-a1（建议）**：`default_column_weights` 的 `action_dim` 默认为 18，但与 `action_bounds` 的行为不同——`action_bounds` 对 `action_dim > 18` 自动补零，`default_column_weights` 对此类情况返回全零权重（因为 `min(10, action_dim)` 和 `min(12, action_dim)` 仅覆盖前 12 列，其余列保持 `np.zeros` 初始值）。建议对齐行为：`action_dim > 18` 时对超出列也赋默认权重（如 0.2）。 + +### 6.2 config.py — 通过，有 1 处 P2 + +**总体评价**：`normalize_mode` 和 `column_weights` 的 CLI 解析、值域校验、checkpoint 兼容性警告均实现完整。 + +**P2-a2（建议）**：`warn_init_from_mismatch` 中 `init_mode = init_data.get("normalize_mode", "zscore")` 对旧 checkpoint 正确回退。但旧 checkpoint 的 `normalize_actions=False` 时，`normalize_mode` 无意义，此时警告 `init_mode != current.data.normalize_mode` 是误报。建议仅在 `init_norm and current.data.normalize_actions` 时比较 `normalize_mode`。 + +### 6.3 dataset.py — 通过 + +固定范围归一化集成正确：`_fixed_range` 标志位在 `__init__` 中设置，`_norm` 方法按标志位分发。stats shape 校验（R2-2 修复）已补。 + +### 6.4 train.py — 通过，有 1 处 P2 行为变更 + +**总体评价**：`_weighted_mse` 实现正确，`weights.mean()` 归一化保证加权 Loss 与无权重 Loss 量级可比。`column_weights` 为 None 时自动取 `default_column_weights`。 + +**P2-a3（行为变更）**：`column_weights` 现在**始终生效**（None 时取默认值），即使 `normalize_actions=False`。这意味着不归一化训练的模型也会按列加权损失——模型对舵机列（权重 1.0）的优化力度高于 spk 列（0.2）。这并非 bug，但行为变更需在 README 注明。 + +### 6.5 evaluate.py — 通过 + +物理界 OOB 与 `action_norm.py` 常量表共用（`_action_bounds` 调用 `action_bounds`）。error_growth 按 episode 分别计算取中位数，消除长短 episode 混杂偏差。实现正确。 + +### 6.6 推理/部署链 — 通过，P0-1 除外 + +| 文件 | 审查结论 | +|------|---------| +| predict.py | `_fixed_range` 从 checkpoint config 读取，normalize/denormalize 正确 | +| onnx_infer.py | `normalize_mode` 构造参数，predict 中 normalize/denormalize 正确 | +| hailo_infer.py | `normalize_mode` 构造参数，predict 中 normalize/denormalize 正确 | +| gen_calibration.py | `fixed_range` 模式校准数据走 `normalize_action` 变换，与推理端一致 | +| export_deploy.py | `check_token_ranges` 按 `normalize_mode` 在正确空间采样，stats_path 逻辑正确 | + +**唯独 pybot 的 `_load_predictor` 未传 `normalize_mode` 给 HailoPredictor（P0-1）**。 + +### 6.7 test_actpred.py — 通过 + +新增 `TestFixedRangeNorm` 类 18 个测试用例，覆盖： +- 已知值归一化（舵机中位/边界、spk 中位） +- 越界 clamp（>180→1.0、<0→0.0） +- denormalize 往返一致性 +- 默认列权重结构 +- 自定义列权重 +- 多维度批次归一化 +- 全 0 输出验证 + +全量 148 passed / 11 skipped。 + +--- + +## 七、修复优先级 + +| 优先级 | 项 | 理由 | +|--------|-----|------| +| **P0** | P0-1 HailoPredictor normalize_mode 缺失 | `fixed_range` HEF 在 pybot 闭环中推理完全错误 | +| **P0** | P0-2 normalize_actions 默认值变更 | 所有无此参数的命令行为改变，README 不一致 | +| P1 | R2-1 NaN val_loss | 训练健壮性，一行 isfinite 修复 | +| P1 | P2-a 后端检测 | 路径大小写敏感，ONNX 无分支 | +| P2 | P2-a2 误报警告 | 旧 checkpoint 的不必要警告 | +| P2 | P2-a1 action_dim>18 | 未来扩展性 | +| P2 | P2-a3 行为变更文档 | README 注明 column_weights 始终生效 | + +--- + +*本报告为归一化+逐列加权方案落地后的专项审计。建议先修复 2 个 P0 再提交。* \ No newline at end of file diff --git a/pybot/pybot/main.py b/pybot/pybot/main.py index 8cde321..82771d8 100644 --- a/pybot/pybot/main.py +++ b/pybot/pybot/main.py @@ -141,7 +141,8 @@ def _load_predictor(checkpoint): """ import os try: - if checkpoint.endswith(".hef"): + lower = checkpoint.lower() + if lower.endswith(".hef"): from actpred.hailo_infer import HailoPredictor as _P hef_dir = os.path.dirname(os.path.abspath(checkpoint)) @@ -160,6 +161,7 @@ def _load_predictor(checkpoint): # predict_delta：从同目录 checkpoint 读取训练配置 predict_delta = False + normalize_mode = "zscore" for name in ("pretrain_best_inference.pt", "best_inference.pt"): ckpt_path = os.path.join(hef_dir, name) @@ -171,9 +173,15 @@ def _load_predictor(checkpoint): cfg = ck.get("config", {}) predict_delta = cfg.get("train", {}).get( "predict_delta", False) + normalize_mode = cfg.get("data", {}).get( + "normalize_mode", "zscore") if predict_delta: logging.info( "predict_delta=True from %s", ckpt_path) + if normalize_mode != "zscore": + logging.info( + "normalize_mode=%s from %s", + normalize_mode, ckpt_path) except Exception: pass break @@ -181,7 +189,8 @@ def _load_predictor(checkpoint): try: return _P(checkpoint, stats_path=stats_path, predict_delta=predict_delta, - pred_token=pred_token) + pred_token=pred_token, + normalize_mode=normalize_mode) except Exception: logging.exception("加载 HEF 推理器失败: %s", checkpoint) return None diff --git a/pybot/pybot/robot.py b/pybot/pybot/robot.py index cd7845d..e339ac6 100644 --- a/pybot/pybot/robot.py +++ b/pybot/pybot/robot.py @@ -14,6 +14,8 @@ import logging import time from collections import deque +import numpy as np + from . import plugin from .hardvars import (AliasVar, CameraVar, I2CVar, MPU6050Var, ServoVar, SpeakerVar) @@ -238,7 +240,6 @@ class RobotStruct: （会阻塞 ~1/framerate 秒，把 hz 锁死在相机帧率之下）。dry_run 或队列空时回退 _grab_frame 合成帧。 """ - import numpy as np camera_var = self._find_camera() if camera_var is None and not self.dry_run: raise ValueError("闭环需要 @Camera 块（或使用 --dry-run 生成合成帧）") @@ -357,7 +358,6 @@ class RobotStruct: 帧源与主循环一致（_loop_recent_frames），避免冷启动与运行期帧源 不一致。history 用零向量填满是保守初值，避免随机姿态跳变损坏舵机。 """ - import numpy as np frames.extend(self._loop_recent_frames(camera_var, frames.maxlen)) for _ in range(history.maxlen): history.append(np.zeros(action_dim, dtype=np.float32)) @@ -368,7 +368,6 @@ class RobotStruct: SpeakerVar.play 强制要求 delay 键；闭环已剔除 delay 列（避免发声 阻塞卡死节拍），此处为其补 delay=0 —— 发声但不阻塞，节拍由 hz 掌控。 """ - import numpy as np if not np.all(np.isfinite(act)): log.warning("action vector contains NaN/Inf, skipping step") return @@ -442,7 +441,6 @@ class RobotStruct: """ if not imu_vars: return None - import numpy as np nk = len(IMU_KEYS) out = np.zeros(nk * len(imu_vars), dtype=np.float32) for k, var in enumerate(imu_vars): @@ -458,7 +456,6 @@ class RobotStruct: @staticmethod def _pad_imu(vec, imu_dim): """拼接姿态向量 → pad/truncate 到 predictor 的 imu_dim（None → 全零）""" - import numpy as np out = np.zeros(imu_dim, dtype=np.float32) if vec is not None: src = np.asarray(vec, dtype=np.float32).ravel() @@ -468,7 +465,6 @@ class RobotStruct: def _build_action_vec(self, vals, action_dim): """从值列表构建动作向量，pad/truncate 到 action_dim""" - import numpy as np vec = np.zeros(action_dim, dtype=np.float32) n = min(len(vals), action_dim) vec[:n] = vals[:n] @@ -476,7 +472,6 @@ class RobotStruct: def _grab_frame(self, camera_var): """从摄像头同步抓一帧；dry_run 或无帧时返回随机噪声帧""" - import numpy as np if camera_var is not None: frame = camera_var.camera.capture() if frame is not None: