Voice Ai Integration

"""
Real-Time Voice Processing Module

Handles real-time audio streaming.
"""

import pyaudio
import numpy as np


class RealTimeVoiceProcessor:
    """Processes real-time voice input/output."""

    def __init__(self, sample_rate: int = 16000, chunk_size: int = 2048):
        """
        Initialize real-time processor.

        Args:
            sample_rate: Audio sample rate in Hz
            chunk_size: Chunk size for processing
        """
        self.sample_rate = sample_rate
        self.chunk_size = chunk_size
        self.audio = pyaudio.PyAudio()

    def stream_audio_input(self):
        """Stream audio from microphone."""
        stream = self.audio.open(
            format=pyaudio.paFloat32,
            channels=1,
            rate=self.sample_rate,
            input=True,
            frames_per_buffer=self.chunk_size
        )

        try:
            while True:
                data = stream.read(self.chunk_size)
                audio_chunk = np.frombuffer(data, dtype=np.float32)
                yield audio_chunk
        finally:
            stream.stop_stream()
            stream.close()

    def stream_audio_output(self, audio_stream):
        """Stream audio to speakers."""
        stream = self.audio.open(
            format=pyaudio.paFloat32,
            channels=1,
            rate=self.sample_rate,
            output=True,
            frames_per_buffer=self.chunk_size
        )

        try:
            for audio_chunk in audio_stream:
                stream.write(audio_chunk.tobytes())
        finally:
            stream.stop_stream()
            stream.close()

    def detect_speech_activity(self, audio_chunk: np.ndarray, threshold: float = 0.02) -> bool:
        """Detect if audio chunk contains speech."""
        # Simple energy-based VAD
        energy = np.sqrt(np.mean(audio_chunk ** 2))
        return energy > threshold

    def __del__(self):
        """Cleanup audio resources."""
        self.audio.terminate()

"""
Speech Recognition Providers Module

Implementations for various speech-to-text services.
"""

import io
import openai
import azure.cognitiveservices.speech as speechsdk
import requests
from typing import Optional


def transcribe_audio_google(audio_file: str) -> str:
    """Transcribe using Google Cloud Speech-to-Text."""
    from google.cloud import speech_v1

    client = speech_v1.SpeechClient()

    with io.open(audio_file, "rb") as audio:
        content = audio.read()

    audio = speech_v1.RecognitionAudio(content=content)
    config = speech_v1.RecognitionConfig(
        encoding=speech_v1.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=16000,
        language_code="en-US",
        enable_automatic_punctuation=True,
    )

    response = client.recognize(config=config, audio=audio)

    transcript = ""
    for result in response.results:
        transcript += result.alternatives[0].transcript

    return transcript


def transcribe_audio_whisper(audio_file: str) -> str:
    """Transcribe using OpenAI Whisper."""
    with open(audio_file, "rb") as f:
        transcript = openai.Audio.transcribe(
            model="whisper-1",
            file=f,
            language="en"
        )
    return transcript["text"]


def transcribe_audio_azure(audio_file: str, subscription_key: str, region: str) -> Optional[str]:
    """Transcribe using Azure Speech Services."""
    speech_config = speechsdk.SpeechConfig(
        subscription=subscription_key,
        region=region
    )
    audio_config = speechsdk.audio.AudioConfig(filename=audio_file)

    recognizer = speechsdk.SpeechRecognizer(
        speech_config=speech_config,
        audio_config=audio_config
    )

    result = recognizer.recognize_once()

    if result.reason == speechsdk.ResultReason.RecognizedSpeech:
        return result.text
    elif result.reason == speechsdk.ResultReason.NoMatch:
        return None
    elif result.reason == speechsdk.ResultReason.Canceled:
        raise Exception(result.cancellation_details.error_details)


def transcribe_audio_assemblyai(audio_file: str, api_key: str) -> str:
    """Transcribe using AssemblyAI."""
    # Upload audio
    with open(audio_file, "rb") as f:
        response = requests.post(
            "https://api.assemblyai.com/v2/upload",
            headers={"authorization": api_key},
            data=f
        )
    audio_url = response.json()["upload_url"]

    # Transcribe
    transcript_response = requests.post(
        "https://api.assemblyai.com/v2/transcript",
        headers={"authorization": api_key},
        json={"audio_url": audio_url}
    )

    transcript_id = transcript_response.json()["id"]

    # Poll for result
    import time
    while True:
        poll_response = requests.get(
            f"https://api.assemblyai.com/v2/transcript/{transcript_id}",
            headers={"authorization": api_key}
        )
        if poll_response.json()["status"] == "completed":
            return poll_response.json()["text"]
        time.sleep(1)

"""
Text-to-Speech Providers Module

Implementations for various TTS services.
"""

import openai
import requests
import azure.cognitiveservices.speech as speechsdk
from pathlib import Path
from typing import Optional


