Audio Fx

src.xil_pipeline.audio_fx

ffmpeg-backed audio treatments for dialogue stems.

The rest of the mixing pipeline uses pydub, which offers only single-pole high/low-pass filters, gain and fades. Treatments that need compression, saturation, parametric EQ or delay are built here instead, as ffmpeg filter graphs. ffmpeg is already a hard requirement (pydub shells out to it for decode/encode), so this adds no new dependency.

Audio is round-tripped through ffmpeg as raw PCM over pipes rather than temp files: there is no lossy intermediate re-encode, and no filesystem churn on drvfs-mounted checkouts where temp-file I/O is dramatically slower than a pipe.

Every treatment is length-preserving. Some ffmpeg filters (notably aecho) extend their output, and the pipeline derives cue positions from unfiltered MP3 header durations — a treatment that changed a stem's length would silently desync the rendered mix from the label/dry-run timeline. :func:run_ffmpeg_filter therefore trims or pads output back to the input length by default.

Failures degrade rather than abort. A DAW export runs ffmpeg once per treated stem across a whole episode; killing a multi-minute render because one invocation failed is worse than emitting an untreated stem plus a warning. Set XIL_STRICT_FX in the environment to raise :class:AudioFxError instead.

Module Attributes

TREATMENTS: Registry of named treatments, keyed by treatment name. STRICT_ENV_VAR: Environment variable that switches failures from warn-and-passthrough to raising :class:AudioFxError.

logger module-attribute

logger = get_logger(__name__)

STRICT_ENV_VAR module-attribute

STRICT_ENV_VAR: str = 'XIL_STRICT_FX'

FILM module-attribute

FILM = Treatment(name='film', summary='Warm, reflective 1990s indie film print — rolled-off top, recessed presence, tape-style saturation and an audible pink-noise grain.', graph='[0:a]highpass=f=90:poles=2,lowpass=f=5500:poles=2,lowshelf=f=180:g=3,equalizer=f=350:w=1.2:t=q:g=2,equalizer=f=2800:w=1.6:t=q:g=-6,highshelf=f=7000:g=-4,acompressor=threshold=-24dB:ratio=3.5:attack=12:release=280:makeup=2:knee=6,volume=12dB,asoftclip=type=tanh:param=1:oversample=4,volume=-12dB,volume=8.4dB[v];anoisesrc=color=pink:amplitude=0.018:sample_rate={rate},aformat=channel_layouts={layout}[n];[v][n]amix=inputs=2:duration=first:normalize=0[out]')

SPEAKERPHONE module-attribute

SPEAKERPHONE = Treatment(name='speakerphone', summary='Narrow-band speakerphone — steep 350 Hz/3.4 kHz skirts, hard AGC, odd-harmonic crunch and a short tabletop slap.', graph='[0:a]highpass=f=350:poles=2,highpass=f=350:poles=2,lowpass=f=3400:poles=2,lowpass=f=3400:poles=2,equalizer=f=700:w=1.0:t=q:g=-4,equalizer=f=1800:w=1.1:t=q:g=5,acompressor=threshold=-24dB:ratio=6:attack=5:release=120:makeup=3:knee=2,volume=8dB,asoftclip=type=atan:param=1:oversample=4,volume=-8dB,aecho=0.9:0.9:55:0.22,volume=13.6dB[out]')

PHONE module-attribute

PHONE = Treatment(name='phone', summary='Mobile call — steep 300 Hz/3.4 kHz skirts, earpiece presence lift, hard AGC and genuine GSM codec grit, sitting just under the room.', graph='[0:a]highpass=f=300:poles=2,highpass=f=300:poles=2,lowpass=f=3400:poles=2,lowpass=f=3400:poles=2,equalizer=f=500:w=1.0:t=q:g=-4,equalizer=f=1700:w=1.2:t=q:g=6,acompressor=threshold=-22dB:ratio=8:attack=3:release=90:makeup=3:knee=2,volume=6dB,asoftclip=type=atan:param=1:oversample=4,volume=-6dB,volume=10dB[out]', codec='libgsm', container='gsm', codec_rate=8000)

TREATMENTS module-attribute

TREATMENTS: dict[str, Treatment] = {t.name: t for t in (FILM, SPEAKERPHONE, PHONE)}

AudioFxError

Bases: RuntimeError

Raised when an ffmpeg treatment fails and strict mode is enabled.

Source code in src/xil_pipeline/audio_fx.py

class AudioFxError(RuntimeError):
    """Raised when an ffmpeg treatment fails and strict mode is enabled."""

Treatment dataclass

A named ffmpeg filter graph applied to dialogue stems.

