#!/usr/bin/env python3 """ Minimal LiveTalker Test Server Demonstrates core functionality without complex dependencies """ import asyncio import json import logging import time from typing import Dict, Any import sys # Try to use system packages first try: from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn FASTAPI_AVAILABLE = True except ImportError: print("FastAPI not available, installing basic packages...") import subprocess subprocess.run([sys.executable, "-m", "pip", "install", "fastapi", "uvicorn[standard]", "python-multipart"]) from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn FASTAPI_AVAILABLE = True logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="LiveTalker Test Server") # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Store active connections active_connections: Dict[str, WebSocket] = {} @app.get("/") async def root(): """Root page with test interface""" html_content = """ LiveTalker Test Interface

๐ŸŽ™๏ธ LiveTalker Test Interface

System Status:

Connecting...
๐ŸŽฏ

Voice Activity Detection

Real-time speech detection with Silero VAD

๐Ÿ”„

Turn-Taking Detection

Prosodic analysis for natural conversation flow

โšก

Interruption Handling

Smart interruption with context preservation

๐Ÿง 

Resume Logic

Continue from last complete thought

Activity Log:

Waiting for connection...

Available Endpoints:

""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "timestamp": time.time(), "version": "LiveTalker Test v1.0", "features": [ "Voice Activity Detection", "Turn-Taking Detection", "Interruption Handling", "Resume from Last Thought", "WebSocket Communication", "Personality Profiles" ] } @app.get("/stats") async def get_stats(): """System statistics""" return { "active_connections": len(active_connections), "server_uptime": time.time(), "components": { "websocket_gateway": "ready", "vad_system": "simulated", "turn_detector": "simulated", "speech_manager": "simulated", "audio_processor": "simulated" }, "test_mode": True, "performance": { "expected_ttfc_ms": "500-650 (with CSM)", "expected_rtf": "2.0-3.5x (with optimizations)", "current_mode": "demonstration" } } @app.get("/widget.js") async def get_widget(): """Return the JavaScript widget""" try: return FileResponse("frontend/widget.js", media_type="application/javascript") except FileNotFoundError: # Return a minimal widget for testing js_content = ''' // LiveTalker Test Widget console.log("LiveTalker widget loaded in test mode"); window.LiveTalkerWidget = class { constructor(config = {}) { console.log("LiveTalker widget initialized:", config); this.createTestButton(); } createTestButton() { const button = document.createElement("button"); button.textContent = "๐ŸŽ™๏ธ LiveTalker (Test)"; button.style.cssText = "position:fixed;bottom:20px;right:20px;z-index:9999;padding:15px;background:#667eea;color:white;border:none;border-radius:50px;cursor:pointer;"; button.onclick = () => window.open("/", "_blank"); document.body.appendChild(button); } }; if (document.currentScript && document.currentScript.hasAttribute('data-agent-id')) { new LiveTalkerWidget({ agentId: document.currentScript.getAttribute('data-agent-id') }); } ''' return HTMLResponse(content=js_content, media_type="application/javascript") @app.get("/demo") async def demo_page(): """Demo page showing widget integration""" demo_html = """ LiveTalker Widget Demo

LiveTalker Widget Integration Demo

This page demonstrates how the LiveTalker widget appears on a website.

Look for the floating voice button in the bottom-right corner!

Integration Code:

<script src="/widget.js" data-agent-id="Demo-Assistant"></script>

Features Demonstrated:

""" return HTMLResponse(content=demo_html) @app.websocket("/media-stream") async def websocket_endpoint(websocket: WebSocket): """WebSocket endpoint for real-time communication""" await websocket.accept() session_id = f"session_{int(time.time() * 1000)}" active_connections[session_id] = websocket logger.info(f"New WebSocket connection: {session_id}") try: # Send welcome message await websocket.send_json({ "type": "config", "session_id": session_id, "message": "LiveTalker Test Server - WebSocket Connected", "features": { "vad": "simulated", "turn_taking": "simulated", "interruption_handling": "simulated", "personality_profiles": "available" } }) async for message in websocket.iter_json(): await handle_websocket_message(websocket, session_id, message) except WebSocketDisconnect: logger.info(f"WebSocket disconnected: {session_id}") except Exception as e: logger.error(f"WebSocket error: {e}") finally: if session_id in active_connections: del active_connections[session_id] async def handle_websocket_message(websocket: WebSocket, session_id: str, message: Dict[str, Any]): """Handle incoming WebSocket messages""" msg_type = message.get("type") logger.info(f"Received message type: {msg_type}") if msg_type == "start_conversation": await websocket.send_json({ "type": "conversation_started", "session_id": session_id, "personality": message.get("config", {}).get("personality", "default"), "message": "Conversation started! (Test mode - simulated responses)" }) elif msg_type == "audio": # Simulate VAD processing await websocket.send_json({ "type": "vad_result", "is_speech": True, "confidence": 0.85, "message": "Audio received and processed (simulated)" }) # Simulate turn-taking detection await asyncio.sleep(0.5) await websocket.send_json({ "type": "turn_detected", "state": "user_speaking", "message": "Turn-taking: User is speaking" }) # Simulate response after 2 seconds await asyncio.sleep(2) await websocket.send_json({ "type": "response_complete", "message": "This is a simulated voice response. In the full implementation, this would be streaming audio from CSM/Sesame.", "segments": 3, "performance": { "ttfc_ms": 650, "rtf": 2.5, "note": "Simulated performance metrics" } }) elif msg_type == "interrupt": await websocket.send_json({ "type": "interrupted", "message": "Assistant interrupted - switching to listening mode", "context_preserved": True }) elif msg_type == "vad_simulation": await websocket.send_json({ "type": "vad_update", "state": message.get("state"), "confidence": message.get("confidence", 0.8), "timestamp": time.time() }) else: await websocket.send_json({ "type": "echo", "original_message": message, "timestamp": time.time() }) if __name__ == "__main__": print("๐Ÿš€ Starting LiveTalker Test Server...") print("๐Ÿ“ Access points:") print(" โ€ข Main Interface: http://localhost:8000") print(" โ€ข Health Check: http://localhost:8000/health") print(" โ€ข System Stats: http://localhost:8000/stats") print(" โ€ข Widget Demo: http://localhost:8000/demo") print(" โ€ข WebSocket: ws://localhost:8000/media-stream") print("") print("๐ŸŽฏ This test server demonstrates:") print(" โ€ข WebSocket real-time communication") print(" โ€ข Voice Activity Detection (simulated)") print(" โ€ข Turn-taking detection (simulated)") print(" โ€ข Interruption handling (simulated)") print(" โ€ข Widget integration") print("") uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info" )