mock_skill.py

# -*- coding: utf-8 -*-
"""Mock skill events publisher：模擬 Skill Manager 發佈 skill_events。

用法：
    python -m tools.mock_skill G1_left_grasp ok --task-id <task_id>
    python -m tools.mock_skill O2_left_rotate_package fail --fault SK-005

對應 interfaces.md §9.2 / §9.3。
"""
from __future__ import annotations

import argparse
import json
import sys
import time

import redis


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Mock skill_events publisher")
    p.add_argument("skill_name", help="例如 G1_left_grasp / O2_left_rotate_package")
    p.add_argument(
        "status",
        choices=["ok", "fail", "timeout", "cancelled"],
        default="ok",
        nargs="?",
    )
    p.add_argument("--phase", default="end", choices=["begin", "progress", "end"])
    p.add_argument("--robot", dest="robot_name", default="humanoid_01")
    p.add_argument("--task-id", default="mock_task")
    p.add_argument("--subtask-id", default="t1")
    p.add_argument("--fault", dest="fault_code", default=None)
    p.add_argument("--duration-ms", type=int, default=1000)
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=6379)
    p.add_argument("--db", type=int, default=0)
    p.add_argument("--password", default=None)
    p.add_argument("--channel", default="skill_events")
    return p.parse_args()


def main() -> None:
    args = parse_args()
    r = redis.Redis(
        host=args.host, port=args.port, db=args.db, password=args.password
    )
    fault_code = args.fault_code
    if args.status != "ok" and not fault_code:
        fault_code = "SK-001"
    payload = {
        "event": "skill_update",
        "ts": time.time(),
        "robot_name": args.robot_name,
        "task_id": args.task_id,
        "subtask_id": args.subtask_id,
        "skill_name": args.skill_name,
        "phase": args.phase,
        "status": args.status,
        "fault_code": fault_code,
        "details": {"duration_ms": args.duration_ms},
    }
    r.publish(args.channel, json.dumps(payload))
    print(f"[mock_skill] sent: {json.dumps(payload, ensure_ascii=False)}")


if __name__ == "__main__":
    sys.exit(main())