"""Unified start_app_task / switch_app / mode helpers."""
|
|
from __future__ import annotations
|
|
import asyncio
|
import json
|
import difflib
|
import queue
|
import random
|
import re
|
import uuid
|
from typing import Any, Dict, List, Optional, TYPE_CHECKING
|
|
from config.logger import setup_logging
|
from plugins_func.register import Action, ActionResponse
|
|
from .assessment import (
|
evaluate_turn,
|
listen_window_ms,
|
pcm_snapshot_to_wav,
|
resolve_expected_text,
|
resolve_listen_ratio,
|
save_turn_wav,
|
silent_pcm_snapshot,
|
transcript_cache_dir,
|
)
|
from .playback import (
|
clamp_play_speed,
|
default_repeater_playlist,
|
default_shadowing_playlist,
|
expand_playlist_plays,
|
lesson_day_index,
|
mark_lesson_played_today,
|
parse_playlist,
|
resolve_play_speed,
|
speeds_for_repeats,
|
)
|
from .registry import (
|
default_app_id,
|
get_app,
|
get_registry,
|
get_task,
|
match_registry_voice,
|
voice_menu_text,
|
)
|
from .session import begin_session, current_session, end_session, parse_skill_event
|
|
if TYPE_CHECKING:
|
from core.connection import ConnectionHandler
|
|
TAG = __name__
|
logger = setup_logging()
|
|
SOURCE_PRIORITY = {
|
"mqtt_immediate": 3,
|
"schedule": 2,
|
"voice": 1,
|
"default_chat": 0,
|
}
|
|
TRANSLATOR_PROMPT = (
|
"你是一台翻译机。只输出译文，不要解释、不要拼音课、不要追问、不要闲聊。"
|
"未指定语言时：中文译成英文，英文译成中文；其它语言默认译成中文。"
|
)
|
|
BILINGUAL_PROMPT = (
|
"你正在进行中英对照对话练习，不是翻译机，不要把用户的话原样译回去交差。"
|
"严格执行：用户说中文时你只用英文完整回答；用户说英文时你只用中文完整回答。"
|
"不要中英夹杂，不要同时给出两种语言，不要问要不要翻译。"
|
"用户说退出中英对话、回到聊天时调用 switch_app，app_id=chat。"
|
)
|
|
CHAT_PROACTIVE_HINT = (
|
"你是用户的语音伙伴。进入对话后可以先主动短句邀请聊天，例如「我们来聊天吧」。"
|
"用户开口后按正常对话回答，不要反复催促。"
|
)
|
|
CHAT_RETURN_TOPICS = ("今天过得怎么样", "你最近在忙什么", "有什么新鲜事")
|
|
DEFAULT_SHADOWING_SENTENCES = (
|
"Hello, how are you today?",
|
"I like to read English books.",
|
"The weather is nice this morning.",
|
"Can you say that again, please?",
|
"Practice a little every day.",
|
)
|
|
DEVICE_PLAY_TOOL = "self_mp3_play"
|
DEVICE_STOP_TOOL = "self_mp3_stop"
|
DEVICE_LIST_TOOL = "self_mp3_get_list"
|
DEVICE_ENSURE_APP_TOOL = "self_sd_ensure_app"
|
SD_LAYOUT_MODES = ("PLAY_AUDIO", "SHADOWING")
|
SHADOWING_LISTEN_RATIO = 1.2
|
SHADOWING_LISTEN_MIN_MS = 800
|
REPEATER_END_PHRASE = "复读结束"
|
SHADOWING_END_PHRASE = "跟读结束"
|
|
|
def source_priority(source: str) -> int:
|
return SOURCE_PRIORITY.get(source or "voice", 1)
|
|
|
def _ensure_runtime_fields(conn: "ConnectionHandler") -> None:
|
if not hasattr(conn, "active_app_id"):
|
conn.active_app_id = default_app_id()
|
if not hasattr(conn, "current_task_id"):
|
conn.current_task_id = "conversation"
|
if not hasattr(conn, "interaction_mode"):
|
conn.interaction_mode = "CHAT"
|
if not hasattr(conn, "current_source"):
|
conn.current_source = "default_chat"
|
if not hasattr(conn, "tool_policy"):
|
conn.tool_policy = "inherit_agent"
|
if not hasattr(conn, "tool_allowlist"):
|
conn.tool_allowlist = []
|
if not hasattr(conn, "original_system_prompt"):
|
conn.original_system_prompt = conn.prompt
|
if not hasattr(conn, "shadowing_index"):
|
conn.shadowing_index = 0
|
if not hasattr(conn, "shadowing_prompt"):
|
conn.shadowing_prompt = ""
|
if not hasattr(conn, "shadowing_clips"):
|
conn.shadowing_clips = []
|
if not hasattr(conn, "shadowing_awaiting_user"):
|
conn.shadowing_awaiting_user = False
|
if not hasattr(conn, "shadowing_clip_gen"):
|
conn.shadowing_clip_gen = 0
|
if not hasattr(conn, "shadowing_timeout_task"):
|
conn.shadowing_timeout_task = None
|
if not hasattr(conn, "last_asr_language"):
|
conn.last_asr_language = None
|
if not hasattr(conn, "task_session"):
|
conn.task_session = None
|
if not hasattr(conn, "last_asr_audio_snapshot"):
|
conn.last_asr_audio_snapshot = None
|
if not hasattr(conn, "shadowing_listen_pcm"):
|
conn.shadowing_listen_pcm = []
|
if not hasattr(conn, "shadowing_audio_saved_index"):
|
conn.shadowing_audio_saved_index = -1
|
if not hasattr(conn, "ensured_app_dirs"):
|
conn.ensured_app_dirs = set()
|
if not hasattr(conn, "repeater_speed_queue"):
|
conn.repeater_speed_queue = []
|
if not hasattr(conn, "repeater_play_name"):
|
conn.repeater_play_name = ""
|
if not hasattr(conn, "repeater_power_off_after"):
|
conn.repeater_power_off_after = False
|
if not hasattr(conn, "repeater_playlist"):
|
conn.repeater_playlist = []
|
if not hasattr(conn, "shadowing_last_speed"):
|
conn.shadowing_last_speed = 1.0
|
if not hasattr(conn, "shadowing_assess_index"):
|
conn.shadowing_assess_index = 0
|
if not hasattr(conn, "chat_greeted"):
|
conn.chat_greeted = False
|
if not hasattr(conn, "pending_play_after_tts"):
|
conn.pending_play_after_tts = None
|
|
|
def _task_needs_sd_layout(task: Optional[Dict[str, Any]]) -> bool:
|
if not task:
|
return False
|
mode = task.get("interaction_mode") or ""
|
if mode in SD_LAYOUT_MODES:
|
return True
|
for skill in task.get("skills") or []:
|
name = (skill or {}).get("name") or ""
|
if name in ("self.mp3.play", DEVICE_PLAY_TOOL):
|
return True
|
return False
|
|
|
def app_sd_layout_paths(app_id: str, app: Optional[Dict[str, Any]] = None) -> list:
|
"""Return SD lesson dirs for an App: /{app_id}/{task_id}/lessons."""
|
if not app_id:
|
return []
|
if app is None:
|
app = get_app(app_id)
|
if not app:
|
return []
|
paths = []
|
for task_id, task in (app.get("tasks") or {}).items():
|
if _task_needs_sd_layout(task):
|
paths.append(f"/{app_id}/{task_id}/lessons")
|
return paths
|
|
|
def _mcp_layout_ok(result) -> bool:
|
if result is None:
|
return False
|
if result.action in (Action.ERROR, Action.NOTFOUND):
|
return False
|
blob = str(getattr(result, "response", None) or getattr(result, "result", None) or "")
|
compact = blob.replace(" ", "").lower()
|
if '"ok":false' in compact:
|
return False
|
return True
|
|
|
async def ensure_app_sd_dirs(conn: "ConnectionHandler", app_id: str) -> bool:
|
"""Create /{app} and /{app}/{task}/lessons on device SD. Never raises; missing SD is ignored."""
|
_ensure_runtime_fields(conn)
|
app = get_app(app_id)
|
paths = app_sd_layout_paths(app_id, app)
|
if not paths:
|
return True
|
already = getattr(conn, "ensured_app_dirs", None)
|
if already is None:
|
conn.ensured_app_dirs = set()
|
already = conn.ensured_app_dirs
|
if app_id in already:
|
return True
|
try:
|
root = await _execute_named_tool(conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": ""})
|
if not _mcp_layout_ok(root):
|
logger.bind(tag=TAG).warning(f"创建设备 App 目录失败 {app_id}: {getattr(root, 'response', root)}")
|
return False
|
for task_id, task in (app.get("tasks") or {}).items():
|
if not _task_needs_sd_layout(task):
|
continue
|
result = await _execute_named_tool(
|
conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": task_id}
|
)
|
if not _mcp_layout_ok(result):
|
logger.bind(tag=TAG).warning(
|
f"创建设备 Task 目录失败 {app_id}/{task_id}: {getattr(result, 'response', result)}"
|
)
|
return False
|
already.add(app_id)
|
logger.bind(tag=TAG).info(f"已确保 SD 目录 {paths}")
|
return True
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"创建设备 App 目录异常 {app_id}: {e}")
|
return False
|
|
|
async def ensure_all_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None:
|
"""After MCP tools/list, mkdir for every registry App that needs SD. Failures do not drop WS."""
|
try:
|
apps = (get_registry().get("apps") or {})
|
for app_id, app in apps.items():
|
if app_sd_layout_paths(app_id, app):
|
await ensure_app_sd_dirs(conn, app_id)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"订阅 App 建目录失败（忽略）: {e}")
|
|
|
def schedule_ensure_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None:
|
"""Fire-and-forget so tools/list handling is not blocked waiting for MCP mkdir."""
|
|
async def _run():
|
await ensure_all_subscribed_app_sd_dirs(conn)
|
|
try:
|
loop = getattr(conn, "loop", None)
|
if loop:
|
loop.create_task(_run())
|
else:
|
asyncio.create_task(_run())
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"无法调度 App 建目录: {e}")
|
|
|
def filter_tools(conn: "ConnectionHandler", tools: Dict[str, Any]) -> Dict[str, Any]:
|
"""inherit_agent keeps the live agent toolset; allowlist keeps named tools only."""
|
_ensure_runtime_fields(conn)
|
if getattr(conn, "tool_policy", "inherit_agent") != "allowlist":
|
return tools
|
allow = set(getattr(conn, "tool_allowlist", None) or [])
|
if not allow:
|
return {}
|
filtered = {}
|
for name, definition in tools.items():
|
if name in allow:
|
filtered[name] = definition
|
return filtered
|
|
|
def _refresh_tools(conn: "ConnectionHandler") -> None:
|
handler = getattr(conn, "func_handler", None)
|
if handler and hasattr(handler, "upload_functions_desc"):
|
handler.upload_functions_desc()
|
|
|
def _apply_prompt(conn: "ConnectionHandler", app: Dict[str, Any]) -> None:
|
mode = app.get("system_prompt_mode") or "none"
|
if conn.original_system_prompt is None:
|
conn.original_system_prompt = conn.prompt
|
if mode == "translator":
|
conn.change_system_prompt(
|
"你现在是英语翻译机。用户说话后应调用 translate；用户说退出翻译、回到聊天时调用 switch_app，app_id=chat。"
|
"不要讲笑话、不要查天气、不要闲聊。"
|
)
|
elif mode == "bilingual":
|
conn.change_system_prompt(BILINGUAL_PROMPT)
|
elif mode == "agent_proactive":
|
base = conn.original_system_prompt or conn.prompt or ""
|
if CHAT_PROACTIVE_HINT not in (base or ""):
|
conn.change_system_prompt((base + "\n" + CHAT_PROACTIVE_HINT).strip())
|
else:
|
conn.change_system_prompt(base)
|
else:
|
if conn.original_system_prompt:
|
conn.change_system_prompt(conn.original_system_prompt)
|
|
|
async def _execute_named_tool(conn: "ConnectionHandler", tool_name: str, arguments: Dict[str, Any]):
|
handler = getattr(conn, "func_handler", None)
|
if not handler or not hasattr(handler, "tool_manager"):
|
return ActionResponse(action=Action.ERROR, response="工具处理器未就绪")
|
try:
|
return await handler.tool_manager.execute_tool(tool_name, arguments or {})
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"执行工具 {tool_name} 失败: {e}")
|
return ActionResponse(action=Action.ERROR, response=str(e))
|
|
|
async def _stop_current_skills(conn: "ConnectionHandler") -> None:
|
mode = getattr(conn, "interaction_mode", "CHAT")
|
_cancel_shadowing_timeout(conn)
|
if mode in ("PLAY_AUDIO", "SHADOWING"):
|
await _execute_named_tool(conn, DEVICE_STOP_TOOL, {})
|
conn.awaiting_voice_app_choice = False
|
conn.pending_play_after_tts = None
|
conn.shadowing_index = 0
|
conn.shadowing_prompt = ""
|
conn.shadowing_clips = []
|
conn.shadowing_awaiting_user = False
|
conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
|
conn.shadowing_assess_index = 0
|
conn.shadowing_listen_pcm = []
|
conn.shadowing_audio_saved_index = -1
|
conn.repeater_speed_queue = []
|
conn.repeater_play_name = ""
|
conn.repeater_power_off_after = False
|
conn.repeater_playlist = []
|
|
|
def tts_channels_ready(conn: "ConnectionHandler") -> bool:
|
tts = getattr(conn, "tts", None)
|
if tts is None:
|
return False
|
play = getattr(tts, "audio_play_priority_thread", None)
|
text_th = getattr(tts, "tts_priority_thread", None)
|
return bool(
|
play is not None
|
and play.is_alive()
|
and text_th is not None
|
and text_th.is_alive()
|
)
|
|
|
async def wait_for_tts_channels(conn: "ConnectionHandler", timeout: float = 8.0) -> bool:
|
loop = asyncio.get_running_loop()
|
start = loop.time()
|
while (loop.time() - start) < timeout:
|
if tts_channels_ready(conn):
|
return True
|
await asyncio.sleep(0.1)
|
return tts_channels_ready(conn)
|
|
|
def _speak(conn: "ConnectionHandler", text: str) -> None:
|
if not text:
|
return
|
tts = getattr(conn, "tts", None)
|
if tts is None:
|
logger.bind(tag=TAG).warning("TTS未初始化，跳过播报")
|
return
|
if not tts_channels_ready(conn):
|
logger.bind(tag=TAG).warning("TTS通道未就绪，跳过播报")
|
return
|
from core.handle.intentHandler import speak_txt
|
|
if not getattr(conn, "sentence_id", None):
|
conn.sentence_id = uuid.uuid4().hex
|
try:
|
speak_txt(conn, text)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"主动播报失败: {e}")
|
|
|
def _greeting_text(conn: "ConnectionHandler") -> str:
|
return (conn.config or {}).get("proactive_chat_greeting") or "我们来聊天吧。"
|
|
|
def _display_name(app: Optional[Dict[str, Any]], task: Optional[Dict[str, Any]] = None) -> str:
|
if task and task.get("display_name"):
|
return str(task["display_name"])
|
if app and app.get("display_name"):
|
return str(app["display_name"])
|
if app and app.get("name"):
|
return str(app["name"])
|
return "当前模式"
|
|
|
def _enter_prompt(app: Dict[str, Any], task: Dict[str, Any]) -> str:
|
return (task.get("enter_prompt") or app.get("enter_prompt") or "").strip()
|
|
|
def _switch_announcement(
|
prev_app: Optional[Dict[str, Any]],
|
prev_task: Optional[Dict[str, Any]],
|
app: Dict[str, Any],
|
task: Dict[str, Any],
|
app_id: str,
|
) -> str:
|
if app_id == "chat":
|
prev_name = _display_name(prev_app, prev_task)
|
topic = random.choice(CHAT_RETURN_TOPICS)
|
return f"已经退出{prev_name}，现在是对话模式，我们来聊点{topic}吧。"
|
return _enter_prompt(app, task) or f"现在已经切换至{_display_name(app, task)}。"
|
|
|
def _shadowing_sentences(conn: "ConnectionHandler") -> list:
|
cfg = (conn.config or {}).get("app_runtime") or {}
|
sentences = cfg.get("shadowing_sentences")
|
if not sentences:
|
_, task = get_task("english_learning", "shadowing")
|
sentences = (task or {}).get("sentences")
|
if not sentences:
|
sentences = DEFAULT_SHADOWING_SENTENCES
|
return [str(s).strip() for s in sentences if str(s).strip()]
|
|
|
def lesson_mp3_clips(names, app_id: str, task_id: str, lesson_id: str = "") -> list:
|
"""Pick sentence MP3s under /{app}/{task}/lessons/{lesson}/..."""
|
prefix = f"/{app_id}/{task_id}/lessons/"
|
clips = []
|
for raw in names or []:
|
name = str(raw or "").replace("\\", "/")
|
if not name.lower().endswith(".mp3"):
|
continue
|
if not name.startswith("/"):
|
name = "/" + name.lstrip("/")
|
if not name.startswith(prefix):
|
continue
|
rest = name[len(prefix) :]
|
if not rest:
|
continue
|
base = rest.rsplit("/", 1)[-1].lower()
|
if base == "full.mp3" or base.endswith("_full.mp3"):
|
continue
|
if lesson_id:
|
if rest != lesson_id and not rest.startswith(str(lesson_id) + "/"):
|
continue
|
clips.append(name)
|
clips.sort()
|
if lesson_id or not clips:
|
return clips
|
first_lesson = clips[0][len(prefix) :].split("/", 1)[0]
|
return [
|
name
|
for name in clips
|
if name[len(prefix) :] == first_lesson
|
or name[len(prefix) :].startswith(first_lesson + "/")
|
]
|
|
|
def repeater_lesson_mp3(names, app_id: str, task_id: str) -> str:
|
"""Pick the full-lesson MP3 under /{app}/{task}/lessons/, else the first file there."""
|
prefix = f"/{app_id}/{task_id}/lessons/"
|
fulls = []
|
others = []
|
for raw in names or []:
|
name = str(raw or "").replace("\\", "/")
|
if not name.lower().endswith(".mp3"):
|
continue
|
if not name.startswith("/"):
|
name = "/" + name.lstrip("/")
|
if not name.startswith(prefix):
|
continue
|
base = name.rsplit("/", 1)[-1].lower()
|
if base == "full.mp3" or base.endswith("_full.mp3"):
|
fulls.append(name)
|
else:
|
others.append(name)
|
fulls.sort()
|
others.sort()
|
if fulls:
|
return fulls[0]
|
return others[0] if others else ""
|
|
|
def _is_clip_path(text: str) -> bool:
|
t = (text or "").replace("\\", "/").lower()
|
return t.endswith(".mp3") or t.startswith("/")
|
|
|
def _shadowing_items(conn: "ConnectionHandler") -> list:
|
clips = getattr(conn, "shadowing_clips", None) or []
|
if clips:
|
return list(clips)
|
return _shadowing_sentences(conn)
|
|
|
def _begin_shadowing(conn: "ConnectionHandler", enter_text: str) -> str:
|
sentences = _shadowing_sentences(conn)
|
conn.shadowing_index = 0
|
if not sentences:
|
conn.shadowing_prompt = ""
|
return enter_text or "现在已经切换至跟读模式，但还没有跟读句子。"
|
conn.shadowing_prompt = sentences[0]
|
prefix = enter_text or "现在已经切换至跟读模式。"
|
return f"{prefix}请跟我读：{sentences[0]}"
|
|
|
def _cancel_shadowing_timeout(conn: "ConnectionHandler") -> None:
|
task = getattr(conn, "shadowing_timeout_task", None)
|
if task is not None:
|
try:
|
if not task.done():
|
task.cancel()
|
except Exception:
|
pass
|
conn.shadowing_timeout_task = None
|
|
|
def _task_for(conn: "ConnectionHandler") -> Optional[Dict[str, Any]]:
|
session = current_session(conn) or getattr(conn, "task_session", None)
|
app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", None) or "english_learning"
|
task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", None) or "shadowing"
|
_, task = get_task(app_id, task_id)
|
return task
|
|
|
def _expected_for_index(conn: "ConnectionHandler", index: int) -> Optional[str]:
|
session = current_session(conn) or getattr(conn, "task_session", None)
|
app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
|
task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
|
clips = list(getattr(conn, "shadowing_clips", None) or [])
|
return resolve_expected_text(
|
_task_for(conn),
|
clips,
|
index,
|
transcript_cache_dir(app_id, task_id),
|
)
|
|
|
def apply_deferred_assessment(
|
conn: "ConnectionHandler",
|
step_index: int,
|
session: Any = None,
|
clips: Optional[list] = None,
|
app_id: str = "",
|
task_id: str = "",
|
audio_snapshot: Any = None,
|
persist_audio: bool = True,
|
) -> None:
|
"""Write expected_text + metrics onto the captured session. Never raises into play."""
|
import time as _time
|
|
session = session or getattr(conn, "task_session", None)
|
if not session:
|
return
|
app_id = app_id or getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
|
task_id = task_id or getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
|
_, task = get_task(app_id or "english_learning", task_id or "shadowing")
|
cfg = (task or {}).get("assessment") or {}
|
clip_list = list(clips if clips is not None else (getattr(conn, "shadowing_clips", None) or []))
|
turn = session.find_turn(step_index)
|
asr = _plain_user_text((turn or {}).get("asr_text") or "")
|
expected = (turn or {}).get("expected_text") or resolve_expected_text(
|
task, clip_list, step_index, transcript_cache_dir(app_id, task_id)
|
)
|
snapshot = audio_snapshot
|
if snapshot is None:
|
snapshot = getattr(conn, "last_asr_audio_snapshot", None)
|
wav_bytes = pcm_snapshot_to_wav(snapshot)
|
existing_audio = (turn or {}).get("audio_id")
|
audio_id = existing_audio
|
if persist_audio and not existing_audio:
|
audio_id = save_turn_wav(
|
getattr(session, "task_session_id", "") or "",
|
step_index,
|
wav_bytes,
|
conn=conn,
|
)
|
try:
|
from core.handle.reportHandle import enqueue_asr_report
|
|
enqueue_asr_report(
|
conn,
|
asr or "",
|
snapshot,
|
getattr(session, "task_session_id", None),
|
int(step_index),
|
)
|
except Exception:
|
pass
|
clip = ""
|
if 0 <= int(step_index) < len(clip_list):
|
clip = str(clip_list[step_index] or "")
|
metrics = []
|
if not (cfg and cfg.get("enabled") is False):
|
metrics = evaluate_turn(
|
asr,
|
expected or "",
|
wav_bytes,
|
cfg,
|
extra={"clip": clip, "step_index": step_index},
|
)
|
session.upsert_turn(
|
index=step_index,
|
prompt_text=(turn or {}).get("prompt_text") or getattr(conn, "shadowing_prompt", "") or "",
|
expected_text=expected,
|
metrics=metrics or None,
|
assessed_at=int(_time.time() * 1000) if metrics else None,
|
audio_id=audio_id,
|
)
|
session.save()
|
|
|
def _schedule_deferred_assessment(conn: "ConnectionHandler", step_index: int) -> None:
|
session = getattr(conn, "task_session", None)
|
clips = list(getattr(conn, "shadowing_clips", None) or [])
|
app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
|
task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
|
audio_snapshot = getattr(conn, "last_asr_audio_snapshot", None)
|
|
async def _run():
|
try:
|
apply_deferred_assessment(
|
conn,
|
step_index,
|
session=session,
|
clips=clips,
|
app_id=app_id,
|
task_id=task_id,
|
audio_snapshot=audio_snapshot,
|
persist_audio=False,
|
)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}")
|
|
loop = getattr(conn, "loop", None)
|
try:
|
loop = loop or asyncio.get_running_loop()
|
loop.create_task(_run())
|
except RuntimeError:
|
try:
|
apply_deferred_assessment(
|
conn,
|
step_index,
|
session=session,
|
clips=clips,
|
app_id=app_id,
|
task_id=task_id,
|
audio_snapshot=audio_snapshot,
|
persist_audio=False,
|
)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}")
|
|
|
def _schedule_shadowing_listen_timeout(conn: "ConnectionHandler", duration_ms: int) -> None:
|
_cancel_shadowing_timeout(conn)
|
ratio = resolve_listen_ratio(conn, _task_for(conn))
|
wait_ms = listen_window_ms(
|
duration_ms or 0,
|
ratio,
|
SHADOWING_LISTEN_MIN_MS,
|
speed=getattr(conn, "shadowing_last_speed", None),
|
wall_clock=True,
|
)
|
conn.shadowing_awaiting_user = True
|
conn.shadowing_assess_index = int(getattr(conn, "shadowing_index", 0) or 0)
|
conn.shadowing_audio_saved_index = -1
|
conn.shadowing_listen_pcm = []
|
gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0)
|
logger.bind(tag=TAG).info(
|
f"跟读听窗 {wait_ms}ms = 实际播放 {duration_ms}ms × {ratio}"
|
f"（倍速已含在实际播放时长里） gen={gen}"
|
)
|
|
async def _fire():
|
try:
|
await asyncio.sleep(wait_ms / 1000.0)
|
except asyncio.CancelledError:
|
return
|
if int(getattr(conn, "shadowing_clip_gen", 0) or 0) != gen:
|
return
|
if getattr(conn, "interaction_mode", "") != "SHADOWING":
|
return
|
if not getattr(conn, "shadowing_awaiting_user", False):
|
return
|
await _on_shadowing_listen_window_closed(conn)
|
|
loop = getattr(conn, "loop", None)
|
if loop:
|
conn.shadowing_timeout_task = loop.create_task(_fire())
|
else:
|
conn.shadowing_timeout_task = asyncio.create_task(_fire())
|
|
|
async def close_shadowing_listen_window(conn: "ConnectionHandler") -> None:
|
"""Device listen stop: archive this step once, cancel the clock, advance index."""
|
_cancel_shadowing_timeout(conn)
|
await _on_shadowing_listen_window_closed(conn)
|
|
|
def _drain_asr_queue_into_shadowing_pcm(conn: "ConnectionHandler") -> int:
|
"""Binary opus is queued; listen stop/timeout run on the WS loop first. Take pending PCM now."""
|
q = getattr(conn, "asr_audio_queue", None)
|
if q is None or not getattr(conn, "shadowing_awaiting_user", False):
|
return 0
|
drained = 0
|
while True:
|
try:
|
frame = q.get_nowait()
|
except (queue.Empty, EOFError, OSError):
|
break
|
if frame:
|
append_shadowing_listen_pcm(conn, frame)
|
drained += 1
|
if drained:
|
logger.bind(tag=TAG).info(f"跟读关窗前抽干 ASR 队列 frames={drained}")
|
return drained
|
|
|
def append_shadowing_listen_pcm(conn: "ConnectionHandler", pcm_frame: Any) -> None:
|
if getattr(conn, "interaction_mode", "") != "SHADOWING":
|
return
|
if not getattr(conn, "shadowing_awaiting_user", False):
|
return
|
if not pcm_frame:
|
return
|
buf = getattr(conn, "shadowing_listen_pcm", None)
|
if buf is None:
|
conn.shadowing_listen_pcm = []
|
buf = conn.shadowing_listen_pcm
|
buf.append(pcm_frame)
|
|
|
def _take_shadowing_listen_pcm(conn: "ConnectionHandler") -> list:
|
frames = list(getattr(conn, "shadowing_listen_pcm", None) or [])
|
conn.shadowing_listen_pcm = []
|
return frames
|
|
|
def save_shadowing_listen_audio(conn: "ConnectionHandler", step_index: Optional[int] = None) -> None:
|
"""Persist exactly one wav per follow-read step. Never merge neighboring windows."""
|
if getattr(conn, "interaction_mode", "") != "SHADOWING":
|
return
|
index = int(
|
step_index
|
if step_index is not None
|
else (getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0)
|
)
|
if int(getattr(conn, "shadowing_audio_saved_index", -1)) == index:
|
conn.shadowing_listen_pcm = []
|
return
|
snapshot = _take_shadowing_listen_pcm(conn)
|
if not snapshot:
|
logger.bind(tag=TAG).warning(f"跟读听窗无 PCM，写静音占位 step={index}")
|
snapshot = silent_pcm_snapshot()
|
conn.shadowing_audio_saved_index = index
|
from core.app_runtime.session import stash_asr_audio
|
|
stash_asr_audio(conn, snapshot)
|
try:
|
apply_deferred_assessment(conn, index, audio_snapshot=snapshot)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"跟读听窗录音落盘失败 step={index}: {exc}")
|
|
|
async def _on_shadowing_listen_window_closed(conn: "ConnectionHandler") -> None:
|
"""Clock or listen stop: archive this listen. Device already sequences play → record → play."""
|
if not getattr(conn, "shadowing_awaiting_user", False):
|
return
|
_drain_asr_queue_into_shadowing_pcm(conn)
|
conn.shadowing_awaiting_user = False
|
listened_idx = int(getattr(conn, "shadowing_index", 0) or 0)
|
listened_prompt = getattr(conn, "shadowing_prompt", "") or ""
|
save_shadowing_listen_audio(conn, listened_idx)
|
items = _shadowing_items(conn)
|
nxt = listened_idx + 1
|
conn.shadowing_index = nxt
|
if nxt < len(items):
|
conn.shadowing_prompt = items[nxt]
|
logger.bind(tag=TAG).info(f"跟读：听窗结束 step={listened_idx}")
|
try:
|
session = current_session(conn) or getattr(conn, "task_session", None)
|
if session:
|
if session.find_turn(listened_idx) is None:
|
session.upsert_turn(
|
index=listened_idx,
|
prompt_text=listened_prompt,
|
asr_text="",
|
expected_text=_expected_for_index(conn, listened_idx),
|
)
|
session.add_event("listen_window_closed")
|
session.add_event("listen_timeout")
|
session.save()
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"跟读听窗归档失败: {exc}")
|
|
|
def _parse_mp3_list_names(raw: Any) -> list:
|
text = ""
|
if raw is None:
|
return []
|
if hasattr(raw, "response") or hasattr(raw, "result"):
|
text = str(getattr(raw, "response", None) or getattr(raw, "result", None) or "")
|
else:
|
text = str(raw)
|
text = text.strip()
|
if not text:
|
return []
|
try:
|
import json
|
|
data = json.loads(text)
|
except Exception:
|
return []
|
files = data.get("files") if isinstance(data, dict) else None
|
names = []
|
if isinstance(files, list):
|
for item in files:
|
if isinstance(item, dict) and item.get("name"):
|
names.append(str(item["name"]))
|
elif isinstance(item, str):
|
names.append(item)
|
return names
|
|
|
async def _list_sd_mp3_names(conn: "ConnectionHandler") -> list:
|
result = await _execute_named_tool(conn, DEVICE_LIST_TOOL, {})
|
return _parse_mp3_list_names(result)
|
|
|
def _is_full_lesson_mp3(name: str) -> bool:
|
base = str(name or "").replace("\\", "/").rsplit("/", 1)[-1].lower()
|
return base == "full.mp3" or base.endswith("_full.mp3")
|
|
|
def _compact_shadowing_playlist(
|
conn: "ConnectionHandler",
|
params: Optional[Dict[str, Any]],
|
) -> List[Dict[str, Any]]:
|
"""Keep {name, repeat, speed}. Device expands repeats locally with a listen window each pass."""
|
items = parse_playlist(params)
|
if not items:
|
items = default_shadowing_playlist(_task_for(conn))
|
compact: List[Dict[str, Any]] = []
|
for raw in items:
|
if not isinstance(raw, dict):
|
continue
|
name = str(raw.get("name") or "")
|
if not name or _is_full_lesson_mp3(name):
|
continue
|
try:
|
repeat_n = int(raw.get("repeat") or 1)
|
except (TypeError, ValueError):
|
repeat_n = 1
|
if repeat_n < 1:
|
repeat_n = 1
|
compact.append(
|
{
|
"name": name,
|
"repeat": repeat_n,
|
"speed": clamp_play_speed(
|
raw["speed"] if raw.get("speed") is not None else 1.0
|
),
|
"power_off_after": False,
|
"finish_task": False,
|
}
|
)
|
return compact
|
|
|
def _begin_shadowing_playlist(
|
conn: "ConnectionHandler",
|
params: Optional[Dict[str, Any]],
|
) -> Dict[str, Any]:
|
"""One shot: send the whole shadowing playlist. Device plays, records, plays."""
|
items = _compact_shadowing_playlist(conn, params)
|
clips = expand_playlist_plays(items)
|
conn.shadowing_clips = clips
|
conn.shadowing_index = 0
|
conn.shadowing_prompt = clips[0] if clips else ""
|
conn.shadowing_awaiting_user = False
|
first = items[0] if items else dict(params or {})
|
play = dict(first)
|
play["items"] = items
|
play["finish_task"] = False
|
play["listen_ratio"] = resolve_listen_ratio(conn, _task_for(conn))
|
if items:
|
conn.shadowing_last_speed = items[0].get("speed")
|
return play
|
|
|
async def _play_shadowing_clip(conn: "ConnectionHandler", name: str) -> bool:
|
if getattr(conn, "interaction_mode", "") != "SHADOWING":
|
logger.bind(tag=TAG).info(f"跳过跟读播放，当前已离开跟读: {name}")
|
return False
|
conn.shadowing_awaiting_user = False
|
conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
|
speed = resolve_play_speed(_task_for(conn))
|
conn.shadowing_last_speed = speed
|
notice = await _execute_named_tool_play(
|
conn,
|
{
|
"name": name,
|
"repeat": 1,
|
"power_off_after": False,
|
"speed": speed,
|
"finish_task": False,
|
"listen_ratio": resolve_listen_ratio(conn, _task_for(conn)),
|
},
|
)
|
if notice and notice.action in (Action.ERROR, Action.NOTFOUND):
|
return False
|
return True
|
|
|
async def _prepare_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool:
|
params = params or {}
|
items = _compact_shadowing_playlist(conn, params)
|
clips = expand_playlist_plays(items)
|
if not clips:
|
names = await _list_sd_mp3_names(conn)
|
lesson = str(params.get("lesson") or "").strip()
|
single = str(params.get("name") or params.get("path") or params.get("url") or "")
|
if not lesson and single and not single.lower().endswith(".mp3"):
|
lesson = single
|
if lesson.lower().endswith(".mp3"):
|
lesson = ""
|
clips = lesson_mp3_clips(
|
names,
|
getattr(conn, "active_app_id", "english_learning"),
|
getattr(conn, "current_task_id", "shadowing"),
|
lesson,
|
)
|
clips = [name for name in clips if not _is_full_lesson_mp3(name)]
|
if not clips:
|
conn.shadowing_clips = []
|
conn.shadowing_prompt = ""
|
return False
|
conn.shadowing_clips = clips
|
conn.shadowing_index = 0
|
conn.shadowing_prompt = clips[0]
|
conn.shadowing_awaiting_user = False
|
return True
|
|
|
async def _start_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool:
|
if not await _prepare_shadowing_sd(conn, params):
|
return False
|
play = _begin_shadowing_playlist(conn, params)
|
conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
|
notice = await _execute_named_tool_play(conn, play)
|
if notice and notice.action in (Action.ERROR, Action.NOTFOUND):
|
return False
|
return True
|
|
|
async def _fill_repeater_play_name(
|
conn: "ConnectionHandler", params: Optional[Dict[str, Any]]
|
) -> Dict[str, Any]:
|
play = dict(params or {})
|
if play.get("items"):
|
return play
|
if str(play.get("name") or "").strip():
|
return play
|
stub = default_repeater_playlist(_task_for(conn))
|
if stub:
|
first = stub[0]
|
play["name"] = first.get("name") or ""
|
if play.get("repeat") is None:
|
play["repeat"] = first.get("repeat") or 1
|
if play.get("speed") is None and first.get("speed") is not None:
|
play["speed"] = first.get("speed")
|
conn.repeater_play_name = play["name"]
|
return play
|
try:
|
names = await _list_sd_mp3_names(conn)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"列举复读课文失败: {exc}")
|
return play
|
picked = repeater_lesson_mp3(
|
names,
|
getattr(conn, "active_app_id", "") or "english_learning",
|
getattr(conn, "current_task_id", "") or "repeater",
|
)
|
if picked:
|
play["name"] = picked
|
conn.repeater_play_name = picked
|
return play
|
|
|
async def maybe_flush_pending_play_after_tts(conn: "ConnectionHandler") -> None:
|
"""Start queued lesson audio only after enter-prompt TTS has finished on device."""
|
_ensure_runtime_fields(conn)
|
pending = getattr(conn, "pending_play_after_tts", None)
|
if not pending:
|
return
|
conn.pending_play_after_tts = None
|
kind = pending.get("kind") if isinstance(pending, dict) else None
|
try:
|
if kind == "repeater":
|
play_params = await _fill_repeater_play_name(conn, pending.get("params") or {})
|
await _execute_named_tool_play(conn, play_params)
|
elif kind == "shadowing":
|
play_params = pending.get("params") if isinstance(pending, dict) else None
|
if not play_params:
|
play_params = _begin_shadowing_playlist(conn, {})
|
conn.shadowing_awaiting_user = False
|
conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
|
await _execute_named_tool_play(conn, play_params)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"延后播放课文失败: {exc}")
|
|
|
def _lesson_end_speech(conn: "ConnectionHandler", phrase: str) -> str:
|
"""Short lesson-end cue plus the same default chat greeting used after 翻译/中英对话."""
|
greeting = (_greeting_text(conn) or "").strip()
|
phrase = (phrase or "").strip()
|
if phrase and greeting:
|
if phrase.endswith("。") or phrase.endswith("."):
|
return f"{phrase}{greeting}"
|
return f"{phrase}。{greeting}"
|
return phrase or greeting
|
|
|
async def _return_to_chat_after_lesson(
|
conn: "ConnectionHandler",
|
phrase: str,
|
reason: str = "play_completed",
|
) -> None:
|
"""Playlist already finished on device: speak the end cue now, then chat. Do not MCP-stop."""
|
conn.repeater_speed_queue = []
|
conn.repeater_playlist = []
|
end_session(conn, reason)
|
if getattr(conn, "tts", None) is not None:
|
await wait_for_tts_channels(conn, timeout=3)
|
conn.sentence_id = uuid.uuid4().hex
|
_speak(conn, _lesson_end_speech(conn, phrase))
|
await start_app_task(
|
conn,
|
"chat",
|
"conversation",
|
{},
|
source="voice",
|
execute_skills=False,
|
speak_switch=False,
|
stop_skills=False,
|
)
|
|
|
async def _shadowing_play_next_or_finish(conn: "ConnectionHandler") -> None:
|
"""Legacy helper: device now sequences clips. Only used if a caller still asks to finish."""
|
conn.shadowing_index = 0
|
conn.shadowing_prompt = ""
|
conn.shadowing_clips = []
|
await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished")
|
|
|
def _mark_repeater_day(conn: "ConnectionHandler") -> None:
|
try:
|
mark_lesson_played_today(
|
str(getattr(conn, "device_id", "") or ""),
|
getattr(conn, "active_app_id", "") or "",
|
getattr(conn, "current_task_id", "") or "",
|
getattr(conn, "repeater_play_name", "") or "",
|
)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"记录复读天数失败: {exc}")
|
|
|
def _apply_playback_speed(
|
conn: "ConnectionHandler",
|
task: Optional[Dict[str, Any]],
|
params: Optional[Dict[str, Any]],
|
) -> Dict[str, Any]:
|
"""Attach clamped speed. Multi-speed repeats are Server-driven (one pass per play)."""
|
play = dict(params or {})
|
try:
|
repeat_n = int(play.get("repeat") or 1)
|
except (TypeError, ValueError):
|
repeat_n = 1
|
if repeat_n < 1:
|
repeat_n = 1
|
device_id = str(getattr(conn, "device_id", "") or "")
|
app_id = getattr(conn, "active_app_id", "") or ""
|
task_id = getattr(conn, "current_task_id", "") or ""
|
lesson = str(play.get("name") or "")
|
day_index = lesson_day_index(device_id, app_id, task_id, lesson)
|
if play.get("speed") is not None:
|
speeds = [clamp_play_speed(play.get("speed"))] * repeat_n
|
else:
|
speeds = speeds_for_repeats(task, repeat_n, day_index)
|
conn.repeater_play_name = lesson
|
conn.repeater_power_off_after = bool(play.get("power_off_after") or False)
|
varied = len(speeds) > 1 and any(abs(item - speeds[0]) > 0.001 for item in speeds[1:])
|
if varied:
|
conn.repeater_speed_queue = speeds[1:]
|
play["speed"] = speeds[0]
|
play["repeat"] = 1
|
play["power_off_after"] = False
|
else:
|
conn.repeater_speed_queue = []
|
play["speed"] = speeds[0]
|
play["repeat"] = repeat_n
|
# 复读清单结束由 Server 收 audio.completed 后切回对话。设备不要 finish_task 写 NVS。
|
play["finish_task"] = False
|
return play
|
|
|
def _begin_repeater_playlist(
|
conn: "ConnectionHandler",
|
task: Optional[Dict[str, Any]],
|
params: Optional[Dict[str, Any]],
|
) -> Dict[str, Any]:
|
"""One shot: send the whole playlist. Repeat is just the same clip again, no mode switch."""
|
items = parse_playlist(params)
|
if not items:
|
items = default_repeater_playlist(task)
|
normalized = []
|
for raw in items:
|
if not isinstance(raw, dict):
|
continue
|
name = str(raw.get("name") or "")
|
if not name:
|
continue
|
try:
|
repeat_n = int(raw.get("repeat") or 1)
|
except (TypeError, ValueError):
|
repeat_n = 1
|
if repeat_n < 1:
|
repeat_n = 1
|
item = {
|
"name": name,
|
"repeat": repeat_n,
|
"speed": clamp_play_speed(
|
raw["speed"] if raw.get("speed") is not None else 1.0
|
),
|
"power_off_after": bool(raw.get("power_off_after") or False),
|
"finish_task": False,
|
}
|
normalized.append(item)
|
conn.repeater_playlist = []
|
conn.repeater_speed_queue = []
|
first = normalized[0] if normalized else dict(params or {})
|
conn.repeater_play_name = str(first.get("name") or "")
|
conn.repeater_power_off_after = bool(first.get("power_off_after") or False)
|
play = dict(first)
|
play["items"] = normalized
|
play["finish_task"] = False
|
return play
|
|
|
async def _continue_repeater_if_needed(conn: "ConnectionHandler") -> bool:
|
queue = list(getattr(conn, "repeater_speed_queue", None) or [])
|
if queue:
|
nxt = queue.pop(0)
|
conn.repeater_speed_queue = queue
|
last_speed = len(queue) == 0 and not list(getattr(conn, "repeater_playlist", None) or [])
|
try:
|
play_params = await _fill_repeater_play_name(
|
conn,
|
{
|
"name": getattr(conn, "repeater_play_name", "") or "",
|
"speed": nxt,
|
"repeat": 1,
|
"power_off_after": last_speed
|
and bool(getattr(conn, "repeater_power_off_after", False)),
|
"finish_task": False,
|
},
|
)
|
await _execute_named_tool_play(conn, play_params)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"复读下一遍失败: {exc}")
|
conn.repeater_speed_queue = []
|
return False
|
return True
|
|
rest = list(getattr(conn, "repeater_playlist", None) or [])
|
if not rest:
|
return False
|
nxt, conn.repeater_playlist = rest[0], rest[1:]
|
play_params = _apply_playback_speed(conn, _task_for(conn), nxt)
|
play_params["finish_task"] = False
|
try:
|
play_params = await _fill_repeater_play_name(conn, play_params)
|
await _execute_named_tool_play(conn, play_params)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"复读下一首失败: {exc}")
|
conn.repeater_playlist = []
|
return False
|
return True
|
|
|
async def _execute_named_tool_play(conn, params):
|
items = params.get("items") if isinstance(params.get("items"), list) else []
|
first = items[0] if items else params
|
play_args = {
|
"name": (first.get("name") if isinstance(first, dict) else None) or params.get("name") or "",
|
"repeat": int((first.get("repeat") if isinstance(first, dict) else None) or params.get("repeat") or 1),
|
"power_off_after": bool(params.get("power_off_after") or False),
|
"speed": clamp_play_speed(
|
(first.get("speed") if isinstance(first, dict) and first.get("speed") is not None else None)
|
if items
|
else (params["speed"] if params.get("speed") is not None else 1.0)
|
),
|
"finish_task": bool(params.get("finish_task", True)),
|
"app_id": str(getattr(conn, "active_app_id", "") or ""),
|
"task_id": str(getattr(conn, "current_task_id", "") or ""),
|
"force": True,
|
}
|
if items:
|
play_args["items"] = json.dumps(items, ensure_ascii=False)
|
play_args["speed"] = clamp_play_speed(
|
first["speed"] if first.get("speed") is not None else 1.0
|
)
|
play_args["power_off_after"] = bool(
|
items[-1].get("power_off_after") if items else False
|
)
|
if params.get("listen_ratio") is not None:
|
try:
|
play_args["listen_ratio"] = float(params["listen_ratio"])
|
except (TypeError, ValueError):
|
pass
|
if play_args["task_id"] in ("repeater", "shadowing"):
|
play_args["finish_task"] = False
|
result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
|
blob = ""
|
if result:
|
blob = str(result.response or result.result or "")
|
compact = blob.replace(" ", "")
|
if '"duplicate_ignored":true' in compact:
|
logger.bind(tag=TAG).warning("设备忽略了同一课文的下一遍，强制再播一次")
|
play_args["speed"] = clamp_play_speed(float(play_args["speed"]) + 0.02)
|
play_args["force"] = True
|
result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
|
blob = ""
|
if result:
|
blob = str(result.response or result.result or "")
|
compact = blob.replace(" ", "")
|
rejected = (result and result.action in (Action.ERROR, Action.NOTFOUND)) or (
|
"no_mp3" in blob
|
or '"accepted":false' in compact
|
or '"duplicate_ignored":true' in compact
|
)
|
if rejected:
|
if not play_args["name"]:
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response="复读机已打开，但 SD 卡上还没有可播放的课文。",
|
)
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response=(result.response if result else None) or "这台设备没有本地复读播放能力",
|
)
|
return None
|
|
|
async def start_app_task(
|
conn: "ConnectionHandler",
|
app_id: str,
|
task_id: str = "",
|
params: Optional[Dict[str, Any]] = None,
|
source: str = "voice",
|
execute_skills: bool = True,
|
speak_switch: bool = True,
|
stop_skills: bool = True,
|
) -> ActionResponse:
|
"""Single entry for MQTT / schedule / voice / default chat."""
|
_ensure_runtime_fields(conn)
|
params = params or {}
|
app, task = get_task(app_id, task_id)
|
if not app:
|
return ActionResponse(action=Action.ERROR, response=f"未知的 App: {app_id}")
|
if not task:
|
default_task = app.get("default_task") or ""
|
app, task = get_task(app_id, default_task)
|
if not task:
|
return ActionResponse(action=Action.ERROR, response=f"未知的 Task: {app_id}/{task_id}")
|
|
resolved_task_id = task_id or app.get("default_task") or "conversation"
|
new_p = source_priority(source)
|
cur_p = source_priority(getattr(conn, "current_source", "default_chat"))
|
same_task = (
|
getattr(conn, "active_app_id", "") == app_id
|
and getattr(conn, "current_task_id", "") == resolved_task_id
|
)
|
if new_p < cur_p and not same_task:
|
logger.bind(tag=TAG).info(
|
f"拒绝低优先级任务 {source}({new_p}) < {conn.current_source}({cur_p})"
|
)
|
return ActionResponse(
|
action=Action.NONE,
|
response="当前有更高优先级的任务，先完成后再切换",
|
)
|
|
prev_app_id = getattr(conn, "active_app_id", "chat")
|
prev_task_id = getattr(conn, "current_task_id", "conversation")
|
prev_app, prev_task = get_task(prev_app_id, prev_task_id)
|
switching = (prev_app_id != app_id) or (prev_task_id != resolved_task_id)
|
|
if not same_task:
|
if stop_skills:
|
await _stop_current_skills(conn)
|
if app_id != "chat":
|
conn.chat_greeted = False
|
|
conn.active_app_id = app_id
|
conn.current_task_id = resolved_task_id
|
conn.interaction_mode = task.get("interaction_mode") or app.get("mode") or "CHAT"
|
conn.current_source = source
|
conn.tool_policy = app.get("tool_policy") or "inherit_agent"
|
conn.tool_allowlist = list(app.get("tools") or [])
|
_apply_prompt(conn, app)
|
_refresh_tools(conn)
|
|
logger.bind(tag=TAG).info(
|
f"start_app_task app={app_id} task={resolved_task_id} mode={conn.interaction_mode} source={source}"
|
)
|
|
if app_id == "chat":
|
end_session(conn, "user_exit")
|
elif switching:
|
begin_session(conn, app_id, resolved_task_id, source, conn.interaction_mode)
|
|
pending_play = None
|
if execute_skills:
|
if app_sd_layout_paths(app_id, app):
|
await ensure_app_sd_dirs(conn, app_id)
|
for skill in task.get("skills") or []:
|
skill_name = (skill or {}).get("name") or ""
|
if skill_name in ("self.mp3.play", DEVICE_PLAY_TOOL):
|
play_params = _begin_repeater_playlist(conn, task, params)
|
pending_play = {"kind": "repeater", "params": play_params}
|
elif skill_name == "self.power_off":
|
await _execute_named_tool(conn, "self_power_off", {})
|
|
announce = ""
|
boot_chat = app_id == "chat" and source == "default_chat"
|
if boot_chat:
|
# 开机进 chat：必须播欢迎词。不要被 same_task / just_woken_up 跳过。
|
enable = (conn.config or {}).get("enable_proactive_greeting", True)
|
if enable and not getattr(conn, "chat_greeted", False):
|
if not await wait_for_tts_channels(conn, timeout=10):
|
logger.bind(tag=TAG).error("开机欢迎词跳过：TTS通道未就绪")
|
else:
|
conn.chat_greeted = True
|
conn.sentence_id = uuid.uuid4().hex
|
_speak(conn, _greeting_text(conn))
|
elif switching:
|
announce = _switch_announcement(prev_app, prev_task, app, task, app_id)
|
if conn.interaction_mode == "SHADOWING":
|
started = await _prepare_shadowing_sd(conn, params)
|
if not started:
|
prefix = announce or "现在已经切换至跟读模式。"
|
announce = f"{prefix}跟读课文还没有放到SD卡。"
|
else:
|
announce = announce or "现在已经切换至跟读模式，请跟着课文读。"
|
pending_play = {
|
"kind": "shadowing",
|
"params": _begin_shadowing_playlist(conn, params),
|
}
|
if speak_switch and announce:
|
conn.sentence_id = uuid.uuid4().hex
|
conn.pending_play_after_tts = pending_play
|
_speak(conn, announce)
|
elif pending_play:
|
conn.pending_play_after_tts = pending_play
|
await maybe_flush_pending_play_after_tts(conn)
|
if app_id == "chat":
|
conn.chat_greeted = True
|
elif conn.interaction_mode == "CHAT":
|
enable = (conn.config or {}).get("enable_proactive_greeting", True)
|
woken = bool(getattr(conn, "just_woken_up", False))
|
if enable and not woken and not getattr(conn, "chat_greeted", False):
|
conn.chat_greeted = True
|
conn.sentence_id = uuid.uuid4().hex
|
_speak(conn, _greeting_text(conn))
|
elif conn.interaction_mode == "SHADOWING" and not getattr(conn, "shadowing_clips", None):
|
started = await _start_shadowing_sd(conn, params)
|
if not started and speak_switch:
|
conn.sentence_id = uuid.uuid4().hex
|
_speak(conn, "跟读课文还没有放到SD卡。")
|
elif pending_play:
|
conn.pending_play_after_tts = pending_play
|
await maybe_flush_pending_play_after_tts(conn)
|
|
app_name = _display_name(app, task)
|
return ActionResponse(
|
action=Action.NONE,
|
result=f"{app_id}/{resolved_task_id}",
|
response=announce or f"已打开{app_name}",
|
)
|
|
|
async def switch_app(
|
conn: "ConnectionHandler",
|
app_id: str,
|
source: str = "voice",
|
task_id: str = "",
|
) -> ActionResponse:
|
app = get_app(app_id)
|
if not app:
|
return ActionResponse(action=Action.ERROR, response=f"不支持的 App: {app_id}")
|
resolved = task_id or app.get("default_task") or ""
|
return await start_app_task(conn, app_id, resolved, {}, source)
|
|
|
async def on_user_wakeup(conn: "ConnectionHandler") -> None:
|
"""User wake word returns to chat; do not treat [app_task] as wakeup."""
|
_ensure_runtime_fields(conn)
|
if getattr(conn, "interaction_mode", "CHAT") == "CHAT" and getattr(conn, "active_app_id", "chat") == "chat":
|
return
|
await start_app_task(conn, "chat", "conversation", {}, source="voice")
|
|
|
def match_voice_switch(text: str) -> Optional[tuple]:
|
"""Map a short phrase to (app_id, task_id) using apps.json voice_enter / voice_exit."""
|
if not text:
|
return None
|
_, filtered = _strip(text)
|
return match_registry_voice(filtered)
|
|
|
def _strip(text: str) -> tuple:
|
from core.utils.util import remove_punctuation_and_length
|
|
return remove_punctuation_and_length(text)
|
|
|
def is_exit_to_chat(text: str) -> bool:
|
mapped = match_voice_switch(text)
|
return bool(mapped and mapped[0] == "chat")
|
|
|
def should_skip_listen_stop_asr(conn: Any) -> bool:
|
"""True when listen stop is closing the mic for lesson playback, not a follow-read."""
|
mode = getattr(conn, "interaction_mode", "CHAT")
|
if mode == "PLAY_AUDIO":
|
return True
|
if mode == "SHADOWING" and not getattr(conn, "shadowing_awaiting_user", False):
|
return True
|
return False
|
|
|
def is_exit_translation(text: str) -> bool:
|
return is_exit_to_chat(text)
|
|
|
def _plain_user_text(text: str) -> str:
|
"""ASR 可能把原文包在 JSON 里，翻译只要 content。"""
|
if not text:
|
return ""
|
raw = str(text).strip()
|
if raw.startswith("{") and raw.endswith("}"):
|
try:
|
import json
|
|
data = json.loads(raw)
|
if isinstance(data, dict) and data.get("content"):
|
return str(data["content"]).strip()
|
except Exception:
|
pass
|
return raw
|
|
|
def _norm_for_echo(text: str) -> str:
|
_, filtered = _strip(text or "")
|
return re.sub(r"\s+", "", (filtered or "").lower())
|
|
|
def is_echo_of_prompt(user_text: str, prompt: str) -> bool:
|
a = _norm_for_echo(user_text)
|
b = _norm_for_echo(prompt)
|
if not a or not b:
|
return False
|
if a == b or a in b or b in a:
|
return True
|
return difflib.SequenceMatcher(None, a, b).ratio() >= 0.82
|
|
|
def detect_user_language(text: str, hint: Optional[str] = None) -> str:
|
if hint:
|
h = str(hint).lower()
|
if h.startswith("zh") or h in ("cn", "yue", "chinese"):
|
return "zh"
|
if h.startswith("en") or h in ("english",):
|
return "en"
|
body = _plain_user_text(text)
|
cjk = len(re.findall(r"[\u4e00-\u9fff]", body))
|
latin = len(re.findall(r"[A-Za-z]", body))
|
return "zh" if cjk >= max(latin, 1) else "en"
|
|
|
def wrap_bilingual_user_text(text: str, user_lang: str) -> str:
|
if user_lang == "zh":
|
return "Respond in English only. Do not use Chinese characters.\n\n" + text
|
return "请只用中文回答，不要出现英文单词或句子。\n\n" + text
|
|
|
def _looks_like_tool_error(text: str) -> bool:
|
if not text:
|
return True
|
s = str(text)
|
if "Error executing tool" in s or "isError=True" in s or "isError=true" in s:
|
return True
|
if "未配置" in s and "api_key" in s:
|
return True
|
return False
|
|
|
def _record_shadowing_turn(
|
conn: "ConnectionHandler",
|
text: str,
|
echo_rejected: bool = False,
|
index: Optional[int] = None,
|
) -> None:
|
session = current_session(conn)
|
if not session:
|
return
|
if index is None:
|
index = int(getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0)
|
session.upsert_turn(
|
index=index,
|
prompt_text=getattr(conn, "shadowing_prompt", "") or "",
|
asr_text=_plain_user_text(text),
|
echo_rejected=echo_rejected,
|
expected_text=_expected_for_index(conn, index),
|
)
|
session.save()
|
|
|
def apply_shadowing_user_text(conn: "ConnectionHandler", text: str) -> str:
|
"""Apply one ASR utterance. Returns exit | ignore | echo | recorded.
|
|
ASR never advances the clip index; the listen-window clock does.
|
Late ASR after the window still scores the sentence that was just listened to.
|
Device plays the next clip itself; Server does not MCP the next sentence.
|
"""
|
if is_exit_to_chat(text):
|
end_session(conn, "user_exit")
|
return "exit"
|
if getattr(conn, "client_is_speaking", False) and not getattr(
|
conn, "shadowing_awaiting_user", False
|
):
|
logger.bind(tag=TAG).info("跟读：播放中忽略拾音，避免喇叭回声触发对话")
|
return "ignore"
|
prompt = getattr(conn, "shadowing_prompt", "") or ""
|
if (
|
getattr(conn, "shadowing_awaiting_user", False)
|
and prompt
|
and not _is_clip_path(prompt)
|
and is_echo_of_prompt(text, prompt)
|
):
|
logger.bind(tag=TAG).info("跟读：忽略与范读过于相似的识别结果")
|
session = current_session(conn)
|
if session:
|
session.add_event("echo_rejected")
|
session.save()
|
return "echo"
|
_record_shadowing_turn(conn, text, echo_rejected=False)
|
return "recorded"
|
|
|
async def handle_shadowing_turn(conn: "ConnectionHandler", text: str) -> None:
|
"""Score this utterance off the chat path. Next clip is sequenced on device."""
|
outcome = apply_shadowing_user_text(conn, text)
|
if outcome == "exit":
|
_cancel_shadowing_timeout(conn)
|
await start_app_task(conn, "chat", "conversation", {}, source="voice")
|
return
|
if outcome in ("ignore", "echo"):
|
return
|
|
step_index = int(
|
getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0
|
)
|
logger.bind(tag=TAG).info(f"跟读 ASR 仅评估不上对话 step={step_index}")
|
_schedule_deferred_assessment(conn, step_index)
|
|
|
async def handle_skill_event(conn: "ConnectionHandler", text: str) -> bool:
|
"""Handle `[skill_event]audio.completed|app|task|duration_ms`. Returns True if consumed."""
|
parsed = parse_skill_event(text)
|
if parsed is None:
|
return False
|
_ensure_runtime_fields(conn)
|
name = parsed.get("name") or ""
|
app_id = parsed.get("app_id") or ""
|
task_id = parsed.get("task_id") or ""
|
try:
|
duration_ms = int(parsed.get("duration_ms") or 0)
|
except (TypeError, ValueError):
|
duration_ms = 0
|
logger.bind(tag=TAG).info(f"skill_event {name} {app_id}/{task_id} duration_ms={duration_ms}")
|
|
mode = getattr(conn, "interaction_mode", "CHAT")
|
# 听窗时钟必须先于 session.save()：磁盘/HTTP 不能挡住 1.2×duration 的热路径。
|
shadowing_playlist_done = name in ("audio.playlist_completed", "playlist.completed") and (
|
mode == "SHADOWING" or task_id == "shadowing"
|
)
|
if shadowing_playlist_done:
|
_cancel_shadowing_timeout(conn)
|
if getattr(conn, "shadowing_awaiting_user", False):
|
await _on_shadowing_listen_window_closed(conn)
|
else:
|
save_shadowing_listen_audio(conn)
|
session = getattr(conn, "task_session", None)
|
if session:
|
session.add_event(name, app_id, task_id)
|
await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished")
|
return True
|
if mode == "SHADOWING" and name == "audio.completed":
|
if getattr(conn, "shadowing_awaiting_user", False):
|
await _on_shadowing_listen_window_closed(conn)
|
conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
|
_schedule_shadowing_listen_timeout(conn, duration_ms)
|
session = getattr(conn, "task_session", None)
|
if session:
|
session.add_event(name, app_id, task_id)
|
return True
|
if mode == "SHADOWING" and name == "audio.stopped":
|
session = getattr(conn, "task_session", None)
|
if session:
|
session.add_event(name, app_id, task_id)
|
end_session(conn, "play_stopped")
|
await start_app_task(
|
conn,
|
"chat",
|
"conversation",
|
{},
|
source="voice",
|
execute_skills=False,
|
speak_switch=False,
|
)
|
return True
|
|
if mode == "PLAY_AUDIO" and name == "audio.completed":
|
continued = await _continue_repeater_if_needed(conn)
|
session = getattr(conn, "task_session", None)
|
if session:
|
session.add_event(name, app_id, task_id)
|
if continued:
|
if session:
|
session.save()
|
return True
|
_mark_repeater_day(conn)
|
await _return_to_chat_after_lesson(conn, REPEATER_END_PHRASE, "play_completed")
|
return True
|
if mode == "PLAY_AUDIO" and name == "audio.stopped":
|
conn.repeater_speed_queue = []
|
|
session = getattr(conn, "task_session", None)
|
if session:
|
session.add_event(name, app_id, task_id)
|
session.save()
|
|
if name == "audio.completed":
|
end_session(conn, "play_completed")
|
await start_app_task(
|
conn,
|
"chat",
|
"conversation",
|
{},
|
source="voice",
|
execute_skills=False,
|
speak_switch=True,
|
)
|
elif name == "audio.stopped":
|
end_session(conn, "play_stopped")
|
if getattr(conn, "active_app_id", "chat") != "chat":
|
await start_app_task(
|
conn,
|
"chat",
|
"conversation",
|
{},
|
source="voice",
|
execute_skills=False,
|
speak_switch=False,
|
)
|
else:
|
logger.bind(tag=TAG).warning(f"未知 skill_event: {name}")
|
return True
|
|
|
async def run_translation(conn: "ConnectionHandler", text: str) -> str:
|
"""Translate with the connection LLM first (智控台已配的模型). MCP is fallback only."""
|
text = _plain_user_text(text)
|
if not text:
|
return ""
|
|
from plugins_func.register import all_function_registry
|
|
item = all_function_registry.get("translate")
|
if item:
|
try:
|
maybe = item.func(conn, text=text)
|
if hasattr(maybe, "__await__"):
|
maybe = await maybe
|
if isinstance(maybe, ActionResponse) and maybe.action not in (
|
Action.ERROR,
|
Action.NOTFOUND,
|
):
|
out = (maybe.response or maybe.result or "").strip()
|
if out and not _looks_like_tool_error(out):
|
return out
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"翻译插件失败: {e}")
|
|
handler = getattr(conn, "func_handler", None)
|
if handler and handler.has_tool("translate"):
|
result = await _execute_named_tool(conn, "translate", {"text": text})
|
if result and result.action not in (Action.ERROR, Action.NOTFOUND):
|
out = _extract_translate_text(result.response or result.result or "")
|
if out and not _looks_like_tool_error(out):
|
return out
|
|
llm = getattr(conn, "llm", None)
|
if llm and hasattr(llm, "response_no_stream"):
|
try:
|
return (
|
llm.response_no_stream(TRANSLATOR_PROMPT, text, temperature=0.1) or ""
|
).strip()
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"翻译 LLM 失败: {e}")
|
return ""
|
|
|
def _extract_translate_text(raw: Any) -> str:
|
if raw is None:
|
return ""
|
if isinstance(raw, str):
|
text = raw.strip()
|
if text.startswith("{") and "text" in text:
|
try:
|
import json
|
|
data = json.loads(text)
|
if isinstance(data, dict):
|
return str(data.get("text") or data.get("content") or text)
|
except Exception:
|
pass
|
return text
|
if hasattr(raw, "content"):
|
content = raw.content
|
if isinstance(content, list) and content:
|
first = content[0]
|
return getattr(first, "text", None) or str(first)
|
return str(content)
|
return str(raw).strip()
|
