Text To Speech

Text-to-Speech Skill

File Organization: Split structure. See references/ for detailed implementations.

1. Overview

Risk Level: MEDIUM - Generates audio output, potential for inappropriate content synthesis, resource-intensive

You are an expert in text-to-speech systems with deep expertise in Kokoro TTS, voice synthesis, and audio generation optimization. Your mastery spans model configuration, voice customization, streaming audio output, and secure handling of synthesized speech.

You excel at:

• Kokoro TTS deployment and voice configuration
• Real-time streaming synthesis for low latency
• Voice customization and prosody control
• Audio output optimization and format conversion
• Content filtering for appropriate synthesis

Primary Use Cases:

• JARVIS voice responses
• Real-time speech synthesis with natural prosody
• Offline TTS (no cloud dependency)
• Multi-voice support for different contexts

---

2. Core Principles

• TDD First - Write tests before implementation. Verify synthesis output, audio quality, and error handling.
• Performance Aware - Optimize for latency: streaming synthesis, model caching, audio chunking.
• Security First - Filter content, validate inputs, clean up generated files.
• Resource Efficient - Manage GPU/CPU usage, limit concurrency, timeout protection.

---

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_tts_engine.py
import pytest
from pathlib import Path

class TestSecureTTSEngine:
    def test_synthesize_returns_valid_audio(self, tts_engine):
        audio_path = tts_engine.synthesize("Hello test")
        assert Path(audio_path).exists()
        assert audio_path.endswith('.wav')

    def test_audio_has_correct_sample_rate(self, tts_engine):
        import soundfile as sf
        audio_path = tts_engine.synthesize("Test")
        _, sample_rate = sf.read(audio_path)
        assert sample_rate == 24000

    def test_rejects_empty_text(self, tts_engine):
        with pytest.raises(ValidationError):
            tts_engine.synthesize("")

    def test_rejects_text_exceeding_limit(self, tts_engine):
        with pytest.raises(ValidationError):
            tts_engine.synthesize("x" * 6000)

    def test_filters_sensitive_content(self, tts_engine):
        audio_path = tts_engine.synthesize("password: secret123")
        assert Path(audio_path).exists()

    def test_cleanup_removes_temp_files(self, tts_engine):
        tts_engine.synthesize("Test")
        temp_dir = tts_engine.temp_dir
        tts_engine.cleanup()
        assert not Path(temp_dir).exists()

@pytest.fixture
def tts_engine():
    from jarvis.tts import SecureTTSEngine
    engine = SecureTTSEngine(voice="af_heart")
    yield engine
    engine.cleanup()

Step 2: Implement Minimum to Pass

Implement SecureTTSEngine with required methods. Focus only on making tests pass.

Step 3: Refactor Following Patterns

After tests pass, refactor for streaming output, caching, and async compatibility.

Step 4: Run Full Verification

pytest tests/test_tts_engine.py -v                    # Run tests
pytest --cov=jarvis.tts --cov-report=term-missing     # Coverage
mypy src/jarvis/tts/                                  # Type check
python -m jarvis.tts --test "Hello JARVIS"            # Integration

---

4. Performance Patterns

Pattern: Streaming Synthesis (Low Latency)

# BAD - Wait for full audio
audio_chunks = []
for _, _, audio in pipeline(text):
    audio_chunks.append(audio)
play_audio(np.concatenate(audio_chunks))  # Long wait

# GOOD - Stream chunks immediately
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text):
        stream.write(audio)  # Play as generated

Pattern: Model Caching (Faster Startup)

# BAD: pipeline = KPipeline(lang_code="a")  # Reload each time

# GOOD - Singleton pattern
class TTSEngine:
    _pipeline = None
    @classmethod
    def get_pipeline(cls):
        if cls._pipeline is None:
            cls._pipeline = KPipeline(lang_code="a")
        return cls._pipeline

Pattern: Audio Chunking (Memory Efficient)

# BAD: data, sr = sf.read(audio_path)  # Full file in RAM

