Audio Spectrum

FFT bars over a synthesised arpeggio

▶ Run in browser

Tags: audio spectrum fft synthesis

An AudioSynth-baked arpeggio loops on an AudioPlayer while a log-spaced bar spectrum dances in time with it. Each frame the demo reads get_playback_position(), slices a 2048-sample window out of the clip’s own PCM buffer (backend_data), and runs np.fft.rfft over it. The engine has no audio output tap, so this analyses the source buffer the example itself generated, synchronised by playback position; it cannot analyse file-backed streams (their backend_data is None).

What it demonstrates

• AudioSynth voices (oscillators + ADSR envelopes + a LowPass drone) baked into an AudioClip via bake() / AudioClip.from_pcm.

• backend_data as a float32 interleaved numpy buffer, indexed at int(get_playback_position() * sample_rate) frames during playback.

• A windowed rfft per frame, folded into log-spaced bands with peak-hold caps, drawn with the immediate-mode 2D renderer.

Controls: SPACE - Pause / resume ESC - Quit

Run: uv run python examples/features/audio/spectrum.py Headless self-check: uv run python examples/features/audio/spectrum.py –test

Source

"""Audio Spectrum: FFT bars over a synthesised arpeggio

An AudioSynth-baked arpeggio loops on an AudioPlayer while a log-spaced bar
spectrum dances in time with it. Each frame the demo reads
get_playback_position(), slices a 2048-sample window out of the clip's own
PCM buffer (backend_data), and runs np.fft.rfft over it. The engine has no
audio output tap, so this analyses the source buffer the example itself
generated, synchronised by playback position; it cannot analyse file-backed
streams (their backend_data is None).

# /// simvx
# tags = ["audio", "spectrum", "fft", "synthesis"]
# ///

## What it demonstrates
- AudioSynth voices (oscillators + ADSR envelopes + a LowPass drone) baked
  into an AudioClip via bake() / AudioClip.from_pcm.
- backend_data as a float32 interleaved numpy buffer, indexed at
  int(get_playback_position() * sample_rate) frames during playback.
- A windowed rfft per frame, folded into log-spaced bands with peak-hold
  caps, drawn with the immediate-mode 2D renderer.

Controls:
  SPACE - Pause / resume
  ESC - Quit

Run: uv run python examples/features/audio/spectrum.py
Headless self-check: uv run python examples/features/audio/spectrum.py --test
"""

import numpy as np

from simvx.core import (
    ADSR,
    AudioClip,
    AudioPlayer,
    AudioSynth,
    Input,
    Key,
    LowPass,
    Node,
    Oscillator,
)
from simvx.graphics import App

WIDTH, HEIGHT = 960, 540

SAMPLE_RATE = 44100
CHANNELS = 2

# An A-minor arpeggio, up and back: A3 C4 E4 A4 C5 A4 E4 C4.
ARPEGGIO_HZ = (220.0, 261.63, 329.63, 440.0, 523.25, 440.0, 329.63, 261.63)
STEP_SECONDS = 0.25
DRONE_HZ = 110.0  # A2, held under the whole loop

FFT_WINDOW = 2048  # ~46 ms of context per frame
BANDS = 48
F_MIN, F_MAX = 60.0, 8000.0
FLOOR_DB = -60.0  # magnitudes at or below this floor draw as zero-height bars

BAR_FALL_PER_S = 1.8  # bars rise instantly, fall at this fraction of full scale
PEAK_FALL_PER_S = 0.35  # peak-hold caps sink slower, tracing the recent maxima

PLOT_LEFT, PLOT_RIGHT = 50, 30
PLOT_TOP, PLOT_BOTTOM = 90, 60
LABEL_HZ = (100, 250, 500, 1000, 2000, 4000, 8000)


def bake_arpeggio() -> AudioClip:
    """Bake the looping demo clip: per-note melody steps over a bass drone.

    Each melody step is its own bake so the ADSR spans exactly one note;
    the release reaches zero at both edges, so concatenating the steps is
    click-free. The drone bakes once across the full duration, and at
    110 Hz over 2.0 s it completes a whole number of cycles, so the loop
    point is seamless too.
    """
    steps = []
    for freq in ARPEGGIO_HZ:
        synth = AudioSynth()
        env = ADSR(attack=0.02, decay=0.06, sustain=0.7, release=0.08)
        synth.add(Oscillator.sine(freq), envelope=env, gain=0.5)
        synth.add(Oscillator.triangle(freq * 2), envelope=env, gain=0.15)
        steps.append(synth.bake(STEP_SECONDS, sample_rate=SAMPLE_RATE, channels=CHANNELS).backend_data)
    melody = np.concatenate(steps)

    drone = AudioSynth()
    drone.add(Oscillator.saw(DRONE_HZ), filter=LowPass(400.0), gain=0.25)
    total = STEP_SECONDS * len(ARPEGGIO_HZ)
    drone_pcm = drone.bake(total, sample_rate=SAMPLE_RATE, channels=CHANNELS).backend_data

    pcm = np.clip(melody + drone_pcm, -1.0, 1.0).astype(np.float32)
    return AudioClip.from_pcm(pcm, sample_rate=SAMPLE_RATE, channels=CHANNELS, name="spectrum_arpeggio")


