""" Audio Cache Module - Unified caching system for TTS engines Provides centralized cache management with engine-specific cache key generation """ import hashlib import torch from typing import Dict, Any, Optional, Tuple, Callable from abc import ABC, abstractmethod # Global audio cache shared across all engines GLOBAL_AUDIO_CACHE = {} class CacheKeyGenerator(ABC): """Abstract base class for engine-specific cache key generation.""" @abstractmethod def generate_cache_key(self, **params) -> str: """Generate cache key from engine-specific parameters.""" pass class F5TTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for F5-TTS engine.""" def generate_cache_key(self, **params) -> str: """Generate F5-TTS cache key from parameters.""" cache_data = { 'text': params.get('text', ''), 'model_name': params.get('model_name', ''), 'device': params.get('device', ''), 'audio_component': params.get('audio_component', ''), 'ref_text': params.get('ref_text', ''), 'temperature': params.get('temperature', 0.8), 'speed': params.get('speed', 1.0), 'target_rms': params.get('target_rms', 0.1), 'cross_fade_duration': params.get('cross_fade_duration', 0.15), 'nfe_step': params.get('nfe_step', 32), 'cfg_strength': params.get('cfg_strength', 2.0), 'seed': params.get('seed', 0), 'character': params.get('character', 'narrator'), 'engine': 'f5tts' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class ChatterBoxCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for ChatterBox engine.""" def generate_cache_key(self, **params) -> str: """Generate ChatterBox cache key from parameters.""" cache_data = { 'text': params.get('text', ''), 'exaggeration': params.get('exaggeration', 1.0), 'temperature': params.get('temperature', 0.8), 'cfg_weight': params.get('cfg_weight', 1.0), 'seed': params.get('seed', 0), 'audio_component': params.get('audio_component', ''), 'model_source': params.get('model_source', ''), 'device': params.get('device', ''), 'language': params.get('language', 'English'), 'character': params.get('character', 'narrator'), 'engine': 'chatterbox', # ChatterBox Official 23-Lang specific parameters 'repetition_penalty': params.get('repetition_penalty', 1.0), 'min_p': params.get('min_p', 0.0), 'top_p': params.get('top_p', 1.0) } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class ChatterBoxOfficial23LangCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for ChatterBox Official 23-Lang engine.""" def generate_cache_key(self, **params) -> str: """Generate ChatterBox Official 23-Lang cache key from parameters.""" cache_data = { 'text': params.get('text', ''), 'exaggeration': params.get('exaggeration', 0.5), 'temperature': params.get('temperature', 0.8), 'cfg_weight': params.get('cfg_weight', 0.5), 'seed': params.get('seed', 0), 'audio_component': params.get('audio_component', ''), 'model_source': params.get('model_source', ''), 'model_version': params.get('model_version', 'v1'), # v1 or v2 - ensures cache invalidation between versions 'device': params.get('device', ''), 'language': params.get('language', 'English'), 'character': params.get('character', 'narrator'), 'engine': 'chatterbox_official_23lang', # ChatterBox Official 23-Lang specific parameters 'repetition_penalty': params.get('repetition_penalty', 1.0), 'min_p': params.get('min_p', 0.0), 'top_p': params.get('top_p', 1.0) } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class HiggsAudioCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Higgs Audio engine.""" def generate_cache_key(self, **params) -> str: """Generate Higgs Audio cache key from parameters.""" cache_data = { 'text': params.get('text', ''), 'model_path': params.get('model_path', ''), 'tokenizer_path': params.get('tokenizer_path', ''), 'device': params.get('device', ''), 'reference_text': params.get('reference_text', ''), 'system_prompt': params.get('system_prompt', ''), 'temperature': params.get('temperature', 1.0), 'top_p': params.get('top_p', 0.95), 'top_k': params.get('top_k', 50), 'max_new_tokens': params.get('max_new_tokens', 2048), 'force_audio_gen': params.get('force_audio_gen', False), 'ras_win_len': params.get('ras_win_len', 7), 'ras_max_num_repeat': params.get('ras_max_num_repeat', 2), 'seed': params.get('seed', 0), 'character': params.get('character', 'narrator'), 'max_chars_per_chunk': params.get('max_chars_per_chunk', 400), # Note: silence_between_chunks_ms excluded - it's post-processing, doesn't affect generation 'engine': 'higgs_audio' } # Add reference audio hash if present if 'reference_audio' in params and params['reference_audio'] is not None: audio_dict = params['reference_audio'] if isinstance(audio_dict, dict) and 'waveform' in audio_dict: # Create hash of audio waveform import hashlib waveform = audio_dict['waveform'] if hasattr(waveform, 'cpu'): # torch tensor audio_hash = hashlib.md5(waveform.cpu().numpy().tobytes()).hexdigest()[:8] cache_data['audio_hash'] = audio_hash cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class HiggsAudioV3CacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Higgs Audio v3 engine.""" def generate_cache_key(self, **params) -> str: temperature = params.get('temperature', 1.0) top_p = params.get('top_p', 0.95) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) if isinstance(top_p, (int, float)): top_p = round(float(top_p), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'model_variant': params.get('model_variant', 'higgs-audio-v3-tts-4b'), 'temperature': temperature, 'top_p': top_p, 'top_k': params.get('top_k', 50), 'max_new_tokens': params.get('max_new_tokens', 2048), 'seed': params.get('seed', 0), 'device': params.get('device', 'auto'), 'dtype': params.get('dtype', 'auto'), 'attention': params.get('attention', 'auto'), 'character': params.get('character', 'narrator'), 'engine': 'higgs_audio_v3', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class HiggsAudioV3CacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Higgs Audio v3 engine.""" def generate_cache_key(self, **params) -> str: temperature = params.get('temperature', 1.0) top_p = params.get('top_p', 0.95) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) if isinstance(top_p, (int, float)): top_p = round(float(top_p), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'model_variant': params.get('model_variant', 'higgs-audio-v3-tts-4b'), 'temperature': temperature, 'top_p': top_p, 'top_k': params.get('top_k', 50), 'max_new_tokens': params.get('max_new_tokens', 2048), 'seed': params.get('seed', 0), 'device': params.get('device', 'auto'), 'dtype': params.get('dtype', 'auto'), 'attention': params.get('attention', 'auto'), 'character': params.get('character', 'narrator'), 'engine': 'higgs_audio_v3', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class VibeVoiceCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for VibeVoice engine.""" def generate_cache_key(self, **params) -> str: """Generate VibeVoice cache key from parameters.""" # Fix floating point precision issues by rounding to 3 decimal places cfg_scale = params.get('cfg_scale', 1.3) temperature = params.get('temperature', 0.95) top_p = params.get('top_p', 0.95) if isinstance(cfg_scale, (int, float)): cfg_scale = round(float(cfg_scale), 3) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) if isinstance(top_p, (int, float)): top_p = round(float(top_p), 3) cache_data = { 'text': params.get('text', ''), 'cfg_scale': cfg_scale, 'temperature': temperature, 'top_p': top_p, 'use_sampling': params.get('use_sampling', False), 'seed': params.get('seed', 42), 'model_source': params.get('model_source', 'vibevoice-1.5B'), 'device': params.get('device', 'auto'), 'max_new_tokens': params.get('max_new_tokens'), 'multi_speaker_mode': params.get('multi_speaker_mode', 'Custom Character Switching'), 'audio_component': params.get('audio_component', ''), 'character': params.get('character', 'narrator'), 'inference_steps': params.get('inference_steps', 20), # Include diffusion inference steps in cache key 'attention_mode': params.get('attention_mode', 'eager'), # Include attention mode in cache key 'quantize': params.get('quantize', False), # Include quantization in cache key 'engine': 'vibevoice' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class StepAudioEditXCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Step Audio EditX engine.""" def generate_cache_key(self, **params) -> str: """Generate Step Audio EditX cache key from parameters.""" # Fix floating point precision issues by rounding to 3 decimal places temperature = params.get('temperature', 0.7) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), # Stable voice hash 'prompt_text': params.get('prompt_text', ''), # Reference text 'temperature': temperature, 'do_sample': params.get('do_sample', True), 'max_new_tokens': params.get('max_new_tokens', 8192), 'seed': params.get('seed', 0), 'model_path': params.get('model_path', 'Step-Audio-EditX'), 'device': params.get('device', 'auto'), 'torch_dtype': params.get('torch_dtype', 'bfloat16'), 'quantization': params.get('quantization', None), 'character': params.get('character', 'narrator'), 'engine': 'step_audio_editx' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class IndexTTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for IndexTTS-2 engine.""" def generate_cache_key(self, **params) -> str: """Generate IndexTTS-2 cache key from parameters.""" # Round floating point values to avoid precision issues temperature = params.get('temperature', 0.8) top_p = params.get('top_p', 0.8) emotion_alpha = params.get('emotion_alpha', 1.0) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) if isinstance(top_p, (int, float)): top_p = round(float(top_p), 3) if isinstance(emotion_alpha, (int, float)): emotion_alpha = round(float(emotion_alpha), 3) cache_data = { 'text': params.get('text', ''), 'speaker_audio': params.get('speaker_audio', ''), 'emotion_audio': params.get('emotion_audio', ''), 'emotion_alpha': emotion_alpha, 'emotion_vector': params.get('emotion_vector'), # List or None 'use_emotion_text': params.get('use_emotion_text', False), 'emotion_text': params.get('emotion_text', ''), 'use_random': params.get('use_random', False), 'seed': params.get('seed', 0), # Seed for reproducible generation 'do_sample': params.get('do_sample', True), # Affects sampling vs greedy decoding 'temperature': temperature, 'top_p': top_p, 'top_k': params.get('top_k', 30), 'length_penalty': params.get('length_penalty', 0.0), 'num_beams': params.get('num_beams', 3), # Beam search quality parameter 'repetition_penalty': params.get('repetition_penalty', 10.0), 'max_mel_tokens': params.get('max_mel_tokens', 1500), 'max_text_tokens_per_segment': params.get('max_text_tokens_per_segment', 120), 'interval_silence': params.get('interval_silence', 200), 'model_name': params.get('model_name', 'IndexTTS-2'), 'model_version': params.get('model_version', '2'), 'model_path': params.get('model_path', ''), 'language': params.get('language', 'English'), 'duration_factor': round(float(params.get('duration_factor', 1.0)), 4), 'text_normalization': params.get('text_normalization', True), 'device': params.get('device', 'auto'), 'character': params.get('character', 'narrator'), 'use_torch_compile': params.get('use_torch_compile', False), # Optimization may affect output precision 'use_accel': params.get('use_accel', False), # Optimization may affect output precision 'stream_return': params.get('stream_return', False), # Streaming mode changes output format 'more_segment_before': params.get('more_segment_before', 0), # Streaming segmentation affects first-chunk timing 'engine': 'index_tts' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class CosyVoiceCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for CosyVoice3 engine.""" def generate_cache_key(self, **params) -> str: """Generate CosyVoice3 cache key from parameters.""" # Round floating point values to avoid precision issues speed = params.get('speed', 1.0) if isinstance(speed, (int, float)): speed = round(float(speed), 3) cache_data = { 'text': params.get('text', ''), 'speaker_audio': params.get('speaker_audio', ''), 'reference_text': params.get('reference_text', ''), 'mode': params.get('mode', 'zero_shot'), 'instruct_text': params.get('instruct_text', ''), 'speed': speed, 'seed': params.get('seed', 0), 'model_path': params.get('model_path', 'Fun-CosyVoice3-0.5B'), 'model_variant': params.get('model_variant', 'standard'), # Invalidate cache on variant switch 'device': params.get('device', 'auto'), 'use_fp16': params.get('use_fp16', True), 'character': params.get('character', 'narrator'), 'engine': 'cosyvoice' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class Qwen3TTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Qwen3-TTS engine.""" def generate_cache_key(self, **params) -> str: """Generate Qwen3-TTS cache key from parameters.""" # Round floating point values to avoid precision issues temperature = params.get('temperature', 0.9) top_p = params.get('top_p', 1.0) repetition_penalty = params.get('repetition_penalty', 1.05) subtalker_temperature = params.get('subtalker_temperature', 0.9) subtalker_top_p = params.get('subtalker_top_p', 1.0) if isinstance(temperature, (int, float)): temperature = round(float(temperature), 3) if isinstance(top_p, (int, float)): top_p = round(float(top_p), 3) if isinstance(repetition_penalty, (int, float)): repetition_penalty = round(float(repetition_penalty), 3) if isinstance(subtalker_temperature, (int, float)): subtalker_temperature = round(float(subtalker_temperature), 3) if isinstance(subtalker_top_p, (int, float)): subtalker_top_p = round(float(subtalker_top_p), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'model_type': params.get('model_type', 'Base'), # CustomVoice, VoiceDesign, Base 'speaker': params.get('speaker', ''), 'instruct': params.get('instruct', ''), 'language': params.get('language', 'Auto'), 'temperature': temperature, 'top_p': top_p, 'top_k': params.get('top_k', 50), 'repetition_penalty': repetition_penalty, 'max_new_tokens': params.get('max_new_tokens', 2048), 'subtalker_temperature': subtalker_temperature, 'subtalker_top_p': subtalker_top_p, 'subtalker_top_k': params.get('subtalker_top_k', 50), 'seed': params.get('seed', 0), 'model_size': params.get('model_size', '1.7B'), 'device': params.get('device', 'auto'), 'dtype': params.get('dtype', 'auto'), 'attn_implementation': params.get('attn_implementation', 'auto'), # Affects output quality 'x_vector_only': params.get('x_vector_only', False), # ICL mode vs x-vector only mode 'character': params.get('character', 'narrator'), 'engine': 'qwen3_tts' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class MossTTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for MOSS-TTS engine.""" def generate_cache_key(self, **params) -> str: audio_temperature = params.get('audio_temperature', params.get('temperature', 1.0)) audio_top_p = params.get('audio_top_p', params.get('top_p', 0.95)) audio_repetition_penalty = params.get( 'audio_repetition_penalty', params.get('repetition_penalty', 1.1) ) if isinstance(audio_temperature, (int, float)): audio_temperature = round(float(audio_temperature), 3) if isinstance(audio_top_p, (int, float)): audio_top_p = round(float(audio_top_p), 3) if isinstance(audio_repetition_penalty, (int, float)): audio_repetition_penalty = round(float(audio_repetition_penalty), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'model_variant': params.get('model_variant', 'MOSS-TTS-Local-Transformer'), 'lora_adapter': params.get('lora_adapter'), 'language': params.get('language', 'auto'), 'instruction': params.get('instruction'), 'quality': params.get('quality'), 'sound_event': params.get('sound_event'), 'ambient_sound': params.get('ambient_sound'), 'duration_tokens': params.get('duration_tokens'), 'audio_temperature': audio_temperature, 'audio_top_p': audio_top_p, 'audio_top_k': params.get('audio_top_k', params.get('top_k', 50)), 'audio_repetition_penalty': audio_repetition_penalty, 'max_new_tokens': params.get('max_new_tokens', 4096), 'n_vq_for_inference': params.get('n_vq_for_inference'), 'seed': params.get('seed', 0), 'device': params.get('device', 'auto'), 'dtype': params.get('dtype', 'auto'), 'attn_implementation': params.get('attn_implementation', 'auto'), 'character': params.get('character', 'narrator'), 'engine': 'moss_tts', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class MossSoundEffectV2CacheKeyGenerator(CacheKeyGenerator): """Cache all controls that materially affect MOSS-SoundEffect v2 output.""" def generate_cache_key(self, **params) -> str: cache_data = { "description": params.get("description", ""), "model": params.get("model", "MOSS-SoundEffect-v2.0"), "duration_seconds": round(float(params.get("duration_seconds", 10.0)), 1), "inference_steps": int(params.get("inference_steps", 100)), "cfg_scale": round(float(params.get("cfg_scale", 4.0)), 3), "sigma_shift": round(float(params.get("sigma_shift", 5.0)), 3), "negative_prompt": params.get("negative_prompt", ""), "seed": int(params.get("seed", 0)), "dtype": params.get("dtype", "auto"), "device": params.get("device", "auto"), "engine": "moss_soundeffect_v2", } return hashlib.md5(str(sorted(cache_data.items())).encode()).hexdigest() class EchoTTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Echo-TTS engine.""" def generate_cache_key(self, **params) -> str: """Generate Echo-TTS cache key from parameters.""" # Round float values to avoid cache misses from tiny precision differences. cfg_scale_text = params.get('cfg_scale_text', 3.0) cfg_scale_speaker = params.get('cfg_scale_speaker', 8.0) cfg_min_t = params.get('cfg_min_t', 0.5) cfg_max_t = params.get('cfg_max_t', 1.0) truncation_factor = params.get('truncation_factor') rescale_k = params.get('rescale_k') rescale_sigma = params.get('rescale_sigma') speaker_kv_scale = params.get('speaker_kv_scale') speaker_kv_min_t = params.get('speaker_kv_min_t') if isinstance(cfg_scale_text, (int, float)): cfg_scale_text = round(float(cfg_scale_text), 3) if isinstance(cfg_scale_speaker, (int, float)): cfg_scale_speaker = round(float(cfg_scale_speaker), 3) if isinstance(cfg_min_t, (int, float)): cfg_min_t = round(float(cfg_min_t), 3) if isinstance(cfg_max_t, (int, float)): cfg_max_t = round(float(cfg_max_t), 3) if isinstance(truncation_factor, (int, float)): truncation_factor = round(float(truncation_factor), 3) if isinstance(rescale_k, (int, float)): rescale_k = round(float(rescale_k), 3) if isinstance(rescale_sigma, (int, float)): rescale_sigma = round(float(rescale_sigma), 3) if isinstance(speaker_kv_scale, (int, float)): speaker_kv_scale = round(float(speaker_kv_scale), 3) if isinstance(speaker_kv_min_t, (int, float)): speaker_kv_min_t = round(float(speaker_kv_min_t), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'model': params.get('model', 'jordand/echo-tts-base'), 'device': params.get('device', 'auto'), 'num_steps': params.get('num_steps', 40), 'cfg_scale_text': cfg_scale_text, 'cfg_scale_speaker': cfg_scale_speaker, 'cfg_min_t': cfg_min_t, 'cfg_max_t': cfg_max_t, 'truncation_factor': truncation_factor, 'rescale_k': rescale_k, 'rescale_sigma': rescale_sigma, 'speaker_kv_scale': speaker_kv_scale, 'speaker_kv_max_layers': params.get('speaker_kv_max_layers'), 'speaker_kv_min_t': speaker_kv_min_t, 'sequence_length': params.get('sequence_length', 640), 'seed': params.get('seed', 0), 'engine': 'echo_tts' } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class DotsTTSCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Dots TTS engine.""" def generate_cache_key(self, **params) -> str: guidance_scale = params.get('guidance_scale', 1.2) speaker_scale = params.get('speaker_scale', 1.5) if isinstance(guidance_scale, (int, float)): guidance_scale = round(float(guidance_scale), 3) if isinstance(speaker_scale, (int, float)): speaker_scale = round(float(speaker_scale), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'prompt_text': params.get('prompt_text', ''), 'model_variant': params.get('model_variant', 'dots.tts-soar'), 'language': params.get('language', 'auto_detect'), 'template_name': params.get('template_name', 'tts'), 'num_steps': params.get('num_steps', 10), 'guidance_scale': guidance_scale, 'speaker_scale': speaker_scale, 'normalize_text': params.get('normalize_text', False), 'max_generate_length': params.get('max_generate_length', 500), 'precision': params.get('precision', 'auto'), 'optimize': params.get('optimize', False), 'seed': params.get('seed', 0), 'device': params.get('device', 'auto'), 'character': params.get('character', 'narrator'), 'engine': 'dots_tts', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class DramaBoxCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for official DramaBox inference.""" def generate_cache_key(self, **params) -> str: cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'model_name': params.get('model_name', 'DramaBox'), 'cfg_scale': round(float(params.get('cfg_scale', 2.5)), 3), 'stg_scale': round(float(params.get('stg_scale', 1.5)), 3), 'duration_multiplier': round(float(params.get('duration_multiplier', 1.1)), 3), 'gen_duration': round(float(params.get('gen_duration', 0.0)), 3), 'ref_duration': round(float(params.get('ref_duration', 10.0)), 3), 'rescale_scale': params.get('rescale_scale', 'auto'), 'watermark': bool(params.get('watermark', False)), 'prompt_template': params.get('prompt_template', '"{seg}"'), 'negative_prompt': params.get('negative_prompt', ''), 'precision': params.get('precision', 'auto'), 'transformer_quantization': params.get('transformer_quantization', 'none'), 'memory_mode': params.get('memory_mode', 'fast'), 'compile_model': bool(params.get('compile_model', False)), 'lora_path': params.get('lora_path', ''), 'lora_strength': round(float(params.get('lora_strength', 1.0)), 4), 'lora_revision': params.get('lora_revision', ''), 'seed': params.get('seed', 42), 'character': params.get('character', 'narrator'), 'engine': 'dramabox', } return hashlib.md5(str(sorted(cache_data.items())).encode()).hexdigest() class FishAudioS2CacheKeyGenerator(CacheKeyGenerator): """Cache key generator for Fish Audio S2 Pro.""" def generate_cache_key(self, **params) -> str: cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'model_variant': params.get('model_variant', 's2-pro'), 'quantization': params.get('quantization', 'none'), 'multi_speaker_mode': params.get('multi_speaker_mode', 'Native Multi-Speaker'), 'temperature': round(float(params.get('temperature', 0.8)), 3), 'top_p': round(float(params.get('top_p', 0.8)), 3), 'repetition_penalty': round(float(params.get('repetition_penalty', 1.1)), 3), 'chunk_length': params.get('chunk_length', 200), 'max_new_tokens': params.get('max_new_tokens', 1024), 'context_length': params.get('context_length', 8192), 'normalize': params.get('normalize', True), 'seed': params.get('seed', 0), 'character': params.get('character', 'narrator'), 'engine': 'fish_audio_s2', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class OmniVoiceCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for OmniVoice engine.""" def generate_cache_key(self, **params) -> str: guidance_scale = params.get('guidance_scale', 2.0) t_shift = params.get('t_shift', 0.1) speed = params.get('speed', 1.0) duration = params.get('duration', 'auto') layer_penalty_factor = params.get('layer_penalty_factor', 5.0) position_temperature = params.get('position_temperature', 5.0) class_temperature = params.get('class_temperature', 0.0) audio_chunk_duration = params.get('audio_chunk_duration', 15.0) audio_chunk_threshold = params.get('audio_chunk_threshold', 30.0) rounded_fields = { 'guidance_scale': guidance_scale, 't_shift': t_shift, 'speed': speed, 'layer_penalty_factor': layer_penalty_factor, 'position_temperature': position_temperature, 'class_temperature': class_temperature, 'audio_chunk_duration': audio_chunk_duration, 'audio_chunk_threshold': audio_chunk_threshold, } for key, value in list(rounded_fields.items()): if isinstance(value, (int, float)): rounded_fields[key] = round(float(value), 3) if isinstance(duration, (int, float)): duration = round(float(duration), 3) cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'prompt_text': params.get('prompt_text', ''), 'model_variant': params.get('model_variant', 'OmniVoice'), 'language': params.get('language', 'none'), 'instruct': params.get('instruct', ''), 'num_step': params.get('num_step', 32), 'guidance_scale': rounded_fields['guidance_scale'], 't_shift': rounded_fields['t_shift'], 'speed': rounded_fields['speed'], 'duration': duration, 'layer_penalty_factor': rounded_fields['layer_penalty_factor'], 'position_temperature': rounded_fields['position_temperature'], 'class_temperature': rounded_fields['class_temperature'], 'denoise': params.get('denoise', True), 'preprocess_prompt': params.get('preprocess_prompt', True), 'postprocess_output': params.get('postprocess_output', True), 'audio_chunk_duration': rounded_fields['audio_chunk_duration'], 'audio_chunk_threshold': rounded_fields['audio_chunk_threshold'], 'dtype': params.get('dtype', 'auto'), 'device': params.get('device', 'auto'), 'seed': params.get('seed', 0), 'character': params.get('character', 'narrator'), 'engine': 'omnivoice', } cache_string = str(sorted(cache_data.items())) return hashlib.md5(cache_string.encode()).hexdigest() class AudioCppCacheKeyGenerator(CacheKeyGenerator): """Cache key generator for the generic audio.cpp server backend.""" def generate_cache_key(self, **params) -> str: cache_data = { 'text': params.get('text', ''), 'audio_component': params.get('audio_component', ''), 'reference_text': params.get('reference_text', ''), 'family': params.get('family', ''), 'package_id': params.get('package_id', ''), 'model_path': params.get('model_path', ''), 'model_id': params.get('model_id', ''), 'task': params.get('task', ''), 'connection_mode': params.get('connection_mode', ''), 'server_url': params.get('server_url', ''), 'binary_path': params.get('binary_path', ''), 'backend': params.get('backend', ''), 'device_index': params.get('device_index', 0), 'language': params.get('language', ''), 'voice_id': params.get('voice_id', ''), 'instruct': params.get('instruct', ''), 'temperature': params.get('temperature'), 'top_p': params.get('top_p'), 'top_k': params.get('top_k'), 'repetition_penalty': params.get('repetition_penalty'), 'max_tokens': params.get('max_tokens'), 'max_steps': params.get('max_steps'), 'num_inference_steps': params.get('num_inference_steps'), 'guidance_scale': params.get('guidance_scale'), 'seed': params.get('seed', 0), 'request_options': params.get('request_options', ''), 'character': params.get('character', 'narrator'), 'engine': 'audio_cpp', } return hashlib.md5(str(sorted(cache_data.items())).encode()).hexdigest() class AudioCache: """Unified audio cache manager for all TTS engines.""" def __init__(self): self.cache_key_generators = { 'f5tts': F5TTSCacheKeyGenerator(), 'chatterbox': ChatterBoxCacheKeyGenerator(), 'chatterbox_official_23lang': ChatterBoxOfficial23LangCacheKeyGenerator(), # Uses specialized generator with advanced params 'higgs_audio': HiggsAudioCacheKeyGenerator(), 'higgs_audio_v3': HiggsAudioV3CacheKeyGenerator(), 'vibevoice': VibeVoiceCacheKeyGenerator(), 'step_audio_editx': StepAudioEditXCacheKeyGenerator(), 'index_tts': IndexTTSCacheKeyGenerator(), 'cosyvoice': CosyVoiceCacheKeyGenerator(), 'qwen3_tts': Qwen3TTSCacheKeyGenerator(), 'dots_tts': DotsTTSCacheKeyGenerator(), 'dramabox': DramaBoxCacheKeyGenerator(), 'fish_audio_s2': FishAudioS2CacheKeyGenerator(), 'audio_cpp': AudioCppCacheKeyGenerator(), 'omnivoice': OmniVoiceCacheKeyGenerator(), 'moss_tts': MossTTSCacheKeyGenerator(), 'moss_soundeffect_v2': MossSoundEffectV2CacheKeyGenerator(), 'echo_tts': EchoTTSCacheKeyGenerator() } def register_cache_key_generator(self, engine_type: str, generator: CacheKeyGenerator): """Register a cache key generator for a specific engine.""" self.cache_key_generators[engine_type] = generator def generate_cache_key(self, engine_type: str, **params) -> str: """Generate cache key for specified engine type.""" if engine_type not in self.cache_key_generators: raise ValueError(f"Unknown engine type: {engine_type}") generator = self.cache_key_generators[engine_type] return generator.generate_cache_key(**params) def get_cached_audio(self, cache_key: str) -> Optional[Tuple[torch.Tensor, float]]: """Retrieve cached audio by cache key.""" return GLOBAL_AUDIO_CACHE.get(cache_key) def cache_audio(self, cache_key: str, audio_tensor: torch.Tensor, duration: float): """Cache audio tensor with duration.""" GLOBAL_AUDIO_CACHE[cache_key] = (audio_tensor.clone(), duration) def create_cache_function(self, engine_type: str, **static_params) -> Callable: """ Create a cache function for use with TTS generation. Args: engine_type: "f5tts" or "chatterbox" **static_params: Parameters that don't change between calls Returns: Cache function that can be called with (text, audio_result=None) """ def cache_fn(text_content: str, audio_result=None): # Combine static params with dynamic text cache_params = static_params.copy() cache_params['text'] = f"{cache_params.get('character', 'narrator')}:{text_content}" # Generate cache key cache_key = self.generate_cache_key(engine_type, **cache_params) if audio_result is None: # Get from cache cached_data = self.get_cached_audio(cache_key) if cached_data: character = cache_params.get('character', 'narrator') language = cache_params.get('language', cache_params.get('model_name', '')) if language and language != 'English': print(f"💾 Using cached audio for '{character}' ({language}): '{text_content[:30]}...'") else: print(f"💾 Using cached audio for '{character}': '{text_content[:30]}...'") return cached_data[0] return None else: # Store in cache duration = self._calculate_duration(audio_result, engine_type) self.cache_audio(cache_key, audio_result, duration) return cache_fn def _calculate_duration(self, audio_tensor: torch.Tensor, engine_type: str) -> float: """Calculate audio duration based on engine type.""" if audio_tensor.dim() == 1: num_samples = audio_tensor.shape[0] elif audio_tensor.dim() == 2: num_samples = audio_tensor.shape[1] else: num_samples = audio_tensor.numel() # Use engine-specific sample rates if engine_type in ('dots_tts', 'dramabox', 'moss_soundeffect_v2'): sample_rate = 48000 elif engine_type in ('f5tts', 'step_audio_editx', 'qwen3_tts', 'moss_tts', 'higgs_audio_v3', 'omnivoice'): sample_rate = 24000 elif engine_type in ('index_tts', 'cosyvoice'): sample_rate = 22050 else: sample_rate = 44100 return num_samples / sample_rate def clear_cache(self): """Clear all cached audio.""" global GLOBAL_AUDIO_CACHE GLOBAL_AUDIO_CACHE.clear() def get_cache_stats(self) -> Dict[str, Any]: """Get cache statistics.""" total_items = len(GLOBAL_AUDIO_CACHE) total_memory = sum( audio.numel() * audio.element_size() + 8 # 8 bytes for duration float for audio, _ in GLOBAL_AUDIO_CACHE.values() ) return { 'total_items': total_items, 'total_memory_bytes': total_memory, 'total_memory_mb': total_memory / (1024 * 1024) } # Global cache instance audio_cache = AudioCache() def get_audio_cache() -> AudioCache: """Get the global audio cache instance.""" return audio_cache def create_cache_function(engine_type: str, **static_params) -> Callable: """ Convenience function to create a cache function. Args: engine_type: "f5tts" or "chatterbox" **static_params: Parameters that don't change between calls Returns: Cache function for use with TTS generation """ return audio_cache.create_cache_function(engine_type, **static_params)