import gradio as gr
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
import asyncio
import json
import logging
from typing import Dict, List, Optional
import os
from datetime import datetime
import httpx
import websockets
# Configuration - use environment variables for deployment
class Config:
def __init__(self):
self.hf_space_url = os.getenv("HF_SPACE_URL", "https://your-space.hf.space")
self.render_url = os.getenv("RENDER_URL", "https://your-app.onrender.com")
self.default_threshold = float(os.getenv("DEFAULT_THRESHOLD", "0.7"))
self.default_max_speakers = int(os.getenv("DEFAULT_MAX_SPEAKERS", "4"))
self.max_speakers_limit = int(os.getenv("MAX_SPEAKERS_LIMIT", "8"))
config = Config()
logger = logging.getLogger(__name__)
class ConnectionManager:
"""Manage WebSocket connections"""
def __init__(self):
self.active_connections: List[WebSocket] = []
self.conversation_history: List[Dict] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
logger.info(f"Client connected. Total connections: {len(self.active_connections)}")
def disconnect(self, websocket: WebSocket):
if websocket in self.active_connections:
self.active_connections.remove(websocket)
logger.info(f"Client disconnected. Total connections: {len(self.active_connections)}")
async def send_personal_message(self, message: str, websocket: WebSocket):
try:
await websocket.send_text(message)
except Exception as e:
logger.error(f"Error sending message: {e}")
self.disconnect(websocket)
async def broadcast(self, message: str):
"""Send message to all connected clients"""
disconnected = []
for connection in self.active_connections:
try:
await connection.send_text(message)
except Exception as e:
logger.error(f"Error broadcasting to connection: {e}")
disconnected.append(connection)
# Clean up disconnected clients
for conn in disconnected:
self.disconnect(conn)
manager = ConnectionManager()
def create_gradio_app():
"""Create the Gradio interface"""
def get_client_js():
"""Return the client-side JavaScript"""
return f"""
"""
with gr.Blocks(
title="Real-time Speaker Diarization",
theme=gr.themes.Soft(),
css="""
.status-indicator { margin: 10px 0; }
.conversation-display {
background: #f8f9fa;
border: 1px solid #dee2e6;
border-radius: 8px;
padding: 20px;
min-height: 400px;
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
overflow-y: auto;
}
"""
) as demo:
# Inject client-side JavaScript
gr.HTML(get_client_js())
# Header
gr.Markdown("# đ¤ Real-time Speaker Diarization")
gr.Markdown("Advanced speech recognition with automatic speaker identification")
# Status indicator
gr.HTML(f"""
Ready to connect
""")
with gr.Row():
with gr.Column(scale=2):
# Conversation display
gr.HTML(f"""
Click 'Start Listening' to begin real-time transcription...
""")
# Control buttons
with gr.Row():
gr.Button(
"âļī¸ Start Listening",
variant="primary",
size="lg",
elem_id="start-btn"
).click(fn=None, js="startListening()")
gr.Button(
"âšī¸ Stop",
variant="stop",
size="lg",
elem_id="stop-btn"
).click(fn=None, js="stopListening()")
gr.Button(
"đī¸ Clear",
variant="secondary",
size="lg",
elem_id="clear-btn"
).click(fn=None, js="clearConversation()")
with gr.Column(scale=1):
gr.Markdown("## âī¸ Settings")
threshold_slider = gr.Slider(
minimum=0.3,
maximum=0.9,
step=0.05,
value=config.default_threshold,
label="Speaker Change Sensitivity",
info="Lower = more sensitive to speaker changes"
)
max_speakers_slider = gr.Slider(
minimum=2,
maximum=config.max_speakers_limit,
step=1,
value=config.default_max_speakers,
label="Maximum Speakers"
)
# Instructions
gr.Markdown("""
## đ How to Use
1. **Start Listening** - Grant microphone access
2. **Speak** - System transcribes and identifies speakers
3. **Stop** when finished
4. **Clear** to reset conversation
## đ¨ Speaker Colors
- đ´ Speaker 1 - đĸ Speaker 2 - đĩ Speaker 3 - đĄ Speaker 4
- â Speaker 5 - đŖ Speaker 6 - đ¤ Speaker 7 - đ Speaker 8
""")
return demo
def create_fastapi_app():
"""Create the FastAPI backend"""
app = FastAPI(title="Speaker Diarization API")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
# Receive audio data
data = await websocket.receive_bytes()
# Process audio data here
# This is where you'd integrate your actual speaker diarization model
result = await process_audio_chunk(data)
# Send result back to client
await manager.send_personal_message(
json.dumps(result),
websocket
)
except WebSocketDisconnect:
manager.disconnect(websocket)
except Exception as e:
logger.error(f"WebSocket error: {e}")
manager.disconnect(websocket)
@app.post("/clear")
async def clear_conversation():
"""Clear the conversation history"""
manager.conversation_history.clear()
await manager.broadcast(json.dumps({
"type": "conversation_cleared"
}))
return {"status": "cleared"}
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"timestamp": datetime.now().isoformat(),
"active_connections": len(manager.active_connections)
}
@app.get("/status")
async def get_status():
"""Get system status"""
return {
"status": "online",
"connections": len(manager.active_connections),
"conversation_length": len(manager.conversation_history)
}
return app
async def process_audio_chunk(audio_data: bytes) -> dict:
"""
Process audio chunk and return diarization result by sending it to the Speaker Diarization backend
"""
try:
# Convert WebM audio to appropriate format if needed
# This step may require additional processing depending on your backend requirements
# Connect to the Speaker Diarization backend via WebSocket
websocket_url = f"wss://{config.hf_space_url.replace('https://', '').replace('http://', '')}/ws_inference"
logger.info(f"Connecting to diarization backend at {websocket_url}")
async with websockets.connect(websocket_url) as websocket:
# Send audio data
await websocket.send(audio_data)
# Receive the response (may need to handle multiple messages)
response = await websocket.recv()
# Parse the response
try:
result = json.loads(response)
# Add to conversation history if it's a transcription
if result.get("type") == "transcription" or result.get("type") == "conversation_update":
if "conversation_html" in result:
manager.conversation_history.append({
"timestamp": datetime.now().isoformat(),
"html": result["conversation_html"]
})
return result
except json.JSONDecodeError:
logger.error(f"Invalid JSON response: {response}")
return {
"type": "error",
"error": "Invalid response from backend",
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.exception(f"Error processing audio chunk: {e}")
return {
"type": "error",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
# Create both apps
fastapi_app = create_fastapi_app()
gradio_app = create_gradio_app()
# Mount Gradio app to FastAPI
fastapi_app.mount("/", gradio_app.app)
if __name__ == "__main__":
import uvicorn
uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)