TTS Best Practices

class OptimizedTTSConnection:
    def __init__(self, provider):
        self.provider = provider
        self.connection = None
        self.last_activity = time.time()
        
    async def get_connection(self):
        # Reuse existing connection if alive
        if self.connection and self.is_connection_alive():
            return self.connection
            
        # Create new connection
        self.connection = await self.create_connection()
        self.last_activity = time.time()
        return self.connection
    
    async def keep_alive(self):
        """Send periodic heartbeat to maintain connection"""
        while self.connection:
            await asyncio.sleep(30)  # Heartbeat every 30 seconds
            if time.time() - self.last_activity > 300:  # 5 min timeout
                await self.close_connection()
                break
            await self.send_heartbeat()

import hashlib
import pickle
from functools import lru_cache

class TTSCache:
    def __init__(self, max_size_mb=100):
        self.cache = {}
        self.max_size = max_size_mb * 1024 * 1024  # Convert to bytes
        self.current_size = 0
    
    def cache_key(self, text, provider, voice, settings):
        """Generate unique cache key"""
        key_data = f"{text}|{provider}|{voice}|{json.dumps(settings, sort_keys=True)}"
        return hashlib.md5(key_data.encode()).hexdigest()
    
    def get(self, key):
        if key in self.cache:
            # Move to end (LRU)
            value = self.cache.pop(key)
            self.cache[key] = value
            return value["audio"]
        return None
    
    def set(self, key, audio_data):
        audio_size = len(audio_data)
        
        # Evict if needed
        while self.current_size + audio_size > self.max_size and self.cache:
            oldest_key = next(iter(self.cache))
            self.evict(oldest_key)
        
        self.cache[key] = {
            "audio": audio_data,
            "size": audio_size,
            "timestamp": time.time()
        }
        self.current_size += audio_size

class SmartTTSCache:
    def __init__(self):
        self.phrase_frequency = defaultdict(int)
        self.cache_hits = 0
        self.cache_misses = 0
        
    def should_cache(self, text):
        """Decide if text is worth caching"""
        # Cache frequently used phrases
        self.phrase_frequency[text] += 1
        
        criteria = [
            len(text) > 20,  # Long enough to be worth caching
            self.phrase_frequency[text] >= 3,  # Used multiple times
            not self.is_time_sensitive(text),  # Not time-specific
            not self.contains_variables(text)  # No dynamic content
        ]
        
        return all(criteria)
    
    def cache_roi_analysis(self):
        """Calculate cache return on investment"""
        cache_savings = self.cache_hits * average_api_cost
        cache_storage_cost = len(self.cache) * storage_cost_per_item
        return cache_savings - cache_storage_cost

def optimize_text_for_cost(text):
    """Reduce character count without losing meaning"""
    
    # Use contractions
    contractions = {
        "you are": "you're",
        "we are": "we're", 
        "it is": "it's",
        "that is": "that's",
        "do not": "don't",
        "will not": "won't"
    }
    
    for full, contracted in contractions.items():
        text = text.replace(full, contracted)
    
    # Remove redundant phrases
    redundant_patterns = [
        r'\bplease note that\b',
        r'\bI would like to\b',
        r'\bI want to\b', 
        r'\bas you can see\b'
    ]
    
    for pattern in redundant_patterns:
        text = re.sub(pattern, '', text, flags=re.IGNORECASE)
    
    # Normalize whitespace
    text = re.sub(r'\s+', ' ', text).strip()
    
    return text

# Example usage
original = "Please note that I would like to inform you that your order has been processed."
optimized = "Your order's been processed."  # 67% character reduction

def calculate_monthly_cost(chars_per_day, provider="elevenlabs"):
    rates = {
        "elevenlabs": 0.00018,  # Flash v2.5 rate
        "deepgram": 0.000135,
        "inworld": 0.00015,
        "resemble": 0.00012
    }
    
    monthly_chars = chars_per_day * 30
    monthly_cost = monthly_chars * rates[provider]
    
    return {
        "monthly_cost": monthly_cost,
        "yearly_cost": monthly_cost * 12,
        "cost_per_call": monthly_cost / (chars_per_day * 30 / 200)  # Assuming 200 chars per call
    }

class TTSFeedbackSystem:
    def __init__(self):
        self.feedback_db = []
        
    def collect_implicit_feedback(self, session_data):
        """Collect implicit feedback from user behavior"""
        signals = {
            "completion_rate": session_data.get("completed", False),
            "repeat_requests": session_data.get("repeats", 0),
            "early_termination": session_data.get("hung_up_early", False),
            "session_duration": session_data.get("duration", 0)
        }
        
        # Convert signals to quality score
        quality_score = self.calculate_implicit_score(signals)
        
        return {
            "session_id": session_data["id"],
            "quality_score": quality_score,
            "feedback_type": "implicit",
            "timestamp": time.time()
        }
    
    def collect_explicit_feedback(self, session_id, rating, comments=None):
        """Collect explicit user feedback"""
        return {
            "session_id": session_id,
            "rating": rating,  # 1-5 scale
            "comments": comments,
            "feedback_type": "explicit",
            "timestamp": time.time()
        }
    
    def aggregate_feedback(self, provider, voice, timeframe_days=30):
        """Aggregate feedback for analysis"""
        cutoff = time.time() - (timeframe_days * 24 * 3600)
        
        relevant_feedback = [
            f for f in self.feedback_db 
            if f["timestamp"] > cutoff 
            and f.get("provider") == provider
            and f.get("voice") == voice
        ]
        
        if relevant_feedback:
            avg_rating = np.mean([f["rating"] for f in relevant_feedback])
            sample_size = len(relevant_feedback)
            
            return {
                "average_rating": avg_rating,
                "sample_size": sample_size,
                "recommendation": self.get_recommendation(avg_rating, sample_size)
            }
        
        return None

