"""Load the Phase 1 App / Task registry from JSON.
|
|
Registry lives on xiaozhi-server only (bundled apps.json, optional data/apps.json).
|
Task voice_enter / voice_exit / enter_prompt live here so new Apps do not need a
|
firmware flash. The device only keeps a small offline MultiNet set (mode switch,
|
repeater, exit repeater, return to chat) plus NVS pending / last-repeater state.
|
"""
|
|
import json
|
import os
|
from typing import Any, Dict, Optional, Tuple
|
|
from config.config_loader import get_project_dir
|
from config.logger import setup_logging
|
|
TAG = __name__
|
logger = setup_logging()
|
|
_REGISTRY: Optional[Dict[str, Any]] = None
|
_VOICE_INDEX: Optional[list] = None
|
_REGISTRY_PATH: Optional[str] = None
|
_REGISTRY_MTIME: Optional[float] = None
|
|
_BUNDLED_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "apps.json")
|
_MENU_PRIO = 0
|
_EXIT_PRIO = 1
|
_CHAT_ENTER_PRIO = 2
|
_ENTER_PRIO = 3
|
_RESERVED_PRIO = 4
|
|
|
def _registry_candidates() -> list:
|
return [
|
os.path.join(get_project_dir(), "data", "apps.json"),
|
_BUNDLED_PATH,
|
]
|
|
|
def _file_mtime(path: str) -> Optional[float]:
|
try:
|
return os.path.getmtime(path)
|
except OSError:
|
return None
|
|
|
def _load_registry() -> Dict[str, Any]:
|
global _REGISTRY_PATH, _REGISTRY_MTIME
|
for path in _registry_candidates():
|
if not os.path.isfile(path):
|
continue
|
try:
|
with open(path, "r", encoding="utf-8") as f:
|
data = json.load(f)
|
if isinstance(data, dict) and data.get("apps"):
|
_REGISTRY_PATH = path
|
_REGISTRY_MTIME = _file_mtime(path)
|
logger.bind(tag=TAG).info(f"已加载 App 注册表: {path}")
|
return data
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"加载 App 注册表失败 {path}: {e}")
|
_REGISTRY_PATH = None
|
_REGISTRY_MTIME = None
|
logger.bind(tag=TAG).warning("未找到 apps.json，使用内置最小注册表")
|
return {
|
"default_app_id": "chat",
|
"apps": {
|
"chat": {
|
"name": "聊天",
|
"mode": "CHAT",
|
"default_task": "conversation",
|
"tool_policy": "inherit_agent",
|
"system_prompt_mode": "agent_proactive",
|
"tasks": {"conversation": {"interaction_mode": "CHAT", "skills": []}},
|
}
|
},
|
}
|
|
|
def get_registry() -> Dict[str, Any]:
|
global _REGISTRY, _VOICE_INDEX
|
if _REGISTRY is None:
|
_REGISTRY = _load_registry()
|
return _REGISTRY
|
if _REGISTRY_PATH:
|
top = next((item for item in _registry_candidates() if os.path.isfile(item)), None)
|
mtime = _file_mtime(_REGISTRY_PATH)
|
if top != _REGISTRY_PATH or (mtime is not None and mtime != _REGISTRY_MTIME):
|
_REGISTRY = _load_registry()
|
_VOICE_INDEX = None
|
return _REGISTRY
|
|
|
def reset_registry_cache() -> None:
|
"""Tests may swap the registry; drop derived indexes."""
|
global _VOICE_INDEX
|
_VOICE_INDEX = None
|
|
|
def iter_tasks():
|
apps = get_registry().get("apps") or {}
|
for app_id, app in apps.items():
|
if not isinstance(app, dict):
|
continue
|
for task_id, task in (app.get("tasks") or {}).items():
|
if isinstance(task, dict):
|
yield app_id, app, task_id, task
|
|
|
def _phrase_list(raw) -> list:
|
if raw is None:
|
return []
|
if isinstance(raw, str):
|
text = raw.strip()
|
return [text] if text else []
|
phrases = []
|
for item in raw:
|
text = str(item or "").strip()
|
if text:
|
phrases.append(text)
|
return phrases
|
|
|
def _build_voice_index() -> list:
|
"""Longer phrase first, then menu/exit/chat-enter/enter/reserved."""
|
entries = []
|
registry = get_registry()
|
for phrase in _phrase_list(registry.get("voice_menu_phrases")):
|
entries.append((phrase, ("__menu__", ""), _MENU_PRIO))
|
for app_id, app, task_id, task in iter_tasks():
|
default_task = app.get("default_task") or task_id
|
for phrase in _phrase_list(task.get("voice_exit")):
|
entries.append((phrase, ("chat", "conversation"), _EXIT_PRIO))
|
for phrase in _phrase_list(task.get("voice_enter")):
|
if app_id == "chat":
|
entries.append((phrase, ("chat", default_task or task_id), _CHAT_ENTER_PRIO))
|
else:
|
entries.append((phrase, (app_id, task_id), _ENTER_PRIO))
|
for phrase in _phrase_list(registry.get("reserved_voice_enter")):
|
entries.append((phrase, ("__reserved__", ""), _RESERVED_PRIO))
|
entries.sort(key=lambda item: (-len(item[0]), item[2]))
|
return entries
|
|
|
def _voice_index() -> list:
|
global _VOICE_INDEX
|
if _VOICE_INDEX is None:
|
_VOICE_INDEX = _build_voice_index()
|
return _VOICE_INDEX
|
|
|
def match_registry_voice(filtered: str) -> Optional[Tuple[str, str]]:
|
"""Map stripped ASR text to (app_id, task_id) or ('__menu__'|'__reserved__', '')."""
|
if not filtered:
|
return None
|
for phrase, target, _prio in _voice_index():
|
if phrase and phrase in filtered:
|
return target
|
return None
|
|
|
def voice_menu_text() -> str:
|
labels = []
|
chat_hint = "我们来聊天吧"
|
for app_id, _app, task_id, task in iter_tasks():
|
enters = _phrase_list(task.get("voice_enter"))
|
if app_id == "chat" and task_id == "conversation" and enters:
|
chat_hint = enters[0]
|
name = str(task.get("menu_name") or "").strip()
|
if name:
|
labels.append(name)
|
if not labels:
|
return f"想聊天？也可以说{chat_hint}。"
|
if len(labels) == 1:
|
body = labels[0]
|
else:
|
body = "、".join(labels[:-1]) + f"，还是{labels[-1]}"
|
return f"想{body}？也可以说{chat_hint}。"
|
|
|
def default_app_id() -> str:
|
return get_registry().get("default_app_id") or "chat"
|
|
|
def get_app(app_id: str) -> Optional[Dict[str, Any]]:
|
if not app_id:
|
return None
|
return get_registry().get("apps", {}).get(app_id)
|
|
|
def get_task(app_id: str, task_id: str) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
|
app = get_app(app_id)
|
if not app:
|
return None, None
|
tasks = app.get("tasks") or {}
|
if not task_id:
|
task_id = app.get("default_task") or ""
|
task = tasks.get(task_id)
|
if not task:
|
return app, None
|
return app, task
|
