import io import os import time import numpy as np import requests import torch from pydub import AudioSegment import logging class ElevenLabsNode: voices_cache = None voices_map = {} models_cache = None last_fetch_time = 0 cache_duration = 3600 # Cache duration in seconds (1 hour) def __init__(self): pass @classmethod def INPUT_TYPES(cls): voices = cls.fetch_elevenlabs_voices() models = cls.fetch_elevenlabs_models() return { "required": { "text": ("STRING", { "multiline": True, "default": "Hello, how are you?" }), "voice": (voices,), "model": (models,), "stability": ("FLOAT", { "default": 0.5, "min": 0.0, "max": 1.0, "step": 0.1 }), "similarity_boost": ("FLOAT", { "default": 0.8, "min": 0.0, "max": 1.0, "step": 0.1 }), "style": ("FLOAT", { "default": 0.0, "min": 0.0, "max": 1.0, "step": 0.1 }), "use_speaker_boost": ("BOOLEAN", { "default": True, }), }, "optional": { "input_text": ("STRING", {"forceInput": True}), } } RETURN_TYPES = ("AUDIO",) FUNCTION = "generate_speech" CATEGORY = "ElevenLabs" @classmethod def _get_api_key(cls): api_key = os.getenv("ELEVENLABS_API_KEY", "").strip() if not api_key: logging.error("ELEVENLABS_API_KEY is not set. Please configure it in your environment.") return None return api_key @classmethod def fetch_elevenlabs_voices(cls): current_time = time.time() if cls.voices_cache is None or (current_time - cls.last_fetch_time > cls.cache_duration): voice_list = [] voice_map = {} api_key = cls._get_api_key() if api_key is None: cls.voices_cache = ["missing_api_key"] cls.voices_map = {} return cls.voices_cache # Fetch voices from the API url = "https://api.elevenlabs.io/v1/voices" try: response = requests.get(url, headers={"xi-api-key": api_key}) response.raise_for_status() voices = response.json()["voices"] name_counts = {} for voice in voices: base_name = voice["name"] count = name_counts.get(base_name, 0) + 1 name_counts[base_name] = count display_name = base_name if count == 1 else f"{base_name} ({count})" voice_list.append(display_name) voice_map[display_name] = voice["voice_id"] cls.voices_cache = voice_list cls.voices_map = voice_map cls.last_fetch_time = current_time except requests.exceptions.RequestException as e: logging.error(f"Error fetching voices: {e}") if cls.voices_cache is None: cls.voices_cache = ["error_fetching_voices"] cls.voices_map = {} return cls.voices_cache @classmethod def fetch_elevenlabs_models(cls): current_time = time.time() if cls.models_cache is None or (current_time - cls.last_fetch_time > cls.cache_duration): cls.models_cache = ["eleven_multilingual_v2", "eleven_english_sts_v2", "eleven_turbo_v2"] cls.last_fetch_time = current_time return cls.models_cache def ensure_3d_tensor(self, tensor): if tensor.dim() == 1: return tensor.unsqueeze(0).unsqueeze(0) elif tensor.dim() == 2: return tensor.unsqueeze(0) elif tensor.dim() > 3: return tensor.squeeze().unsqueeze(0) return tensor def _load_audio_from_bytes(self, audio_bytes: bytes): """Decode ElevenLabs MP3 response into torch waveform.""" if not audio_bytes: logging.error("Empty audio response received.") return None, None try: audio = AudioSegment.from_file(io.BytesIO(audio_bytes), format="mp3") except Exception as e: preview = audio_bytes[:80] try: preview_text = preview.decode("utf-8", errors="replace") except Exception: preview_text = str(preview) logging.error(f"Error decoding MP3 audio: {e}. Preview: {preview_text}") return None, None sample_rate = audio.frame_rate sample_width = audio.sample_width channels = audio.channels samples = np.array(audio.get_array_of_samples()) if channels > 1: samples = samples.reshape(-1, channels).T # channels x samples else: samples = samples.reshape(1, -1) max_val = float(2 ** (8 * sample_width - 1)) waveform = torch.from_numpy(samples).unsqueeze(0).float() / max_val waveform = self.ensure_3d_tensor(waveform) return waveform, sample_rate def generate_speech(self, text, voice, model, stability, similarity_boost, style, use_speaker_boost, input_text=None): # Use input_text if provided, otherwise use the text from the textbox final_text = input_text if input_text is not None else text voice_id = self.voices_map.get(voice) if not voice_id: logging.error(f"Voice ID not found for selection '{voice}'. Refreshing voices cache.") self.fetch_elevenlabs_voices() voice_id = self.voices_map.get(voice) if not voice_id: return ({"waveform": torch.zeros(1, 1, 1).float(), "sample_rate": 44100},) api_key = self._get_api_key() if api_key is None: return ({"waveform": torch.zeros(1, 1, 1).float(), "sample_rate": 44100},) headers = { "Accept": "audio/mpeg", "xi-api-key": api_key, "Content-Type": "application/json" } # Text-to-Speech url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}" payload = { "text": final_text, "model_id": model, "voice_settings": { "stability": stability, "similarity_boost": similarity_boost, "style": style, "use_speaker_boost": use_speaker_boost } } try: response = requests.post(url, headers=headers, json=payload) except requests.exceptions.RequestException as e: logging.error(f"Error in text-to-speech: {str(e)}") return ({"waveform": torch.zeros(1, 1, 1).float(), "sample_rate": 44100},) if response.status_code == 200: waveform, sample_rate = self._load_audio_from_bytes(response.content) if waveform is None or sample_rate is None: return ({"waveform": torch.zeros(1, 1, 1).float(), "sample_rate": 44100},) # Ensure the waveform is 3D if waveform.dim() != 3: logging.info(f"Unexpected tensor dimension {waveform.dim()}. Reshaping to 3D.") waveform = waveform.view(1, 1, -1) return ({"waveform": waveform, "sample_rate": sample_rate},) else: logging.error(f"API Error: {response.status_code} - {response.text}") return ({"waveform": torch.zeros(1, 1, 1).float(), "sample_rate": 44100},) @classmethod def IS_CHANGED(cls, text, voice, model, stability, similarity_boost, style, use_speaker_boost, input_text=None): return (text, voice, model, stability, similarity_boost, style, use_speaker_boost, input_text) # Node class mappings NODE_CLASS_MAPPINGS = { "ElevenLabsNode": ElevenLabsNode } # Node display name mappings NODE_DISPLAY_NAME_MAPPINGS = { "ElevenLabsNode": "ElevenLabs TTS Node" }