"""CLI 入口 —— 移植自 bot4cj/src/main.cj 用法（与仓颉版管道方式兼容）： cat scripts/std.mt | python3 -m pybot # stdin 管道 python3 -m pybot scripts/std.mt # 位置参数（增强） python3 -m pybot --dry-run scripts/std.mt # Mock 总线，动作序列到 stdout python3 -m pybot --test roundtrip in.mt out.mt # 解析↔序列化往返 与仓颉版的差异（见方案 §4.7）： --test nn/cnn/cnn1 移除（NN 由 actpred 承担）； --test matrix_o/matrix_2 合并为 --test roundtrip（路径改为参数）； 新增 --dry-run / --verbose；-i / -o 保留参数位（reserved，对接 actpred）。 日志规范（S02 根治）：全部诊断走 logging(stderr)，stdout 保持纯净。 """ import argparse import logging import sys from .hd import i2c, spi from .parser import MatrixParser, MatrixSaver from .robot import RobotStruct def _build_argparser(): p = argparse.ArgumentParser( prog="pybot", description="机器人 .mt 动作脚本解析器（bot4cj 的 Python 迁移版）") p.add_argument("script", nargs="?", default=None, help=".mt 脚本路径（缺省从 stdin 读取）") p.add_argument("--test", nargs=3, metavar=("MODE", "IN", "OUT"), help="自检：roundtrip 解析↔序列化往返") p.add_argument("--dry-run", action="store_true", help="Mock 总线运行（不碰硬件），动作序列输出到 stdout") p.add_argument("--verbose", action="store_true", help="DEBUG 级日志（stderr）") p.add_argument("-i", metavar="FIFO", default=None, help="(reserved) 传感器数据输出管道，对接 actpred") p.add_argument("-o", metavar="FIFO", default=None, help="(reserved) 动作指令输入管道，对接 actpred") p.add_argument("--record", metavar="DIR", default=None, help="录制 actpred 格式 episode 到 DIR（需要 @Camera 块；" "--dry-run 下用合成帧）") p.add_argument("--action-dim", type=int, default=18, help="动作向量维度（默认 18，与 actpred 一致）") p.add_argument("--loop", metavar="CKPT", default=None, help="实时预测闭环：加载 actpred checkpoint，" "采样→推理→执行→回写（不执行脚本 @Action 行，" "仅用其表头做动作向量索引映射）") p.add_argument("--hz", type=float, default=10.0, help="闭环控制频率 Hz（默认 10，须 ≥ 推理+执行耗时）") p.add_argument("--loop-steps", type=int, default=None, metavar="N", help="闭环步数上限（默认无限，调试用）") return p def _roundtrip(in_path, out_path): """解析 in.mt 再序列化到 out.mt（对应仓颉版 --test matrix_o/matrix_2）""" objects = [] with open(in_path, encoding="utf-8") as f: parser = MatrixParser(f) while True: obj = parser.parse() if obj is None: break objects.append(obj) with open(out_path, "w", encoding="utf-8") as f: MatrixSaver(f).save(objects) logging.info("roundtrip: %d blocks, %s -> %s", len(objects), in_path, out_path) return 0 def _dump_journal(*mocks): """--dry-run 结束后把 Mock 总线动作序列输出到 stdout（等价性验证工件）""" for mock in mocks: for entry in mock.journal: print(" ".join(str(x) for x in entry)) def main(argv=None): args = _build_argparser().parse_args(argv) logging.basicConfig( stream=sys.stderr, level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s %(name)s: %(message)s") if args.i or args.o: logging.warning("-i/-o are reserved for actpred integration, ignored") if args.test: mode, in_path, out_path = args.test if mode != "roundtrip": logging.error("unknown --test mode: %s (supported: roundtrip)", mode) return 2 return _roundtrip(in_path, out_path) mocks = () if args.dry_run: mocks = (i2c.use_mock(), spi.use_mock()) predictor = None if args.loop: predictor = _load_predictor(args.loop) if predictor is None: return 2 source = open(args.script, encoding="utf-8") if args.script else sys.stdin try: robot = RobotStruct(source, dry_run=args.dry_run) recorder = None if args.record: from .recorder import EpisodeRecorder recorder = EpisodeRecorder(args.record, action_dim=args.action_dim) with robot: if predictor is not None: robot.run_loop(predictor, hz=args.hz, max_steps=args.loop_steps, recorder=recorder) else: robot.run(recorder=recorder) finally: if args.script: source.close() if args.dry_run: _dump_journal(*mocks) return 0 def _load_predictor(checkpoint): """进程内构建 actpred 推理器；失败返回 None pybot 不硬依赖 actpred：仅 --loop 时才尝试 import，actpred 不在 sys.path 时给出可操作提示。checkpoint 后缀决定后端（.hef → Hailo NPU， .onnx → ONNX Runtime，否则 → PyTorch），推理类由 actpred 提供。 HEF 后端会自动查找同目录下的 pred_token.npy 和 companion checkpoint （pretrain_best_inference.pt / best_inference.pt），以恢复 pred_token 默认值和 predict_delta 训练配置——否则 Hailo 推理输出与训练语义不一致。 ONNX 后端的 normalize_mode/predict_delta 由模型元数据自动识别。 """ import os try: lower = checkpoint.lower() if lower.endswith(".onnx"): from actpred.onnx_infer import OnnxPredictor as _P onnx_dir = os.path.dirname(os.path.abspath(checkpoint)) # stats_path：zscore 归一化模型的 action_stats.npy stats_path = None sp = os.path.join(onnx_dir, "action_stats.npy") if os.path.exists(sp): stats_path = sp try: pred = _P(checkpoint, stats_path=stats_path) except Exception: logging.exception("加载 ONNX 推理器失败: %s", checkpoint) return None if pred.normalize_mode == "zscore" and stats_path is None: logging.warning("ONNX 元数据为 zscore 归一化但同目录无 " "action_stats.npy，动作历史将不归一化: %s", checkpoint) return pred if lower.endswith(".hef"): from actpred.hailo_infer import HailoPredictor as _P hef_dir = os.path.dirname(os.path.abspath(checkpoint)) # pred_token：export_deploy.py --pred-token-out 的产物 pred_token = None pt_path = os.path.join(hef_dir, "pred_token.npy") if os.path.exists(pt_path): pred_token = pt_path logging.info("found pred_token: %s", pt_path) # stats_path：归一化模型的 action_stats.npy stats_path = None sp = os.path.join(hef_dir, "action_stats.npy") if os.path.exists(sp): stats_path = sp # 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) if os.path.exists(ckpt_path): try: import torch ck = torch.load(ckpt_path, map_location="cpu", weights_only=True) 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 try: return _P(checkpoint, stats_path=stats_path, predict_delta=predict_delta, pred_token=pred_token, normalize_mode=normalize_mode) except Exception: logging.exception("加载 HEF 推理器失败: %s", checkpoint) return None else: from actpred.predict import ActionPredictorInference as _P except ImportError: logging.error("--loop 需要 actpred 模块（把 blackbird 根目录加入 " "PYTHONPATH，或在 actpred/ 同级运行）") return None try: return _P(checkpoint) except Exception: logging.exception("加载 actpred checkpoint 失败: %s", checkpoint) return None if __name__ == "__main__": sys.exit(main())