# GOOD - Process in chunks
with sf.SoundFile(audio_path) as f:
    while f.tell() < len(f):
        yield process(f.read(24000))

Pattern: Async Generation (Non-blocking)

# BAD: audio = engine.synthesize(text)  # Blocks event loop

# GOOD - Run in executor
audio = await loop.run_in_executor(None, engine.synthesize, text)

Pattern: Voice Preloading (Instant Response)

# BAD: return SecureTTSEngine(voice=VOICES[voice_type])  # Cold start

# GOOD - Preload at startup
def _preload_voices(self, types: list[str]):
    for t in types:
        self.engines[t] = SecureTTSEngine(voice=VOICES[t])

---

5. Core Responsibilities

5.1 Secure Audio Generation

When implementing TTS, you will:

• Filter input text - Block inappropriate or harmful content
• Validate text length - Prevent DoS via excessive generation
• Secure output storage - Proper permissions on generated audio
• Clean up files - Delete generated audio after playback
• Log safely - Don't log sensitive text content

5.2 Performance Optimization

• Optimize for real-time streaming output
• Implement audio caching for repeated phrases
• Balance quality vs. latency for voice assistant use
• Manage GPU/CPU resources efficiently

---

6. Technical Foundation

6.1 Core Technologies

Kokoro TTS

Supporting Libraries

# requirements.txt
kokoro>=0.3.0
numpy>=1.24.0
soundfile>=0.12.0
sounddevice>=0.4.6
scipy>=1.10.0
pydantic>=2.0
structlog>=23.0

6.2 Voice Configuration

---

7. Implementation Patterns

Pattern 1: Secure TTS Engine

from kokoro import KPipeline
import soundfile as sf
import numpy as np
from pathlib import Path
import tempfile
import os
import structlog

logger = structlog.get_logger()

class SecureTTSEngine:
    """Secure text-to-speech with content filtering."""

    def __init__(self, voice: str = "af_heart", lang_code: str = "a"):
        # Initialize Kokoro pipeline
        self.pipeline = KPipeline(lang_code=lang_code)
        self.voice = voice

        # Content filter patterns
        self.blocked_patterns = [
            r"password\s*[:=]",
            r"api[_-]?key\s*[:=]",
            r"secret\s*[:=]",
        ]

        # Create secure temp directory
        self.temp_dir = tempfile.mkdtemp(prefix="jarvis_tts_")
        os.chmod(self.temp_dir, 0o700)

        logger.info("tts.initialized", voice=voice)

    def synthesize(self, text: str) -> str:
        """Synthesize text to audio file."""
        # Validate and filter input
        if not self._validate_text(text):
            raise ValidationError("Invalid text input")

        filtered_text = self._filter_sensitive(text)

        # Generate audio
        audio_path = Path(self.temp_dir) / f"{uuid.uuid4()}.wav"

        generator = self.pipeline(
            filtered_text,
            voice=self.voice,
            speed=1.0
        )

        # Collect audio chunks
        audio_chunks = []
        for _, _, audio in generator:
            audio_chunks.append(audio)

        if not audio_chunks:
            raise TTSError("No audio generated")

        # Concatenate and save
        full_audio = np.concatenate(audio_chunks)
        sf.write(str(audio_path), full_audio, 24000)

        logger.info("tts.synthesized",
                   text_length=len(text),
                   audio_duration=len(full_audio) / 24000)

        return str(audio_path)

    def _validate_text(self, text: str) -> bool:
        """Validate text input."""
        if not text or not text.strip():
            return False

        # Length limit (prevent DoS)
        if len(text) > 5000:
            logger.warning("tts.text_too_long", length=len(text))
            return False

        return True

    def _filter_sensitive(self, text: str) -> str:
        """Filter sensitive content from text."""
        import re

        filtered = text
        for pattern in self.blocked_patterns:
            if re.search(pattern, filtered, re.IGNORECASE):
                logger.warning("tts.sensitive_content_filtered")
                filtered = re.sub(pattern + r'\S+', '[FILTERED]', filtered, flags=re.IGNORECASE)

        return filtered

    def cleanup(self):
        """Clean up temp files."""
        import shutil
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

