feat(api,models): FastAPI app with /api/health, /api/models, activate, /api/generate + chatterbox-en adapter

server/main.py

@@ -0,0 +1,161 @@
+"""FastAPI application factory."""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import torch
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, Response
+from fastapi.staticfiles import StaticFiles
+from sse_starlette.sse import EventSourceResponse
+
+from server.audio import AudioValidationError, validate_reference_clip
+from server.device import select_device
+from server.registry import Registry
+from server.zerogpu import decorate
+
+
+STATIC_DIR = Path(__file__).parent / "static"
+
+
+def _discover_adapter_classes() -> dict[str, type]:
+    """Lazily import adapter modules. Empty dict during early scaffolding."""
+    classes: dict[str, type] = {}
+    for module_name in ("chatterbox_en", "chatterbox_turbo", "chatterbox_mtl"):
+        try:
+            mod = __import__(f"server.models.{module_name}", fromlist=["Adapter"])
+        except ImportError:
+            continue
+        cls = getattr(mod, "Adapter", None)
+        if cls is not None:
+            classes[cls.id] = cls
+    return classes
+
+
+def build_app() -> FastAPI:
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        device = select_device()
+        app.state.registry = Registry(
+            adapter_classes=_discover_adapter_classes(),
+            device=device,
+        )
+        yield
+
+    app = FastAPI(title="Chatterbox Voice Studio", lifespan=lifespan)
+
+    origins = os.getenv(
+        "CORS_ORIGINS",
+        "http://localhost:5173,http://127.0.0.1:5173",
+    ).split(",")
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=origins,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.get("/api/health")
+    def health() -> dict:
+        registry = app.state.registry
+        return {
+            "device": registry.device,
+            "torch_version": torch.__version__,
+            "model_status": registry.status()["status"],
+        }
+
+    @app.get("/api/models")
+    def list_models() -> list[dict]:
+        return app.state.registry.list_models()
+
+    @app.get("/api/models/active")
+    def active_model() -> dict:
+        return app.state.registry.status()
+
+    @app.post("/api/models/{model_id}/activate")
+    async def activate_model(model_id: str):
+        try:
+            await app.state.registry.get_or_load(model_id)
+        except KeyError:
+            raise HTTPException(
+                status_code=404,
+                detail={"error": {"code": "model_not_found", "message": model_id}},
+            )
+        except Exception as exc:
+            return JSONResponse(
+                status_code=503,
+                content={"error": {"code": "model_load_failed", "message": str(exc)}},
+            )
+        return {"ok": True}
+
+    @app.get("/api/models/active/events")
+    async def active_events():
+        async def gen():
+            async for evt in app.state.registry.stream_events():
+                yield {"data": json.dumps(evt)}
+
+        return EventSourceResponse(gen())
+
+    @app.post("/api/generate")
+    async def generate(
+        text: str = Form(...),
+        model_id: str = Form(...),
+        params: str = Form("{}"),
+        language: str | None = Form(None),
+        reference_wav: UploadFile | None = File(None),
+    ):
+        try:
+            adapter = await app.state.registry.get_or_load(model_id)
+        except KeyError:
+            raise HTTPException(
+                status_code=404,
+                detail={"error": {"code": "model_not_found", "message": model_id}},
+            )
+
+        ref_path: str | None = None
+        if reference_wav is not None:
+            data = await reference_wav.read()
+            try:
+                validate_reference_clip(data)
+            except AudioValidationError as exc:
+                return JSONResponse(
+                    status_code=400,
+                    content={"error": {"code": "reference_invalid", "message": str(exc)}},
+                )
+            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
+            tmp.write(data)
+            tmp.flush()
+            tmp.close()
+            ref_path = tmp.name
+
+        gen_fn = decorate(adapter.generate)
+        try:
+            wav_bytes, _sr = gen_fn(text, ref_path, language, json.loads(params or "{}"))
+        except Exception as exc:
+            return JSONResponse(
+                status_code=500,
+                content={"error": {"code": "generation_failed", "message": str(exc)}},
+            )
+        return Response(content=wav_bytes, media_type="audio/wav")
+
+    @app.exception_handler(HTTPException)
+    async def _http_exc(request, exc: HTTPException):
+        if isinstance(exc.detail, dict) and "error" in exc.detail:
+            return JSONResponse(status_code=exc.status_code, content=exc.detail)
+        return JSONResponse(
+            status_code=exc.status_code,
+            content={"error": {"code": "http_error", "message": str(exc.detail)}},
+        )
+
+    if STATIC_DIR.exists():
+        app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")
+
+    return app
+
+
+app = build_app()

