supertonic.server¶

def create_app(
    *,
    state: Optional[ServerState] = None,
    model: str = DEFAULT_MODEL,
    custom_styles_dir: Optional[Path] = None,
    cors_origins: Optional[Iterable[str]] = None,
) -> FastAPI:
    """Build a configured FastAPI app.

    Args:
        state: Pre-built state to reuse. When provided, the lifespan does *not*
            instantiate :class:`supertonic.TTS` — useful for tests that inject
            a fake. Pass ``None`` for normal use.
        model: Model name to load if ``state.tts`` is ``None``.
        custom_styles_dir: Override the on-disk location of user-imported
            voice styles. Defaults to
            :func:`supertonic.server.styles_store.default_custom_styles_dir`.
        cors_origins: If non-empty, install ``CORSMiddleware`` for these
            origins. Browser-extension or Electron clients need this; n8n and
            curl do not.
    """
    if state is None:
        state = ServerState(model=model, custom_styles_dir=custom_styles_dir)
    elif custom_styles_dir is not None:
        state.custom_styles_dir = Path(custom_styles_dir)

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        if state.tts is None:
            # Import here so that ``supertonic.server`` import does not pull
            # the model loader into hot paths or test harnesses that mock it.
            from ..pipeline import TTS

            logger.info("Loading TTS model %r ...", state.model)
            state.tts = TTS(model=state.model)
        state.custom_styles = styles_store.scan(state.custom_styles_dir)
        state.is_ready = True
        logger.info(
            "supertonic serve ready: model=%s builtin=%d custom=%d",
            state.model,
            len(state.tts.voice_style_names) if state.tts else 0,
            len(state.custom_styles),
        )
        try:
            yield
        finally:
            state.is_ready = False

    app = FastAPI(
        title="Supertonic TTS",
        description=(
            "Local HTTP server for Supertonic TTS. Exposes a native /v1/* "
            "namespace plus an OpenAI Audio Speech-compatible alias at "
            "POST /v1/audio/speech so existing clients work with just a "
            "base-URL change."
        ),
        version=__version__,
        lifespan=lifespan,
    )
    app.state.server_state = state

    # Note: middlewares execute in reverse order of addition (the *last*
    # added wraps everything below it). Add the size limit last so it
    # short-circuits before FastAPI's routing/dependency layers start
    # buffering the multipart body.
    if cors_origins:
        from fastapi.middleware.cors import CORSMiddleware

        app.add_middleware(
            CORSMiddleware,
            allow_origins=list(cors_origins),
            allow_credentials=False,
            allow_methods=["*"],
            allow_headers=["*"],
        )
    app.add_middleware(StyleImportSizeLimit, max_bytes=MAX_STYLE_IMPORT_BYTES)

    register_routes(app)
    return app

def default_custom_styles_dir(model: str = DEFAULT_MODEL) -> Path:
    """Resolve the on-disk directory for user-imported voice styles.

    Priority:

    1. ``$SUPERTONIC_CUSTOM_STYLES_DIR`` — explicit override, applies to every
       model (the user opted into a single shared location).
    2. ``<model cache dir>/custom_styles/`` — e.g. ``~/.cache/supertonic3/
       custom_styles/`` for ``supertonic-3``. Respects ``$SUPERTONIC_CACHE_DIR``
       through :func:`supertonic.loader.get_cache_dir`.
    """
    env = os.getenv("SUPERTONIC_CUSTOM_STYLES_DIR")
    if env:
        return Path(env).expanduser()
    return get_cache_dir(model) / "custom_styles"

def sanitize_name(name: str) -> str:
    name = (name or "").strip()
    if not _NAME_RE.fullmatch(name):
        raise InvalidStyleName(f"Invalid style name {name!r}: must match [A-Za-z0-9_-]{{1,64}}")
    return name

