""" Piper TTS service for generating multi-voice audiobooks. """ import logging import io import os import subprocess from typing import Dict, List, Tuple, Optional import numpy as np import soundfile as sf from pydub import AudioSegment from pydub.effects import speedup logger = logging.getLogger(__name__) class PiperTTSService: """ Service for text-to-speech synthesis using Piper TTS. Piper is a fast, local neural TTS system that supports multiple voices and allows pitch modification through audio processing. """ # Default voice for unmapped roles DEFAULT_VOICE = "en_US-lessac-medium" # Sample rate for Piper TTS (typically 22050) SAMPLE_RATE = 22050 # Available Piper voices with their characteristics AVAILABLE_VOICES = { # English voices "en_US-lessac-medium": {"lang": "en", "gender": "male", "quality": "medium"}, "en_US-libritts-high": {"lang": "en", "gender": "neutral", "quality": "high"}, "en_US-amy-medium": {"lang": "en", "gender": "female", "quality": "medium"}, "en_GB-alan-medium": {"lang": "en", "gender": "male", "quality": "medium"}, "en_GB-alba-medium": {"lang": "en", "gender": "female", "quality": "medium"}, } def __init__(self): """Initializes Piper TTS service.""" self.models_dir = os.path.join(os.path.dirname(__file__), "..", "piper_models") os.makedirs(self.models_dir, exist_ok=True) logger.info("PiperTTSService initialized") def _check_piper_installed(self) -> bool: """Check if Piper is installed and accessible.""" try: result = subprocess.run( ["piper", "--version"], capture_output=True, text=True, timeout=5 ) return result.returncode == 0 except (subprocess.SubprocessError, FileNotFoundError): return False def _download_voice_model(self, voice: str) -> bool: """ Download a Piper voice model if not already present. Args: voice: Voice ID to download Returns: True if model is available, False otherwise """ model_path = os.path.join(self.models_dir, f"{voice}.onnx") config_path = os.path.join(self.models_dir, f"{voice}.onnx.json") # Check if model already exists if os.path.exists(model_path) and os.path.exists(config_path): logger.info(f"Voice model {voice} already exists") return True logger.info(f"Downloading Piper voice model: {voice}") logger.info(f"Models directory: {self.models_dir}") # Try to download using wget/curl as Piper's download might not work try: # Piper models are hosted on HuggingFace base_url = "https://huggingface.co/rhasspy/piper-voices/resolve/v1.0.0" # Determine language prefix from voice ID lang_prefix = voice.split('-')[0] # e.g., "en_US" from "en_US-lessac-medium" model_url = f"{base_url}/{lang_prefix}/{voice}.onnx" config_url = f"{base_url}/{lang_prefix}/{voice}.onnx.json" logger.info(f"Downloading from: {model_url}") # Try using curl/wget first (more reliable) try: # Try curl subprocess.run( ["curl", "-L", "-o", model_path, model_url], check=True, timeout=120, capture_output=True ) subprocess.run( ["curl", "-L", "-o", config_path, config_url], check=True, timeout=120, capture_output=True ) except (subprocess.SubprocessError, FileNotFoundError): # Fall back to urllib with SSL context import urllib.request import ssl # Create SSL context that doesn't verify certificates (for development) ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE urllib.request.urlretrieve(model_url, model_path, context=ssl_context) urllib.request.urlretrieve(config_url, config_path, context=ssl_context) if os.path.exists(model_path) and os.path.exists(config_path): logger.info(f"Successfully downloaded voice model: {voice}") return True else: logger.error(f"Download completed but files not found") return False except Exception as e: logger.error(f"Failed to download voice model {voice}: {e}") logger.error(f"Please manually download from: https://github.com/rhasspy/piper/releases/tag/v1.2.0") logger.error(f"See backend/PIPER_SETUP.md for detailed instructions") return False def synthesize_with_pitch( self, text: str, voice: str, speed: float = 1.0, pitch_shift: int = 0 ) -> np.ndarray: """ Synthesizes text with Piper and applies pitch shifting. Args: text: Text to synthesize voice: Piper voice ID speed: Playback speed (0.5-2.0) pitch_shift: Semitones to shift pitch (-12 to +12) Returns: Audio samples as numpy array """ # Ensure voice model is available if not self._download_voice_model(voice): error_msg = ( f"Piper voice model '{voice}' not found. " f"Please download voice models manually. " f"See backend/PIPER_SETUP.md for instructions." ) logger.error(error_msg) raise FileNotFoundError(error_msg) model_path = os.path.join(self.models_dir, f"{voice}.onnx") # Check if model file exists if not os.path.exists(model_path): logger.error(f"Model file not found: {model_path}") return np.array([], dtype=np.float32) # Generate audio with Piper try: logger.info(f"Running Piper with model: {model_path}") # Run piper command process = subprocess.Popen( ["piper", "--model", model_path, "--output-raw"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # Send text and get audio stdout, stderr = process.communicate(input=text.encode('utf-8'), timeout=30) if process.returncode != 0: error_msg = stderr.decode() if stderr else "Unknown error" logger.error(f"Piper synthesis failed (exit code {process.returncode}): {error_msg}") return np.array([], dtype=np.float32) # Convert raw audio to numpy array audio_data = np.frombuffer(stdout, dtype=np.int16) audio_float = audio_data.astype(np.float32) / 32768.0 # Apply pitch shift and speed change using pydub if pitch_shift != 0 or speed != 1.0: audio_float = self._apply_audio_effects( audio_float, self.SAMPLE_RATE, speed, pitch_shift ) return audio_float except subprocess.TimeoutExpired: logger.error("Piper synthesis timed out") return np.array([], dtype=np.float32) except Exception as e: logger.error(f"Error in Piper synthesis: {e}") return np.array([], dtype=np.float32) def _apply_audio_effects( self, audio: np.ndarray, sample_rate: int, speed: float, pitch_shift: int ) -> np.ndarray: """ Apply speed and pitch effects to audio using pydub. Args: audio: Audio samples sample_rate: Sample rate speed: Speed multiplier pitch_shift: Semitones to shift Returns: Processed audio samples """ try: # Convert numpy to pydub AudioSegment audio_int16 = (audio * 32768.0).astype(np.int16) audio_bytes = audio_int16.tobytes() audio_segment = AudioSegment( audio_bytes, frame_rate=sample_rate, sample_width=2, channels=1 ) # Apply pitch shift if pitch_shift != 0: # Pitch shift by changing frame rate then resampling new_sample_rate = int(sample_rate * (2.0 ** (pitch_shift / 12.0))) audio_segment = audio_segment._spawn( audio_segment.raw_data, overrides={'frame_rate': new_sample_rate} ) audio_segment = audio_segment.set_frame_rate(sample_rate) # Apply speed change if speed != 1.0: audio_segment = speedup(audio_segment, playback_speed=speed) # Convert back to numpy samples = np.array(audio_segment.get_array_of_samples()) audio_float = samples.astype(np.float32) / 32768.0 return audio_float except Exception as e: logger.error(f"Error applying audio effects: {e}") return audio def get_available_voices(self) -> Dict[str, Dict[str, str]]: """ Returns dictionary of available Piper voices. Returns: Dict mapping voice IDs to their characteristics """ return self.AVAILABLE_VOICES.copy()