Attributes:

• name (str) –

  Registry key, as written in a cast config filter field.

• graph (str) –

  ffmpeg -filter_complex graph producing a [out] pad. May contain {rate} and {layout} placeholders, substituted with the input segment's sample rate and channel layout.

• summary (str) –

  One-line description of the sound, for docs and logs.

• codec (str | None) –

  Optional ffmpeg encoder to round-trip the filtered audio through, for treatments whose character comes from real codec artifacts rather than from EQ alone. None skips the round-trip entirely, which is the behaviour every filter-only treatment relies on.

• container (str | None) –

  Muxer for that round-trip. Codec and container are not freely interchangeable — libgsm needs the raw gsm format and errors out inside a WAV container.

• codec_rate (int | None) –

  Sample rate to encode at. This is where a telephone treatment gets its hard ceiling: encoding at 8 kHz brickwalls at 4 kHz far more steeply than any practical filter cascade.

Source code in src/xil_pipeline/audio_fx.py

@dataclass(frozen=True)
class Treatment:
    """A named ffmpeg filter graph applied to dialogue stems.

    Attributes:
        name: Registry key, as written in a cast config ``filter`` field.
        graph: ffmpeg ``-filter_complex`` graph producing a ``[out]`` pad.
            May contain ``{rate}`` and ``{layout}`` placeholders, substituted
            with the input segment's sample rate and channel layout.
        summary: One-line description of the sound, for docs and logs.
        codec: Optional ffmpeg encoder to round-trip the filtered audio through,
            for treatments whose character comes from real codec artifacts
            rather than from EQ alone.  ``None`` skips the round-trip entirely,
            which is the behaviour every filter-only treatment relies on.
        container: Muxer for that round-trip.  Codec and container are not
            freely interchangeable — ``libgsm`` needs the raw ``gsm`` format and
            errors out inside a WAV container.
        codec_rate: Sample rate to encode at.  This is where a telephone
            treatment gets its hard ceiling: encoding at 8 kHz brickwalls at
            4 kHz far more steeply than any practical filter cascade.
    """

    name: str
    graph: str
    summary: str
    codec: str | None = None
    container: str | None = None
    codec_rate: int | None = None

name instance-attribute

name: str

graph instance-attribute

graph: str

summary instance-attribute

summary: str

codec class-attribute instance-attribute

codec: str | None = None

container class-attribute instance-attribute

container: str | None = None

codec_rate class-attribute instance-attribute

codec_rate: int | None = None

__init__

__init__(name: str, graph: str, summary: str, codec: str | None = None, container: str | None = None, codec_rate: int | None = None) -> None

ffmpeg_available

ffmpeg_available() -> bool

Report whether the configured ffmpeg binary can be executed.

Returns:

• bool –

  True if ffmpeg responds to -version, False otherwise.

Source code in src/xil_pipeline/audio_fx.py