server/models/chatterbox_en.py

@@ -0,0 +1,76 @@
+"""Chatterbox English adapter (ResembleAI/chatterbox)."""
+from __future__ import annotations
+
+import io
+from typing import Any, ClassVar
+
+import soundfile as sf
+
+from server.schemas import Lang, ParamSpec
+
+
+class Adapter:
+    id: ClassVar[str] = "chatterbox-en"
+    label: ClassVar[str] = "Chatterbox (English)"
+    description: ClassVar[str] = (
+        "Original Chatterbox English voice cloning with CFG and exaggeration controls."
+    )
+    languages: ClassVar[list[Lang]] = [Lang(code="en", label="English")]
+    paralinguistic_tags: ClassVar[list[str]] = []
+    supports_voice_clone: ClassVar[bool] = True
+    params: ClassVar[list[ParamSpec]] = [
+        ParamSpec(
+            name="exaggeration", label="Exaggeration", type="float",
+            default=0.5, min=0.0, max=2.0, step=0.05,
+            help="Higher = more expressive prosody.",
+        ),
+        ParamSpec(
+            name="cfg_weight", label="CFG weight", type="float",
+            default=0.5, min=0.0, max=1.0, step=0.05,
+        ),
+        ParamSpec(
+            name="temperature", label="Temperature", type="float",
+            default=0.8, min=0.1, max=1.5, step=0.05,
+        ),
+    ]
+
+    def __init__(self, device: str) -> None:
+        self.device = device
+        self._model = None
+
+    def load(self) -> None:
+        from chatterbox.tts import ChatterboxTTS
+
+        self._model = ChatterboxTTS.from_pretrained(device=self.device)
+
+    def unload(self) -> None:
+        self._model = None
+
+    def generate(
+        self,
+        text: str,
+        reference_wav_path: str | None,
+        language: str | None,
+        params: dict[str, Any],
+    ) -> tuple[bytes, int]:
+        if self._model is None:
+            raise RuntimeError("model not loaded")
+        wav = self._model.generate(
+            text,
+            audio_prompt_path=reference_wav_path,
+            exaggeration=float(params.get("exaggeration", 0.5)),
+            cfg_weight=float(params.get("cfg_weight", 0.5)),
+            temperature=float(params.get("temperature", 0.8)),
+        )
+        import numpy as np
+        import torch
+
+        if hasattr(wav, "detach"):
+            wav = wav.detach().cpu().numpy()
+        if isinstance(wav, torch.Tensor):  # pragma: no cover
+            wav = wav.numpy()
+        arr = np.asarray(wav).squeeze()
+        sr = getattr(self._model, "sr", 24000)
+        buf = io.BytesIO()
+        sf.write(buf, arr, sr, format="WAV", subtype="PCM_16")
+        return buf.getvalue(), sr

tests/conftest.py

@@ -1,7 +1,7 @@
 """Shared test fixtures."""
 from __future__ import annotations
 
-import asyncio
+from contextlib import asynccontextmanager
 
 import pytest
 
@@ -49,3 +49,10 @@ class FakeAdapterB(FakeAdapter):
 def fake_classes():
     FakeAdapter.instances.clear()
     return {FakeAdapter.id: FakeAdapter, FakeAdapterB.id: FakeAdapterB}
+
+
+@asynccontextmanager
+async def lifespan_ctx(app):
+    """Run an ASGI app's lifespan startup/shutdown around an `httpx.AsyncClient`."""
+    async with app.router.lifespan_context(app):
+        yield

tests/test_adapter_contract.py

@@ -0,0 +1,21 @@
+import importlib
+
+import pytest
+
+from server.models.base import is_valid_adapter
+from server.schemas import ParamSpec
+
+
+ADAPTER_MODULES = [
+    "server.models.chatterbox_en",
+]
+
+
+@pytest.mark.parametrize("module_name", ADAPTER_MODULES)
+def test_adapter_class_attributes_valid(module_name):
+    mod = importlib.import_module(module_name)
+    cls = getattr(mod, "Adapter")
+    assert is_valid_adapter(cls)
+    assert cls.id
+    for p in cls.params:
+        assert isinstance(p, ParamSpec)

