from typing import TYPE_CHECKING from plugins_func.register import register_function, ToolType, ActionResponse, Action if TYPE_CHECKING: from core.connection import ConnectionHandler start_repeater_function_desc = { "type": "function", "function": { "name": "start_repeater", "description": "在设备 SD 卡上复读指定 MP3。name 必须与 self.mp3.get_list 返回的 files[].name 完全一致。repeat 是整首循环次数。不要用 play_music。", "parameters": { "type": "object", "properties": { "name": { "type": "string", "description": "SD 卡 MP3 文件名,必须与 get_list 的 files[].name 一致", }, "repeat": { "type": "integer", "description": "整首循环次数,默认 1", }, "power_off_after": { "type": "boolean", "description": "播完后是否软关机,默认 false", }, }, "required": ["name"], }, }, } @register_function("start_repeater", start_repeater_function_desc, ToolType.SYSTEM_CTL) async def start_repeater( conn: "ConnectionHandler", name: str, repeat: int = 1, power_off_after: bool = False, ): from core.app_runtime.runtime import start_app_task try: repeat_n = int(repeat or 1) except (TypeError, ValueError): repeat_n = 1 if repeat_n < 1: repeat_n = 1 result = await start_app_task( conn, "english_learning", "repeater", { "name": name, "repeat": repeat_n, "power_off_after": bool(power_off_after), }, source="voice", ) if result.action in (Action.ERROR, Action.RESPONSE): return ActionResponse( action=Action.RESPONSE, result=result.result, response=result.response or "无法开始复读", ) return ActionResponse( action=Action.NONE, result=result.result, response=result.response, )