Files
esh-pfi-infrastructure/stacks/stable-audio-open/server.py
T
vh 4a4c09177f ace-step + stable-audio-open: deploy music + SFX generation to irv-ml1
Two new audio-generation stacks alongside the TTS slate:

ace-step :8210 — Apache 2.0 music generation foundation model
(hybrid diffusion + LLM). Lyric-aware multi-minute songs. ~10-12 GB
VRAM during inference, A6000-pinned. Custom Dockerfile patches
upstream's torch/cu126 resolution bug (--extra-index-url cu126 was
falling back to pypi-default cu13 wheels, mismatching torchvision).

stable-audio-open :8211 — Stability AI 1.21B latent-diffusion SFX +
ambience. Up to 47s clips at 44.1 kHz. ~6 GB VRAM in fp16,
A6000-pinned. Custom FastAPI shim around diffusers' StableAudioPipeline
(no upstream HTTP server). Dockerfile pins torchsde explicitly —
diffusers doesn't pull it as a hard dep but
CosineDPMSolverMultistepScheduler needs it.
2026-04-28 09:11:23 -07:00

81 lines
2.2 KiB
Python

# FastAPI shim around diffusers' StableAudioPipeline.
# Single endpoint POST /v1/audio/sfx returns a WAV blob.
# Model is loaded once on startup and held in process memory.
import io
import os
import time
from contextlib import asynccontextmanager
from typing import Optional
import soundfile as sf
import torch
from diffusers import StableAudioPipeline
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
MODEL_ID = os.environ.get("SAO_MODEL", "stabilityai/stable-audio-open-1.0")
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
DTYPE = torch.float16 if DEVICE == "cuda" else torch.float32
state: dict = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
print(f"[sao] loading {MODEL_ID} on {DEVICE} ({DTYPE})", flush=True)
t0 = time.time()
pipe = StableAudioPipeline.from_pretrained(MODEL_ID, torch_dtype=DTYPE)
pipe = pipe.to(DEVICE)
state["pipe"] = pipe
print(f"[sao] loaded in {time.time() - t0:.1f}s", flush=True)
yield
state.clear()
app = FastAPI(lifespan=lifespan)
class SfxRequest(BaseModel):
prompt: str = Field(..., min_length=1)
negative_prompt: Optional[str] = "Low quality."
duration: float = Field(10.0, gt=0.5, le=47.0)
steps: int = Field(100, ge=10, le=300)
seed: Optional[int] = None
cfg_scale: float = Field(7.0, gt=0.0, le=20.0)
@app.get("/health")
def health():
return {
"status": "ok",
"model": MODEL_ID,
"device": DEVICE,
"loaded": "pipe" in state,
}
@app.post("/v1/audio/sfx")
def sfx(req: SfxRequest):
pipe = state.get("pipe")
if pipe is None:
raise HTTPException(503, "model not loaded yet")
generator = None
if req.seed is not None:
generator = torch.Generator(DEVICE).manual_seed(req.seed)
audio = pipe(
req.prompt,
negative_prompt=req.negative_prompt,
num_inference_steps=req.steps,
audio_end_in_s=req.duration,
num_waveforms_per_prompt=1,
generator=generator,
).audios
waveform = audio[0].T.float().cpu().numpy()
buf = io.BytesIO()
sf.write(buf, waveform, pipe.vae.sampling_rate, format="WAV")
buf.seek(0)
return Response(content=buf.read(), media_type="audio/wav")