"""
RealtimeSTT FastAPI server — deploy to Hugging Face Spaces.
POST /transcribe — upload audio, returns {"text": "..."}
GET /health — health check
GET / — browser recording web UI
"""
import io
import logging
import os
import tempfile
import threading
from pathlib import Path
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse, JSONResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("realtimestt")
app = FastAPI(title="RealtimeSTT HF Space")
INDEX_HTML = r"""
RealtimeSTT
"""
# --- Model management ---
_model = None
_model_lock = threading.Lock()
def get_model():
global _model
if _model is None:
with _model_lock:
if _model is None:
from faster_whisper import WhisperModel
model_size = os.getenv("STT_MODEL", "tiny")
device = os.getenv("STT_DEVICE", "cpu")
compute_type = os.getenv("STT_COMPUTE", "int8" if device == "cpu" else "float16")
logger.info(f"Loading faster-whisper {model_size} ({device}/{compute_type})...")
_model = WhisperModel(model_size, device=device, compute_type=compute_type)
logger.info("Model loaded.")
return _model
# --- Routes ---
@app.get("/")
async def index():
return HTMLResponse(INDEX_HTML)
@app.get("/health")
async def health():
return JSONResponse({"ok": True, "ready": _model is not None})
@app.post("/transcribe")
async def transcribe(file: UploadFile = File(...)):
import soundfile as sf
import numpy as np
model = get_model()
contents = await file.read()
try:
# Try to decode as WAV or other soundfile-supported format
try:
data, sr = sf.read(io.BytesIO(contents))
except Exception:
# WebM/Opus — convert via ffmpeg pipe
import subprocess
proc = subprocess.run(
["ffmpeg", "-y", "-i", "pipe:0", "-ar", "16000", "-ac", "1",
"-sample_fmt", "s16", "-f", "wav", "pipe:1"],
input=contents, capture_output=True, timeout=30, check=True
)
data, sr = sf.read(io.BytesIO(proc.stdout))
if len(data) == 0:
return JSONResponse({"text": ""})
# Ensure mono, float32, normalized to [-1, 1]
if data.ndim > 1:
data = data.mean(axis=1)
data = data.astype(np.float32)
max_val = data.max()
if max_val > 1.0:
data /= 32768.0
# Resample to 16kHz if needed (Whisper expects 16kHz)
if sr != 16000:
from scipy.signal import resample_poly
data = resample_poly(data, 16000, sr)
sr = 16000
max_val = data.max()
if max_val > 1.0:
data /= 32768.0
logger.info(f"Audio: {len(data)} samples, sr={sr}, "
f"rms={np.sqrt(np.mean(data**2)):.5f}, "
f"max={data.max():.5f}, min={data.min():.5f}")
transcribe_kwargs = {"beam_size": 5, "no_speech_threshold": 0.99, "compression_ratio_threshold": 2.0}
segments, info = model.transcribe(data, **transcribe_kwargs)
seg_texts = []
for seg in segments:
seg_texts.append(seg.text)
logger.info(f"Segment [{seg.start:.2f}-{seg.end:.2f}]: {seg.text.strip()} "
f"(no_speech={seg.no_speech_prob:.2f})")
text = " ".join(seg_texts)
return JSONResponse({"text": text.strip()})
except Exception as e:
logger.error(f"Transcription failed: {e}")
return JSONResponse({"text": "", "error": str(e)}, status_code=500)
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", "7860"))
uvicorn.run(app, host="0.0.0.0", port=port)