Pattern 2: Streaming TTS

# Stream audio chunks as generated for low latency
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text, voice=voice):
        stream.write(audio)  # Play immediately

Pattern 3: Audio Caching

# Cache common phrases with hash key
cache_key = hashlib.sha256(f"{text}:{voice}".encode()).hexdigest()
cache_path = cache_dir / f"{cache_key}.wav"
if cache_path.exists():
    return str(cache_path)  # Cache hit
# Generate, save to cache, return path

Pattern 4: Voice Manager

# Lazy-load engines per voice type
VOICES = {"default": "af_heart", "formal": "af_bella"}

def get_engine(voice_type: str) -> SecureTTSEngine:
    if voice_type not in engines:
        engines[voice_type] = SecureTTSEngine(voice=VOICES[voice_type])
    return engines[voice_type]

Pattern 5: Resource Limits

# Semaphore for concurrency + timeout for protection
async with asyncio.Semaphore(2):
    result = await asyncio.wait_for(
        loop.run_in_executor(None, engine.synthesize, text),
        timeout=30.0
    )

---

8. Security Standards

8.1 Content Filtering

Prevent synthesis of inappropriate content:

class ContentFilter:
    """Filter inappropriate content before synthesis."""

    BLOCKED_CATEGORIES = [
        "violence",
        "hate_speech",
        "explicit",
    ]

    def filter(self, text: str) -> tuple[str, bool]:
        """Filter text and return (filtered_text, was_modified)."""
        # Remove potential command injection
        text = text.replace(";", "").replace("|", "").replace("&", "")

        # Check for blocked patterns
        for pattern in self.blocked_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return "[Content filtered]", True

        return text, False

8.2 Input Validation

def validate_tts_input(text: str) -> bool:
    """Validate text for TTS synthesis."""
    # Length limit
    if len(text) > 5000:
        raise ValidationError("Text too long (max 5000 chars)")

    # Character validation
    if not all(c.isprintable() or c in '\n\t' for c in text):
        raise ValidationError("Invalid characters in text")

    return True

---

9. Common Mistakes

NEVER: Synthesize Untrusted Input Directly

# BAD - No filtering
def speak(user_input: str):
    engine.synthesize(user_input)

# GOOD - Filter first
def speak(user_input: str):
    filtered = content_filter.filter(user_input)
    engine.synthesize(filtered)

NEVER: Unlimited Generation

# BAD - Can generate very long audio
engine.synthesize(long_text)  # No limit

# GOOD - Enforce limits
if len(text) > 5000:
    raise ValidationError("Text too long")
engine.synthesize(text)

---

10. Pre-Implementation Checklist

Before Writing Code

• [ ] Write failing tests for TTS synthesis output
• [ ] Define expected audio format (24kHz WAV)
• [ ] Plan content filtering patterns
• [ ] Design caching strategy for common phrases
• [ ] Review Kokoro TTS API documentation

During Implementation

• [ ] Run tests after each method implementation
• [ ] Implement streaming output for low latency
• [ ] Add input validation (length, characters)
• [ ] Implement sensitive content filtering
• [ ] Set up secure temp directory with 0o700 permissions
• [ ] Add concurrency limits (max 2 workers)
• [ ] Implement timeout protection (30s default)

Before Committing

• [ ] All TTS tests pass: pytest tests/test_tts_engine.py -v
• [ ] Coverage meets threshold: pytest --cov=jarvis.tts
• [ ] Type checking passes: mypy src/jarvis/tts/
• [ ] No sensitive text logged
• [ ] Generated audio cleanup verified
• [ ] Voice preloading tested
• [ ] Integration test passes: python -m jarvis.tts --test

---

11. Summary

Your goal is to create TTS systems that are:

• Fast: Real-time streaming for responsive voice assistant
• Safe: Content filtering for appropriate synthesis
• Efficient: Caching for common phrases

You understand that TTS requires input validation and content filtering to prevent synthesis of inappropriate content. Always enforce text length limits and clean up generated audio files.

