#!/usr/bin/env python3
"""
LiveTalker Voice Chat Server - GPU Ready
Direct voice chat interface with WebSocket support
"""
import asyncio
import logging
import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import uvicorn
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="LiveTalker Voice Chat", version="1.0.0")
# Voice Chat HTML Interface
VOICE_CHAT_HTML = """
LiveTalker Voice Chat
đī¸ LiveTalker
AI Voice Assistant
Disconnected
đ GPU Mode Ready: Full voice-to-voice conversation with ultra-low latency!
Welcome to LiveTalker!
Click the microphone to speak, or type your message below.
Voice input and output âĸ Real-time conversation
"""
@app.get("/", response_class=HTMLResponse)
async def voice_chat():
"""Main voice chat interface"""
return VOICE_CHAT_HTML
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""WebSocket endpoint for real-time voice chat"""
await websocket.accept()
logger.info('Voice chat WebSocket connected')
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
logger.info(f'Received message: {message}')
if message.get('type') == 'text':
user_text = message.get('text', '')
# Enhanced response with personality
if "hello" in user_text.lower() or "hi" in user_text.lower():
response = f"Hello! Great to meet you. You said: '{user_text}'. I'm LiveTalker, your AI voice assistant. How can I help you today?"
elif "how are you" in user_text.lower():
response = "I'm doing excellent! I'm running in GPU mode for ultra-fast voice responses. What would you like to chat about?"
elif "what can you do" in user_text.lower():
response = "I can have voice conversations, answer questions, help with tasks, and chat about anything you'd like. Try speaking to me using the microphone button!"
else:
response = f'You said: "{user_text}". I\'m LiveTalker, ready to help with anything you need. I can hear you clearly and respond with both text and voice!'
await websocket.send_text(json.dumps({
'type': 'text',
'text': response,
'speaker': 'assistant'
}))
except WebSocketDisconnect:
logger.info('Voice chat WebSocket disconnected')
except Exception as e:
logger.error(f'WebSocket error: {e}')
@app.get("/health")
async def health():
"""Health check endpoint"""
return {
'status': 'ok',
'mode': 'gpu_voice_ready',
'features': ['voice_input', 'voice_output', 'real_time_chat']
}
@app.get("/stats")
async def stats():
"""System statistics"""
import torch
gpu_available = torch.cuda.is_available()
return {
'mode': 'gpu_ready' if gpu_available else 'cpu_fallback',
'gpu_available': gpu_available,
'voice_features': {
'speech_recognition': True,
'text_to_speech': True,
'real_time_chat': True
}
}
def main():
"""Start the voice chat server"""
logger.info("đī¸ Starting LiveTalker Voice Chat Server")
uvicorn.run(app, host="0.0.0.0", port=8001, log_level="info")
if __name__ == "__main__":
main()