def ffmpeg_available() -> bool:
    """Report whether the configured ffmpeg binary can be executed.

    Returns:
        True if ffmpeg responds to ``-version``, False otherwise.
    """
    try:
        proc = subprocess.run(
            [_ffmpeg_binary(), "-hide_banner", "-version"],
            capture_output=True,
            timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return False
    return proc.returncode == 0

encoder_available

encoder_available(name: str) -> bool

Report whether this ffmpeg build ships a given encoder.

ffmpeg builds differ in what they bundle, and the codec-backed treatments degrade quietly when an encoder is absent rather than failing. This is what lets a caller — a test, or someone wondering why a treatment sounds different on another machine — tell the two situations apart.

Notably libgsm is present in Debian's ffmpeg but not in the GitHub macOS or Windows runner builds, so phone renders band-limited but without its codec grit on those platforms.

Parameters:

• name (str) –

  Encoder name, e.g. "libgsm".

Returns:

• bool –

  True if ffmpeg lists the encoder, False if it does not or cannot run.

Source code in src/xil_pipeline/audio_fx.py

def encoder_available(name: str) -> bool:
    """Report whether this ffmpeg build ships a given encoder.

    ffmpeg builds differ in what they bundle, and the codec-backed treatments
    degrade quietly when an encoder is absent rather than failing.  This is what
    lets a caller — a test, or someone wondering why a treatment sounds
    different on another machine — tell the two situations apart.

    Notably ``libgsm`` is present in Debian's ffmpeg but not in the GitHub
    macOS or Windows runner builds, so ``phone`` renders band-limited but
    without its codec grit on those platforms.

    Args:
        name: Encoder name, e.g. ``"libgsm"``.

    Returns:
        True if ffmpeg lists the encoder, False if it does not or cannot run.
    """
    try:
        proc = subprocess.run(
            [_ffmpeg_binary(), "-hide_banner", "-encoders"],
            capture_output=True,
            timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return False
    if proc.returncode != 0:
        return False
    # Encoder rows are " V..... name  description"; match the name as a token
    # so a substring such as "gsm" cannot match "libgsm_ms".
    return name in proc.stdout.decode("utf-8", "replace").split()

missing_codecs

missing_codecs(names: Iterable[str]) -> dict[str, str]

Map each named treatment to the encoder it needs but cannot find.

A treatment whose codec is missing still renders — it degrades to the filter-only result — so nothing fails and the difference is easy to miss until two machines produce masters that do not match. This is what lets a render stage say so up front instead of one warning buried mid-run.

Parameters:

• names (Iterable[str]) –

  Treatment names actually in use for this render.

Returns:

• dict[str, str] –

  {treatment_name: encoder_name} for treatments that declare a codec

• dict[str, str] –

  this ffmpeg build does not ship. Empty when everything is available,

• dict[str, str] –

  including when no named treatment uses a codec at all.

Source code in src/xil_pipeline/audio_fx.py

def missing_codecs(names: Iterable[str]) -> dict[str, str]:
    """Map each named treatment to the encoder it needs but cannot find.

    A treatment whose codec is missing still renders — it degrades to the
    filter-only result — so nothing fails and the difference is easy to miss
    until two machines produce masters that do not match.  This is what lets a
    render stage say so up front instead of one warning buried mid-run.

    Args:
        names: Treatment names actually in use for this render.

    Returns:
        ``{treatment_name: encoder_name}`` for treatments that declare a codec
        this ffmpeg build does not ship.  Empty when everything is available,
        including when no named treatment uses a codec at all.
    """
    missing: dict[str, str] = {}
    for name in sorted(set(names)):
        treatment = TREATMENTS.get(name)
        if treatment is None or not treatment.codec:
            continue
        if treatment.codec in _encoder_probe_cache:
            available = _encoder_probe_cache[treatment.codec]
        else:
            # Each probe spawns ffmpeg, so cache per process: a render asks
            # about the same handful of codecs for every episode.
            available = encoder_available(treatment.codec)
            _encoder_probe_cache[treatment.codec] = available
        if not available:
            missing[name] = treatment.codec
    return missing

run_ffmpeg_filter

run_ffmpeg_filter(segment: AudioSegment, graph: str, *, preserve_length: bool = True, label: str = '', codec: str | None = None, container: str | None = None, codec_rate: int | None = None) -> AudioSegment

Push a segment through an ffmpeg -filter_complex graph.

Parameters:

• segment (AudioSegment) –

  Input audio.

• graph (str) –

  Filter graph producing an [out] pad. {rate} and {layout} placeholders are substituted from segment.

• preserve_length (bool, default: True ) –

  Trim or pad the result back to the input length. Leave enabled unless the caller genuinely wants a length change; the pipeline's cue timeline assumes stems keep their duration.

• label (str, default: '' ) –

  Treatment name, used in log messages and cache keys.

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

  Optional encoder to round-trip the filtered audio through, for treatments whose character is a real codec artifact.

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

  Muxer for that round-trip; required alongside codec.

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

  Sample rate to encode at, or None to keep the input's.

Returns:

• AudioSegment –

  The treated segment, or segment unchanged if ffmpeg is unavailable

• AudioSegment –

  or the invocation fails and strict mode is off.

Raises:

• AudioFxError –

  On failure when XIL_STRICT_FX is set.

Source code in src/xil_pipeline/audio_fx.py

def run_ffmpeg_filter(
    segment: AudioSegment,
    graph: str,
    *,
    preserve_length: bool = True,
    label: str = "",
    codec: str | None = None,
    container: str | None = None,
    codec_rate: int | None = None,
) -> AudioSegment:
    """Push a segment through an ffmpeg ``-filter_complex`` graph.

    Args:
        segment: Input audio.
        graph: Filter graph producing an ``[out]`` pad.  ``{rate}`` and
            ``{layout}`` placeholders are substituted from ``segment``.
        preserve_length: Trim or pad the result back to the input length.
            Leave enabled unless the caller genuinely wants a length change;
            the pipeline's cue timeline assumes stems keep their duration.
        label: Treatment name, used in log messages and cache keys.
        codec: Optional encoder to round-trip the filtered audio through, for
            treatments whose character is a real codec artifact.
        container: Muxer for that round-trip; required alongside *codec*.
        codec_rate: Sample rate to encode at, or ``None`` to keep the input's.

    Returns:
        The treated segment, or ``segment`` unchanged if ffmpeg is unavailable
        or the invocation fails and strict mode is off.

    Raises:
        AudioFxError: On failure when ``XIL_STRICT_FX`` is set.
    """
    raw_fmt = _RAW_FORMATS.get(segment.sample_width)
    if raw_fmt is None:
        _fail(
            label,
            "sample_width",
            "Unsupported sample width %d bytes for treatment %r",
            segment.sample_width,
            label,
        )
        return segment

    resolved = graph.format(
        rate=segment.frame_rate,
        layout="mono" if segment.channels == 1 else "stereo",
    )

    source = segment.raw_data
    cache_key = (
        hashlib.blake2b(source, digest_size=16).digest(),
        segment.frame_rate,
        segment.channels,
        segment.sample_width,
        resolved,
        preserve_length,
        codec,
        container,
        codec_rate,
    )
    cached = _fx_cache.get(cache_key)
    if cached is not None:
        return segment._spawn(cached)

    cmd = [
        _ffmpeg_binary(),
        "-hide_banner", "-loglevel", "error", "-nostdin", "-y",
        "-f", raw_fmt,
        "-ar", str(segment.frame_rate),
        "-ac", str(segment.channels),
        "-i", "pipe:0",
        "-filter_complex", resolved,
        "-map", "[out]",
        "-vn", "-sn", "-dn",
        "-f", raw_fmt,
        "-ar", str(segment.frame_rate),
        "-ac", str(segment.channels),
        "pipe:1",
    ]

    try:
        proc = subprocess.run(cmd, input=source, capture_output=True)
    except FileNotFoundError:
        _fail(label, "missing", "ffmpeg not found for treatment %r", label)
        return segment
    except OSError as exc:
        _fail(label, "oserror", "ffmpeg failed for treatment %r: %s", label, exc)
        return segment

    if proc.returncode != 0 or not proc.stdout:
        detail = proc.stderr.decode("utf-8", "replace").strip().splitlines()
        _fail(
            label,
            "returncode",
            "ffmpeg treatment %r failed (rc=%s): %s",
            label,
            proc.returncode,
            detail[-1] if detail else "no stderr output",
        )
        return segment

    out = proc.stdout
    if codec:
        # Degrades to the filter-only result, not to the dry input: losing the
        # codec should cost the grit, not the whole treatment.
        out = _codec_round_trip(
            out, segment, raw_fmt, label,
            codec=codec, container=container, codec_rate=codec_rate,
        )
    if preserve_length:
        out = _fit_length(out, len(source), segment.sample_width)

    if len(_fx_cache) >= _CACHE_MAX_ENTRIES:
        # FIFO eviction: dicts preserve insertion order.
        del _fx_cache[next(iter(_fx_cache))]
    _fx_cache[cache_key] = out

    return segment._spawn(out)

apply_treatment

apply_treatment(segment: AudioSegment, name: str) -> AudioSegment

Apply a named treatment from :data:TREATMENTS to a segment.

Parameters:

• segment (AudioSegment) –

  Input audio.

• name (str) –

  Treatment name, e.g. "film" or "speakerphone".

Returns:

• AudioSegment –

  The treated segment, or segment unchanged if the name is unknown

• AudioSegment –

  or the ffmpeg invocation fails outside strict mode.

Source code in src/xil_pipeline/audio_fx.py

def apply_treatment(segment: AudioSegment, name: str) -> AudioSegment:
    """Apply a named treatment from :data:`TREATMENTS` to a segment.

    Args:
        segment: Input audio.
        name: Treatment name, e.g. ``"film"`` or ``"speakerphone"``.

    Returns:
        The treated segment, or ``segment`` unchanged if the name is unknown
        or the ffmpeg invocation fails outside strict mode.
    """
    treatment = TREATMENTS.get(name)
    if treatment is None:
        _warn_once(
            name,
            "unknown",
            "Unknown ffmpeg treatment %r — known treatments: %s",
            name,
            ", ".join(sorted(TREATMENTS)),
        )
        return segment
    return run_ffmpeg_filter(
        segment,
        treatment.graph,
        label=treatment.name,
        codec=treatment.codec,
        container=treatment.container,
        codec_rate=treatment.codec_rate,
    )

clear_cache

clear_cache() -> None

Empty the treatment result cache.

Intended for tests; the cache is otherwise self-limiting.

Source code in src/xil_pipeline/audio_fx.py

def clear_cache() -> None:
    """Empty the treatment result cache.

    Intended for tests; the cache is otherwise self-limiting.
    """
    _fx_cache.clear()
    _encoder_probe_cache.clear()