tests/test_main_activate.py

@@ -0,0 +1,45 @@
+import asyncio
+
+import httpx
+import pytest
+
+from server.main import build_app
+
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_activate_then_status_loaded(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    from tests.conftest import lifespan_ctx
+    transport = httpx.ASGITransport(app=app)
+    async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
+        r = await c.post("/api/models/fake/activate")
+        assert r.status_code in (200, 202)
+        for _ in range(20):
+            s = (await c.get("/api/models/active")).json()
+            if s["status"] == "loaded":
+                break
+            await asyncio.sleep(0.05)
+        assert s["id"] == "fake"
+        assert s["status"] == "loaded"
+
+
+async def test_activate_unknown_returns_404(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    from tests.conftest import lifespan_ctx
+    transport = httpx.ASGITransport(app=app)
+    async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
+        r = await c.post("/api/models/nope/activate")
+    assert r.status_code == 404
+    assert r.json()["error"]["code"] == "model_not_found"
+
+
+# Note: integration test for /api/models/active/events SSE stream is omitted.
+# Registry event emission is unit-tested in tests/test_registry.py
+# (test_emits_loading_then_loaded_events). The /api/models/active/events
+# endpoint is a thin sse-starlette wrapper around that generator.

tests/test_main_generate.py

@@ -0,0 +1,59 @@
+import httpx
+import pytest
+
+from server.main import build_app
+
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_generate_returns_wav_bytes(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    from tests.conftest import lifespan_ctx
+    transport = httpx.ASGITransport(app=app)
+    async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
+        r = await c.post(
+            "/api/generate",
+            data={
+                "text": "hello world",
+                "model_id": "fake",
+                "params": "{}",
+            },
+        )
+    assert r.status_code == 200
+    assert r.headers["content-type"].startswith("audio/wav")
+    assert r.content == b"FAKEWAV"
+
+
+async def test_generate_unknown_model_404(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    from tests.conftest import lifespan_ctx
+    transport = httpx.ASGITransport(app=app)
+    async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
+        r = await c.post(
+            "/api/generate",
+            data={"text": "x", "model_id": "nope", "params": "{}"},
+        )
+    assert r.status_code == 404
+    assert r.json()["error"]["code"] == "model_not_found"
+
+
+async def test_generate_invalid_reference_returns_400(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    from tests.conftest import lifespan_ctx
+    transport = httpx.ASGITransport(app=app)
+    bad = b"not a wav"
+    async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
+        r = await c.post(
+            "/api/generate",
+            data={"text": "x", "model_id": "fake", "params": "{}"},
+            files={"reference_wav": ("ref.wav", bad, "audio/wav")},
+        )
+    assert r.status_code == 400
+    assert r.json()["error"]["code"] == "reference_invalid"

tests/test_main_health.py

@@ -0,0 +1,16 @@
+from fastapi.testclient import TestClient
+
+from server.main import build_app
+
+
+def test_health_returns_device_and_status(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    with TestClient(app) as client:
+        r = client.get("/api/health")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["device"] == "cpu"
+    assert data["model_status"] == "idle"
+    assert "torch_version" in data

tests/test_main_models.py

@@ -0,0 +1,30 @@
+from fastapi.testclient import TestClient
+
+from server.main import build_app
+
+
+def test_models_list_returns_registered(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    with TestClient(app) as client:
+        r = client.get("/api/models")
+    assert r.status_code == 200
+    items = r.json()
+    ids = sorted(m["id"] for m in items)
+    assert ids == ["fake", "fake-b"]
+    fake = next(m for m in items if m["id"] == "fake")
+    assert fake["paralinguistic_tags"] == ["[laugh]"]
+    assert fake["params"][0]["name"] == "t"
+
+
+def test_active_model_initially_idle(monkeypatch, fake_classes):
+    monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
+    monkeypatch.setattr("server.main.select_device", lambda: "cpu")
+    app = build_app()
+    with TestClient(app) as client:
+        r = client.get("/api/models/active")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["id"] is None
+    assert body["status"] == "idle"