Skip to content

Sfx Backends

src.xil_pipeline.sfx_backends

Pluggable backends for SFX / music / ambience asset generation.

The pipeline generates every non-silence sound effect through the ElevenLabs Sound Effects API. This module keeps a thin :class:SfxBackend adapter so the shared generation path in :mod:xil_pipeline.sfx_common does not talk to the ElevenLabs client directly:

  • :class:ElevenLabsSfxBackend — wraps client.text_to_sound_effects.convert with stream-to-temp, atomic rename, and 429 / 5xx / network retry handling.

Two local diffusion backends (AudioLDM 2, Stable Audio Open) were removed in #62 after both trials produced unusable audio. The adapter and factory survive them deliberately — they are the seam a future backend plugs into, and collapsing them into a direct ElevenLabs call would have to be undone to add one.

The contract is::

backend.generate_to(out_path, prompt, duration_seconds, prompt_influence)
backend.close()

Use :func:make_sfx_backend to construct the right backend from a CLI flag.

logger module-attribute 

logger = get_logger(__name__)

SfxBackend

Bases: Protocol

Minimal contract for a sound-effect generation backend.

Source code in src/xil_pipeline/sfx_backends.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@runtime_checkable
class SfxBackend(Protocol):
    """Minimal contract for a sound-effect generation backend."""

    name: str

    def generate_to(
        self,
        out_path: str,
        prompt: str,
        duration_seconds: float,
        prompt_influence: float,
    ) -> None:
        """Generate audio for *prompt* and write it to *out_path*."""
        ...

    def close(self) -> None:
        """Release any resources (subprocess, sockets). No-op for stateless backends."""
        ...

name instance-attribute 

name: str

generate_to 

generate_to(out_path: str, prompt: str, duration_seconds: float, prompt_influence: float) -> None

Generate audio for prompt and write it to out_path.

Source code in src/xil_pipeline/sfx_backends.py 
54
55
56
57
58
59
60
61
62
def generate_to(
    self,
    out_path: str,
    prompt: str,
    duration_seconds: float,
    prompt_influence: float,
) -> None:
    """Generate audio for *prompt* and write it to *out_path*."""
    ...

close 

close() -> None

Release any resources (subprocess, sockets). No-op for stateless backends.

Source code in src/xil_pipeline/sfx_backends.py 
64
65
66
def close(self) -> None:
    """Release any resources (subprocess, sockets). No-op for stateless backends."""
    ...

ElevenLabsSfxBackend

SFX backend backed by the ElevenLabs Sound Effects API.

Wraps client.text_to_sound_effects.convert with the streaming download, atomic temp-file rename, and retry behaviour that previously lived inline in :func:xil_pipeline.sfx_common.ensure_shared_sfx. Retries 429 (rate limit), 5xx (server error), and network transport errors up to five times with linear backoff (10s, 20s, …).

