""" NovaAi – TTS-Engine-Hub engines/piper.py Version: v0.1.1 Description: Piper TTS engine adapter: real CLI invocation + output as WAV, OGG, or MP3. Synthesizes WAV via Piper, converts to OGG/MP3 via ffmpeg-python if needed. Uses dynamic model path: ./models/piper/[model]/model.onnx Author: Abby (ChatGPT) Date: 2025-07-23 Canvas: piper.py """ import asyncio import subprocess import tempfile import os import shutil import json import logging from .engine_base import TTSEngineBase from app.config import settings import ffmpeg logger = logging.getLogger(__name__) class PiperEngine(TTSEngineBase): def __init__(self): self.piper_executable = shutil.which("piper") self.ffmpeg_executable = shutil.which("ffmpeg") self.PIPER_TIMEOUT_SECONDS = settings.PIPER_TIMEOUT_SECONDS self.FFMPEG_TIMEOUT_SECONDS = settings.FFMPEG_TIMEOUT_SECONDS def _load_config(self, model: str): """Load the model config JSON file to get speaker mappings.""" model_dir = f"/models/piper/{model}" config_file = os.path.join(model_dir, f"{model}.onnx.json") if os.path.isfile(config_file): with open(config_file, 'r') as f: return json.load(f) return {} def _get_speaker_id(self, speaker: str, model: str): """Convert speaker name to speaker ID using the model's config.""" if not speaker or speaker == "default": return None if speaker.isdigit(): return speaker config = self._load_config(model) speaker_id_map = config.get('speaker_id_map', {}) return str(speaker_id_map.get(speaker)) def _run_ffmpeg_blocking(self, input_path, output_path): """ Wrapper for the blocking ffmpeg call with error capture. Bug #5 fix: Captures stderr for better error reporting. """ try: stdout, stderr = ( ffmpeg .input(input_path) .output(output_path) .run(overwrite_output=True, capture_stdout=True, capture_stderr=True) ) # Log stderr even on success (ffmpeg writes info there) if stderr: logger.debug(f"FFmpeg output: {stderr.decode('utf-8', errors='replace')}") except ffmpeg.Error as e: stderr_output = e.stderr.decode('utf-8', errors='replace') if e.stderr else "No error output" logger.error(f"FFmpeg conversion failed: {input_path} -> {output_path}. Error: {stderr_output}") raise RuntimeError( f"FFmpeg conversion failed: {input_path} -> {output_path}. " f"Error: {stderr_output}" ) async def synthesize(self, text: str, speaker: str = None, model: str = None, fmt: str = "ogg"): """ Synthesize speech from text using Piper TTS. Comprehensive bug fixes applied: - Bug #1: Remove invalid --stdin flag - Bug #2: Use mkstemp to avoid file handle race - Bug #3: Add timeouts to prevent hung processes - Bug #4: Comprehensive temp file cleanup - Bug #5: Capture ffmpeg errors properly - Bug #6: Validate config file upfront - Bug #7: Enhanced error context and logging """ # Validation if not self.piper_executable: raise RuntimeError("Piper executable not found. Please install it and ensure it's in your PATH.") if not model: raise ValueError("Model must be specified for Piper.") # Validate model and config files (Bug #6) model_dir = f"/models/piper/{model}" model_file = os.path.join(model_dir, f"{model}.onnx") config_file = os.path.join(model_dir, f"{model}.onnx.json") if not os.path.isfile(model_file): raise FileNotFoundError(f"Piper model not found: {model_file}") if not os.path.isfile(config_file): if speaker and speaker != "default": # Config required for speaker mapping raise FileNotFoundError( f"Piper config file not found: {config_file}. " f"Config required for speaker '{speaker}' selection." ) # Just warn if no speaker requested logger.warning( f"Piper config file not found: {config_file}. " f"Speaker selection will not be available for model '{model}'." ) # Validate speaker exists in config if specified if speaker and speaker != "default": speaker_id = self._get_speaker_id(speaker, model) if speaker_id is None or speaker_id == "None": # Load config to get available speakers for error message config = self._load_config(model) available_speakers = list(config.get('speaker_id_map', {}).keys()) raise ValueError( f"Speaker '{speaker}' not found for model '{model}'. " f"Available speakers: {available_speakers or ['default']}" ) # Track temp files for cleanup (Bug #4) temp_files_to_cleanup = [] try: # Create WAV temp file (Bug #2 - use mkstemp) fd, output_wav_path = tempfile.mkstemp(suffix=".wav", prefix="piper_") os.close(fd) temp_files_to_cleanup.append(output_wav_path) # Build command (Bug #1 - no --stdin flag, piper reads stdin by default) cmd = [self.piper_executable, "--model", model_file, "--output-file", output_wav_path] if speaker: speaker_id = self._get_speaker_id(speaker, model) if speaker_id: cmd += ["--speaker", speaker_id] # Log command (Bug #7) text_preview = text[:100] + "..." if len(text) > 100 else text logger.debug(f"Executing piper command: {' '.join(cmd)}") logger.debug(f"Input text ({len(text)} chars): {text_preview}") # Execute with timeout (Bug #3) process = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) try: stdout, stderr = await asyncio.wait_for( process.communicate(input=text.encode('utf-8')), timeout=self.PIPER_TIMEOUT_SECONDS ) except asyncio.TimeoutError: process.kill() await process.wait() logger.error( f"Piper synthesis timed out. Command: {' '.join(cmd)}, " f"Text length: {len(text)}, Model: {model}, Speaker: {speaker}" ) raise RuntimeError( f"Piper synthesis timed out after {self.PIPER_TIMEOUT_SECONDS}s. " f"Text length: {len(text)} chars. Model: {model}" ) # Enhanced error reporting (Bug #7) if process.returncode != 0: stderr_text = stderr.decode('utf-8', errors='replace') stdout_text = stdout.decode('utf-8', errors='replace') error_msg = ( f"Piper synthesis failed (exit code {process.returncode})\\n" f"Command: {' '.join(cmd)}\\n" f"Model: {model}\\n" f"Speaker: {speaker}\\n" f"Text length: {len(text)} chars\\n" f"Text preview: {text_preview}\\n" f"Stderr: {stderr_text}\\n" f"Stdout: {stdout_text}" ) logger.error(error_msg) raise RuntimeError( f"Piper synthesis failed (exit code {process.returncode}): {stderr_text}. " f"Command: {' '.join(cmd)}. See logs for full details." ) # Verify output file was created (Bug #7) if not os.path.exists(output_wav_path): error_msg = ( f"Piper synthesis failed: output file not created.\\n" f"Command: {' '.join(cmd)}\\n" f"Return code: {process.returncode} (success)\\n" f"This may indicate a bug in piper or incorrect command parameters." ) logger.error(error_msg) raise RuntimeError(error_msg) if os.path.getsize(output_wav_path) == 0: error_msg = ( f"Piper synthesis failed: output file is empty.\\n" f"Command: {' '.join(cmd)}\\n" f"This may indicate invalid input or model issues." ) logger.error(error_msg) raise RuntimeError(error_msg) logger.info( f"Piper synthesis succeeded: {len(text)} chars -> " f"{os.path.getsize(output_wav_path)} bytes. Model: {model}, Speaker: {speaker}" ) # Return WAV if requested fmt = (fmt or "ogg").lower() if fmt == "wav": temp_files_to_cleanup.remove(output_wav_path) return output_wav_path # FFmpeg conversion if not self.ffmpeg_executable: raise RuntimeError("ffmpeg not found, cannot convert audio format.") # Create converted file temp path (Bug #2) fd_conv, output_other_path = tempfile.mkstemp(suffix=f'.{fmt}', prefix="piper_conv_") os.close(fd_conv) temp_files_to_cleanup.append(output_other_path) logger.debug(f"Converting WAV to {fmt}: {output_wav_path} -> {output_other_path}") # Convert with timeout (Bug #3, #5) try: await asyncio.wait_for( asyncio.to_thread(self._run_ffmpeg_blocking, output_wav_path, output_other_path), timeout=self.FFMPEG_TIMEOUT_SECONDS ) except asyncio.TimeoutError: logger.error( f"FFmpeg conversion timed out after {self.FFMPEG_TIMEOUT_SECONDS}s. " f"Input size: {os.path.getsize(output_wav_path)} bytes" ) raise RuntimeError( f"FFmpeg conversion timed out after {self.FFMPEG_TIMEOUT_SECONDS}s. " f"Input size: {os.path.getsize(output_wav_path)} bytes" ) # Verify conversion succeeded if not os.path.exists(output_other_path) or os.path.getsize(output_other_path) == 0: raise RuntimeError("FFmpeg conversion failed: output file not created or empty") logger.info( f"FFmpeg conversion succeeded: {os.path.getsize(output_wav_path)} bytes (WAV) -> " f"{os.path.getsize(output_other_path)} bytes ({fmt})" ) # Success! Remove converted file from cleanup (we're returning it) temp_files_to_cleanup.remove(output_other_path) return output_other_path finally: # Cleanup all temp files (Bug #4) for temp_file in temp_files_to_cleanup: try: if os.path.exists(temp_file): os.remove(temp_file) logger.debug(f"Cleaned up temp file: {temp_file}") except Exception as e: # Log but don't raise - we're in cleanup logger.warning(f"Failed to cleanup temp file {temp_file}: {e}") def list_models(self): models_dir = "/models/piper/" if not os.path.isdir(models_dir): return [] return [name for name in os.listdir(models_dir) if os.path.isdir(os.path.join(models_dir, name))] def list_voices(self, model: str = None): if not model: return ["default"] config = self._load_config(model) speaker_id_map = config.get('speaker_id_map', {}) if speaker_id_map: return ["default"] + sorted(speaker_id_map.keys()) return ["default"] def healthcheck(self): status = "ok" if not self.piper_executable: status = "missing_piper_executable" return {"status": status, "engine": "piper"} async def selftest(self): if not self.piper_executable: return {"selftest": False, "error": "Piper executable not found.", "engine": "piper"} try: models = self.list_models() if not models: return {"selftest": False, "error": "No Piper models found.", "engine": "piper"} test_text = "This is a selftest." first_model = models[0] voices = self.list_voices(first_model) test_voice = voices[0] if voices else None audio_file = await self.synthesize(test_text, speaker=test_voice, model=first_model, fmt="wav") selftest_passed = os.path.exists(audio_file) and os.path.getsize(audio_file) > 0 if selftest_passed: os.remove(audio_file) return {"selftest": selftest_passed, "models": models, "engine": "piper"} except Exception as e: return {"selftest": False, "error": str(e), "engine": "piper"} if __name__ == "__main__": async def main(): engine = PiperEngine() print("Selftest:", await engine.selftest()) print("Models:", engine.list_models()) print("Voices:", engine.list_voices(engine.list_models()[0])) print("Healthcheck:", engine.healthcheck()) asyncio.run(main())