Paul Bailey
d3e72edd7f
- Fixed preview endpoint to work with both Kokoro and Piper engines - Improved voice model download with curl fallback and SSL handling - Added better error messages when models are missing - Created PIPER_SETUP.md with manual download instructions - Added test_piper_direct.py for testing Piper installation - Updated README with voice model download requirements Note: Piper voice models require manual download from GitHub releases due to HuggingFace URL structure. See PIPER_SETUP.md for instructions.
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
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()
|