text-to-speech

What this skill does

# 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

```python
# 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

```bash
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)

```python
# 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)

```python
# 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)

```python
# 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)

```python
# 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)

```python
# 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**

| Use Case | Version | Notes |
|----------|---------|-------|
| **Production** | kokoro>=0.3.0 | Latest stable |

**Supporting Libraries**

```python
# 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

| Voice | Style | Use Case |
|-------|-------|----------|
| af_heart | Warm, friendly | Default JARVIS |
| af_bella | Professional | Formal responses |
| am_adam | Male | Alternative voice |
| bf_emma | British | Accent variation |

---

## 7. Implementation Patterns

### Pattern 1: Secure TTS Engine

```python
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_d