真机一键侦察、抓动作表与点火器源码精讲

#!/usr/bin/env python3
"""真机 SSH 环境探测、机上 YAML 动作表拉取、技能包上传与一键运行脚本（纯 Python 跨平台版）。"""

import argparse
import datetime
import shutil
import subprocess
import sys
from pathlib import Path

TOOLS_ROOT = Path(__file__).resolve().parent.parent
WORKSPACE_ROOT = TOOLS_ROOT.parent
DEPLOY_SCRIPT = TOOLS_ROOT / "primehub_dev_kit/deploy_test.py"
DIST_ROOT = TOOLS_ROOT / "dist"
REPORTS_DIR = WORKSPACE_ROOT / "Doc/reports"


def run_remote(target: str, cmd: str) -> str:
    res = subprocess.run(
        ["ssh", target, cmd],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    if res.returncode != 0:
        raise RuntimeError(f"远程命令执行失败：{cmd}\n{res.stdout}\n{res.stderr}")
    return (res.stdout + res.stderr).strip()


def main() -> int:
    parser = argparse.ArgumentParser(description="启元 Q1 / T1 真机探测、传包与技能运行工具")
    parser.add_argument("--model", required=True, choices=["q1", "t1"], help="机器人型号：q1 或 t1")
    parser.add_argument("--ip", required=True, help="机器人局域网 IP，例如 192.168.112.222")
    parser.add_argument(
        "--skill",
        default="universal_skill",
        choices=["universal_skill", "energy_station"],
        help="要上传/运行的技能名称（默认 universal_skill）",
    )
    parser.add_argument("--run", action="store_true", help="上传后立即在真机上启动技能并抓取硬件日志")
    args = parser.parse_args()

    for cmd in ("ssh", "scp"):
        if not shutil.which(cmd):
            raise SystemExit(f"找不到 {cmd} 命令，请先确认系统已安装 OpenSSH 客户端。")

    pkg_path = DIST_ROOT / f"{args.skill}_{args.model}.rpk"
    if not pkg_path.exists():
        raise SystemExit(f"找不到技能包 {pkg_path}，请先运行 python Tools/scripts/build.py 构建。")

    REPORTS_DIR.mkdir(parents=True, exist_ok=True)
    ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = REPORTS_DIR / f"robot_probe_{args.model}_{ts}.txt"
    target = f"run@{args.ip}"
    remote_pkg = f"/tmp/{args.skill}_{args.model}.rpk"

    lines = [
        f"# 真机探测记录 {datetime.datetime.now().isoformat(timespec='seconds')}",
        f"型号: {args.model.upper()}",
        f"机器人: {args.ip}",
        f"技能包: {pkg_path.name}",
    ]

    def record(title: str, cmd: str) -> None:
        print(f"== {title} ==")
        out = run_remote(target, cmd)
        lines.extend([f"\n## {title}", f"> {cmd}", out])

    record("连接探测", "echo ROBOT_SSH_OK")
    record(
        "运行环境与残留任务",
        "python3 --version; curl -fsS --max-time 5 http://127.0.0.1:58080/packages; echo; "
        "curl -fsS --max-time 5 http://127.0.0.1:58080/tasks/running",
    )

    if args.model == "q1":
        record("Q1 upper_body 机上动作表", "cat /robot/software/task_engine/bin/cfg/motions/q1/upper_body.yaml")
    else:
        record("T1 机上动作表", "cat /robot/software/task_engine/bin/cfg/motions/t1/*.yaml")

    print(f"== 上传调试工具与 {pkg_path.name} ==")
    subprocess.run(["scp", str(DEPLOY_SCRIPT), f"{target}:/tmp/deploy_test.py"], check=True)
    subprocess.run(["scp", str(pkg_path), f"{target}:{remote_pkg}"], check=True)
    lines.extend(["\n## 已上传文件", "deploy_test.py -> /tmp/deploy_test.py", f"{pkg_path.name} -> {remote_pkg}"])

    if args.run:
        record("真机技能运行结果", f"python3 /tmp/deploy_test.py {remote_pkg} 120")
        record(
            "硬件动作日志核验",
            'grep "Animation task成功" /robot/persist/log/task_engine/task_engine.log | tail -12',
        )
    else:
        print("探测与传包已完成。核对动作表与安全环境后，追加 --run 即可启动真机运行。")

    report_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"探测日志已归档：{report_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())