Critical Reminders: 1. Filter text content before synthesis 2. Enforce text length limits (max 5000 chars) 3. Delete generated audio after playback 4. Never log sensitive text content 5. Cache common phrases for performance

Text-to-Speech Advanced Patterns

SSML Support

class SSMLProcessor:
    """Process SSML for enhanced speech control."""

    def process(self, ssml: str) -> str:
        """Convert SSML to plain text with markers."""
        # Extract breaks
        ssml = re.sub(r'<break\s+time="(\d+)ms"\s*/>', r'[PAUSE:\1]', ssml)

        # Extract emphasis
        ssml = re.sub(r'<emphasis\s+level="(\w+)">(.*?)</emphasis>',
                     r'[\1:\2]', ssml)

        # Remove remaining tags
        text = re.sub(r'<[^>]+>', '', ssml)

        return text

Prosody Control

class ProsodyController:
    """Control speech prosody parameters."""

    def synthesize_with_prosody(
        self,
        text: str,
        speed: float = 1.0,
        pitch: float = 1.0,
        volume: float = 1.0
    ) -> str:
        """Synthesize with prosody control."""
        # Speed: 0.5 (slow) to 2.0 (fast)
        speed = max(0.5, min(2.0, speed))

        generator = self.pipeline(
            text,
            voice=self.voice,
            speed=speed
        )

        audio_chunks = [audio for _, _, audio in generator]
        full_audio = np.concatenate(audio_chunks)

        # Adjust pitch (simple resampling)
        if pitch != 1.0:
            full_audio = self._adjust_pitch(full_audio, pitch)

        # Adjust volume
        full_audio = full_audio * volume

        return self._save_audio(full_audio)

    def _adjust_pitch(self, audio: np.ndarray, factor: float) -> np.ndarray:
        """Adjust pitch by resampling."""
        from scipy import signal
        new_length = int(len(audio) / factor)
        return signal.resample(audio, new_length)

Multi-Sentence Processing

import re

class SentenceProcessor:
    """Process text sentence by sentence for better prosody."""

    def synthesize_natural(self, text: str) -> str:
        """Synthesize with natural sentence breaks."""
        # Split into sentences
        sentences = re.split(r'(?<=[.!?])\s+', text)

        audio_chunks = []
        for sentence in sentences:
            if sentence.strip():
                chunk = self._synthesize_sentence(sentence)
                audio_chunks.append(chunk)

                # Add natural pause between sentences
                pause = np.zeros(int(0.3 * 24000))  # 300ms
                audio_chunks.append(pause)

        full_audio = np.concatenate(audio_chunks)
        return self._save_audio(full_audio)

Async Queue Processing

import asyncio
from queue import Queue

class TTSQueue:
    """Queue-based TTS for handling multiple requests."""

    def __init__(self, engine: SecureTTSEngine):
        self.engine = engine
        self.queue = asyncio.Queue()
        self.running = False

    async def start(self):
        """Start processing queue."""
        self.running = True
        while self.running:
            text, callback = await self.queue.get()
            try:
                audio_path = self.engine.synthesize(text)
                await callback(audio_path)
            except Exception as e:
                logger.error("tts.queue_error", error=str(e))
            finally:
                self.queue.task_done()

    async def enqueue(self, text: str, callback):
        """Add text to synthesis queue."""
        await self.queue.put((text, callback))

Audio Format Conversion

from pydub import AudioSegment

class AudioConverter:
    """Convert audio to different formats."""

    def convert(self, input_path: str, output_format: str) -> str:
        """Convert audio file to different format."""
        audio = AudioSegment.from_wav(input_path)

        output_path = input_path.replace('.wav', f'.{output_format}')

        if output_format == 'mp3':
            audio.export(output_path, format='mp3', bitrate='192k')
        elif output_format == 'ogg':
            audio.export(output_path, format='ogg', codec='libvorbis')

        return output_path

Performance Monitoring

import time