Source code in src/xil_pipeline/sfx_backends.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
class ElevenLabsSfxBackend:
    """SFX backend backed by the ElevenLabs Sound Effects API.

    Wraps ``client.text_to_sound_effects.convert`` with the streaming
    download, atomic temp-file rename, and retry behaviour that previously
    lived inline in :func:`xil_pipeline.sfx_common.ensure_shared_sfx`.
    Retries 429 (rate limit), 5xx (server error), and network transport
    errors up to five times with linear backoff (10s, 20s, …).
    """

    name = "elevenlabs"

    def __init__(self, client) -> None:
        self._client = client

    def generate_to(
        self,
        out_path: str,
        prompt: str,
        duration_seconds: float,
        prompt_influence: float,
    ) -> None:
        if self._client is None:
            raise ValueError(
                "ElevenLabs client is required to generate SFX "
                "(set ELEVENLABS_API_KEY)."
            )
        logger.info("   [api] text-to-sound-effects → %r (%.1fs)", prompt, duration_seconds)
        tmp_path = None
        try:
            max_retries, delay = 5, 10
            for attempt in range(1, max_retries + 1):
                try:
                    audio_stream = self._client.text_to_sound_effects.convert(
                        text=prompt,
                        duration_seconds=duration_seconds,
                        prompt_influence=prompt_influence,
                    )
                    tmp_fd, tmp_path = tempfile.mkstemp(
                        dir=os.path.dirname(out_path) or ".", suffix=".tmp"
                    )
                    with os.fdopen(tmp_fd, "wb") as f:
                        for chunk in audio_stream:
                            if chunk:
                                f.write(chunk)
                    os.replace(tmp_path, out_path)
                    tmp_path = None
                    logger.info("   [api] saved %s", os.path.basename(out_path))
                    return
                except (ApiError, httpx.TransportError) as exc:
                    if tmp_path is not None:
                        with contextlib.suppress(FileNotFoundError):
                            os.unlink(tmp_path)
                        tmp_path = None
                    is_rate_limit = isinstance(exc, ApiError) and exc.status_code == 429
                    is_server_error = (
                        isinstance(exc, ApiError)
                        and exc.status_code is not None
                        and exc.status_code >= 500
                    )
                    is_network_error = isinstance(exc, httpx.TransportError)
                    is_retryable = is_rate_limit or is_server_error or is_network_error
                    if is_retryable and attempt < max_retries:
                        wait = delay * attempt
                        if is_rate_limit:
                            reason = "429 rate limited"
                        elif is_server_error:
                            reason = f"{exc.status_code} server error"
                        else:
                            reason = f"network error ({type(exc).__name__})"
                        logger.warning(
                            "[%s] — retrying in %ds (attempt %d/%d)",
                            reason, wait, attempt, max_retries,
                        )
                        time.sleep(wait)
                    else:
                        raise
        finally:
            if tmp_path is not None:
                with contextlib.suppress(FileNotFoundError):
                    os.unlink(tmp_path)

    def close(self) -> None:
        # The ElevenLabs client is owned by the caller; nothing to release here.
        return

name class-attribute instance-attribute 

name = 'elevenlabs'

__init__ 

__init__(client) -> None
Source code in src/xil_pipeline/sfx_backends.py 
84
85
def __init__(self, client) -> None:
    self._client = client

generate_to 

generate_to(out_path: str, prompt: str, duration_seconds: float, prompt_influence: float) -> None
Source code in src/xil_pipeline/sfx_backends.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def generate_to(
    self,
    out_path: str,
    prompt: str,
    duration_seconds: float,
    prompt_influence: float,
) -> None:
    if self._client is None:
        raise ValueError(
            "ElevenLabs client is required to generate SFX "
            "(set ELEVENLABS_API_KEY)."
        )
    logger.info("   [api] text-to-sound-effects → %r (%.1fs)", prompt, duration_seconds)
    tmp_path = None
    try:
        max_retries, delay = 5, 10
        for attempt in range(1, max_retries + 1):
            try:
                audio_stream = self._client.text_to_sound_effects.convert(
                    text=prompt,
                    duration_seconds=duration_seconds,
                    prompt_influence=prompt_influence,
                )
                tmp_fd, tmp_path = tempfile.mkstemp(
                    dir=os.path.dirname(out_path) or ".", suffix=".tmp"
                )
                with os.fdopen(tmp_fd, "wb") as f:
                    for chunk in audio_stream:
                        if chunk:
                            f.write(chunk)
                os.replace(tmp_path, out_path)
                tmp_path = None
                logger.info("   [api] saved %s", os.path.basename(out_path))
                return
            except (ApiError, httpx.TransportError) as exc:
                if tmp_path is not None:
                    with contextlib.suppress(FileNotFoundError):
                        os.unlink(tmp_path)
                    tmp_path = None
                is_rate_limit = isinstance(exc, ApiError) and exc.status_code == 429
                is_server_error = (
                    isinstance(exc, ApiError)
                    and exc.status_code is not None
                    and exc.status_code >= 500
                )
                is_network_error = isinstance(exc, httpx.TransportError)
                is_retryable = is_rate_limit or is_server_error or is_network_error
                if is_retryable and attempt < max_retries:
                    wait = delay * attempt
                    if is_rate_limit:
                        reason = "429 rate limited"
                    elif is_server_error:
                        reason = f"{exc.status_code} server error"
                    else:
                        reason = f"network error ({type(exc).__name__})"
                    logger.warning(
                        "[%s] — retrying in %ds (attempt %d/%d)",
                        reason, wait, attempt, max_retries,
                    )
                    time.sleep(wait)
                else:
                    raise
    finally:
        if tmp_path is not None:
            with contextlib.suppress(FileNotFoundError):
                os.unlink(tmp_path)

