ElevenLabs

pip install elevenlabs netra-sdk

import os
from netra import Netra

Netra.init(
    app_name="elevenlabs-tts-service",
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}"
)

from elevenlabs import ElevenLabs
from netra import task
import os

client = ElevenLabs(api_key=os.environ.get("ELEVENLABS_API_KEY"))

@task()
def synthesize_speech(text: str, voice_id: str) -> bytes:
    """Generate speech from text using ElevenLabs."""
    audio = client.text_to_speech.convert(
        voice_id=voice_id,
        text=text,
        model_id="eleven_turbo_v2",
        voice_settings={
            "stability": 0.5,
            "similarity_boost": 0.75
        }
    )
    
    # Collect audio chunks
    audio_bytes = b""
    for chunk in audio:
        audio_bytes += chunk
    
    return audio_bytes

@task()
def synthesize_with_emotions(text: str, voice_id: str, emotion: str) -> bytes:
    """Generate emotional speech using advanced models."""
    audio = client.text_to_speech.convert(
        voice_id=voice_id,
        text=text,
        model_id="eleven_turbo_v2_5",
        voice_settings={
            "stability": 0.6,
            "similarity_boost": 0.8,
            "style": 0.5,
            "use_speaker_boost": True
        }
    )
    
    audio_bytes = b""
    for chunk in audio:
        audio_bytes += chunk
    
    return audio_bytes

# Usage
audio_data = synthesize_speech(
    "Hello, this is a test of ElevenLabs speech synthesis.",
    "pNInz6obpgDQGcFmaJgB"
)

from elevenlabs import ElevenLabs
from netra import SpanWrapper, ActionModel, UsageModel
import os
import time

client = ElevenLabs(api_key=os.environ.get("ELEVENLABS_API_KEY"))

def synthesize_with_tracking(text: str, voice_id: str) -> bytes:
    """Generate speech with detailed tracking."""
    span = SpanWrapper("elevenlabs-synthesis")
    span.start()
    
    try:
        start_time = time.time_ns()
        span.set_attribute("voice_id", voice_id)
        span.set_attribute("text_length", len(text))
        span.set_attribute("model", "eleven_turbo_v2")
        
        audio = client.text_to_speech.convert(
            voice_id=voice_id,
            text=text,
            model_id="eleven_turbo_v2",
            voice_settings={
                "stability": 0.5,
                "similarity_boost": 0.75
            }
        )
        
        # Collect audio chunks
        audio_bytes = b""
        for chunk in audio:
            audio_bytes += chunk
        
        end_time = time.time_ns()
        duration_ms = (end_time - start_time) / 1_000_000
        
        # Track the TTS API operation
        action = ActionModel(
            start_time=str(start_time),
            action="API",
            action_type="TTS_SYNTHESIS",
            metadata={
                "provider": "elevenlabs",
                "voice_id": voice_id,
                "model": "eleven_turbo_v2",
                "text_length": str(len(text)),
                "audio_size_bytes": str(len(audio_bytes)),
                "duration_ms": str(duration_ms)
            },
            success=True
        )
        span.set_action([action])
        
        # Track usage
        usage = UsageModel(
            model="eleven_turbo_v2",
            usage_type="characters",
            units_used=len(text),
            cost_in_usd=len(text) * 0.00003  # $0.30 per 1000 characters
        )
        span.set_usage([usage])
        
        span.set_status({"code": 1, "message": "Success"})
        span.end()
        
        return audio_bytes
        
    except Exception as e:
        span.set_error(e)
        span.set_status({"code": 2, "message": "Error"})
        span.end()
        raise

# Usage
audio_data = synthesize_with_tracking(
    "This is tracked speech synthesis with detailed metrics.",
    "pNInz6obpgDQGcFmaJgB"
)