《启元百变互动官》核心源码精讲

"""启元百变互动官（Q1 / T1 通用全能技能模板，纯 Python 3.10 标准库）。

设计目标：
1. 开箱即用：可直接打包部署到 Q1 或 T1 真机运行，完成「迎宾 → 每日金句(联网+断网兜底) → 随机抽签互动 → 谢幕」完整闭环。
2. 极易改编：只需修改顶部的配置区（台词、动作别名、抽签池、是否开启联网/灯光），即可快速改成迎宾、算命、播报、小剧场等任意技能。
3. 严守契约：不导入 robot / ROS2，内蔵 TTS 0.7 秒防抖轮询与超时保护，自动适配 Q1 upper_body 与 T1 双足/四足动作白名单。
"""

import json
import random
import time
import urllib.request

# ==============================================================================
# 【第一部分：积木配置区 —— 改这里就能变成你的新技能！】
# ==============================================================================

# 构建脚本 build.py 会在打包时自动将此值替换为 "q1" 或 "t1"
ROBOT_MODEL = "q1"

# 功能开关
ENABLE_NET_QUOTE = False         # True=尝试联网拉取每日一言（失败自动用离线兜底）；False=纯离线最稳模式
ENABLE_RANDOM_DRAW = True        # True=从 LUCKY_POOL 随机抽签；False=固定使用第一支签
ENABLE_MOOD_EFFECTS = True       # True=尝试联动呼吸灯带与面部表情（自带安全容错）

# 联网 API 配置与断网兜底金句
QUOTE_API_URL = "https://v1.hitokoto.cn/?c=i&encode=json"
FALLBACK_QUOTE = "代码写得稳，真机跑得准，每一次动手都是超酷的创造！"

# 随机互动签文池：(签文台词, 动作语义名)
LUCKY_POOL = (
    ("叮！为你抽到「真机丝滑上上签」！今天你的所有动作指令全部一次通过！", "cheer"),
    ("叮！为你抽到「脑洞大开幸运签」！你的创意全场最亮眼，评委看了都点赞！", "heart"),
    ("叮！为你抽到「零Bug欧皇签」！编译不报错，网络不超时，好运直接拉满！", "greet"),
)

# 主流程固定剧本：(动作语义名, 台词文本, 备注说明)
# 支持占位符：{quote} 会自动填入联网/兜底金句，{lucky_text} 会填入抽签结果
OPENING_LINE = "哈喽大家好！我是启元百变互动官，很高兴在活动现场见到你！"
CLOSING_LINE = "今日互动圆满完成！祝你玩得开心，作品大放异彩，我们下次见！"

# 语义化动作映射表：把好记的名字映射到 Q1 / T1 的安全动作 ID 和保持秒数 (动作ID, 等待秒数)
# - Q1 严格只用室内模式允许的 upper_body 动作（3000 系）
# - T1 四足/双足均默认使用四足白名单兼容动作（26 喜悦 / 43 律动），也可在 biped 下扩展
ACTION_TABLE = {
    "q1": {
        "greet": (3001, 6.5),   # 招手
        "cheer": (3008, 6.5),   # 加油
        "heart": (3007, 6.5),   # 双手比心
    },
    "t1_quadruped": {
        "greet": (26, 16.0),    # 喜悦（四足安全白名单）
        "cheer": (43, 16.0),    # 律动（四足安全白名单）
        "heart": (26, 16.0),    # 喜悦（四足安全白名单）
    },
    "t1_biped": {
        "greet": (26, 16.0),    # 喜悦（双足同样安全可用）
        "cheer": (43, 16.0),    # 律动
        "heart": (26, 16.0),    # 喜悦
    },
}

POLL_SECONDS = 0.1
IDLE_SECONDS = 0.7


# ==============================================================================
# 【第二部分：安全硬件底座 —— 防吞字轮询、形态校验、网络兜底（一般无需修改）】
# ==============================================================================

def audio_state() -> int:
    """读取并校验音频状态：1=播放中，2=空闲。"""
    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:
    """说话并等待真实播完（含 4 秒起声监控 + 0.7 秒防抖滤波 + 超时保护）。"""
    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 trigger_mood(color: str = "#FF6B2B", mode: int = 3) -> None:
    """可选的灯带氛围触发（带安全容错，不影响主流程与离线替身测试）。"""
    if not ENABLE_MOOD_EFFECTS:
        return
    play_light = getattr(robot, "play_light", None)
    if callable(play_light):
        try:
            play_light(color, mode)
        except Exception as exc:
            print("LIGHT_SKIP " + str(exc), flush=True)


def fetch_quote() -> str:
    """获取今日金句：若开启联网则请求 API，超时或断网时 100% 自动回退到离线金句。"""
    if not ENABLE_NET_QUOTE:
        return FALLBACK_QUOTE
    try:
        req = urllib.request.Request(QUOTE_API_URL, headers={"User-Agent": "PrimeBot-Skill/1.0"})
        with urllib.request.urlopen(req, timeout=2.5) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            hitokoto = data.get("hitokoto")
            if isinstance(hitokoto, str) and hitokoto.strip():
                return hitokoto.strip()
    except Exception as exc:
        print("NET_FALLBACK " + str(exc), flush=True)
    return FALLBACK_QUOTE


def resolve_action_profile() -> dict:
    """根据机型（Q1 / T1）与实时形态（biped / quadruped）选择安全动作表。"""
    if ROBOT_MODEL == "q1":
        return ACTION_TABLE["q1"]
    if ROBOT_MODEL == "t1":
        form = robot.get_t1_robot_form()
        if form not in ("biped", "quadruped"):
            raise RuntimeError("T1_FORM_UNKNOWN")
        print("T1_FORM " + form, flush=True)
        return ACTION_TABLE["t1_" + form]
    raise RuntimeError("ROBOT_MODEL_UNKNOWN")


# ==============================================================================
# 【第三部分：主演出流程】
# ==============================================================================

def main() -> None:
    profile = resolve_action_profile()
    print("UNIVERSAL_SKILL_START model=" + ROBOT_MODEL, flush=True)
    trigger_mood("#FF6B2B", 3)

    # 1. 开场问候
    speak(OPENING_LINE)

    # 2. 组装三幕互动内容（迎宾问候 → 今日灵感金句 → 幸运抽签）
    quote_text = fetch_quote()
    lucky_line, lucky_action = (
        random.choice(LUCKY_POOL) if ENABLE_RANDOM_DRAW else LUCKY_POOL[0]
    )
    steps = (
        ("greet", "第一幕，元气问候。先挥个手，给你打满今日份的开发者Buff！"),
        ("cheer", "第二幕，灵感广播。送你一句话：" + quote_text),
        (lucky_action, "第三幕，好运盲盒。" + lucky_line),
    )

    # 3. 逐幕执行：先说话播完 → 再触发安全动作 → 等待动作完成
    for index, (action_key, line) in enumerate(steps, 1):
        speak(line)
        animation_id, hold = profile[action_key]
        accepted = robot.play_animation(animation_id, interrupt=False)
        print("ANIMATION step=%d key=%s id=%d accepted=%s" % (index, action_key, animation_id, accepted), flush=True)
        if not accepted:
            raise RuntimeError("ANIMATION_REJECTED id=%d" % animation_id)
        time.sleep(hold)

    # 4. 收场谢幕与完成标记
    speak(CLOSING_LINE)
    print("UNIVERSAL_SKILL_DONE steps=%d" % len(steps), flush=True)


# 机器人运行时在注入全局 robot 对象后直接执行
main()