def text_to_speech_google(text: str, output_file: str = "output.mp3") -> str:
    """Convert text to speech using Google Cloud TTS."""
    from google.cloud import tts_v1

    client = tts_v1.TextToSpeechClient()

    input_text = tts_v1.SynthesisInput(text=text)

    voice = tts_v1.VoiceSelectionParams(
        language_code="en-US",
        name="en-US-Neural2-C",
    )

    audio_config = tts_v1.AudioConfig(
        audio_encoding=tts_v1.AudioEncoding.MP3
    )

    response = client.synthesize_speech(
        input=input_text,
        voice=voice,
        audio_config=audio_config
    )

    with open(output_file, "wb") as out:
        out.write(response.audio_content)

    return output_file


def text_to_speech_openai(text: str, output_file: str = "output.mp3") -> str:
    """Convert text to speech using OpenAI TTS."""
    response = openai.Audio.create(
        model="tts-1-hd",
        voice="nova",
        input=text
    )

    response.stream_to_file(output_file)
    return output_file


def text_to_speech_openai_streaming(text: str):
    """Stream text-to-speech from OpenAI."""
    response = openai.Audio.create(
        model="tts-1",
        voice="nova",
        input=text,
        stream=True
    )
    return response


def text_to_speech_azure(text: str, output_file: str, subscription_key: str, region: str) -> Optional[str]:
    """Convert text to speech using Azure Speech Services."""
    speech_config = speechsdk.SpeechConfig(
        subscription=subscription_key,
        region=region
    )

    audio_config = speechsdk.audio.AudioOutputConfig(filename=output_file)

    synthesizer = speechsdk.SpeechSynthesizer(
        speech_config=speech_config,
        audio_config=audio_config
    )

    result = synthesizer.speak_text(text)

    if result.reason == speechsdk.ResultReason.SynthesizingAudioCompleted:
        return output_file
    else:
        raise Exception(f"Speech synthesis failed: {result.reason}")


def text_to_speech_elevenlabs(text: str, voice_id: str = "21m00Tcm4TlvDq8ikWAM",
                              api_key: str = None) -> str:
    """Convert text to speech using Eleven Labs."""
    url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream"

    headers = {
        "xi-api-key": api_key,
        "Content-Type": "application/json"
    }

    data = {
        "text": text,
        "model_id": "eleven_monolingual_v1",
        "voice_settings": {
            "stability": 0.5,
            "similarity_boost": 0.75
        }
    }

    response = requests.post(url, json=data, headers=headers, stream=True)

    with open("output.mp3", "wb") as f:
        for chunk in response.iter_content(chunk_size=1024):
            f.write(chunk)

    return "output.mp3"

"""
Voice Assistant Module

Complete voice processing pipeline.
"""

import asyncio
from typing import Optional


class VoiceAssistant:
    """Complete voice assistant implementation."""

    def __init__(self, stt_provider: str = "openai", tts_provider: str = "openai"):
        """
        Initialize voice assistant.

        Args:
            stt_provider: Speech-to-text provider
            tts_provider: Text-to-speech provider
        """
        self.stt_provider = stt_provider
        self.tts_provider = tts_provider
        self.conversation_history = []
        self.llm = None  # Will be initialized with LLM

    async def process_voice_input(self, audio_file: str) -> str:
        """Convert voice to text."""
        from .speech_recognition_providers import (
            transcribe_audio_whisper, transcribe_audio_google
        )

        if self.stt_provider == "openai":
            text = transcribe_audio_whisper(audio_file)
        elif self.stt_provider == "google":
            text = transcribe_audio_google(audio_file)
        else:
            raise ValueError(f"Unknown STT provider: {self.stt_provider}")

        return text

    async def generate_response(self, user_input: str) -> str:
        """Generate AI response."""
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        if self.llm:
            response = self.llm(self.conversation_history)
        else:
            response = f"You said: {user_input}"

        self.conversation_history.append({
            "role": "assistant",
            "content": response
        })

        return response

    async def synthesize_response(self, text: str) -> str:
        """Convert text to voice."""
        from .text_to_speech_providers import (
            text_to_speech_openai, text_to_speech_google
        )

        if self.tts_provider == "openai":
            audio_file = text_to_speech_openai(text)
        elif self.tts_provider == "google":
            audio_file = text_to_speech_google(text)
        else:
            raise ValueError(f"Unknown TTS provider: {self.tts_provider}")

        return audio_file

    async def chat(self, audio_input: str) -> str:
        """Complete voice chat pipeline."""
        # Speech to text
        user_text = await self.process_voice_input(audio_input)
        print(f"User said: {user_text}")

        # Generate response
        response_text = await self.generate_response(user_text)

        # Text to speech
        audio_output = await self.synthesize_response(response_text)

        return audio_output

    def get_conversation_history(self):
        """Get conversation history."""
        return self.conversation_history

    def clear_history(self):
        """Clear conversation history."""
        self.conversation_history = []