def scan(directory: Path) -> Dict[str, Path]:
    """Return ``{stem: path}`` for every well-formed JSON in ``directory``.

    A file that fails :func:`validate_voice_style_format` is skipped with a
    warning rather than crashing startup — the server should still come up.
    """
    out: Dict[str, Path] = {}
    if not directory.exists():
        return out
    for p in sorted(directory.glob("*.json")):
        try:
            with p.open("r", encoding="utf-8") as f:
                data = json.load(f)
            if not validate_voice_style_format(data):
                logger.warning("Skipping invalid voice style file: %s", p)
                continue
        except (OSError, json.JSONDecodeError) as e:
            logger.warning("Skipping unreadable voice style file %s: %s", p, e)
            continue
        out[p.stem] = p
    return out

def save(
    directory: Path,
    name: str,
    payload: dict,
    *,
    builtin_names: Iterable[str] = (),
    overwrite: bool = False,
) -> Path:
    """Persist a validated style payload to ``directory / f"{name}.json"``.

    Args:
        directory: target directory (created if missing).
        name: requested style name; sanitized via :func:`sanitize_name`.
        payload: parsed JSON; must pass :func:`validate_voice_style_format`.
        builtin_names: names reserved by the bundled model; conflict → 400.
        overwrite: if False, conflict with an existing custom name → 409.

    Returns:
        The path the style was written to.
    """
    name = sanitize_name(name)
    if name in set(builtin_names):
        raise StyleNameConflict(f"Name {name!r} is a built-in voice and cannot be overwritten")
    if not validate_voice_style_format(payload):
        # Re-using the SDK error type so server handlers can map uniformly.
        raise ValueError("voice style JSON is missing required keys/fields")

    directory.mkdir(parents=True, exist_ok=True)
    target = directory / f"{name}.json"
    if target.exists() and not overwrite:
        raise StyleNameConflict(f"Style {name!r} already exists")
    tmp = target.with_suffix(".json.tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(payload, f)
    tmp.replace(target)
    return target

def format_to_mime(fmt: str) -> str:
    entry = _FORMATS.get(fmt)
    if entry is None:
        raise UnsupportedAudioFormat(fmt)
    return entry[2]

def encode_audio(wav: np.ndarray, sample_rate: int, fmt: str) -> bytes:
    """Encode a synthesized waveform into ``fmt`` bytes.

    Args:
        wav: ndarray of shape ``(1, num_samples)`` or ``(num_samples,)`` —
            the shape produced by :meth:`supertonic.TTS.synthesize`.
        sample_rate: model sample rate (e.g. 44100).
        fmt: one of :data:`SUPPORTED_FORMATS`.
    """
    entry = _FORMATS.get(fmt)
    if entry is None:
        raise UnsupportedAudioFormat(fmt)
    sf_format, subtype, _ = entry

    if wav.ndim == 2:
        # soundfile expects (frames,) or (frames, channels). The pipeline
        # returns (1, num_samples), so squeeze the leading singleton.
        wav = wav.squeeze(0)

    buf = io.BytesIO()
    sf.write(buf, wav, sample_rate, format=sf_format, subtype=subtype)
    return buf.getvalue()

def duration_seconds(wav: np.ndarray, sample_rate: int) -> float:
    return float(wav.shape[-1]) / float(sample_rate)

def coerce_response_format(value: Optional[str]) -> str:
    """Validate and normalize a user-supplied ``response_format``.

    ``None`` → ``"wav"`` (sensible default for local-host integrations). An
    unsupported value raises :class:`UnsupportedAudioFormat` so handlers can
    return a 400 with a stable error code.
    """
    if value is None:
        return "wav"
    v = value.lower().strip()
    if v not in _FORMATS:
        raise UnsupportedAudioFormat(value)
    return v

def register_routes(app: FastAPI) -> None:
    """Attach all `/v1/...` routes to ``app``.

    Called from :func:`supertonic.server.app.create_app` after the lifespan and
    ``app.state.server_state`` have been set up.
    """
    router = APIRouter()

    @router.get("/v1/health", response_model=HealthResponse)
    def health(request: Request):
        state = _state(request)
        if not state.is_ready or state.tts is None:
            return JSONResponse(
                status_code=503,
                content=HealthResponse(
                    status="loading",
                    model=state.model,
                    version=__version__,
                    voices_loaded=0,
                ).model_dump(),
            )
        return HealthResponse(
            status="ok",
            model=state.model,
            sample_rate=state.tts.sample_rate,
            version=__version__,
            voices_loaded=len(state.tts.voice_style_names) + len(state.custom_styles),
        )

    @router.get("/v1/styles", response_model=StylesResponse)
    def list_styles(request: Request):
        state = _state(request)
        if state.tts is None:
            return _error(503, "server not ready", "not_ready", type_="server_error")
        builtin = [StyleInfo(name=n, kind="builtin") for n in state.tts.voice_style_names]
        custom = [
            StyleInfo(name=n, kind="custom", path=str(p))
            for n, p in sorted(state.custom_styles.items())
        ]
        return StylesResponse(styles=builtin + custom)

    @router.post("/v1/styles/import", response_model=StyleImportResponse)
    async def import_style(
        request: Request,
        overwrite: bool = False,
        file: Optional[UploadFile] = File(None),
        name: Optional[str] = Form(None),
    ):
        state = _state(request)
        if state.tts is None:
            return _error(503, "server not ready", "not_ready", type_="server_error")

        ct = request.headers.get("content-type", "")
        chosen_name: Optional[str]
        if ct.startswith("multipart/form-data"):
            if file is None:
                return _error(400, "missing 'file' part", "missing_file")
            # Read with an explicit cap as a fallback for chunked uploads
            # that bypass the middleware's Content-Length pre-flight check.
            raw = await file.read(MAX_STYLE_IMPORT_BYTES + 1)
            if len(raw) > MAX_STYLE_IMPORT_BYTES:
                return _error(
                    413,
                    f"uploaded voice style exceeds {MAX_STYLE_IMPORT_BYTES} bytes",
                    "payload_too_large",
                )
            try:
                data = json.loads(raw)
            except json.JSONDecodeError as e:
                return _error(400, f"invalid JSON in uploaded file: {e}", "invalid_json")
            chosen_name = name or Path(file.filename or "").stem or "imported"
        else:
            try:
                body = await request.json()
            except json.JSONDecodeError:
                return _error(400, "invalid JSON body", "invalid_json")
            if not isinstance(body, dict):
                return _error(400, "JSON body must be an object", "invalid_body")
            chosen_name = body.get("name")
            if not chosen_name:
                return _error(400, "missing 'name' in JSON body", "missing_name")
            data = {k: body[k] for k in ("style_ttl", "style_dp") if k in body}

        try:
            target = styles_store.save(
                state.custom_styles_dir,
                chosen_name,
                data,
                builtin_names=state.tts.voice_style_names,
                overwrite=overwrite,
            )
        except styles_store.InvalidStyleName as e:
            return _error(400, str(e), "invalid_style_name")
        except styles_store.StyleNameConflict as e:
            status = 409 if "already exists" in str(e) else 400
            return _error(status, str(e), "style_name_conflict")
        except ValueError as e:
            return _error(400, str(e), "invalid_style_payload")

        state.custom_styles[target.stem] = target
        return StyleImportResponse(name=target.stem, stored_at=str(target))

    @router.post("/v1/tts")
    def synth_native(req: TTSRequest, request: Request):
        state = _state(request)
        if state.tts is None:
            return _error(503, "server not ready", "not_ready", type_="server_error")
        try:
            fmt = coerce_response_format(req.response_format)
        except UnsupportedAudioFormat as e:
            return _error(
                400,
                f"unsupported response_format {str(e)!r}",
                "unsupported_response_format",
            )
        err = _validate_lang(req.lang)
        if err is not None:
            return err
        try:
            wav, dur = _do_synthesize(
                state,
                text=req.text,
                voice=req.voice,
                lang=req.lang,
                speed=req.speed,
                steps=req.steps,
                max_chunk_length=req.max_chunk_length,
                silence_duration=req.silence_duration,
            )
        except UnknownVoice as e:
            return _error(400, f"unknown voice {str(e)!r}", "unknown_voice")
        except Exception as e:  # noqa: BLE001 — surface as 500 with code
            logger.exception("synthesis failed")
            return _error(500, f"synthesis failed: {e}", "synthesis_failed", type_="server_error")
        return _audio_response(state, wav, fmt, dur)

    @router.post("/v1/audio/speech")
    def openai_compat_speech(req: OpenAISpeechRequest, request: Request):
        # Validate ``model`` against AVAILABLE_MODELS but only *accept* the
        # model currently loaded — switching at request time is out of scope.
        state = _state(request)
        if req.model not in OpenAISpeechRequest.valid_models():
            return _error(
                400,
                f"unknown model {req.model!r}; valid: {', '.join(OpenAISpeechRequest.valid_models())}",
                "unknown_model",
            )
        if req.model != state.model:
            return _error(
                400,
                f"this server serves {state.model!r}; request asked for {req.model!r}. "
                f"Restart with --model {req.model} to switch.",
                "model_not_loaded",
            )
        if state.tts is None:
            return _error(503, "server not ready", "not_ready", type_="server_error")
        # OpenAI clients default to ``response_format='mp3'`` — surface a
        # clear error rather than silently emitting WAV.
        try:
            fmt = coerce_response_format(req.response_format)
        except UnsupportedAudioFormat as e:
            return _error(
                400,
                f"unsupported response_format {str(e)!r}; "
                f"set response_format to one of: {', '.join(SUPPORTED_FORMATS)}",
                "unsupported_response_format",
            )
        err = _validate_lang(req.lang)
        if err is not None:
            return err
        try:
            wav, dur = _do_synthesize(
                state,
                text=req.input,
                voice=req.voice,
                lang=req.lang,
                speed=req.speed,
                steps=None,
                max_chunk_length=None,
                silence_duration=None,
            )
        except UnknownVoice as e:
            return _error(400, f"unknown voice {str(e)!r}", "unknown_voice")
        except Exception as e:  # noqa: BLE001
            logger.exception("synthesis failed")
            return _error(500, f"synthesis failed: {e}", "synthesis_failed", type_="server_error")
        return _audio_response(state, wav, fmt, dur)

    @router.post("/v1/tts/batch", response_model=BatchResponse)
    def synth_batch(req: BatchRequest, request: Request):
        state = _state(request)
        if state.tts is None:
            return _error(503, "server not ready", "not_ready", type_="server_error")
        try:
            fmt = coerce_response_format(req.response_format)
        except UnsupportedAudioFormat as e:
            return _error(
                400,
                f"unsupported response_format {str(e)!r}",
                "unsupported_response_format",
            )
        defaults = req.defaults
        results: list[BatchResultItem] = []
        for idx, item in enumerate(req.items):
            voice = item.voice or (defaults.voice if defaults else None) or "M1"
            lang = item.lang or (defaults.lang if defaults else None)
            speed = item.speed if item.speed is not None else (defaults.speed if defaults else None)
            steps = item.steps if item.steps is not None else (defaults.steps if defaults else None)
            mcl = (
                item.max_chunk_length
                if item.max_chunk_length is not None
                else (defaults.max_chunk_length if defaults else None)
            )
            sil = (
                item.silence_duration
                if item.silence_duration is not None
                else (defaults.silence_duration if defaults else None)
            )
            if lang is not None and lang not in AVAILABLE_LANGUAGES:
                return _error(
                    400,
                    f"items[{idx}].lang: unsupported lang {lang!r}",
                    "unsupported_lang",
                )
            try:
                wav, dur = _do_synthesize(
                    state,
                    text=item.text,
                    voice=voice,
                    lang=lang,
                    speed=speed,
                    steps=steps,
                    max_chunk_length=mcl,
                    silence_duration=sil,
                )
            except UnknownVoice as e:
                return _error(
                    400,
                    f"items[{idx}]: unknown voice {str(e)!r}",
                    "unknown_voice",
                )
            except Exception as e:  # noqa: BLE001
                logger.exception("batch item %d synthesis failed", idx)
                return _error(
                    500,
                    f"items[{idx}]: synthesis failed: {e}",
                    "synthesis_failed",
                    type_="server_error",
                )
            body = encode_audio(wav, state.tts.sample_rate, fmt)
            results.append(
                BatchResultItem(
                    audio_base64=base64.b64encode(body).decode("ascii"),
                    duration_s=dur,
                    format=fmt,
                    sample_rate=state.tts.sample_rate,
                )
            )
        return BatchResponse(items=results)

    app.include_router(router)