class SpectrumAnalyser:
    """FFT band magnitudes over a PCM buffer, addressed by playback position.

    Owns the mono mixdown of an interleaved float32 buffer plus the band
    layout, so ``magnitudes(position)`` is the whole per-frame cost: one
    windowed rfft and a fold into log-spaced bands.
    """

    def __init__(
        self,
        pcm: np.ndarray,
        *,
        sample_rate: int,
        channels: int,
        bands: int = BANDS,
        window: int = FFT_WINDOW,
        f_min: float = F_MIN,
        f_max: float = F_MAX,
        floor_db: float = FLOOR_DB,
    ):
        # Interleaved frames -> mono. Indexing this by frame is the same
        # as indexing the raw buffer at frame * channels.
        self._mono = pcm.reshape(-1, channels).mean(axis=1).astype(np.float32)
        self._rate = sample_rate
        self._window = window
        self._taper = np.hanning(window).astype(np.float32)
        self._floor_db = floor_db
        self.bands = bands

        self._freqs = np.fft.rfftfreq(window, 1.0 / sample_rate)
        self.band_edges_hz = np.geomspace(f_min, f_max, bands + 1)
        self._bin_edges = np.searchsorted(self._freqs, self.band_edges_hz)

    @property
    def bin_width_hz(self) -> float:
        return self._rate / self._window

    def _spectrum(self, position: float) -> np.ndarray:
        """Magnitude spectrum of the window centred on `position` seconds."""
        centre = int(position * self._rate)  # frame index into the buffer
        offsets = np.arange(self._window) + centre - self._window // 2
        chunk = self._mono[offsets % self._mono.shape[0]]  # wrap: the clip loops
        # Hann window, then normalise so a full-scale sine peaks near 1.0.
        return np.abs(np.fft.rfft(chunk * self._taper)) / (self._window / 4.0)

    def magnitudes(self, position: float) -> np.ndarray:
        """Per-band heights in [0, 1]: dB magnitude above the floor."""
        spectrum = self._spectrum(position)
        out = np.empty(self.bands, dtype=np.float32)
        for i in range(self.bands):
            lo = self._bin_edges[i]
            hi = max(lo + 1, self._bin_edges[i + 1])
            out[i] = spectrum[lo:hi].max()
        db = 20.0 * np.log10(out + 1e-9)
        return np.clip((db - self._floor_db) / -self._floor_db, 0.0, 1.0)

    def peak_hz(self, position: float, *, lo_hz: float = 0.0, hi_hz: float | None = None) -> float:
        """Frequency of the strongest bin within [lo_hz, hi_hz]."""
        spectrum = self._spectrum(position)
        mask = self._freqs >= lo_hz
        if hi_hz is not None:
            mask &= self._freqs <= hi_hz
        idx = np.flatnonzero(mask)
        return float(self._freqs[idx[np.argmax(spectrum[idx])]])


class SpectrumDemo(Node):
    """Looping arpeggio with a live log-spaced FFT bar display."""

    dynamic = True  # the bars animate every frame

    input_actions = {"quit": [Key.ESCAPE], "pause": [Key.SPACE]}

    def on_ready(self):
        clip = bake_arpeggio()
        self._analyser = SpectrumAnalyser(clip.backend_data, sample_rate=SAMPLE_RATE, channels=CHANNELS)
        self._bars = np.zeros(BANDS, dtype=np.float32)
        self._peaks = np.zeros(BANDS, dtype=np.float32)

        self._player = self.add_child(AudioPlayer(name="Player"))
        self._player.loop = True  # takes effect on the next play(), so set first
        self._player.stream = clip
        self._player.play()

    def on_update(self, dt: float):
        if Input.is_action_just_pressed("quit"):
            self.app.quit()
            return
        if Input.is_action_just_pressed("pause"):
            if self._player.is_playing():
                self._player.pause()
            elif self._player.is_paused():
                self._player.play()

        if self._player.is_playing():
            target = self._analyser.magnitudes(self._player.get_playback_position())
        else:
            target = 0.0
        self._bars = np.maximum(target, self._bars - BAR_FALL_PER_S * dt)
        self._peaks = np.maximum(self._bars, self._peaks - PEAK_FALL_PER_S * dt)

    def _band_colour(self, i: int) -> tuple[float, float, float, float]:
        t = i / (BANDS - 1)
        return (0.25 + 0.75 * t, 0.55 - 0.15 * t, 1.0 - 0.7 * t, 0.95)

    def on_draw(self, renderer):
        plot_w = WIDTH - PLOT_LEFT - PLOT_RIGHT
        plot_h = HEIGHT - PLOT_TOP - PLOT_BOTTOM
        base_y = HEIGHT - PLOT_BOTTOM

        gap = 3.0
        bar_w = (plot_w - gap * (BANDS - 1)) / BANDS
        for i in range(BANDS):
            x = PLOT_LEFT + i * (bar_w + gap)
            h = float(self._bars[i]) * plot_h
            if h > 1.0:
                renderer.draw_rect((x, base_y - h), (bar_w, h), colour=self._band_colour(i), filled=True)
            peak_y = base_y - float(self._peaks[i]) * plot_h
            renderer.draw_rect((x, peak_y - 2), (bar_w, 2), colour=(0.95, 0.95, 1.0, 0.8), filled=True)

        renderer.draw_line((PLOT_LEFT, base_y), (WIDTH - PLOT_RIGHT, base_y), colour=(0.5, 0.5, 0.6, 1.0))

        # Frequency labels along the log axis.
        span = np.log(F_MAX / F_MIN)
        for hz in LABEL_HZ:
            x = PLOT_LEFT + plot_w * np.log(hz / F_MIN) / span
            label = f"{hz // 1000}k" if hz >= 1000 else str(hz)
            renderer.draw_line((x, base_y), (x, base_y + 6), colour=(0.5, 0.5, 0.6, 1.0))
            renderer.draw_text(label, (x - 4 * len(label), base_y + 12), colour=(0.6, 0.6, 0.7))

        renderer.draw_text("Audio Spectrum", (PLOT_LEFT, 18), colour=(1.0, 1.0, 1.0), scale=2)
        state = "playing" if self._player.is_playing() else "paused"
        pos = self._player.get_playback_position()
        renderer.draw_text(
            f"Synthesised arpeggio, {state} at {pos:4.2f}s   SPACE: pause   ESC: quit",
            (PLOT_LEFT, 52),
            colour=(0.7, 0.7, 0.75),
        )


