#!/usr/bin/env python3 """ LiveTalker HTTPS Voice Server With proper SSL/TLS support for microphone access """ import asyncio import json import logging import time import base64 import numpy as np import ssl import os from typing import Dict, Any from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="LiveTalker HTTPS Voice Server") # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) active_connections: Dict[str, Dict] = {} def simple_vad(audio_data: np.ndarray, threshold: float = 0.01) -> tuple: """Simple energy-based voice activity detection""" if len(audio_data) == 0: return False, 0.0 # Calculate RMS energy rms = np.sqrt(np.mean(audio_data ** 2)) # Simple threshold-based detection is_speech = rms > threshold confidence = min(rms / threshold, 1.0) if threshold > 0 else 0.0 return is_speech, confidence def create_self_signed_cert(): """Create a self-signed certificate for HTTPS""" try: from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization import datetime # Generate private key private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, ) # Create certificate subject = issuer = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "CA"), x509.NameAttribute(NameOID.LOCALITY_NAME, "San Francisco"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LiveTalker"), x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), ]) cert = x509.CertificateBuilder().subject_name( subject ).issuer_name( issuer ).public_key( private_key.public_key() ).serial_number( x509.random_serial_number() ).not_valid_before( datetime.datetime.utcnow() ).not_valid_after( datetime.datetime.utcnow() + datetime.timedelta(days=365) ).add_extension( x509.SubjectAlternativeName([ x509.DNSName("localhost"), x509.DNSName("*.localhost"), x509.IPAddress("127.0.0.1"), x509.IPAddress("100.118.75.128"), ]), critical=False, ).sign(private_key, hashes.SHA256()) # Write certificate and key cert_path = "server.crt" key_path = "server.key" with open(cert_path, "wb") as f: f.write(cert.public_bytes(serialization.Encoding.PEM)) with open(key_path, "wb") as f: f.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() )) return cert_path, key_path except ImportError: logger.warning("cryptography package not available, using HTTP only") return None, None @app.get("/") async def root(): """Main interface with browser compatibility checks""" html_content = """ LiveTalker - Voice Assistant

🎙️ LiveTalker Voice Assistant

🔗 Connection Status

Ready to connect

🎤 Microphone Status

Not checked

🎵 Voice Activity Detection

Click microphone to start
LiveTalker ready! Enable microphone to start voice conversation...

📊 Activity Log

Initializing...
""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): """Health check with HTTPS info""" return { "status": "healthy", "timestamp": time.time(), "protocol": "HTTPS" if os.path.exists("server.crt") else "HTTP", "features": { "microphone_input": True, "browser_compatibility": True, "https_support": os.path.exists("server.crt"), "simple_vad": True } } @app.websocket("/media-stream") async def websocket_endpoint(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": [], "conversation": [], "is_listening": False } active_connections[session_id] = session try: await websocket.send_json({ "type": "config", "session_id": session_id, "message": "Voice processing ready with browser compatibility" }) async for message in websocket.iter_json(): if message.get("type") == "start_conversation": session["is_listening"] = True await websocket.send_json({ "type": "conversation_started", "message": "Listening for your voice!" }) elif message.get("type") == "audio" and session["is_listening"]: try: audio_data = base64.b64decode(message["data"]) audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0 if len(audio_np) > 0: is_speech, confidence = simple_vad(audio_np) await websocket.send_json({ "type": "vad_result", "is_speech": is_speech, "confidence": confidence, "timestamp": time.time() }) if is_speech: session["audio_buffer"].extend(audio_np.tolist()) if len(session["audio_buffer"]) > 16000: # ~1 second duration = len(session["audio_buffer"]) / 16000 text = f"Voice detected: {duration:.1f}s speech segment" await websocket.send_json({ "type": "speech_to_text", "text": text, "confidence": 0.8 }) response = f"I heard {duration:.1f} seconds of speech! Your voice is being processed successfully." await websocket.send_json({ "type": "ai_response", "text": response }) session["audio_buffer"] = [] except Exception as e: await websocket.send_json({ "type": "error", "error": f"Audio processing error: {str(e)}" }) elif message.get("type") == "stop_listening": session["is_listening"] = False except WebSocketDisconnect: pass except Exception as e: logger.error(f"WebSocket error: {e}") finally: if session_id in active_connections: del active_connections[session_id] if __name__ == "__main__": print("🎙️ Starting LiveTalker HTTPS Voice Server...") # Try to create SSL certificate cert_path, key_path = create_self_signed_cert() ssl_config = None if cert_path and key_path: ssl_config = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_config.load_cert_chain(cert_path, key_path) print("✅ Created self-signed certificate for HTTPS") print("📍 HTTPS URL: https://localhost:8000") print("📍 HTTPS URL: https://100.118.75.128:8000") else: print("⚠️ Running HTTP only - microphone may not work on some browsers") print("📍 HTTP URL: http://localhost:8000") print("") print("🔧 Features:") print(" ✅ Browser compatibility checking") print(" ✅ Microphone permission handling") print(" ✅ HTTPS support for modern browsers") print(" ✅ Fallback HTTP for localhost") print(" ✅ Real voice activity detection") print("") if ssl_config: uvicorn.run( app, host="0.0.0.0", port=8000, ssl_context=ssl_config, log_level="info" ) else: uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info" )