close 

close() -> None
Source code in src/xil_pipeline/sfx_backends.py 
154
155
156
def close(self) -> None:
    # The ElevenLabs client is owned by the caller; nothing to release here.
    return

MMAudioSfxBackend

Local SFX backend backed by MMAudio (text-to-audio mode).

The model weights are CC BY-NC 4.0 — non-commercial use only. The code is MIT, the checkpoints are not. Audio produced here must not end up in a monetised episode. Construction therefore requires an explicit accept_noncommercial=True, every session logs the constraint, and generated assets carry it in their ID3 comment (:attr:asset_comment) on top of the .mmaudio filename infix that :func:xil_pipeline.sfx_common.shared_sfx_path already applies. Between the two, an asset stays identifiable even if it is renamed.

Duration handling is the interesting part. MMAudio is trained at 8 seconds and the project warns that a large deviation degrades quality, but SFX cues here are typically shorter. So generation always runs at the native duration and the result is trimmed afterwards to the caller's duration_seconds. The trim cannot be left to the mixer: for a prompt-generated cue duration_seconds is the requested generation length and there is no mix-time clip (that only applies to source= cues), so an untrimmed asset would simply play long.

Source code in src/xil_pipeline/sfx_backends.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class MMAudioSfxBackend:
    """Local SFX backend backed by MMAudio (text-to-audio mode).

    **The model weights are CC BY-NC 4.0 — non-commercial use only.**  The code
    is MIT, the checkpoints are not.  Audio produced here must not end up in a
    monetised episode.  Construction therefore requires an explicit
    ``accept_noncommercial=True``, every session logs the constraint, and
    generated assets carry it in their ID3 comment (:attr:`asset_comment`) on
    top of the ``.mmaudio`` filename infix that
    :func:`xil_pipeline.sfx_common.shared_sfx_path` already applies.  Between
    the two, an asset stays identifiable even if it is renamed.

    Duration handling is the interesting part.  MMAudio is trained at 8 seconds
    and the project warns that a large deviation degrades quality, but SFX cues
    here are typically shorter.  So generation always runs at the native
    duration and the result is **trimmed afterwards** to the caller's
    ``duration_seconds``.  The trim cannot be left to the mixer: for a
    prompt-generated cue ``duration_seconds`` is the *requested generation
    length* and there is no mix-time clip (that only applies to ``source=``
    cues), so an untrimmed asset would simply play long.
    """

    name = "mmaudio"

    #: Written into every generated asset's ID3 comment.
    asset_comment = (
        "Generated by MMAudio (hkchengrex/MMAudio). Model weights are "
        "CC BY-NC 4.0 — NON-COMMERCIAL USE ONLY."
    )

    #: MMAudio's training duration. Generating here and trimming beats asking
    #: the model for a short clip directly.
    NATIVE_DURATION_S = 8.0

    def __init__(self, client: _MMAudioClient, *,
                 accept_noncommercial: bool = False,
                 native_duration: float = NATIVE_DURATION_S) -> None:
        if not accept_noncommercial:
            raise ValueError(
                "MMAudio weights are CC BY-NC 4.0 (non-commercial only). Pass "
                "--mmaudio-accept-noncommercial to acknowledge that generated "
                "audio must not be used in a monetised production."
            )
        self._client = client
        self._native_duration = native_duration
        logger.warning(
            "MMAudio weights are CC BY-NC 4.0 — NON-COMMERCIAL USE ONLY. "
            "Generated assets are tagged .mmaudio and carry the notice in ID3."
        )

    def generate_to(
        self,
        out_path: str,
        prompt: str,
        duration_seconds: float,
        prompt_influence: float,
    ) -> None:
        """Generate at MMAudio's native duration, then trim to *duration_seconds*.

        ``prompt_influence`` has no direct MMAudio analogue; it is not silently
        dropped — the client's ``cfg_strength`` plays the equivalent role and is
        set from ``--mmaudio-cfg``.
        """
        target = max(0.0, float(duration_seconds or 0.0))
        gen_seconds = max(self._native_duration, target)
        logger.info(
            "   [mmaudio] %r — generating %.1fs (native), trimming to %.1fs",
            prompt, gen_seconds, target or gen_seconds,
        )
        self._client.generate(prompt, out_path, gen_seconds)
        if target and target < gen_seconds:
            _trim_audio_file(out_path, target)

    def close(self) -> None:
        self._client.close()

