#!/usr/bin/env python3 """ Simple LiveTalker Voice Chat Server Works with basic dependencies and provides HTTPS + Tailscale integration """ import ssl import uvicorn import asyncio import json import logging import time import base64 from typing import Dict, Any from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="LiveTalker Simple Voice Chat") # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) active_sessions: Dict[str, Dict] = {} def simple_vad(audio_data, threshold: float = 0.01) -> tuple: """Simple energy-based voice activity detection""" if not audio_data: return False, 0.0 # Calculate simple energy level energy = sum(abs(x) for x in audio_data) / len(audio_data) if audio_data else 0 is_speech = energy > threshold confidence = min(energy / threshold, 1.0) if threshold > 0 else 0.0 return is_speech, confidence @app.get("/") async def serve_main_page(): """Serve main test page with voice chat widget""" html_content = """ LiveTalker Voice Chat - HTTPS

🎙️ LiveTalker Voice Chat

✅ HTTPS Enabled - Voice chat ready!

🎵 Voice Activity

Click "Connect" to start
Ready to start voice conversation...
""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): return { "status": "healthy", "timestamp": time.time(), "https_enabled": True, "features": ["voice_chat", "https", "tailscale_ready"] } @app.websocket("/voice") async def voice_websocket(websocket: WebSocket): """WebSocket for voice processing""" await websocket.accept() session_id = f"session_{int(time.time() * 1000)}" session = { "id": session_id, "websocket": websocket, "audio_buffer": [], "is_listening": False } active_sessions[session_id] = session try: await websocket.send_json({ "type": "status", "message": "Voice chat connected! Enable microphone to start." }) async for message in websocket.iter_json(): if message.get("type") == "start_listening": session["is_listening"] = True await websocket.send_json({ "type": "status", "message": "🎤 Listening for voice..." }) elif message.get("type") == "audio" and session["is_listening"]: try: # Decode audio data audio_data = base64.b64decode(message["data"]) # Convert to list for simple processing audio_values = list(audio_data) if audio_values: # Simple voice activity detection is_speech, confidence = simple_vad(audio_values) await websocket.send_json({ "type": "vad_result", "is_speech": is_speech, "confidence": confidence }) if is_speech and confidence > 0.3: session["audio_buffer"].extend(audio_values) # Simulate speech recognition after 2 seconds of audio if len(session["audio_buffer"]) > 32000: # ~2 seconds at 16kHz duration = len(session["audio_buffer"]) / 16000 # Simple mock transcription await websocket.send_json({ "type": "transcription", "text": f"Voice detected for {duration:.1f} seconds" }) # Simple mock response await websocket.send_json({ "type": "response", "text": f"I heard you speaking for {duration:.1f} seconds! The voice chat is working perfectly." }) session["audio_buffer"] = [] except Exception as e: await websocket.send_json({ "type": "status", "message": f"Audio processing error: {str(e)}" }) elif message.get("type") == "stop_listening": session["is_listening"] = False await websocket.send_json({ "type": "status", "message": "Stopped listening" }) except WebSocketDisconnect: pass finally: if session_id in active_sessions: del active_sessions[session_id] def main(): print("🎙️ Starting LiveTalker Simple Voice Chat...") # Check for HTTPS certificates cert_file = "livetalker.crt" key_file = "livetalker.key" if not (Path(cert_file).exists() and Path(key_file).exists()): print(f"❌ HTTPS certificates not found ({cert_file}, {key_file})") print("Running in HTTP mode (microphone may not work in browsers)") uvicorn.run(app, host="0.0.0.0", port=8000) else: print("✅ HTTPS certificates found") print("🌟 Starting with HTTPS support") print("") print("📍 Access URLs:") print(" Local HTTPS: https://localhost:8000") print(" Setup Tailscale: ./scripts/setup_tailscale.sh") print("") uvicorn.run( app, host="0.0.0.0", port=8000, ssl_certfile=cert_file, ssl_keyfile=key_file ) if __name__ == "__main__": main()