别被机器人吓到！从 0 到 1 玩转启元 Q1 & T1 技能整活

"""社交能量补给站：基于官方示例的 TTS 轮询与动作编排，纯标准库。"""

import time

ROBOT_MODEL = "q1"
POLL_SECONDS = 0.1
IDLE_SECONDS = 0.7
STEPS = (
    "第一格，勇气充电。今天你已经迈出了第一步，给自己加个油！",
    "第二格，灵感充电。先让一个小点子跑起来，再慢慢把它变精彩！",
    "第三格，快乐充电。请收下这份机械式认真，和百分之百的祝福！",
)
# 等待时长为保守初值，现场按动作权威表、任务日志与实际观感调整。
MOTIONS = {
    "q1": ((3001, 7.0), (3008, 7.0), (3007, 7.0)),
    "t1": ((26, 18.0), (43, 18.0), (26, 18.0)),
}


def audio_state() -> int:
    snapshot = robot.get_audio_play_state()
    if isinstance(snapshot, dict):
        state = snapshot.get("state")
        if isinstance(state, dict) and state.get("value") in (1, 2):
            return state["value"]
    raise RuntimeError("AUDIO_STATE_INVALID")


def speak(text: str) -> None:
    """必须观测到起声及稳定空闲；超时终止，避免下一句截断当前音频。"""
    print("SPEAK " + text, flush=True)
    robot.play_tts(text)
    start_deadline = time.monotonic() + 4.0
    while audio_state() != 1:
        if time.monotonic() >= start_deadline:
            raise RuntimeError("TTS_NOT_STARTED")
        time.sleep(POLL_SECONDS)

    finish_deadline = time.monotonic() + max(len(text) * 0.45, 4.0) + 8.0
    idle_since = None
    while time.monotonic() < finish_deadline:
        if audio_state() == 1:
            idle_since = None
        elif idle_since is None:
            idle_since = time.monotonic()
        elif time.monotonic() - idle_since >= IDLE_SECONDS:
            return
        time.sleep(POLL_SECONDS)
    raise RuntimeError("TTS_NOT_FINISHED")


def choreography() -> tuple:
    if ROBOT_MODEL == "t1":
        form = robot.get_t1_robot_form()
        if form not in ("biped", "quadruped"):
            raise RuntimeError("T1_FORM_UNKNOWN")
        # 两种形态均使用 26/43，属于手册所列的四足允许集合。
        print("T1_FORM " + form, flush=True)
    return MOTIONS[ROBOT_MODEL]


def main() -> None:
    motions = choreography()
    print("ENERGY_STATION_START model=" + ROBOT_MODEL, flush=True)
    speak("欢迎来到社交能量补给站！不用扫码，送你三格好心情。")
    for index, (text, (animation_id, hold)) in enumerate(zip(STEPS, motions), 1):
        speak(text)
        accepted = robot.play_animation(animation_id, interrupt=False)
        print("ANIMATION step=%d id=%d accepted=%s" % (index, animation_id, accepted), flush=True)
        if not accepted:
            raise RuntimeError("ANIMATION_REJECTED id=%d" % animation_id)
        time.sleep(hold)
    speak("充电完成！愿你带着自己的点子出发。掌声送给每一个敢动手的人！")
    print("ENERGY_STATION_DONE steps=%d" % len(STEPS), flush=True)


# 与官方示例一样，在运行时 exec 注入 robot 后直接执行。
main()

"""从 Example/ 下的技能源码构建 Q1/T1 双机型 .rpk 包及可审阅的发布元数据。"""

import importlib.util
import json
import shutil
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
WORKSPACE = ROOT.parent
DEV_KIT = ROOT / "primehub_dev_kit"
EXAMPLE_ROOT = WORKSPACE / "Example"
OUTPUT = ROOT / "dist"

SKILL_SPECS = (
    {
        "folder": "energy_station",
        "pkg_prefix": "cans518_energy_station",
        "publish_prefix": "publish",
        "title": "社交能量补给站",
        "short_desc": "三格好心情，给每个敢动手的人充个电",
        "long_desc_tpl": (
            "离线小剧场：开场、勇气/灵感/快乐三格充电、谢幕。"
            "每格短台词配动作，无需联网或语音输入。"
            "仅适配%s，使用预设动作，运行时请留出周围1米空间。"
            "基于官方示例改编；台词和编排可在源码中修改。"
        ),
        "icon": "skill-game.svg",
    },
    {
        "folder": "universal_skill",
        "pkg_prefix": "cans518_universal_skill",
        "publish_prefix": "publish_universal",
        "title": "启元百变互动官",
        "short_desc": "迎宾问候+灵感广播+好运盲盒抽签的通用互动剧场",
        "long_desc_tpl": (
            "Q1/T1通用互动技能模板：包含元气迎宾、每日灵感金句（支持联网API与断网自动兜底）、"
            "随机好运抽签及谢幕。仅适配%s，内置形态检测与TTS防抖轮询，运行时请留出周围1米空间。"
        ),
        "icon": "skill-verify.svg",
    },
)


def build_skill(pack_module, spec_cfg: dict) -> None:
    source_dir = EXAMPLE_ROOT / spec_cfg["folder"]
    for model, model_id, model_code in (("q1", 100, "robot_q"), ("t1", 101, "robot_t")):
        with tempfile.TemporaryDirectory(prefix="qiyuan_build_") as staging:
            skill_dir = Path(staging) / (spec_cfg["folder"] + "_" + model)
            shutil.copytree(source_dir, skill_dir, ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
            script = skill_dir / "src/main.py"
            source = script.read_text(encoding="utf-8")
            assignment = 'ROBOT_MODEL = "q1"'
            if source.count(assignment) != 1:
                raise ValueError("期望恰好一个 ROBOT_MODEL 声明: " + str(script))
            script.write_text(source.replace(assignment, 'ROBOT_MODEL = "%s"' % model), encoding="utf-8")
            config = skill_dir / "cfg/pkg.yaml"
            package_name = "%s_%s_20261024" % (spec_cfg["pkg_prefix"], model)
            lines = []
            for line in config.read_text(encoding="utf-8").splitlines():
                if line.startswith("package_name:"):
                    line = "package_name: " + package_name
                elif line.startswith("robot_model:"):
                    line = "robot_model: " + model
                lines.append(line)
            config.write_text("\n".join(lines) + "\n", encoding="utf-8")
            pack_module.pack(skill_dir, OUTPUT)
        metadata = {
            "skillName": spec_cfg["title"] + " " + model.upper(),
            "version": "0.1.0",
            "skillDesc": spec_cfg["short_desc"],
            "long_description": spec_cfg["long_desc_tpl"] % model.upper(),
            "categoryIds": [104],
            "categoryCodeList": ["action"],
            "tagsIds": [127],
            "tagsCodeList": ["companion"],
            "robotModelIds": [model_id],
            "robotModelCodeList": [model_code],
            "iconUrl": spec_cfg["icon"],
        }
        publish_file = "%s_%s.json" % (spec_cfg["publish_prefix"], model)
        (OUTPUT / publish_file).write_text(
            json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )


def build() -> None:
    spec = importlib.util.spec_from_file_location("primehub_pack", DEV_KIT / "pack.py")
    pack_module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(pack_module)
    OUTPUT.mkdir(exist_ok=True)
    for spec_cfg in SKILL_SPECS:
        build_skill(pack_module, spec_cfg)


if __name__ == "__main__":
    build()