name class-attribute instance-attribute 

name = 'mmaudio'

asset_comment class-attribute instance-attribute 

asset_comment = 'Generated by MMAudio (hkchengrex/MMAudio). Model weights are CC BY-NC 4.0 — NON-COMMERCIAL USE ONLY.'

NATIVE_DURATION_S class-attribute instance-attribute 

NATIVE_DURATION_S = 8.0

__init__ 

__init__(client: _MMAudioClient, *, accept_noncommercial: bool = False, native_duration: float = NATIVE_DURATION_S) -> None
Source code in src/xil_pipeline/sfx_backends.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def __init__(self, client: _MMAudioClient, *,
             accept_noncommercial: bool = False,
             native_duration: float = NATIVE_DURATION_S) -> None:
    if not accept_noncommercial:
        raise ValueError(
            "MMAudio weights are CC BY-NC 4.0 (non-commercial only). Pass "
            "--mmaudio-accept-noncommercial to acknowledge that generated "
            "audio must not be used in a monetised production."
        )
    self._client = client
    self._native_duration = native_duration
    logger.warning(
        "MMAudio weights are CC BY-NC 4.0 — NON-COMMERCIAL USE ONLY. "
        "Generated assets are tagged .mmaudio and carry the notice in ID3."
    )

generate_to 

generate_to(out_path: str, prompt: str, duration_seconds: float, prompt_influence: float) -> None

Generate at MMAudio's native duration, then trim to duration_seconds.

prompt_influence has no direct MMAudio analogue; it is not silently dropped — the client's cfg_strength plays the equivalent role and is set from --mmaudio-cfg.

Source code in src/xil_pipeline/sfx_backends.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def generate_to(
    self,
    out_path: str,
    prompt: str,
    duration_seconds: float,
    prompt_influence: float,
) -> None:
    """Generate at MMAudio's native duration, then trim to *duration_seconds*.

    ``prompt_influence`` has no direct MMAudio analogue; it is not silently
    dropped — the client's ``cfg_strength`` plays the equivalent role and is
    set from ``--mmaudio-cfg``.
    """
    target = max(0.0, float(duration_seconds or 0.0))
    gen_seconds = max(self._native_duration, target)
    logger.info(
        "   [mmaudio] %r — generating %.1fs (native), trimming to %.1fs",
        prompt, gen_seconds, target or gen_seconds,
    )
    self._client.generate(prompt, out_path, gen_seconds)
    if target and target < gen_seconds:
        _trim_audio_file(out_path, target)

close 

close() -> None
Source code in src/xil_pipeline/sfx_backends.py 
365
366
def close(self) -> None:
    self._client.close()

make_sfx_backend 

make_sfx_backend(name: str, client: ElevenLabs | None = None, *, mmaudio_python: str | None = None, device: str = 'cuda', mmaudio_cfg: float = 4.5, mmaudio_steps: int = 25, mmaudio_negative_prompt: str = '', mmaudio_seed: int | None = None, mmaudio_duration: float = MMAudioSfxBackend.NATIVE_DURATION_S, accept_noncommercial: bool = False) -> SfxBackend

Construct an :class:SfxBackend for the given backend name.

"elevenlabs" calls the Sound Effects API; "mmaudio" runs MMAudio locally in venv-mmaudio (CC BY-NC 4.0 weights — non-commercial only, hence accept_noncommercial).

The factory is kept rather than inlined because it is the seam a backend plugs into — the audioldm2/stableaudio trials were removed through it in #62 and MMAudio was added through it in #64 without touching call sites.