class TTSHealthChecker:
    def __init__(self):
        self.providers = ["elevenlabs", "deepgram", "inworld", "resemble"]
        self.health_status = {}
        
    async def comprehensive_health_check(self):
        """Run comprehensive health checks"""
        results = {}
        
        for provider in self.providers:
            try:
                # Test basic connectivity
                connectivity = await self.test_connectivity(provider)
                
                # Test response time
                latency = await self.test_latency(provider)
                
                # Test quality with standard phrase
                quality = await self.test_quality(provider, "Hello, this is a test.")
                
                results[provider] = {
                    "status": "healthy" if all([connectivity, latency < 5.0, quality > 0.8]) else "degraded",
                    "connectivity": connectivity,
                    "latency": latency,
                    "quality": quality,
                    "timestamp": time.time()
                }
                
            except Exception as e:
                results[provider] = {
                    "status": "unhealthy",
                    "error": str(e),
                    "timestamp": time.time()
                }
        
        return results
    
    async def setup_monitoring_alerts(self):
        """Setup automated monitoring and alerting"""
        while True:
            health_results = await self.comprehensive_health_check()
            
            for provider, result in health_results.items():
                if result["status"] != "healthy":
                    await self.send_alert(provider, result)
            
            await asyncio.sleep(300)  # Check every 5 minutes

class RobustTTSService:
    def __init__(self):
        self.retry_config = {
            "max_retries": 3,
            "backoff_multiplier": 2,
            "base_delay": 1.0
        }
        
    async def synthesize_with_retry(self, text, provider_config):
        """Synthesize with automatic retry and fallback"""
        last_exception = None
        
        for attempt in range(self.retry_config["max_retries"]):
            try:
                return await self.synthesize_text(text, provider_config)
                
            except RateLimitError as e:
                # Handle rate limiting specially
                delay = self.calculate_rate_limit_delay(e)
                await asyncio.sleep(delay)
                last_exception = e
                
            except (ConnectionError, TimeoutError) as e:
                # Handle network issues
                delay = self.retry_config["base_delay"] * (
                    self.retry_config["backoff_multiplier"] ** attempt
                )
                await asyncio.sleep(delay)
                last_exception = e
                
            except Exception as e:
                # Other errors - try fallback provider
                if attempt == self.retry_config["max_retries"] - 1:
                    return await self.try_fallback_provider(text)
                last_exception = e
        
        # All retries failed
        raise TTSServiceError(f"All retries failed: {last_exception}")
    
    async def try_fallback_provider(self, text):
        """Try alternative provider as fallback"""
        fallback_order = ["deepgram", "elevenlabs", "inworld"]
        
        for provider in fallback_order:
            try:
                return await self.synthesize_text(text, {"provider": provider})
            except Exception:
                continue
        
        raise TTSServiceError("All providers failed")

class TTSLoadBalancer:
    def __init__(self):
        self.provider_weights = {
            "elevenlabs": 0.4,  # 40% of traffic
            "deepgram": 0.3,    # 30% of traffic  
            "inworld": 0.2,     # 20% of traffic
            "resemble": 0.1     # 10% of traffic
        }
        self.health_status = {}
        
    def select_provider(self, requirements=None):
        """Select provider based on requirements and health"""
        
        # Filter by requirements
        candidates = self.filter_by_requirements(requirements)
        
        # Filter by health
        healthy_candidates = [
            p for p in candidates 
            if self.health_status.get(p, {}).get("status") == "healthy"
        ]
        
        if not healthy_candidates:
            # Fallback to any healthy provider
            healthy_candidates = [
                p for p, status in self.health_status.items()
                if status.get("status") == "healthy"
            ]
        
        # Weighted selection
        return self.weighted_random_selection(healthy_candidates)
    
    def adjust_weights_by_performance(self):
        """Dynamically adjust weights based on performance"""
        performance_metrics = self.get_performance_metrics()
        
        for provider in self.provider_weights:
            metrics = performance_metrics.get(provider, {})
            
            # Adjust weight based on performance
            if metrics.get("avg_latency", float('inf')) > 2.0:
                self.provider_weights[provider] *= 0.9  # Reduce weight
            elif metrics.get("error_rate", 1.0) < 0.01:
                self.provider_weights[provider] *= 1.1  # Increase weight
        
        # Normalize weights
        total_weight = sum(self.provider_weights.values())
        for provider in self.provider_weights:
            self.provider_weights[provider] /= total_weight