from __future__ import annotations import argparse import json import re import subprocess from pathlib import Path from .audio_preprocess import SherpaOnnxDenoiserPreprocessor from .assets import validate_pet_assets from .config import AppConfig from .conversation import ConversationContext from .llm import MockLlmProvider, OpenAICompatibleLlmProvider from .models import AudioFrame, ProviderError from .pipeline import VoicePipeline from .real_live_check import run_real_live_check from .runtime import build_live_runtime from .simulation import run_simulated_live from .speech_models import check_speech_models, model_status_errors from .stt import MetadataSttProvider, SherpaOnnxSttProvider from .transport import MemoryAudioTransport, sounddevice_device_report from .tts import SineTtsProvider from .vad import EnergyVadProvider, SherpaOnnxVadProvider, VadRecorder from .wakeword import KeywordWakeWordProvider, SherpaOnnxKeywordWakeWordProvider def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="owner-voice-pet") parser.add_argument("--env-file", default=".env", help="Path to .env config file") parser.add_argument("--show-config", action="store_true", help="Print non-secret config summary") subparsers = parser.add_subparsers(dest="command") subparsers.add_parser("acceptance", help="Run deterministic end-to-end pipeline acceptance") subparsers.add_parser("validate-assets", help="Validate project pet assets") subparsers.add_parser("security-check", help="Scan tracked files for leaked API keys") model_check = subparsers.add_parser("model-check", help="Validate local speech model files") model_check.add_argument("--models-dir", default=None, help="Speech models directory. Defaults to .env or models") subparsers.add_parser("device-check", help="Validate local microphone and speaker availability") live = subparsers.add_parser("run-live", help="Run real repeated live voice conversation") live.add_argument("--once", action="store_true", help="Run one completed live turn and exit") simulate = subparsers.add_parser("simulate-live", help="Run live pipeline with simulated microphone frames") simulate.add_argument("--turns", type=int, default=2, help="Number of simulated turns. Default: 2") simulate.add_argument("--fixture", default=None, help="Replay simulated microphone frames from JSONL") simulate.add_argument("--write-fixture", default=None, help="Write generated simulated microphone frames to JSONL") real_check = subparsers.add_parser("real-live-check", help="Run generated-audio live check with real providers") real_check.add_argument("--turns", type=int, default=2, help="Number of generated live turns. Default: 2") real_check.add_argument("--voice", default="Tingting", help="macOS say voice used for generated microphone input") real_check.add_argument("--wake-text", default="小杰小杰。", help="Generated wake utterance") real_check.add_argument("--question", action="append", default=None, help="Generated user question; can be repeated") real_check.add_argument("--no-playback", action="store_true", help="Synthesize but do not play generated TTS output") smoke = subparsers.add_parser("llm-smoke", help="Call configured OpenAI/NewAPI endpoint") smoke.add_argument("--message", default="用一句中文回复：小杰在线。") smoke.add_argument("--no-stream", action="store_true") args = parser.parse_args(argv) if args.show_config: config = AppConfig.from_dotenv(args.env_file) print( json.dumps( { "wake_word": config.wake_word, "sample_rate": config.sample_rate, "channels": config.channels, "llm_base_url": config.llm_base_url, "llm_model": config.llm_model, "llm_api_style": config.llm_api_style, "llm_stream": config.llm_stream, "realtime_transcript_enabled": config.realtime_transcript_enabled, "llm_api_key_present": bool(config.llm_api_key), "asset_dir": str(config.asset_dir), "wake_provider": config.wake_provider, "wake_keywords_file": str(config.wake_keywords_file) if config.wake_keywords_file else "", "wake_kws_threshold": config.wake_kws_threshold, "wake_kws_score": config.wake_kws_score, "wake_ack_text": config.wake_ack_text, "post_playback_drain_ms": config.post_playback_drain_ms, "pipeline_mode": config.pipeline_mode, "endpoint_mode": config.endpoint_mode, "noise_filter_enabled": config.noise_filter_enabled, "noise_filter_provider": config.noise_filter_provider, "wake_denoise_enabled": config.wake_denoise_enabled, "speaker_profile_ms": config.speaker_profile_ms, "speaker_profile_min_ms": config.speaker_profile_min_ms, "speaker_absent_ms": config.speaker_absent_ms, "speaker_similarity_threshold": config.speaker_similarity_threshold, "speaker_min_rms": config.speaker_min_rms, "vad_provider": config.vad_provider, "vad_threshold": config.vad_threshold, "vad_min_duration_ms": config.vad_min_duration_ms, "vad_end_silence_ms": config.vad_end_silence_ms, "vad_no_speech_timeout_ms": config.vad_no_speech_timeout_ms, "vad_max_recording_ms": config.vad_max_recording_ms, "speech_provider": config.speech_provider, "asr_model": config.asr_model, "tts_model": config.tts_model, "tts_voice": config.tts_voice, "speech_models_dir": str(config.speech_models_dir), "context_mode": config.context_mode, }, ensure_ascii=False, sort_keys=True, ) ) return 0 if args.command == "validate-assets": infos = validate_pet_assets(AppConfig.from_dotenv(args.env_file).asset_dir) print(json.dumps({"assets": len(infos), "valid": True}, ensure_ascii=False, sort_keys=True)) return 0 if args.command == "security-check": leaks = find_secret_leaks(Path.cwd()) print(json.dumps({"secret_leaks": leaks, "valid": not leaks}, ensure_ascii=False, sort_keys=True)) return 1 if leaks else 0 if args.command == "model-check": config = AppConfig.from_dotenv(args.env_file) models_dir = Path(args.models_dir) if args.models_dir else config.speech_models_dir status = check_speech_models(models_dir, require_sherpa=True) errors = model_status_errors(status) provider_load_checked = False if not errors: try: SherpaOnnxKeywordWakeWordProvider( models_dir, keyword=config.wake_word, keywords_file=config.wake_keywords_file, threshold=config.wake_kws_threshold, score=config.wake_kws_score, ).load() SherpaOnnxVadProvider(models_dir).load() SherpaOnnxSttProvider(str(models_dir)).load() SherpaOnnxDenoiserPreprocessor(models_dir).load() provider_load_checked = True except ProviderError as exc: errors.append(exc) data = status.to_json() data["errors"] = [str(error) for error in errors] data["provider_load_checked"] = provider_load_checked print(json.dumps(data, ensure_ascii=False, sort_keys=True)) return 1 if errors else 0 if args.command == "device-check": report = sounddevice_device_report() print(json.dumps(report, ensure_ascii=False, sort_keys=True)) return 0 if report["ok"] else 1 if args.command == "run-live": config = AppConfig.from_dotenv(args.env_file) try: summary = build_live_runtime(config).run(once=args.once) except ProviderError as exc: print(json.dumps({"ok": False, "code": exc.code.value, "message": exc.message}, ensure_ascii=False, sort_keys=True)) return 1 return 0 if summary.completed_turns > 0 or summary.interrupted else 1 if args.command == "simulate-live": try: data = run_simulated_live( turns=args.turns, fixture_path=args.fixture, write_fixture=args.write_fixture, ) except (ProviderError, ValueError) as exc: print(json.dumps({"success": False, "error": str(exc)}, ensure_ascii=False, sort_keys=True)) return 1 print(json.dumps(data, ensure_ascii=False, sort_keys=True)) return 0 if data["success"] else 1 if args.command == "real-live-check": config = AppConfig.from_dotenv(args.env_file) try: data = run_real_live_check( config=config, turns=args.turns, voice=args.voice, wake_text=args.wake_text, questions=args.question, play_audio=not args.no_playback, ) except (ProviderError, ValueError) as exc: print(json.dumps({"success": False, "error": str(exc)}, ensure_ascii=False, sort_keys=True)) return 1 print(json.dumps(data, ensure_ascii=False, sort_keys=True)) return 0 if data["success"] else 1 if args.command == "acceptance": result = run_acceptance() print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if result["success"] else 1 if args.command == "llm-smoke": config = AppConfig.from_dotenv(args.env_file) if args.no_stream: config = AppConfig( wake_word=config.wake_word, sample_rate=config.sample_rate, channels=config.channels, llm_base_url=config.llm_base_url, llm_api_key=config.llm_api_key, llm_model=config.llm_model, llm_api_style=config.llm_api_style, llm_stream=False, realtime_transcript_enabled=config.realtime_transcript_enabled, audio_input_device=config.audio_input_device, audio_output_device=config.audio_output_device, asset_dir=config.asset_dir, log_dir=config.log_dir, wake_provider=config.wake_provider, wake_keywords_file=config.wake_keywords_file, wake_kws_threshold=config.wake_kws_threshold, wake_kws_score=config.wake_kws_score, wake_ack_text=config.wake_ack_text, post_playback_drain_ms=config.post_playback_drain_ms, pipeline_mode=config.pipeline_mode, endpoint_mode=config.endpoint_mode, noise_filter_enabled=config.noise_filter_enabled, noise_filter_provider=config.noise_filter_provider, wake_denoise_enabled=config.wake_denoise_enabled, speaker_profile_ms=config.speaker_profile_ms, speaker_profile_min_ms=config.speaker_profile_min_ms, speaker_absent_ms=config.speaker_absent_ms, speaker_similarity_threshold=config.speaker_similarity_threshold, speaker_min_rms=config.speaker_min_rms, vad_provider=config.vad_provider, vad_threshold=config.vad_threshold, vad_min_duration_ms=config.vad_min_duration_ms, vad_end_silence_ms=config.vad_end_silence_ms, vad_no_speech_timeout_ms=config.vad_no_speech_timeout_ms, vad_max_recording_ms=config.vad_max_recording_ms, speech_provider=config.speech_provider, asr_model=config.asr_model, tts_model=config.tts_model, tts_voice=config.tts_voice, speech_models_dir=config.speech_models_dir, context_mode=config.context_mode, context_max_messages=config.context_max_messages, context_max_chars=config.context_max_chars, ) try: provider = OpenAICompatibleLlmProvider(config, timeout_s=30) messages = [ConversationContext().build_llm_messages()[0]] messages.append(__import__("owner_voice_pet.models", fromlist=["Message"]).Message("user", args.message, 1.0)) text = "".join(delta.text_delta for delta in provider.stream_reply(messages)).strip() print(json.dumps({"ok": bool(text), "reply_preview": text[:80]}, ensure_ascii=False, sort_keys=True)) return 0 if text else 1 except ProviderError as exc: print(json.dumps({"ok": False, "code": exc.code.value, "message": exc.message}, ensure_ascii=False, sort_keys=True)) return 1 parser.print_help() return 0 def run_acceptance() -> dict[str, object]: validate_pet_assets(AppConfig.from_dotenv().asset_dir) frames = [ AudioFrame(b"\x80\x80", 16000, 1, 0, 0, {"duration_ms": 20, "wake_word": "小杰小杰", "wake_confidence": 0.99}), AudioFrame(b"\xff\xff", 16000, 1, 20, 1, {"duration_ms": 20, "speech": True, "transcript": "你是谁"}), AudioFrame(b"\xff\xff", 16000, 1, 40, 2, {"duration_ms": 20, "speech": True}), AudioFrame(b"\x80\x80", 16000, 1, 60, 3, {"duration_ms": 20, "speech": False}), AudioFrame(b"\x80\x80", 16000, 1, 80, 4, {"duration_ms": 20, "speech": False}), ] transport = MemoryAudioTransport(frames) pipeline = VoicePipeline( transport=transport, wakeword=KeywordWakeWordProvider(), vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40), stt=MetadataSttProvider(), context=ConversationContext(), llm=MockLlmProvider(["我是小杰桌宠，已经在线。"]), tts=SineTtsProvider(), ) pipeline.load() result = pipeline.run_once() return { "success": result.success, "transcript": result.transcript, "assistant_text": result.assistant_text, "played_segments": result.played_segments, "states": [state.value for state in result.states], } def find_secret_leaks(root: Path) -> list[str]: completed = subprocess.run(["git", "ls-files"], cwd=root, check=True, stdout=subprocess.PIPE, text=True) pattern = re.compile(r"(?:sk|tp)-[A-Za-z0-9_\\-]{16,}") leaks: list[str] = [] for rel in completed.stdout.splitlines(): path = root / rel if not path.exists(): continue if path.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}: continue try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: continue if pattern.search(text): leaks.append(rel) return leaks