Parameters:

  • name (str) –

    "elevenlabs" or "mmaudio".

  • client (ElevenLabs | None, default: None ) –

    ElevenLabs client (used only for "elevenlabs").

  • mmaudio_python (str | None, default: None ) –

    Explicit venv-mmaudio interpreter; auto-detected when None.

  • device (str, default: 'cuda' ) –

    "cuda" (default) or "cpu" for the local backend.

  • mmaudio_cfg (float, default: 4.5 ) –

    Classifier-free guidance strength.

  • mmaudio_steps (int, default: 25 ) –

    Flow-matching sampling steps.

  • mmaudio_negative_prompt (str, default: '' ) –

    Optional negative prompt.

  • mmaudio_seed (int | None, default: None ) –

    Reproducibility seed (None = nondeterministic).

  • mmaudio_duration (float, default: NATIVE_DURATION_S ) –

    Generation length before trimming (default: the 8 s training duration).

  • accept_noncommercial (bool, default: False ) –

    Required acknowledgement of the CC BY-NC weights.

Returns:

Raises:

  • ValueError –

    For an unknown name, or for "mmaudio" without accept_noncommercial — a stale script must fail loudly rather than quietly producing audio that cannot be used commercially.

Source code in src/xil_pipeline/sfx_backends.py 
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
def make_sfx_backend(
    name: str,
    client: ElevenLabs | None = None,
    *,
    mmaudio_python: str | None = None,
    device: str = "cuda",
    mmaudio_cfg: float = 4.5,
    mmaudio_steps: int = 25,
    mmaudio_negative_prompt: str = "",
    mmaudio_seed: int | None = None,
    mmaudio_duration: float = MMAudioSfxBackend.NATIVE_DURATION_S,
    accept_noncommercial: bool = False,
) -> SfxBackend:
    """Construct an :class:`SfxBackend` for the given backend *name*.

    ``"elevenlabs"`` calls the Sound Effects API; ``"mmaudio"`` runs MMAudio
    locally in ``venv-mmaudio`` (**CC BY-NC 4.0 weights — non-commercial only**,
    hence ``accept_noncommercial``).

    The factory is kept rather than inlined because it is the seam a backend
    plugs into — the audioldm2/stableaudio trials were removed through it in #62
    and MMAudio was added through it in #64 without touching call sites.

    Args:
        name: ``"elevenlabs"`` or ``"mmaudio"``.
        client: ElevenLabs client (used only for ``"elevenlabs"``).
        mmaudio_python: Explicit venv-mmaudio interpreter; auto-detected when ``None``.
        device: ``"cuda"`` (default) or ``"cpu"`` for the local backend.
        mmaudio_cfg: Classifier-free guidance strength.
        mmaudio_steps: Flow-matching sampling steps.
        mmaudio_negative_prompt: Optional negative prompt.
        mmaudio_seed: Reproducibility seed (``None`` = nondeterministic).
        mmaudio_duration: Generation length before trimming (default: the 8 s
            training duration).
        accept_noncommercial: Required acknowledgement of the CC BY-NC weights.

    Returns:
        A ready-to-use backend instance.

    Raises:
        ValueError: For an unknown name, or for ``"mmaudio"`` without
            ``accept_noncommercial`` — a stale script must fail loudly rather
            than quietly producing audio that cannot be used commercially.
    """
    if name == "elevenlabs":
        return ElevenLabsSfxBackend(client)
    if name == "mmaudio":
        # Check the licence acknowledgement BEFORE resolving the venv: someone
        # who has not accepted the CC BY-NC terms should be told that, not sent
        # off to install a venv they may then be unable to use.
        if not accept_noncommercial:
            raise ValueError(
                "MMAudio weights are CC BY-NC 4.0 (non-commercial only). Pass "
                "--mmaudio-accept-noncommercial to acknowledge that generated "
                "audio must not be used in a monetised production."
            )
        worker = _MMAudioClient(
            python_path=_find_mmaudio_python(mmaudio_python),
            device=device,
            guidance=mmaudio_cfg,
            steps=mmaudio_steps,
            negative_prompt=mmaudio_negative_prompt,
            seed=mmaudio_seed,
        )
        return MMAudioSfxBackend(
            worker,
            accept_noncommercial=accept_noncommercial,
            native_duration=mmaudio_duration,
        )
    raise ValueError(f"Unknown sfx backend: {name!r}")