From 659c470eba542383987aad227b7f918100f131fa Mon Sep 17 00:00:00 2001
From: wangzhibo <wangzhibo@shsening.com>
Date: 星期五, 11 九月 2026 15:09:05 +0800
Subject: [PATCH] 小智终端 5种 场景模式 OK
---
xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py | 1477 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 1,426 insertions(+), 51 deletions(-)
diff --git a/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py b/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py
index e62fdcf..71dc2af 100644
--- a/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py
+++ b/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py
@@ -2,13 +2,48 @@
from __future__ import annotations
+import asyncio
+import json
+import difflib
+import queue
+import random
+import re
import uuid
-from typing import Any, Dict, Optional, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
from config.logger import setup_logging
from plugins_func.register import Action, ActionResponse
-from .registry import default_app_id, get_app, get_task
+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
@@ -28,19 +63,37 @@
"鏈寚瀹氳瑷�鏃讹細涓枃璇戞垚鑻辨枃锛岃嫳鏂囪瘧鎴愪腑鏂囷紱鍏跺畠璇█榛樿璇戞垚涓枃銆�"
)
+BILINGUAL_PROMPT = (
+ "浣犳鍦ㄨ繘琛屼腑鑻卞鐓у璇濈粌涔狅紝涓嶆槸缈昏瘧鏈猴紝涓嶈鎶婄敤鎴风殑璇濆師鏍疯瘧鍥炲幓浜ゅ樊銆�"
+ "涓ユ牸鎵ц锛氱敤鎴疯涓枃鏃朵綘鍙敤鑻辨枃瀹屾暣鍥炵瓟锛涚敤鎴疯鑻辨枃鏃朵綘鍙敤涓枃瀹屾暣鍥炵瓟銆�"
+ "涓嶈涓嫳澶规潅锛屼笉瑕佸悓鏃剁粰鍑轰袱绉嶈瑷�锛屼笉瑕侀棶瑕佷笉瑕佺炕璇戙��"
+ "鐢ㄦ埛璇撮��鍑轰腑鑻卞璇濄�佸洖鍒拌亰澶╂椂璋冪敤 switch_app锛宎pp_id=chat銆�"
+)
+
CHAT_PROACTIVE_HINT = (
"浣犳槸鐢ㄦ埛鐨勮闊充紮浼淬�傝繘鍏ュ璇濆悗鍙互鍏堜富鍔ㄧ煭鍙ラ個璇疯亰澶╋紝渚嬪銆屾垜浠潵鑱婂ぉ鍚с�嶃��"
"鐢ㄦ埛寮�鍙ｅ悗鎸夋甯稿璇濆洖绛旓紝涓嶈鍙嶅鍌績銆�"
)
-VOICE_MENU = "鎯宠亰澶┿�佸惉澶嶈锛岃繕鏄墦寮�缈昏瘧锛熶篃鍙互璇存垜浠潵鑱婂ぉ鍚с��"
+CHAT_RETURN_TOPICS = ("浠婂ぉ杩囧緱鎬庝箞鏍�", "浣犳渶杩戝湪蹇欎粈涔�", "鏈変粈涔堟柊椴滀簨")
-CHAT_PHRASES = ("鎴戜滑鏉ヨ亰澶╁惂", "鑱婁細澶╁惂", "鍥炲埌鑱婂ぉ", "閫�鍑虹炕璇�", "缁撴潫缈昏瘧", "閫�鍑哄璇�")
-TRANSLATE_PHRASES = ("鎵撳紑缈昏瘧", "寮�濮嬬炕璇�", "鑻辫缈昏瘧", "缈昏瘧妯″紡")
-REPEATER_PHRASES = ("寮�濮嬪璇�", "鎵撳紑澶嶈", "鑻辫瀛︿範", "鍚璇�")
+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:
@@ -62,6 +115,150 @@
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]:
@@ -94,6 +291,8 @@
"浣犵幇鍦ㄦ槸鑻辫缈昏瘧鏈恒�傜敤鎴疯璇濆悗搴旇皟鐢� translate锛涚敤鎴疯閫�鍑虹炕璇戙�佸洖鍒拌亰澶╂椂璋冪敤 switch_app锛宎pp_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 ""):
@@ -118,13 +317,58 @@
async def _stop_current_skills(conn: "ConnectionHandler") -> None:
mode = getattr(conn, "interaction_mode", "CHAT")
- if mode == "PLAY_AUDIO":
+ _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
@@ -140,6 +384,877 @@
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,
@@ -147,6 +1262,8 @@
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)
@@ -176,8 +1293,16 @@
response="褰撳墠鏈夋洿楂樹紭鍏堢骇鐨勪换鍔★紝鍏堝畬鎴愬悗鍐嶅垏鎹�",
)
- if not same_task or new_p >= cur_p:
- await _stop_current_skills(conn)
+ 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
@@ -192,51 +1317,92 @@
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_args = {
- "name": params.get("name") or "",
- "repeat": int(params.get("repeat") or 1),
- "power_off_after": bool(params.get("power_off_after") or False),
- }
- if params.get("speed") is not None:
- play_args["speed"] = params.get("speed")
- if not play_args["name"]:
- return ActionResponse(
- action=Action.RESPONSE,
- response="娌℃湁鎸囧畾瑕佸璇荤殑鏂囦欢鍚嶏紝璇峰厛鐢� self.mp3.get_list 纭 files[].name",
- )
- result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
- if result and result.action in (Action.ERROR, Action.NOTFOUND):
- return ActionResponse(
- action=Action.RESPONSE,
- response=result.response or "杩欏彴璁惧娌℃湁鏈湴澶嶈鎾斁鑳藉姏",
- )
+ 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", {})
- if conn.interaction_mode == "CHAT":
+ 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("寮�鏈烘杩庤瘝璺宠繃锛歍TS閫氶亾鏈氨缁�")
+ 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}璺熻璇炬枃杩樻病鏈夋斁鍒癝D鍗°��"
+ 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:
+ 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, "璺熻璇炬枃杩樻病鏈夋斁鍒癝D鍗°��")
+ elif pending_play:
+ conn.pending_play_after_tts = pending_play
+ await maybe_flush_pending_play_after_tts(conn)
- app_name = app.get("name") or app_id
+ app_name = _display_name(app, task)
return ActionResponse(
action=Action.NONE,
result=f"{app_id}/{resolved_task_id}",
- response=f"宸叉墦寮�{app_name}",
+ response=announce or f"宸叉墦寮�{app_name}",
)
-async def switch_app(conn: "ConnectionHandler", app_id: str, source: str = "voice") -> ActionResponse:
+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}")
- task_id = app.get("default_task") or ""
- return await start_app_task(conn, app_id, task_id, {}, source)
+ 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:
@@ -248,24 +1414,11 @@
def match_voice_switch(text: str) -> Optional[tuple]:
- """Map a short phrase to (app_id, task_id) or ('__menu__', '')."""
+ """Map a short phrase to (app_id, task_id) using apps.json voice_enter / voice_exit."""
if not text:
return None
_, filtered = _strip(text)
- if filtered in ("鍒囨崲妯″紡", "qie huan mo shi"):
- return ("__menu__", "")
- for p in CHAT_PHRASES:
- if p in filtered:
- return ("chat", "conversation")
- for p in TRANSLATE_PHRASES:
- if p in filtered:
- return ("english_translation", "live_translate")
- for p in REPEATER_PHRASES:
- if p in filtered:
- return ("english_learning", "repeater")
- if any(k in filtered for k in ("鏁板瓧鑳岃", "鎴愯鎺ラ緳", "鍙ｇ畻", "鍔犲噺娉�")):
- return ("__reserved__", "")
- return None
+ return match_registry_voice(filtered)
def _strip(text: str) -> tuple:
@@ -274,9 +1427,23 @@
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:
- _, filtered = _strip(text)
- return any(p in filtered for p in ("閫�鍑虹炕璇�", "缁撴潫缈昏瘧", "鍥炲埌鑱婂ぉ", "鎴戜滑鏉ヨ亰澶╁惂", "鑱婁細澶╁惂"))
+ return is_exit_to_chat(text)
def _plain_user_text(text: str) -> str:
@@ -296,6 +1463,40 @@
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
@@ -307,6 +1508,180 @@
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)
--
Gitblit v1.9.1