"""Web server for displaying transcriptions in a browser (for OBS browser source)."""
import asyncio
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
from typing import List, Optional
import json
from datetime import datetime
class TranscriptionWebServer:
"""Web server for displaying transcriptions."""
def __init__(self, host: str = "127.0.0.1", port: int = 8080, show_timestamps: bool = True, fade_after_seconds: int = 10):
"""
Initialize web server.
Args:
host: Server host address
port: Server port
show_timestamps: Whether to show timestamps in transcriptions
fade_after_seconds: Time in seconds before transcriptions fade out (0 = never fade)
"""
self.host = host
self.port = port
self.show_timestamps = show_timestamps
self.fade_after_seconds = fade_after_seconds
self.app = FastAPI()
self.active_connections: List[WebSocket] = []
self.transcriptions = [] # Store recent transcriptions
# Setup routes
self._setup_routes()
def _setup_routes(self):
"""Setup FastAPI routes."""
@self.app.get("/", response_class=HTMLResponse)
async def get_display():
"""Serve the transcription display page."""
return self._get_html()
@self.app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""WebSocket endpoint for real-time updates."""
await websocket.accept()
self.active_connections.append(websocket)
try:
# Send recent transcriptions
for trans in self.transcriptions[-20:]: # Last 20
await websocket.send_json(trans)
# Keep connection alive
while True:
# Wait for ping/pong to keep connection alive
await websocket.receive_text()
except:
self.active_connections.remove(websocket)
def _get_html(self) -> str:
"""Generate HTML for transcription display."""
return f"""
Transcription Display
"""
async def broadcast_transcription(self, text: str, user_name: str = "", timestamp: Optional[datetime] = None):
"""
Broadcast a transcription to all connected clients.
Args:
text: Transcription text
user_name: User/speaker name
timestamp: Timestamp of transcription
"""
if timestamp is None:
timestamp = datetime.now()
trans_data = {
"text": text,
"user_name": user_name,
}
# Only include timestamp if enabled
if self.show_timestamps:
trans_data["timestamp"] = timestamp.strftime("%H:%M:%S")
# Store transcription
self.transcriptions.append(trans_data)
if len(self.transcriptions) > 100:
self.transcriptions.pop(0)
# Broadcast to all connected clients
disconnected = []
for connection in self.active_connections:
try:
await connection.send_json(trans_data)
except:
disconnected.append(connection)
# Remove disconnected clients
for conn in disconnected:
self.active_connections.remove(conn)
async def start(self):
"""Start the web server."""
import uvicorn
config = uvicorn.Config(
self.app,
host=self.host,
port=self.port,
log_level="warning"
)
server = uvicorn.Server(config)
await server.serve()