《社交能量补给站》极简源码精讲

"""社交能量补给站：基于官方示例的 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()