src/speech.py

# File: src/speech.py
# Purpose: Speech-to-Text via Groq Whisper and Text-to-Speech via ElevenLabs

import tempfile
from pathlib import Path

import httpx
from groq import Groq

from config import (
    GROQ_API_KEY, GROQ_STT_MODEL,
    ELEVENLABS_API_KEY, ELEVENLABS_VOICE_ID, ELEVENLABS_MODEL_ID,
)

_groq = Groq(api_key=GROQ_API_KEY)

# ── Speech-to-Text (Groq Whisper) ─────────────────────────────────────────────

def transcribe_audio(audio_bytes: bytes, content_type: str = "audio/wav") -> str:
    """Transcribe audio bytes using Groq Whisper. Returns transcript string."""
    suffix_map = {
        "audio/wav": ".wav",
        "audio/mpeg": ".mp3",
        "audio/webm": ".webm",
        "audio/ogg": ".ogg",
    }
    suffix = suffix_map.get(content_type, ".wav")

    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
        tmp.write(audio_bytes)
        tmp_path = tmp.name

    try:
        with open(tmp_path, "rb") as f:
            transcript = _groq.audio.transcriptions.create(
                model=GROQ_STT_MODEL,
                file=(Path(tmp_path).name, f, content_type),
                language="en",
            )
        return transcript.text.strip()
    finally:
        Path(tmp_path).unlink(missing_ok=True)


# ── Text-to-Speech (ElevenLabs) ───────────────────────────────────────────────

ELEVENLABS_TTS_URL = f"https://api.elevenlabs.io/v1/text-to-speech/{ELEVENLABS_VOICE_ID}"

def synthesize_speech(text: str) -> bytes:
    """Convert text to speech using ElevenLabs. Returns raw MP3 bytes."""
    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
        "Content-Type": "application/json",
        "Accept": "audio/mpeg",
    }
    payload = {
        "text": text,
        "model_id": ELEVENLABS_MODEL_ID,
        "voice_settings": {
            "stability": 0.55,
            "similarity_boost": 0.75,
            "style": 0.0,
            "use_speaker_boost": True,
        },
    }
    response = httpx.post(ELEVENLABS_TTS_URL, headers=headers, json=payload, timeout=30)
    response.raise_for_status()
    return response.content