def _selftest() -> bool:
    """Logic-level checks on the baked clip and the analyser: no window, no audio device."""
    ok = True

    def check(label: str, passed: bool, detail: str) -> None:
        nonlocal ok
        ok = ok and passed
        print(f"{'ok  ' if passed else 'FAIL'} {label}: {detail}")

    clip = bake_arpeggio()
    pcm = clip.backend_data
    frames_per_step = int(STEP_SECONDS * SAMPLE_RATE)
    expected = frames_per_step * len(ARPEGGIO_HZ) * CHANNELS

    check(
        "bake produces a float32 interleaved buffer",
        isinstance(pcm, np.ndarray) and pcm.dtype == np.float32 and pcm.size == expected,
        f"dtype={getattr(pcm, 'dtype', None)}, size={getattr(pcm, 'size', None)} (expected {expected})",
    )
    check(
        "the clip carries its rate and channel count",
        clip.sample_rate == SAMPLE_RATE and clip.channels == CHANNELS,
        f"sample_rate={clip.sample_rate}, channels={clip.channels}",
    )

    analyser = SpectrumAnalyser(pcm, sample_rate=SAMPLE_RATE, channels=CHANNELS)
    tolerance = 2.0 * analyser.bin_width_hz

    # At the middle of each step the ADSR is in its sustain phase, so the
    # dominant frequency above the drone's register must be that step's note.
    misses = []
    for i, freq in enumerate(ARPEGGIO_HZ):
        pos = (i + 0.5) * STEP_SECONDS
        peak = analyser.peak_hz(pos, lo_hz=150.0)
        if abs(peak - freq) > tolerance:
            misses.append(f"step {i}: {peak:.0f}Hz for {freq:.0f}Hz")
    check(
        "each arpeggio note dominates its own step",
        not misses,
        f"{len(ARPEGGIO_HZ) - len(misses)}/{len(ARPEGGIO_HZ)} within {tolerance:.0f}Hz"
        + (f"; {misses}" if misses else ""),
    )

    drone_peak = analyser.peak_hz(0.5 * STEP_SECONDS, lo_hz=60.0, hi_hz=150.0)
    check(
        "the bass drone shows up under the melody",
        abs(drone_peak - DRONE_HZ) <= tolerance,
        f"low-band peak at {drone_peak:.0f}Hz (drone is {DRONE_HZ:.0f}Hz)",
    )

    bars = analyser.magnitudes(0.6)
    check(
        "band magnitudes are normalised heights",
        bars.shape == (BANDS,)
        and bool(np.isfinite(bars).all())
        and float(bars.min()) >= 0.0
        and float(bars.max()) <= 1.0,
        f"shape={bars.shape}, range=[{bars.min():.2f}, {bars.max():.2f}]",
    )
    check(
        "a mid-note window actually lights bars up",
        float(bars.max()) > 0.5,
        f"max height {bars.max():.2f}",
    )

    # The clip loops, so a position past the end (or a window straddling the
    # loop point) must wrap instead of running off the buffer.
    total = STEP_SECONDS * len(ARPEGGIO_HZ)
    wrapped = analyser.magnitudes(total + 0.6)
    check(
        "positions wrap at the loop point",
        bool(np.allclose(wrapped, bars)),
        f"max deviation {float(np.abs(wrapped - bars).max()):.4f}",
    )

    print("SELFTEST:", "PASS" if ok else "FAIL")
    return ok


if __name__ == "__main__":
    import sys

    if "--test" in sys.argv:
        sys.exit(0 if _selftest() else 1)
    App(title="Audio Spectrum", width=WIDTH, height=HEIGHT).run(SpectrumDemo())