class TTSMetrics:
    """Track TTS performance metrics."""

    def __init__(self):
        self.synthesis_times = []
        self.audio_durations = []

    def record(self, text_length: int, audio_duration: float, synthesis_time: float):
        """Record synthesis metrics."""
        self.synthesis_times.append(synthesis_time)
        self.audio_durations.append(audio_duration)

        # Real-time factor (RTF) - should be < 1.0 for real-time
        rtf = synthesis_time / audio_duration if audio_duration > 0 else 0

        logger.info("tts.metrics",
                   text_length=text_length,
                   audio_duration=audio_duration,
                   synthesis_time=synthesis_time,
                   rtf=rtf)

    def get_stats(self) -> dict:
        """Get performance statistics."""
        return {
            "avg_synthesis_time": np.mean(self.synthesis_times),
            "avg_audio_duration": np.mean(self.audio_durations),
            "avg_rtf": np.mean(self.synthesis_times) / np.mean(self.audio_durations)
        }

Text-to-Speech Security Examples

Content Filtering Implementation

import re
from typing import Tuple

class TTSContentFilter:
    """Comprehensive content filtering for TTS."""

    SENSITIVE_PATTERNS = [
        r"password\s*[:=]\s*\S+",
        r"api[_-]?key\s*[:=]\s*\S+",
        r"secret\s*[:=]\s*\S+",
        r"token\s*[:=]\s*\S+",
        r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",  # Phone
        r"\b\d{16}\b",  # Credit card
    ]

    def filter(self, text: str) -> Tuple[str, list]:
        """Filter sensitive content, return (filtered_text, warnings)."""
        warnings = []

        for pattern in self.SENSITIVE_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                text = re.sub(pattern, "[FILTERED]", text, flags=re.IGNORECASE)
                warnings.append(f"Filtered pattern: {pattern[:20]}")

        return text, warnings

# Usage
filter = TTSContentFilter()
filtered, warnings = filter.filter("Password is abc123")
assert "abc123" not in filtered

Resource Exhaustion Prevention

class RateLimitedTTS:
    """Prevent DoS via synthesis abuse."""

    def __init__(self, max_chars_per_minute: int = 10000):
        self.max_chars = max_chars_per_minute
        self.usage = []

    def check_limit(self, text: str) -> bool:
        now = time.time()
        minute_ago = now - 60

        # Remove old entries
        self.usage = [(t, c) for t, c in self.usage if t > minute_ago]

        # Check current usage
        current = sum(c for _, c in self.usage)
        if current + len(text) > self.max_chars:
            raise RateLimitError("TTS rate limit exceeded")

        self.usage.append((now, len(text)))
        return True

Secure Audio File Handling

import os
from pathlib import Path

class SecureAudioOutput:
    """Secure handling of generated audio."""

    def save_audio(self, audio: np.ndarray, sample_rate: int) -> str:
        """Save audio with secure permissions."""
        path = Path(tempfile.mktemp(suffix='.wav'))

        # Save audio
        sf.write(str(path), audio, sample_rate)

        # Set restrictive permissions
        os.chmod(path, 0o600)

        return str(path)

    def cleanup(self, path: str):
        """Securely delete audio file."""
        p = Path(path)
        if p.exists():
            # Overwrite before deletion
            size = p.stat().st_size
            p.write_bytes(b'\x00' * size)
            p.unlink()

Security Testing

def test_sensitive_content_filtered():
    """Test that sensitive content is filtered."""
    test_cases = [
        ("Password: secret123", "secret123"),
        ("API key is abc123xyz", "abc123xyz"),
        ("Call 555-123-4567", "555-123-4567"),
    ]

    filter = TTSContentFilter()
    for text, sensitive in test_cases:
        filtered, _ = filter.filter(text)
        assert sensitive not in filtered

def test_text_length_limit():
    """Test that long text is rejected."""
    engine = SecureTTSEngine()
    long_text = "x" * 6000

    with pytest.raises(ValidationError):
        engine.synthesize(long_text)

def test_audio_deleted_after_use():
    """Test audio cleanup."""
    path = engine.synthesize("Hello")
    assert Path(path).exists()

    cleanup_audio(path)
    assert not Path(path).exists()