Audio & Voice

from gaia.audio.audio_client import AudioClient

class AudioClient:
    """Handles all audio-related functionality including TTS, ASR, and voice chat."""

    def __init__(
        self,
        whisper_model_size="base",
        audio_device_index=None,
        silence_threshold=0.5,
        mic_threshold=0.003,
        enable_tts=True,
        logging_level="INFO",
        use_claude=False,
        use_chatgpt=False,
        system_prompt=None,
        model=None,
        claude_model="claude-sonnet-5",
        base_url=None,
    ):
        """
        Initialize audio client.

        Args:
            whisper_model_size: Whisper model size (tiny, base, small, medium, large, turbo)
            audio_device_index: Audio input device index (None = default)
            silence_threshold: Silence duration before processing (seconds)
            mic_threshold: Microphone amplitude threshold for voice-activity
                detection (passed through to WhisperAsr as silence_threshold).
            enable_tts: Enable text-to-speech output
            logging_level: Logging level (DEBUG, INFO, WARNING, ERROR)
            use_claude: Use Claude API for LLM
            use_chatgpt: Removed; True raises migration guidance
            system_prompt: Default system prompt for LLM
            model: Model ID for the local/OpenAI provider (None = provider default)
            claude_model: Claude model ID, used when use_claude=True
            base_url: Lemonade server URL (None = LEMONADE_BASE_URL or default)
        """
        pass

    async def start_voice_chat(self, message_processor_callback):
        """
        Start a voice-based chat session.

        Args:
            message_processor_callback: Async callback to process transcribed text

        Example:
            >>> async def process_message(text: str):
            ...     await audio.process_voice_input(text)
            >>> await audio.start_voice_chat(process_message)
        """
        pass

    async def process_voice_input(self, text: str, get_stats_callback=None):
        """
        Process transcribed voice input and get AI response.

        Capture stays paused until streamed playback has fully finished;
        transcriptions queued meanwhile are discarded before capture resumes.

        Args:
            text: Transcribed text to process
            get_stats_callback: Optional callback for performance stats
        """
        pass

    async def speak_text(self, text: str):
        """Speak text using TTS.

        Returns once playback has finished. The microphone is paused for the
        whole utterance and anything transcribed meanwhile is discarded, so
        the assistant never hears its own reply as the next user turn.
        Playback cleanup is bounded by synthesized audio duration plus five
        seconds. On a device timeout, the call raises and capture stays paused;
        restart voice chat after checking the output device.
        """
        pass

    async def halt_generation(self):
        """Send a request to halt the current generation."""
        pass

from gaia.audio.whisper_asr import WhisperAsr

class WhisperAsr:
    """Live-microphone speech-to-text, transcribed by Lemonade's Whisper."""

    def __init__(
        self,
        model_size="base",
        device_index=None,
        transcription_queue=None,
        silence_threshold=None,
        min_audio_length=None,
        base_url=None,
        language="en",
        client=None,
        say=None,
    ):
        """
        Initialize Whisper ASR and pull its Lemonade model if needed.

        Args:
            model_size: tiny, base, small, medium, large or turbo — mapped to
                Whisper-Tiny … Whisper-Large-v3-Turbo on Lemonade
            device_index: Audio input device index (None = default)
            transcription_queue: Queue to receive transcriptions
            silence_threshold: Custom voice-activity energy threshold
            min_audio_length: Minimum utterance length in samples
            base_url: Lemonade API root (default: LEMONADE_BASE_URL)
            language: ISO 639-1 code sent with every utterance
            client: A ready LemonadeASRClient (tests inject one)
            say: Called with a line before a model download

        Raises:
            ConnectionError: Lemonade Server is not reachable
            LemonadeASRError: Lemonade cannot serve the Whisper model
        """
        pass

    def transcribe_utterance(self, audio) -> str:
        """Send one 16 kHz float utterance to Lemonade; return its text."""
        pass

    def start_recording(self):
        """Start recording audio."""
        pass

    def stop_recording(self):
        """Stop recording audio."""
        pass

    def pause_recording(self):
        """Pause audio recording (e.g., during TTS playback)."""
        pass

    def resume_recording(self):
        """Resume audio recording after pause."""
        pass

    def get_device_name(self) -> str:
        """Get the name of the current audio input device."""
        pass

from gaia.audio.kokoro_tts import KokoroTTS

class KokoroTTS:
    """Kokoro-based Text-to-Speech synthesis."""

    def __init__(self, base_url=None, client=None, say=None):
        """
        Initialize Kokoro TTS and pull Lemonade's kokoro-v1 model if needed.

        Note:
            - Default voice: af_bella (A- quality)
            - 17+ voices available
            - Playback needs sounddevice ([talk] extra); synthesis does not
        """
        pass

    def generate_speech(self, text, stream_callback=None) -> tuple:
        """Return (24 kHz float32 samples, stats), synthesized sentence by sentence."""
        pass

    def generate_speech_streaming(
        self,
        text_queue,
        status_callback=None,
        interrupt_event=None,
    ):
        """
        Generate and play speech from streaming text queue.

        Args:
            text_queue: Queue containing text chunks and control signals
            status_callback: Callback for speaking state (True=speaking, False=done)
            interrupt_event: Event to signal interruption

        Control Signals:
            "__END__": End of text stream
            "__HALT__": Immediate stop
        """
        pass

    def list_available_voices(self) -> dict:
        """
        Get available voice configurations.

        Returns:
            Dictionary of voice configs with quality/duration ratings
        """
        pass

    def set_voice(self, voice_id: str):
        """
        Set the TTS voice.

        Args:
            voice_id: Voice identifier (e.g., "af_bella", "am_michael")
        """
        pass

# pyproject.toml
[project.optional-dependencies]
talk = [
    "sounddevice",   # Microphone capture & speaker playback
    "websockets",    # Staged for Lemonade Realtime transcription (#372)
]

import asyncio
from gaia.audio.audio_client import AudioClient

async def main():
    audio = AudioClient(
        whisper_model_size="base",
        enable_tts=True,
        system_prompt="You are a helpful assistant"
    )

    async def process_message(text):
        await audio.process_voice_input(text)

    await audio.start_voice_chat(process_message)

asyncio.run(main())

from gaia.audio.audio_client import AudioClient

audio = AudioClient(enable_tts=True)
audio.initialize_tts()

# Change voice
audio.tts.set_voice("am_michael")  # Male voice

# Speak text
await audio.speak_text("Hello in a different voice")

import queue
from gaia.audio.whisper_asr import WhisperAsr

# Create transcription queue
transcription_queue = queue.Queue()

# Initialize ASR
asr = WhisperAsr(
    model_size="small",  # Whisper-Small on Lemonade
    transcription_queue=transcription_queue,
)

# Start recording
asr.start_recording()

# Process transcriptions
try:
    while True:
        text = transcription_queue.get(timeout=1)
        print(f"Transcribed: {text}")
except KeyboardInterrupt:
    asr.stop_recording()

import queue
import threading
from gaia.audio.kokoro_tts import KokoroTTS

tts = KokoroTTS()
text_queue = queue.Queue()

# Start TTS thread
thread = threading.Thread(
    target=tts.generate_speech_streaming,
    args=(text_queue,)
)
thread.start()

# Send text
text_queue.put("Hello")
text_queue.put(" world!")
text_queue.put("__END__")

thread.join()