734 lines
40 KiB
Python
734 lines
40 KiB
Python
"""
|
|
IndexTTS-2 Processor - Handles TTS generation orchestration
|
|
Called by unified TTS nodes when using IndexTTS-2 engine
|
|
"""
|
|
|
|
import torch
|
|
from typing import Dict, Any, Optional, List, Tuple
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import torchaudio
|
|
|
|
# Add project root to path
|
|
current_dir = os.path.dirname(__file__)
|
|
engines_dir = os.path.dirname(current_dir)
|
|
project_root = os.path.dirname(engines_dir)
|
|
if project_root not in sys.path:
|
|
sys.path.insert(0, project_root)
|
|
|
|
|
|
def save_audio_safe(filepath: str, waveform: torch.Tensor, sample_rate: int):
|
|
"""
|
|
Save audio to file, using soundfile fallback if torchaudio fails (PyTorch 2.9+ TorchCodec issue).
|
|
|
|
Args:
|
|
filepath: Path to save to
|
|
waveform: Audio tensor
|
|
sample_rate: Sample rate in Hz
|
|
"""
|
|
try:
|
|
# Try torchaudio first (preferred)
|
|
torchaudio.save(filepath, waveform, sample_rate)
|
|
except RuntimeError as e:
|
|
if "torchcodec" in str(e).lower() or "libtorchcodec" in str(e).lower():
|
|
# TorchCodec error - fallback to soundfile
|
|
try:
|
|
import soundfile as sf
|
|
# Convert tensor to numpy and handle shape
|
|
audio_np = waveform.cpu().detach().numpy()
|
|
# soundfile expects (samples, channels) format
|
|
if audio_np.ndim == 2:
|
|
if audio_np.shape[0] <= 2:
|
|
# Likely (channels, samples) - transpose
|
|
audio_np = audio_np.T
|
|
sf.write(filepath, audio_np, sample_rate)
|
|
except Exception as fallback_err:
|
|
raise RuntimeError(f"Both torchaudio and soundfile save failed: {e} -> {fallback_err}")
|
|
else:
|
|
# Different error - re-raise
|
|
raise
|
|
|
|
from utils.text.chunking import ImprovedChatterBoxChunker
|
|
from utils.audio.processing import AudioProcessingUtils
|
|
from utils.audio.edit_post_processor import process_segments as apply_edit_post_processing
|
|
from utils.text.character_parser import CharacterParser
|
|
from utils.text.pause_processor import PauseTagProcessor
|
|
from utils.text.segment_parameters import apply_segment_parameters
|
|
from utils.text.index_tts_emotion import resolve_inline_emotion
|
|
from utils.text.step_audio_editx_special_tags import get_edit_tags_for_segment
|
|
from utils.voice.discovery import get_character_mapping
|
|
from engines.adapters.index_tts_adapter import IndexTTSAdapter
|
|
|
|
|
|
class IndexTTSProcessor:
|
|
"""
|
|
Internal processor for IndexTTS-2 TTS generation.
|
|
Handles emotion control, character processing, and generation orchestration.
|
|
"""
|
|
|
|
def __init__(self, engine_config: Dict[str, Any]):
|
|
"""
|
|
Initialize IndexTTS-2 processor.
|
|
|
|
Args:
|
|
engine_config: Engine configuration from IndexTTS-2 Engine node
|
|
"""
|
|
self.config = engine_config
|
|
self.adapter = IndexTTSAdapter()
|
|
language_defaults = {
|
|
"English": "en",
|
|
"Chinese": "zh",
|
|
"Japanese": "ja",
|
|
"Spanish": "es",
|
|
"Arabic": "ar",
|
|
}
|
|
configured_language = str(engine_config.get("language", "English"))
|
|
self.character_parser = CharacterParser(
|
|
default_language=language_defaults.get(
|
|
configured_language,
|
|
configured_language.lower(),
|
|
)
|
|
)
|
|
self.pause_processor = PauseTagProcessor()
|
|
self.sample_rate = 22050 # IndexTTS-2 native sample rate
|
|
|
|
# Set up character parser with available characters
|
|
self._setup_character_parser()
|
|
|
|
# Initialize adapter with engine config
|
|
self.adapter.initialize_engine(
|
|
model_path=engine_config.get('model_path'),
|
|
device=engine_config.get('device', 'auto'),
|
|
use_fp16=engine_config.get('use_fp16', True),
|
|
use_cuda_kernel=engine_config.get('use_cuda_kernel'),
|
|
use_deepspeed=engine_config.get('use_deepspeed', False),
|
|
use_torch_compile=engine_config.get('use_torch_compile', False),
|
|
use_accel=engine_config.get('use_accel', False),
|
|
low_vram=engine_config.get('low_vram', False)
|
|
)
|
|
|
|
def _setup_character_parser(self):
|
|
"""Set up character parser with available characters and aliases."""
|
|
from utils.voice.discovery import get_available_characters, voice_discovery
|
|
|
|
# Get available characters and aliases
|
|
available_chars = get_available_characters()
|
|
character_aliases = voice_discovery.get_character_aliases()
|
|
|
|
# Build complete available set
|
|
all_available = set()
|
|
if available_chars:
|
|
all_available.update(available_chars)
|
|
for alias, target in character_aliases.items():
|
|
all_available.add(alias.lower())
|
|
all_available.add(target.lower())
|
|
|
|
self.character_parser.set_available_characters(list(all_available))
|
|
|
|
# Set language defaults
|
|
char_lang_defaults = voice_discovery.get_character_language_defaults()
|
|
for char, lang in char_lang_defaults.items():
|
|
self.character_parser.set_character_language_default(char, lang)
|
|
|
|
def _process_dynamic_emotion_template(self, emotion_text: str, segment_text: str) -> str:
|
|
"""
|
|
Process dynamic emotion template by replacing {seg} with actual segment text.
|
|
|
|
Args:
|
|
emotion_text: Template with {seg} placeholder
|
|
segment_text: Actual text segment content
|
|
|
|
Returns:
|
|
Processed emotion text for QwenEmotion analysis
|
|
"""
|
|
if "{seg}" in emotion_text:
|
|
processed_text = emotion_text.replace("{seg}", segment_text)
|
|
print(f"🌈 Dynamic Text Emotion: '{segment_text[:30]}...' → '{processed_text[:50]}...'")
|
|
return processed_text
|
|
return emotion_text
|
|
|
|
def process_text(self,
|
|
text: str,
|
|
speaker_audio: Optional[Dict] = None,
|
|
reference_text: str = "",
|
|
seed: int = 1,
|
|
enable_chunking: bool = True,
|
|
max_chars_per_chunk: int = 400,
|
|
silence_between_chunks_ms: int = 100,
|
|
return_info: bool = False):
|
|
"""
|
|
Process text and generate audio with IndexTTS-2.
|
|
|
|
Args:
|
|
text: Input text with potential character tags and emotions
|
|
speaker_audio: Speaker reference audio tensor dict
|
|
reference_text: Reference text for voice cloning
|
|
seed: Random seed for generation
|
|
enable_chunking: Whether to chunk long text (may be disabled for IndexTTS-2)
|
|
max_chars_per_chunk: Maximum characters per chunk
|
|
silence_between_chunks_ms: Silence between segments
|
|
return_info: If True, return (audio, chunk_info) tuple
|
|
|
|
Returns:
|
|
Generated audio tensor, or (tensor, chunk_info) if return_info=True
|
|
"""
|
|
try:
|
|
# Validate text input - prevent None text which causes NoneType errors
|
|
if text is None:
|
|
raise ValueError("Text input cannot be None. Please provide valid text to generate speech.")
|
|
|
|
if isinstance(text, str):
|
|
text = text.strip()
|
|
if not text:
|
|
raise ValueError("Text input is empty. Please provide text to generate speech.")
|
|
|
|
print(f"🤖 IndexTTS-2: Processing text with emotion support")
|
|
|
|
# Check if IndexTTS-2's native chunking should override our chunking
|
|
max_text_tokens_per_segment = self.config.get('max_text_tokens_per_segment', 120)
|
|
if max_text_tokens_per_segment > 0:
|
|
# IndexTTS-2 has its own token-based chunking, disable our character-based chunking
|
|
print(f"📝 IndexTTS-2: Using native token chunking ({max_text_tokens_per_segment} tokens), disabling character chunking")
|
|
enable_chunking = False
|
|
|
|
# Parse character segments with emotion support and parameters
|
|
character_segment_objects = self.character_parser.parse_text_segments(text)
|
|
character_segments = [(seg.character, seg.text, seg.language, seg.emotion) for seg in character_segment_objects]
|
|
|
|
any_inline_edit_tags = False
|
|
for seg in character_segment_objects:
|
|
_, seg_edit_tags = get_edit_tags_for_segment(seg.text)
|
|
if seg_edit_tags:
|
|
any_inline_edit_tags = True
|
|
all_characters = set(char for char, _, _, _ in character_segments)
|
|
all_characters.add("narrator")
|
|
|
|
# Build character mapping for emotion references (IndexTTS only needs audio files)
|
|
character_mapping = get_character_mapping(list(all_characters), engine_type="audio_only")
|
|
|
|
print(f"🎭 IndexTTS-2: Processing {len(character_segments)} character segment(s) - {', '.join(sorted(all_characters))}")
|
|
|
|
# Build voice references with narrator fallback
|
|
narrator_voice_dict = None
|
|
narrator_ref_text = ""
|
|
|
|
if speaker_audio is not None:
|
|
# Handle ComfyUI audio tensor dimensions (3D -> 2D)
|
|
waveform = speaker_audio["waveform"]
|
|
if waveform.dim() == 3:
|
|
waveform = waveform.squeeze(0) # Remove batch dimension
|
|
if waveform.shape[0] > 1:
|
|
waveform = torch.mean(waveform, dim=0, keepdim=True) # Convert stereo to mono
|
|
narrator_voice_dict = {"waveform": waveform, "sample_rate": speaker_audio["sample_rate"]}
|
|
narrator_ref_text = reference_text or ""
|
|
print(f"📖 Using connected narrator voice | Ref: '{narrator_ref_text[:50]}...'")
|
|
else:
|
|
# Check mapped narrator
|
|
mapped_narrator = character_mapping.get("narrator", (None, None))
|
|
if mapped_narrator[0] and os.path.exists(mapped_narrator[0]):
|
|
from utils.audio.processing import AudioProcessingUtils
|
|
waveform, sample_rate = AudioProcessingUtils.safe_load_audio(mapped_narrator[0])
|
|
if waveform.shape[0] > 1:
|
|
waveform = torch.mean(waveform, dim=0, keepdim=True)
|
|
narrator_voice_dict = {"waveform": waveform, "sample_rate": sample_rate}
|
|
narrator_ref_text = mapped_narrator[1] or ""
|
|
print(f"📖 Using mapped narrator voice: {mapped_narrator[0]} | Ref: '{narrator_ref_text[:50]}...'")
|
|
else:
|
|
print(f"⚠️ No narrator voice available - using IndexTTS-2 without speaker reference")
|
|
|
|
# Build voice references for all characters
|
|
voice_refs = {}
|
|
if narrator_voice_dict and 'waveform' in narrator_voice_dict:
|
|
voice_refs['narrator'] = narrator_voice_dict
|
|
|
|
for character in all_characters:
|
|
if character.lower() == "narrator":
|
|
continue
|
|
|
|
audio_path, char_ref_text = character_mapping.get(character, (None, None))
|
|
if audio_path and os.path.exists(audio_path):
|
|
from utils.audio.processing import AudioProcessingUtils
|
|
waveform, sample_rate = AudioProcessingUtils.safe_load_audio(audio_path)
|
|
if waveform.shape[0] > 1:
|
|
waveform = torch.mean(waveform, dim=0, keepdim=True)
|
|
voice_refs[character] = {"waveform": waveform, "sample_rate": sample_rate}
|
|
print(f"🎭 {character}: Loaded voice reference")
|
|
else:
|
|
# Fallback to narrator voice
|
|
voice_refs[character] = narrator_voice_dict
|
|
if narrator_voice_dict:
|
|
print(f"🎭 {character}: Using narrator voice fallback")
|
|
else:
|
|
print(f"⚠️ {character}: No voice available")
|
|
|
|
# Define TTS generation function for pause processor
|
|
def tts_generate_func(
|
|
text_content: str,
|
|
segment_params: Optional[Dict[str, Any]] = None,
|
|
character_name: Optional[str] = None,
|
|
emotion_reference: Optional[str] = None,
|
|
segment_language: Optional[str] = None,
|
|
) -> torch.Tensor:
|
|
# Import references for nested function scope
|
|
import torchaudio as ta
|
|
import tempfile as tf
|
|
"""TTS generation function for pause tag processor with per-segment parameters"""
|
|
|
|
def generate_with_context(context: str, **kwargs):
|
|
"""Add segment context without hiding the original engine error."""
|
|
try:
|
|
return self.adapter.generate(**kwargs)
|
|
except Exception as exc:
|
|
# Avoid leaking temporary WAV files when an engine
|
|
# failure happens before the normal success cleanup.
|
|
for path in (kwargs.get("speaker_audio"), kwargs.get("emotion_audio")):
|
|
if (
|
|
isinstance(path, str)
|
|
and path.startswith(tf.gettempdir())
|
|
and os.path.exists(path)
|
|
):
|
|
try:
|
|
os.unlink(path)
|
|
except OSError:
|
|
pass
|
|
raise RuntimeError(
|
|
f"IndexTTS-2 segment failed ({context}): "
|
|
f"{type(exc).__name__}: {exc}"
|
|
) from exc
|
|
|
|
# Apply segment parameters if provided
|
|
current_config = dict(self.config)
|
|
if segment_params:
|
|
current_config = apply_segment_parameters(current_config, segment_params, "index_tts")
|
|
current_config, _ = resolve_inline_emotion(current_config)
|
|
|
|
if '[' in text_content and ']' in text_content:
|
|
# Handle character switching with emotion parsing using modularized parser
|
|
char_segments = self.character_parser.split_by_character_with_emotions(text_content)
|
|
segment_audio_parts = []
|
|
|
|
for character, segment_text, language, emotion in char_segments:
|
|
# Character segment processing (debug info removed for cleaner logs)
|
|
# Get character voice reference
|
|
char_audio_dict = voice_refs.get(character)
|
|
speaker_audio_path = None
|
|
if char_audio_dict and 'waveform' in char_audio_dict:
|
|
# Convert tensor back to temporary file for IndexTTS-2
|
|
# IndexTTS-2 adapter expects file paths, not tensors
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
waveform = char_audio_dict['waveform']
|
|
# Ensure 2D tensor for torchaudio.save (channels, samples)
|
|
if waveform.dim() == 3:
|
|
waveform = waveform.squeeze(0) # Remove batch dimension
|
|
save_audio_safe(tmp_file.name, waveform, char_audio_dict['sample_rate'])
|
|
speaker_audio_path = tmp_file.name
|
|
elif character.lower() == "narrator" and narrator_voice_dict and 'waveform' in narrator_voice_dict:
|
|
# Fallback to narrator voice for narrator character
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
waveform = narrator_voice_dict['waveform']
|
|
# Ensure 2D tensor for torchaudio.save (channels, samples)
|
|
if waveform.dim() == 3:
|
|
waveform = waveform.squeeze(0) # Remove batch dimension
|
|
save_audio_safe(tmp_file.name, waveform, narrator_voice_dict['sample_rate'])
|
|
speaker_audio_path = tmp_file.name
|
|
print(f"✅ Using fallback narrator voice for character: {character}")
|
|
|
|
# Check for emotion reference from character mapping or config
|
|
emotion_audio_path = None
|
|
if emotion:
|
|
# Emotion from character tag like [Alice:Bob] - use character parser's emotion resolution
|
|
emotion_audio_path = self.character_parser.get_emotion_voice_path(emotion)
|
|
if emotion_audio_path:
|
|
print(f"😊 Using emotion reference from tag: {emotion} -> {emotion_audio_path}")
|
|
else:
|
|
print(f"🐛 Could not resolve emotion reference '{emotion}'")
|
|
|
|
# Fall back to connected emotion audio if no character
|
|
# reference was specified. Inline vector/text emotion
|
|
# can now be blended with this audio source.
|
|
if not emotion_audio_path:
|
|
emotion_from_config = self.config.get('emotion_audio')
|
|
# Process emotion audio for this character
|
|
if emotion_from_config:
|
|
if isinstance(emotion_from_config, dict) and 'waveform' in emotion_from_config:
|
|
# Convert tensor to temporary file
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
# Ensure correct tensor dimensions for torchaudio.save
|
|
emotion_waveform = emotion_from_config['waveform']
|
|
if emotion_waveform.dim() == 3:
|
|
emotion_waveform = emotion_waveform.squeeze(0)
|
|
elif emotion_waveform.dim() == 1:
|
|
emotion_waveform = emotion_waveform.unsqueeze(0)
|
|
save_audio_safe(tmp_file.name, emotion_waveform, emotion_from_config['sample_rate'])
|
|
emotion_audio_path = tmp_file.name
|
|
print(f"🎭 Using connected engine emotion audio for character: {character}")
|
|
else:
|
|
emotion_audio_path = emotion_from_config
|
|
print(f"🎭 Using connected engine emotion audio for character: {character}")
|
|
else:
|
|
print(f"🎭 No emotion audio for character: {character} (no tag emotion, no connected engine emotion)")
|
|
|
|
# Audio and vector/text controls are blended by
|
|
# IndexTTS-2 when both are present.
|
|
character_emotion_vector = current_config.get('emotion_vector')
|
|
character_use_emotion_text = current_config.get('use_emotion_text', False)
|
|
character_emotion_text = current_config.get('emotion_text')
|
|
|
|
# Handle dynamic QwenEmotion template
|
|
if character_use_emotion_text and character_emotion_text and current_config.get('is_dynamic_template', False):
|
|
character_emotion_text = self._process_dynamic_emotion_template(character_emotion_text, segment_text)
|
|
|
|
# Generate audio for this character segment (use parameters from current_config with segment overrides)
|
|
segment_result = generate_with_context(
|
|
f"character '{character}', text={segment_text[:120]!r}",
|
|
text=segment_text,
|
|
speaker_audio=speaker_audio_path,
|
|
emotion_audio=emotion_audio_path,
|
|
emotion_alpha=current_config.get('emotion_alpha', 1.0),
|
|
emotion_vector=character_emotion_vector,
|
|
use_emotion_text=character_use_emotion_text,
|
|
emotion_text=character_emotion_text,
|
|
use_random=current_config.get('use_random', False),
|
|
interval_silence=current_config.get('interval_silence', 200),
|
|
max_text_tokens_per_segment=current_config.get('max_text_tokens_per_segment', 120),
|
|
seed=current_config.get('seed', seed),
|
|
do_sample=current_config.get('do_sample', True),
|
|
temperature=current_config.get('temperature', 0.8),
|
|
top_p=current_config.get('top_p', 0.8),
|
|
top_k=current_config.get('top_k', 30),
|
|
length_penalty=current_config.get('length_penalty', 0.0),
|
|
num_beams=current_config.get('num_beams', 3),
|
|
repetition_penalty=current_config.get('repetition_penalty', 10.0),
|
|
max_mel_tokens=current_config.get('max_mel_tokens', 1500),
|
|
language=language or current_config.get('language', 'English'),
|
|
duration_factor=current_config.get('duration_factor', 1.0),
|
|
text_normalization=current_config.get('text_normalization', True),
|
|
stream_return=current_config.get('stream_return', False),
|
|
more_segment_before=current_config.get('more_segment_before', 0)
|
|
)
|
|
|
|
# Clean up temp files
|
|
if speaker_audio_path and os.path.exists(speaker_audio_path):
|
|
os.unlink(speaker_audio_path)
|
|
if emotion_audio_path and isinstance(emotion_audio_path, str) and os.path.exists(emotion_audio_path) and emotion_audio_path.startswith(tempfile.gettempdir()):
|
|
os.unlink(emotion_audio_path)
|
|
|
|
# Ensure correct dimensions
|
|
if segment_result.dim() == 1:
|
|
segment_result = segment_result.unsqueeze(0)
|
|
elif segment_result.dim() == 3:
|
|
segment_result = segment_result.squeeze(0)
|
|
|
|
segment_audio_parts.append(segment_result)
|
|
|
|
# Combine character segments
|
|
if segment_audio_parts:
|
|
return torch.cat(segment_audio_parts, dim=-1)
|
|
else:
|
|
return torch.zeros(1, 0)
|
|
else:
|
|
# Character tags are removed before this branch is called
|
|
# for parameterized segments. Keep the parsed character
|
|
# here so [Bob:emotion] does not silently use the narrator.
|
|
character_audio = voice_refs.get(character_name) if character_name else None
|
|
narrator_audio = character_audio or voice_refs.get("narrator")
|
|
speaker_audio_path = None
|
|
if narrator_audio and 'waveform' in narrator_audio:
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
save_audio_safe(tmp_file.name, narrator_audio['waveform'], narrator_audio['sample_rate'])
|
|
speaker_audio_path = tmp_file.name
|
|
|
|
# Ensure we have a speaker audio path - always use narrator if available
|
|
if not speaker_audio_path:
|
|
if narrator_voice_dict and 'waveform' in narrator_voice_dict:
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
save_audio_safe(tmp_file.name, narrator_voice_dict['waveform'], narrator_voice_dict['sample_rate'])
|
|
speaker_audio_path = tmp_file.name
|
|
|
|
# Character emotion references are preserved on the parsed
|
|
# segment even after its tag has been stripped. Resolve
|
|
# that reference before falling back to the node's global
|
|
# emotion audio.
|
|
emotion_audio_path = None
|
|
if emotion_reference:
|
|
emotion_audio_path = self.character_parser.get_emotion_voice_path(emotion_reference)
|
|
if emotion_audio_path:
|
|
print(
|
|
f"😊 Using emotion reference from segment: "
|
|
f"{emotion_reference} -> {emotion_audio_path}"
|
|
)
|
|
else:
|
|
print(f"⚠️ Could not resolve emotion reference '{emotion_reference}'")
|
|
if not emotion_audio_path:
|
|
emotion_audio_path = self.config.get('emotion_audio')
|
|
# Process emotion audio from config
|
|
if emotion_audio_path and isinstance(emotion_audio_path, dict) and 'waveform' in emotion_audio_path:
|
|
# Convert tensor to temporary file
|
|
with tf.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
|
|
# Ensure correct tensor dimensions for torchaudio.save
|
|
emotion_waveform = emotion_audio_path['waveform']
|
|
if emotion_waveform.dim() == 3:
|
|
emotion_waveform = emotion_waveform.squeeze(0)
|
|
elif emotion_waveform.dim() == 1:
|
|
emotion_waveform = emotion_waveform.unsqueeze(0)
|
|
save_audio_safe(tmp_file.name, emotion_waveform, emotion_audio_path['sample_rate'])
|
|
emotion_audio_path = tmp_file.name
|
|
print(f"🎭 Using connected engine emotion audio for simple text segment")
|
|
elif emotion_audio_path:
|
|
print(f"🎭 Using connected engine emotion audio for simple text segment")
|
|
else:
|
|
print(f"🎭 No emotion audio for simple text segment (no connected engine emotion)")
|
|
|
|
# Audio and vector/text controls are blended by
|
|
# IndexTTS-2 when both are present.
|
|
simple_emotion_vector = current_config.get('emotion_vector')
|
|
simple_use_emotion_text = current_config.get('use_emotion_text', False)
|
|
simple_emotion_text = current_config.get('emotion_text')
|
|
|
|
# Handle dynamic QwenEmotion template
|
|
if simple_use_emotion_text and simple_emotion_text and current_config.get('is_dynamic_template', False):
|
|
simple_emotion_text = self._process_dynamic_emotion_template(simple_emotion_text, text_content)
|
|
|
|
result = generate_with_context(
|
|
f"text={text_content[:120]!r}",
|
|
text=text_content,
|
|
speaker_audio=speaker_audio_path,
|
|
emotion_audio=emotion_audio_path,
|
|
emotion_alpha=current_config.get('emotion_alpha', 1.0),
|
|
emotion_vector=simple_emotion_vector,
|
|
use_emotion_text=simple_use_emotion_text,
|
|
emotion_text=simple_emotion_text,
|
|
use_random=current_config.get('use_random', False),
|
|
interval_silence=current_config.get('interval_silence', 200),
|
|
max_text_tokens_per_segment=current_config.get('max_text_tokens_per_segment', 120),
|
|
seed=current_config.get('seed', seed),
|
|
do_sample=current_config.get('do_sample', True),
|
|
temperature=current_config.get('temperature', 0.8),
|
|
top_p=current_config.get('top_p', 0.8),
|
|
top_k=current_config.get('top_k', 30),
|
|
length_penalty=current_config.get('length_penalty', 0.0),
|
|
num_beams=current_config.get('num_beams', 3),
|
|
repetition_penalty=current_config.get('repetition_penalty', 10.0),
|
|
max_mel_tokens=current_config.get('max_mel_tokens', 1500),
|
|
language=segment_language or current_config.get('language', 'English'),
|
|
duration_factor=current_config.get('duration_factor', 1.0),
|
|
text_normalization=current_config.get('text_normalization', True),
|
|
stream_return=current_config.get('stream_return', False),
|
|
more_segment_before=current_config.get('more_segment_before', 0)
|
|
)
|
|
|
|
# Clean up temp files
|
|
if speaker_audio_path and os.path.exists(speaker_audio_path):
|
|
os.unlink(speaker_audio_path)
|
|
if emotion_audio_path and isinstance(emotion_audio_path, str) and os.path.exists(emotion_audio_path) and emotion_audio_path.startswith(tempfile.gettempdir()):
|
|
os.unlink(emotion_audio_path)
|
|
|
|
return result
|
|
|
|
# Check if we have multiple character segments with different parameters
|
|
# If so, generate each character segment separately to preserve per-segment parameters
|
|
has_parameter_changes = False
|
|
if len(character_segment_objects) > 1:
|
|
for i in range(1, len(character_segment_objects)):
|
|
if character_segment_objects[i].parameters != character_segment_objects[i-1].parameters:
|
|
has_parameter_changes = True
|
|
break
|
|
|
|
if has_parameter_changes or any_inline_edit_tags:
|
|
# Generate each parsed segment separately so inline edit tags can be
|
|
# stripped before IndexTTS tokenization, then restored via post-processing.
|
|
segment_records = []
|
|
for seg_idx, seg_obj in enumerate(character_segment_objects):
|
|
print(f" 📊 Segment {seg_idx + 1}: Character '{seg_obj.character}' with params {seg_obj.parameters}")
|
|
pause_segments, _ = self.pause_processor.parse_pause_tags(seg_obj.text)
|
|
if not pause_segments:
|
|
pause_segments = [("text", seg_obj.text)]
|
|
|
|
for subseg_type, subseg_content in pause_segments:
|
|
if subseg_type == "pause":
|
|
silence_audio = self.pause_processor.create_silence_segment(subseg_content, self.sample_rate)
|
|
segment_records.append({
|
|
"waveform": silence_audio,
|
|
"sample_rate": self.sample_rate,
|
|
"text": f"[pause:{subseg_content}s]",
|
|
"original_text": f"[pause:{subseg_content}s]",
|
|
"edit_tags": [],
|
|
"character": seg_obj.character,
|
|
})
|
|
continue
|
|
|
|
clean_segment_text, edit_tags = get_edit_tags_for_segment(subseg_content)
|
|
if not clean_segment_text and edit_tags:
|
|
print(f" ⚠️ Segment {seg_idx + 1}: Only inline edit tags found, skipping empty TTS segment")
|
|
continue
|
|
|
|
segment_audio = tts_generate_func(
|
|
clean_segment_text or subseg_content,
|
|
seg_obj.parameters,
|
|
seg_obj.character,
|
|
seg_obj.emotion,
|
|
seg_obj.language,
|
|
)
|
|
if isinstance(segment_audio, torch.Tensor) and segment_audio.numel() > 0:
|
|
if segment_audio.dim() == 1:
|
|
segment_audio = segment_audio.unsqueeze(0)
|
|
elif segment_audio.dim() == 3:
|
|
segment_audio = segment_audio.squeeze(0)
|
|
segment_records.append({
|
|
"waveform": segment_audio,
|
|
"sample_rate": self.sample_rate,
|
|
"text": clean_segment_text or subseg_content,
|
|
"original_text": subseg_content,
|
|
"edit_tags": edit_tags,
|
|
"character": seg_obj.character,
|
|
})
|
|
|
|
if any(seg.get("edit_tags") for seg in segment_records):
|
|
print("🎨 Applying Step Audio EditX inline edit tags post-processing...")
|
|
segment_records = apply_edit_post_processing(segment_records, self.config)
|
|
|
|
if segment_records:
|
|
result = torch.cat([seg["waveform"] for seg in segment_records], dim=-1)
|
|
else:
|
|
result = torch.zeros(1, self.sample_rate)
|
|
else:
|
|
# No parameter changes between segments, use pause processor for pause tag handling
|
|
segments, clean_text = self.pause_processor.parse_pause_tags(text)
|
|
|
|
# Create a wrapper that handles pause segments
|
|
def tts_generate_with_params(text_content: str) -> torch.Tensor:
|
|
"""Wrapper for pause processor that passes segment parameters"""
|
|
# For pause-based processing, use first segment's parameters
|
|
segment_params = character_segment_objects[0].parameters if character_segment_objects and character_segment_objects[0].parameters else None
|
|
character_name = character_segment_objects[0].character if character_segment_objects else None
|
|
emotion_reference = character_segment_objects[0].emotion if character_segment_objects else None
|
|
segment_language = character_segment_objects[0].language if character_segment_objects else None
|
|
return tts_generate_func(text_content, segment_params, character_name, emotion_reference, segment_language)
|
|
|
|
# Generate audio with pauses
|
|
if segments:
|
|
result = self.pause_processor.generate_audio_with_pauses(
|
|
segments=segments,
|
|
tts_generate_func=tts_generate_with_params,
|
|
sample_rate=self.sample_rate
|
|
)
|
|
else:
|
|
# No pause tags, generate directly
|
|
segment_params = character_segment_objects[0].parameters if character_segment_objects and character_segment_objects[0].parameters else None
|
|
character_name = character_segment_objects[0].character if character_segment_objects else None
|
|
emotion_reference = character_segment_objects[0].emotion if character_segment_objects else None
|
|
segment_language = character_segment_objects[0].language if character_segment_objects else None
|
|
result = tts_generate_func(text, segment_params, character_name, emotion_reference, segment_language)
|
|
|
|
# Ensure correct tensor format
|
|
if isinstance(result, torch.Tensor):
|
|
if result.dim() == 1:
|
|
result = result.unsqueeze(0) # Add channel dimension
|
|
elif result.dim() == 3:
|
|
result = result.squeeze(0) # Remove batch dimension
|
|
|
|
# Build chunk info if requested
|
|
if return_info:
|
|
# Parse segments from text for timing estimation
|
|
import re
|
|
segments = re.split(r'(\[.*?\])', text)
|
|
text_segments = [s for s in segments if s and not s.startswith('[')]
|
|
|
|
total_duration = result.shape[-1] / self.sample_rate
|
|
chunk_info = {
|
|
"method_used": "concatenate",
|
|
"total_chunks": len(text_segments) if text_segments else 1,
|
|
"chunk_timings": []
|
|
}
|
|
|
|
# Estimate timing per segment (rough approximation)
|
|
if text_segments:
|
|
for i, seg_text in enumerate(text_segments):
|
|
start_time = (i / len(text_segments)) * total_duration
|
|
end_time = ((i + 1) / len(text_segments)) * total_duration
|
|
chunk_info["chunk_timings"].append({
|
|
"start": start_time,
|
|
"end": end_time,
|
|
"text": seg_text.strip()
|
|
})
|
|
else:
|
|
chunk_info["chunk_timings"].append({
|
|
"start": 0.0,
|
|
"end": total_duration,
|
|
"text": text
|
|
})
|
|
|
|
return result, chunk_info
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"❌ IndexTTS-2 processor error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise RuntimeError(f"IndexTTS-2 generation failed: {e}") from e
|
|
|
|
def combine_audio_segments(self,
|
|
segments: List[torch.Tensor],
|
|
method: str = "concatenate",
|
|
silence_ms: int = 100,
|
|
text_chunks: Optional[List[str]] = None,
|
|
text_length: int = 0,
|
|
return_info: bool = False):
|
|
"""
|
|
Combine audio segments using unified ChunkCombiner.
|
|
|
|
Args:
|
|
segments: List of audio tensors
|
|
method: Combination method
|
|
silence_ms: Silence between segments
|
|
text_chunks: Text for each segment
|
|
text_length: Total text length
|
|
return_info: If True, return (audio, chunk_info) tuple
|
|
|
|
Returns:
|
|
Combined audio tensor, or (tensor, chunk_info) if return_info=True
|
|
"""
|
|
if not segments:
|
|
empty = torch.zeros(1, 0)
|
|
if return_info:
|
|
return empty, {"method_used": "none", "total_chunks": 0, "chunk_timings": []}
|
|
return empty
|
|
|
|
# Single segment
|
|
if len(segments) == 1:
|
|
if return_info:
|
|
chunk_info = {
|
|
"method_used": "none",
|
|
"total_chunks": 1,
|
|
"chunk_timings": [{
|
|
"start": 0.0,
|
|
"end": segments[0].shape[-1] / self.sample_rate,
|
|
"text": text_chunks[0] if text_chunks else ""
|
|
}]
|
|
}
|
|
return segments[0], chunk_info
|
|
return segments[0]
|
|
|
|
# Use unified ChunkCombiner
|
|
from utils.audio.chunk_combiner import ChunkCombiner
|
|
result = ChunkCombiner.combine_chunks(
|
|
audio_segments=segments,
|
|
method=method,
|
|
silence_ms=silence_ms,
|
|
crossfade_duration=0.1,
|
|
sample_rate=self.sample_rate,
|
|
text_length=text_length,
|
|
original_text=" ".join(text_chunks) if text_chunks else "",
|
|
text_chunks=text_chunks,
|
|
return_info=return_info
|
|
)
|
|
|
|
return result
|
|
|
|
def cleanup(self):
|
|
"""Clean up resources"""
|
|
if self.adapter:
|
|
self.adapter.unload()
|