Just confirming that this works! I’ve successfully deployed the Parakeet ASR model on the DGX Spark using Portainer.
Here is the Portainer stack YAML I am using. One quick heads-up: depending on your specific Docker environment, you may need to change or remove the networks block (dgx_net external network) at the bottom to match your own setup - you can do a search and replace to match your own network.
Hope this helps anyone else trying to get this running!
0. Get your Api Key here
Search for the two instances of PUT-IN-YOUR-NVIDIA-API-KEY-HERE and use your generated API key.
For the main part this is AI Generated.
Create a Directory for the Build
mkdir -p ~/Docker/parakeet-spark/app
cd ~/Docker/parakeet-spark
Then create each file:
1. Dockerfile
FROM nvcr.io/nvidia/pytorch:25.11-py3
ENV DEBIAN_FRONTEND=noninteractive
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
RUN apt-get update && apt-get install -y --no-install-recommends \
ffmpeg libsndfile1 libsndfile1-dev sox \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /opt/parakeet
COPY requirements.txt .
RUN pip install --break-system-packages --no-cache-dir -r requirements.txt
RUN python3 -c "\
import nemo.collections.asr as nemo_asr; \
print('Downloading parakeet-tdt-0.6b-v3...'); \
model = nemo_asr.models.ASRModel.from_pretrained('nvidia/parakeet-tdt-0.6b-v3'); \
print('Model downloaded successfully'); \
del model"
COPY app/ /opt/parakeet/app/
HEALTHCHECK --interval=30s --timeout=10s --start-period=120s --retries=3 \
CMD python3 -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
EXPOSE 8000
CMD ["python3", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]
2. docker-compose.yml or Portainer Stack
# =============================================================================
# PORTAINER STACK — Multilingual Voice Agent
# DGX Spark (ARM64 / CUDA 13 / Blackwell GB10)
#
# IMPORTANT: Before deploying this stack in Portainer, you must first
# build the ASR image on your Spark by running:
#
# bash ~/parakeet-spark/build-and-deploy.sh
#
# That script builds the image: parakeet-tdt-v3-spark:latest
# This stack then references that local image.
#
# In Portainer:
# 1. Stacks → Add Stack → name: "voice-agent"
# 2. Paste this YAML
# 3. Add environment variable: NGC_API_KEY = PUT-IN-YOUR-NVIDIA-API-KEY-HERE
# 4. Deploy
# =============================================================================
version: '3.8'
services:
# ─── ASR: Parakeet TDT 0.6b v3 ────────────────────────────────────────────
# Custom NeMo-based container with OpenAI-compatible API
# 25 European languages with auto-detection
# Endpoint: http://<spark-ip>:8010/v1/audio/transcriptions
parakeet-asr:
image: parakeet-tdt-v3-spark:latest
container_name: parakeet-asr
restart: unless-stopped
environment:
- NGC_API_KEY=PUT-IN-YOUR-NVIDIA-API-KEY-HERE
- PARAKEET_MODEL=nvidia/parakeet-tdt-0.6b-v3
- MAX_SEGMENT_SECONDS=1200
- MAX_UPLOAD_MB=200
- HF_HOME=/cache/huggingface
- TRANSFORMERS_CACHE=/cache/huggingface
- CUDA_VISIBLE_DEVICES=0
- NCCL_P2P_DISABLE=1
ports:
- "8010:8000"
volumes:
- parakeet-model-cache:/cache/huggingface
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: all
capabilities: [gpu]
shm_size: 16gb
healthcheck:
test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 30s
timeout: 10s
start_period: 180s
retries: 3
networks:
- dgx_net
volumes:
parakeet-model-cache:
nim-tts-cache:
# nim-asr-cache: # Uncomment if using RNNT NIM
networks:
dgx_net:
external: true
```
`
3. requirements.txt
# Requirements for Parakeet TDT v3 on DGX Spark (ARM64)
#
# NOTE: torch, torchaudio, and CUDA libs are already in the NGC base image.
# Do NOT install them here or you'll break the ARM64 GPU support.
# =============================================================================
# ── NeMo ASR toolkit ─────────────────────────────────────────────────────────
# The [asr] extra pulls in all speech recognition dependencies
nemo_toolkit[asr]>=2.1.0
# ── Pin lhotse to avoid DynamicCutSampler TypeError ──────────────────────────
# Several DGX Spark users hit this bug with newer lhotse versions.
# Pin to a known-good version that works with NeMo 2.x on ARM64.
lhotse>=1.24.0,<1.30.0
# ── Audio processing ─────────────────────────────────────────────────────────
librosa>=0.10.0
soundfile>=0.12.0
pydub>=0.25.1
# ── FastAPI server ───────────────────────────────────────────────────────────
fastapi>=0.115.0
uvicorn[standard]>=0.30.0
python-multipart>=0.0.9
# ── Utilities ────────────────────────────────────────────────────────────────
pyyaml>=6.0
aiofiles>=24.1.0```
4. build-and-deploy.sh
# =============================================================================
# build-and-deploy.sh
#
# Complete setup script for Parakeet TDT 0.6b v3 + Magpie TTS
# on NVIDIA DGX Spark (ARM64, CUDA 13, Blackwell GB10)
#
# Run this script on your DGX Spark.
# =============================================================================
set -euo pipefail
# ── Colors for output ────────────────────────────────────────────────────────
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
CYAN='\033[0;36m'
NC='\033[0m' # No Color
info() { echo -e "${CYAN}[INFO]${NC} $*"; }
warn() { echo -e "${YELLOW}[WARN]${NC} $*"; }
ok() { echo -e "${GREEN}[OK]${NC} $*"; }
fail() { echo -e "${RED}[FAIL]${NC} $*"; exit 1; }
# =============================================================================
# STEP 0: Verify we're on DGX Spark
# =============================================================================
echo ""
echo "============================================================"
echo " Parakeet TDT v3 + Magpie TTS — DGX Spark Deployment"
echo "============================================================"
echo ""
info "Checking system..."
# Check architecture
ARCH=$(uname -m)
if [[ "$ARCH" != "aarch64" ]]; then
warn "Expected aarch64, got $ARCH. This script is designed for DGX Spark ARM64."
read -p "Continue anyway? [y/N] " -n 1 -r; echo
[[ $REPLY =~ ^[Yy]$ ]] || exit 1
fi
# Check NVIDIA GPU
if ! command -v nvidia-smi &>/dev/null; then
fail "nvidia-smi not found. Is the NVIDIA driver installed?"
fi
info "GPU detected:"
nvidia-smi --query-gpu=name,memory.total,driver_version --format=csv,noheader
echo ""
# Check Docker
if ! command -v docker &>/dev/null; then
fail "Docker not found. Install Docker first."
fi
# Check NVIDIA Container Toolkit
if ! docker info 2>/dev/null | grep -q "nvidia"; then
warn "NVIDIA Container Toolkit may not be configured."
warn "Ensure 'nvidia' runtime is available in Docker."
fi
# Check CUDA version
if command -v nvcc &>/dev/null; then
info "CUDA version: $(nvcc --version | grep release | awk '{print $6}')"
fi
ok "System checks passed"
# =============================================================================
# STEP 1: Set up project directory
# =============================================================================
echo ""
info "Setting up project directory..."n
PROJECT_DIR="~/Docker/parakeet-spark"
# If script is run from a different location, copy files
if [[ ! -f "$PROJECT_DIR/Dockerfile" ]]; then
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if [[ -f "$SCRIPT_DIR/Dockerfile" ]]; then
info "Copying project files to $PROJECT_DIR..."
mkdir -p "$PROJECT_DIR"
cp -r "$SCRIPT_DIR"/* "$PROJECT_DIR/"
else
fail "Cannot find Dockerfile. Run this script from the project directory."
fi
fi
cd "$PROJECT_DIR"
ok "Project directory: $PROJECT_DIR"
# =============================================================================
# STEP 2: Create the Docker network (if not exists)
# =============================================================================
echo ""
info "Ensuring Docker network 'dgx_net' exists..."
if ! docker network inspect dgx_net &>/dev/null; then
docker network create dgx_net
ok "Created network 'dgx_net'"
else
ok "Network 'dgx_net' already exists"
fi
# =============================================================================
# STEP 3: Log in to NGC (needed for base image + TTS NIM)
# =============================================================================
echo ""
info "Logging in to NVIDIA NGC registry..."
if [[ -z "${NGC_API_KEY:-}" ]]; then
echo ""
echo " You need an NGC API key to pull NVIDIA containers."
echo " Get one at: https://org.ngc.nvidia.com/setup/api-key"
echo ""
read -sp " Enter your NGC API key: " NGC_API_KEY
echo ""
fi
echo "$NGC_API_KEY" | docker login nvcr.io -u '$oauthtoken' --password-stdin
ok "Logged in to nvcr.io"
# Save for docker-compose
export NGC_API_KEY
# =============================================================================
# STEP 4: Pull the base image
# =============================================================================
echo ""
info "Pulling NVIDIA PyTorch base image (ARM64)..."
info "This is ~15 GB and may take a while on first pull."
echo ""
docker pull nvcr.io/nvidia/pytorch:25.11-py3
ok "Base image pulled"
# =============================================================================
# STEP 5: Build the Parakeet ASR container
# =============================================================================
echo ""
info "Building Parakeet TDT v3 container..."
info "This will:"
info " 1. Install NeMo toolkit + dependencies"
info " 2. Download the Parakeet TDT 0.6b v3 model (~1.2 GB)"
info " 3. Set up the FastAPI server"
info ""
info "Expected build time: 15-30 minutes (first time)"
echo ""
docker build \
--tag parakeet-tdt-v3-spark:latest \
--progress=plain \
--shm-size=8g \
-f Dockerfile \
.
ok "Parakeet ASR container built: parakeet-tdt-v3-spark:latest"
# =============================================================================
# STEP 6: Quick smoke test of ASR container
# =============================================================================
echo ""
info "Running smoke test..."
# Start container briefly to verify GPU access and model loading
SMOKE_ID=$(docker run -d --rm \
--gpus all \
--shm-size=8g \
--name parakeet-smoke-test \
-e CUDA_VISIBLE_DEVICES=0 \
parakeet-tdt-v3-spark:latest \
python3 -c "
import torch
print(f'CUDA available: {torch.cuda.is_available()}')
if torch.cuda.is_available():
print(f'GPU: {torch.cuda.get_device_name(0)}')
print(f'Memory: {torch.cuda.get_device_properties(0).total_mem / 1e9:.1f} GB')
print('Smoke test passed!')
")
# Wait for it to finish
docker wait "$SMOKE_ID" 2>/dev/null || true
# Show output
docker logs parakeet-smoke-test 2>/dev/null || true
ok "Smoke test complete"
# =============================================================================
# STEP 7: Pull the Magpie TTS NIM (optional)
# =============================================================================
# echo ""
# read -p "Pull Magpie TTS Multilingual NIM as well? [Y/n] " -n 1 -r; echo
# if [[ ! $REPLY =~ ^[Nn]$ ]]; then
# info "Pulling Magpie TTS NIM (ARM64)..."
# docker pull nvcr.io/nim/nvidia/magpie-tts-multilingual:latest
# ok "Magpie TTS NIM pulled"
#else
# warn "Skipping Magpie TTS. You can pull it later with:"
# warn " docker pull nvcr.io/nim/nvidia/magpie-tts-multilingual:latest"
#fi
# =============================================================================
# STEP 8: Create .env file
# =============================================================================
echo ""
info "Creating .env file..."
if [[ ! -f "$PROJECT_DIR/.env" ]]; then
cat > "$PROJECT_DIR/.env" <<EOF
NGC_API_KEY=${NGC_API_KEY}
EOF
chmod 600 "$PROJECT_DIR/.env"
ok "Created .env with your NGC API key"
else
ok ".env already exists"
fi
# =============================================================================
# STEP 9: Deploy with docker compose
# =============================================================================
echo ""
echo "============================================================"
echo " Deployment Options"
echo "============================================================"
echo ""
echo " A) Deploy now with docker compose (recommended for testing)"
echo " B) I'll deploy via Portainer (show me the steps)"
echo " C) Skip — I'll deploy later"
echo ""
read -p " Choose [A/B/C]: " -n 1 -r DEPLOY_CHOICE; echo
case $DEPLOY_CHOICE in
[Aa])
echo ""
info "Starting services with docker compose..."
docker compose --env-file .env up -d
echo ""
ok "Services starting!"
info "Watch logs with:"
info " docker compose logs -f parakeet-asr"
info " docker compose logs -f magpie-tts"
;;
[Bb])
echo ""
echo "============================================================"
echo " Portainer Deployment Steps"
echo "============================================================"
echo ""
echo " 1. Open Portainer: https://<your-spark-ip>:9443"
echo ""
echo " 2. Go to: Stacks → Add Stack"
echo ""
echo " 3. Name: voice-agent"
echo ""
echo " 4. Build method: Upload"
echo " Upload: $PROJECT_DIR/docker-compose.yml"
echo ""
echo " 5. Environment variables → Add:"
echo " NGC_API_KEY = $NGC_API_KEY"
echo ""
echo " 6. Click 'Deploy the stack'"
echo ""
echo " NOTE: The parakeet-asr image was already built locally."
echo " Portainer will find it by the image name."
echo ""
;;
*)
info "Skipping deployment."
;;
esac
# =============================================================================
# STEP 10: Print summary
# =============================================================================
echo ""
echo "============================================================"
echo " Setup Complete!"
echo "============================================================"
echo ""
echo " Built images:"
echo " • parakeet-tdt-v3-spark:latest (Parakeet ASR)"
if docker images --format '{{.Repository}}:{{.Tag}}' | grep -q "magpie-tts"; then
echo " • nvcr.io/nim/nvidia/magpie-tts-multilingual:latest (TTS)"
fi
echo ""
echo " Endpoints (after deployment):"
echo " ASR Transcription: http://<spark-ip>:8010/v1/audio/transcriptions"
echo " ASR Health: http://<spark-ip>:8010/health"
echo " ASR Info: http://<spark-ip>:8010/"
echo " TTS (gRPC): <spark-ip>:50053"
echo " TTS (HTTP): http://<spark-ip>:8030"
echo ""
echo " Test ASR:"
echo ' curl -s http://localhost:8010/v1/audio/transcriptions \'
echo ' -F file="@test.wav" -F language=auto | python3 -m json.tool'
echo ""
echo " Test with German audio:"
echo ' curl -s http://localhost:8010/v1/audio/transcriptions \'
echo ' -F file="@german.wav" -F language=de | python3 -m json.tool'
echo ""
echo " Supported languages: bg, cs, da, de, el, en, es, et, fi, fr,"
echo " hr, hu, it, lt, lv, mt, nl, pl, pt, ro, ru, sk, sl, sv, uk"
echo ""
echo " Project dir: $PROJECT_DIR"
echo " Logs: docker compose logs -f"
echo ""
echo "============================================================"```
change to the app folder
cd app
create these files
5. app/__init__.py
# Parakeet TDT 0.6b v3 — FastAPI ASR Service
6. app/transcriber.py
import io
import os
import logging
import tempfile
import subprocess
from pathlib import Path
from typing import Optional
import torch
import soundfile as sf
import numpy as np
logger = logging.getLogger(__name__)
MODEL_NAME = os.getenv("PARAKEET_MODEL", "nvidia/parakeet-tdt-0.6b-v3")
MAX_SEGMENT_SECONDS = int(os.getenv("MAX_SEGMENT_SECONDS", "1200"))
TARGET_SAMPLE_RATE = 16000
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
class ParakeetTranscriber:
def __init__(self):
self.model = None
self._loaded = False
def load_model(self):
if self._loaded:
return
logger.info(f"Loading model {MODEL_NAME} on {DEVICE}...")
logger.info(f"CUDA available: {torch.cuda.is_available()}")
if torch.cuda.is_available():
logger.info(f"GPU: {torch.cuda.get_device_name(0)}")
logger.info(f"GPU memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB")
import nemo.collections.asr as nemo_asr
self.model = nemo_asr.models.ASRModel.from_pretrained(MODEL_NAME)
self.model.eval()
if DEVICE == "cuda":
self.model = self.model.cuda()
self._loaded = True
logger.info(f"Model loaded successfully on {DEVICE}")
def _convert_to_wav_16k_mono(self, audio_bytes: bytes, original_filename: str) -> str:
suffix = Path(original_filename).suffix.lower() if original_filename else ".wav"
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp_in:
tmp_in.write(audio_bytes)
tmp_in_path = tmp_in.name
tmp_out_path = tmp_in_path + ".converted.wav"
try:
cmd = [
"ffmpeg", "-y", "-i", tmp_in_path,
"-ac", "1", "-ar", "16000", "-sample_fmt", "s16", "-f", "wav",
tmp_out_path
]
result = subprocess.run(cmd, capture_output=True, timeout=300)
if result.returncode != 0:
stderr = result.stderr.decode("utf-8", errors="replace")
raise RuntimeError(f"ffmpeg conversion failed: {stderr[:500]}")
return tmp_out_path
finally:
try:
os.unlink(tmp_in_path)
except OSError:
pass
def _chunk_audio_file(self, wav_path: str) -> list[str]:
data, sr = sf.read(wav_path)
total_seconds = len(data) / sr
if total_seconds <= MAX_SEGMENT_SECONDS:
return [wav_path]
logger.info(f"Audio is {total_seconds:.0f}s, chunking into {MAX_SEGMENT_SECONDS}s segments...")
chunk_samples = MAX_SEGMENT_SECONDS * sr
chunks = []
offset = 0
while offset < len(data):
end = min(offset + chunk_samples, len(data))
chunk_data = data[offset:end]
chunk_path = wav_path.replace(".wav", f".chunk{len(chunks):04d}.wav")
sf.write(chunk_path, chunk_data, sr)
chunks.append(chunk_path)
offset = end
logger.info(f"Created {len(chunks)} chunks")
return chunks
def transcribe(
self,
audio_bytes: bytes,
filename: str = "audio.wav",
language: Optional[str] = None,
timestamps: bool = False,
) -> dict:
if not self._loaded:
self.load_model()
wav_path = None
chunk_paths = []
try:
wav_path = self._convert_to_wav_16k_mono(audio_bytes, filename)
data, sr = sf.read(wav_path)
duration = len(data) / sr
logger.info(f"Audio duration: {duration:.1f}s ({filename})")
chunk_paths = self._chunk_audio_file(wav_path)
all_texts = []
time_offset = 0.0
for i, chunk_path in enumerate(chunk_paths):
logger.info(f"Transcribing chunk {i + 1}/{len(chunk_paths)}...")
with torch.no_grad():
output = self.model.transcribe([chunk_path], batch_size=1)
if isinstance(output, list):
for item in output:
if isinstance(item, str):
all_texts.append(item)
elif hasattr(item, 'text'):
all_texts.append(item.text)
elif isinstance(item, list) and len(item) > 0:
all_texts.append(str(item[0]))
elif isinstance(output, str):
all_texts.append(output)
chunk_data, chunk_sr = sf.read(chunk_path)
time_offset += len(chunk_data) / chunk_sr
if DEVICE == "cuda":
torch.cuda.empty_cache()
full_text = " ".join(all_texts).strip()
return {
"text": full_text,
"duration": round(duration, 3),
"model": MODEL_NAME,
"device": DEVICE,
}
finally:
for path in chunk_paths:
if path != wav_path:
try:
os.unlink(path)
except OSError:
pass
if wav_path:
try:
os.unlink(wav_path)
except OSError:
pass
transcriber = ParakeetTranscriber()
7. app/main.py
import os
import time
import logging
from contextlib import asynccontextmanager
from typing import Optional
import torch
from fastapi import FastAPI, File, Form, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from app.transcriber import transcriber, MODEL_NAME, DEVICE
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("parakeet-api")
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("=" * 60)
logger.info("Parakeet TDT 0.6b v3 — Starting up")
logger.info(f" Model: {MODEL_NAME}")
logger.info(f" Device: {DEVICE}")
if torch.cuda.is_available():
logger.info(f" GPU: {torch.cuda.get_device_name(0)}")
mem = torch.cuda.get_device_properties(0).total_memory / (1024**3)
logger.info(f" VRAM: {mem:.1f} GB")
logger.info("=" * 60)
transcriber.load_model()
logger.info("Model ready — accepting requests")
yield
logger.info("Shutting down...")
app = FastAPI(
title="Parakeet ASR API",
description="OpenAI-compatible Speech-to-Text powered by NVIDIA Parakeet TDT 0.6b v3",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def root():
return {
"service": "parakeet-asr",
"model": MODEL_NAME,
"device": DEVICE,
"cuda_available": torch.cuda.is_available(),
"gpu": torch.cuda.get_device_name(0) if torch.cuda.is_available() else None,
"endpoints": {
"transcribe": "/v1/audio/transcriptions",
"models": "/v1/models",
"health": "/health",
},
"supported_languages": [
"en", "de", "fr", "es", "it", "pt", "nl", "pl", "ru", "uk",
"cs", "sk", "sl", "hr", "bg", "ro", "hu", "el", "da", "sv",
"fi", "et", "lt", "lv", "mt",
],
}
@app.get("/health")
async def health():
gpu_mem = None
if torch.cuda.is_available():
allocated = torch.cuda.memory_allocated(0) / (1024**3)
total = torch.cuda.get_device_properties(0).total_memory / (1024**3)
gpu_mem = {"allocated_gb": round(allocated, 2), "total_gb": round(total, 1)}
return {
"status": "ready" if transcriber._loaded else "loading",
"model": MODEL_NAME,
"device": DEVICE,
"gpu_memory": gpu_mem,
}
@app.get("/v1/models")
async def list_models():
return {
"object": "list",
"data": [
{
"id": "parakeet-tdt-0.6b-v3",
"object": "model",
"created": 1723593600,
"owned_by": "nvidia",
},
{
"id": "whisper-1",
"object": "model",
"created": 1723593600,
"owned_by": "nvidia",
},
],
}
@app.post("/v1/audio/transcriptions")
async def transcribe(
file: UploadFile = File(...),
model: Optional[str] = Form(default="parakeet-tdt-0.6b-v3"),
language: Optional[str] = Form(default=None),
response_format: Optional[str] = Form(default="json"),
temperature: Optional[float] = Form(default=0.0),
timestamp_granularities: Optional[str] = Form(default=None),
prompt: Optional[str] = Form(default=None),
):
if not transcriber._loaded:
raise HTTPException(status_code=503, detail="Model is still loading.")
if not file.filename:
raise HTTPException(status_code=400, detail="No file provided")
try:
audio_bytes = await file.read()
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to read audio file: {e}")
if len(audio_bytes) == 0:
raise HTTPException(status_code=400, detail="Empty audio file")
max_size = int(os.getenv("MAX_UPLOAD_MB", "200")) * 1024 * 1024
if len(audio_bytes) > max_size:
raise HTTPException(status_code=413, detail=f"File too large. Max: {max_size // (1024*1024)} MB")
want_timestamps = response_format in ("verbose_json", "srt", "vtt")
logger.info(f"Request: {file.filename} ({len(audio_bytes) / 1024:.0f} KB, lang={language})")
start_time = time.time()
try:
result = transcriber.transcribe(
audio_bytes=audio_bytes,
filename=file.filename,
language=language,
timestamps=want_timestamps,
)
except Exception as e:
logger.exception("Transcription failed")
raise HTTPException(status_code=500, detail=f"Transcription failed: {e}")
elapsed = time.time() - start_time
rtfx = result.get("duration", 0) / elapsed if elapsed > 0 else 0
logger.info(f"Done: {result.get('duration', 0):.1f}s audio in {elapsed:.1f}s ({rtfx:.0f}x realtime)")
if response_format == "text":
return JSONResponse(content=result["text"], media_type="text/plain")
if response_format == "verbose_json":
return {
"task": "transcribe",
"language": language or "auto",
"duration": result.get("duration", 0),
"text": result["text"],
"segments": result.get("segments", []),
}
if response_format == "srt":
return JSONResponse(content=_to_srt(result), media_type="text/plain")
if response_format == "vtt":
return JSONResponse(content=_to_vtt(result), media_type="text/plain")
return {"text": result["text"]}
@app.post("/v1/audio/translations")
async def translate(
file: UploadFile = File(...),
model: Optional[str] = Form(default="parakeet-tdt-0.6b-v3"),
language: Optional[str] = Form(default=None),
response_format: Optional[str] = Form(default="json"),
temperature: Optional[float] = Form(default=0.0),
prompt: Optional[str] = Form(default=None),
):
return await transcribe(file=file, model=model, language=language, response_format=response_format, temperature=temperature)
def _fmt_srt(s: float) -> str:
h, m = int(s // 3600), int((s % 3600) // 60)
return f"{h:02d}:{m:02d}:{int(s % 60):02d},{int((s % 1) * 1000):03d}"
def _fmt_vtt(s: float) -> str:
h, m = int(s // 3600), int((s % 3600) // 60)
return f"{h:02d}:{m:02d}:{int(s % 60):02d}.{int((s % 1) * 1000):03d}"
def _to_srt(result: dict) -> str:
segs = result.get("segments", [])
if not segs:
return f"1\n00:00:00,000 --> 99:59:59,999\n{result['text']}\n"
return "\n".join(f"{i}\n{_fmt_srt(s.get('start',0))} --> {_fmt_srt(s.get('end',0))}\n{s.get('text','')}\n" for i, s in enumerate(segs, 1))
def _to_vtt(result: dict) -> str:
segs = result.get("segments", [])
hdr = "WEBVTT\n\n"
if not segs:
return hdr + f"00:00:00.000 --> 99:59:59.999\n{result['text']}\n"
return hdr + "\n".join(f"{_fmt_vtt(s.get('start',0))} --> {_fmt_vtt(s.get('end',0))}\n{s.get('text','')}\n" for s in segs)
Build it
you can automate the process using this build and deploy script. Just save it as build-and-deploy.sh in the root of your project folder, make it executable
chmod +x build-and-deploy.sh
and run it:
./build-and-deploy.sh