Music Streaming

file-backed loops that crossfade and duck under dialogue.

📄 Docs only

Tags: audio music streaming crossfade fades

What it demonstrates

• Two short music loops synthesised at startup and written as WAV files into a temp directory with the stdlib wave module. Only file-backed clips can stream (synthetic AudioClip.tone/from_pcm cannot), so this is the recipe for streaming music without shipping an asset.

• AudioPlayer(stream_mode="streaming", loop=True): the file is decoded in chunks each frame instead of being loaded whole, which is the mode for music.

• a.crossfade(b, 2.0): one call hands playback over on an equal-power curve, so the pair holds a steady loudness through the middle of the fade.

• Ducking under a fake dialogue line with fade_to(0.25, 0.3), then restoring with fade_to(1.0, 0.5) once the line ends.

• Fades and the streaming feed both tick in on_update, so the players must be in the tree; a detached player’s fade never advances.

Controls

• SPACE: crossfade to the other track (2 s, equal power).

• D: play a dialogue line: duck the music to a quarter amplitude, restore after.

• ESC: quit.

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

Source

"""Music Streaming: file-backed loops that crossfade and duck under dialogue.

# /// simvx
# tags = ["audio", "music", "streaming", "crossfade", "fades"]
# [web]
# disabled = true
# reason = "stream_mode='streaming' needs the native backend's chunk-fed decoder, absent in the browser runtime."
# ///

## What it demonstrates
- Two short music loops synthesised at startup and written as WAV files into a
  temp directory with the stdlib `wave` module. Only file-backed clips can
  stream (synthetic `AudioClip.tone`/`from_pcm` cannot), so this is the recipe
  for streaming music without shipping an asset.
- `AudioPlayer(stream_mode="streaming", loop=True)`: the file is decoded in
  chunks each frame instead of being loaded whole, which is the mode for music.
- `a.crossfade(b, 2.0)`: one call hands playback over on an equal-power curve,
  so the pair holds a steady loudness through the middle of the fade.
- Ducking under a fake dialogue line with `fade_to(0.25, 0.3)`, then restoring
  with `fade_to(1.0, 0.5)` once the line ends.
- Fades and the streaming feed both tick in `on_update`, so the players must be
  in the tree; a detached player's fade never advances.

## Controls
- `SPACE`: crossfade to the other track (2 s, equal power).
- `D`: play a dialogue line: duck the music to a quarter amplitude, restore after.
- `ESC`: quit.

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

import math
import tempfile
import wave
from pathlib import Path

import numpy as np

from simvx.core import AudioClip, AudioPlayer, Input, Key, Node, Text2D, Vec2, wait
from simvx.graphics import App

SAMPLE_RATE = 44100
CROSSFADE_SECONDS = 2.0
DUCK_GAIN = 0.25  # a quarter of the amplitude, about -12 dB
DUCK_FADE_SECONDS = 0.3
RESTORE_FADE_SECONDS = 0.5
DIALOGUE_SECONDS = 1.8

# Both loops are 2.4 s and wrap seamlessly: every pluck decays inside its own
# slot, and each bass drone completes a whole number of cycles over the loop
# (110 Hz x 2.4 s = 264; 82.5 Hz x 2.4 s = 198), so the join is inaudible.
TRACK_A = {
    "notes": [220.00, 261.63, 329.63, 392.00, 440.00, 392.00, 329.63, 261.63],
    "note_len": 0.3,
    "bass": 110.0,
    "harmonics": ((1, 1.0), (2, 0.35)),  # warm: fundamental plus one octave
}
TRACK_B = {
    "notes": [329.63, 392.00, 493.88, 587.33, 493.88, 392.00, 329.63, 392.00, 493.88, 659.25, 587.33, 493.88],
    "note_len": 0.2,
    "bass": 82.5,
    "harmonics": ((1, 1.0), (3, 0.33), (5, 0.2)),  # reedy: odd harmonics
}


def _pluck(freq: float, duration: float, harmonics) -> np.ndarray:
    """One exponentially decaying note built from (multiple, amplitude) partials."""
    t = np.arange(int(SAMPLE_RATE * duration)) / SAMPLE_RATE
    tone = sum(amp * np.sin(2.0 * np.pi * freq * mult * t) for mult, amp in harmonics)
    return np.exp(-5.0 * t / duration) * tone


def _render_loop(notes, note_len, bass, harmonics) -> np.ndarray:
    """A normalised mono loop: an arpeggio of plucks over a sine bass drone."""
    melody = np.concatenate([_pluck(freq, note_len, harmonics) for freq in notes])
    t = np.arange(len(melody)) / SAMPLE_RATE
    mix = 0.6 * melody + 0.22 * np.sin(2.0 * np.pi * bass * t)
    return mix / np.max(np.abs(mix))


def _write_wav(path: Path, mono: np.ndarray) -> None:
    """Write a mono float array as a 16-bit stereo WAV with the stdlib."""
    samples = (np.clip(mono, -1.0, 1.0) * 0.8 * 32767).astype(np.int16)
    stereo = np.column_stack((samples, samples))
    with wave.open(str(path), "wb") as out:
        out.setnchannels(2)
        out.setsampwidth(2)
        out.setframerate(SAMPLE_RATE)
        out.writeframes(stereo.tobytes())


def _write_tracks() -> tuple[Path, Path]:
    """Render both loops into a fresh temp directory and return their paths.

    Streaming needs real files on disk, so the PCM is written out rather than
    wrapped in a synthetic clip. mkdtemp keeps the example's own directory
    clean (it may be a read-only site-packages install); the OS reaps /tmp.
    """
    folder = Path(tempfile.mkdtemp(prefix="simvx_music_"))
    path_a, path_b = folder / "track_a.wav", folder / "track_b.wav"
    _write_wav(path_a, _render_loop(**TRACK_A))
    _write_wav(path_b, _render_loop(**TRACK_B))
    return path_a, path_b


class MusicDemo(Node):
    """Two streaming music players: crossfade between them, duck for dialogue."""

    input_actions = {
        "quit": [Key.ESCAPE],
        "crossfade": [Key.SPACE],
        "dialogue": [Key.D],
    }

    def on_ready(self):
        path_a, path_b = _write_tracks()
        # stream_mode is [next_play], so it is set here, before the first play().
        # Only a file-backed AudioClip can stream; a synthetic one raises at play.
        self._a = self.add_child(
            AudioPlayer(AudioClip(path_a), name="TrackA", stream_mode="streaming", loop=True, bus="Music")
        )
        self._b = self.add_child(
            AudioPlayer(AudioClip(path_b), name="TrackB", stream_mode="streaming", loop=True, bus="Music")
        )
        self._active = self._a
        self._dialogue = ""
        self._status = self.add_child(Text2D(text="", position=Vec2(40, 40), name="Status"))
        self._a.play()

    def on_update(self, dt: float):
        if Input.is_action_just_pressed("quit"):
            self.app.quit()
            return
        if Input.is_action_just_pressed("crossfade"):
            self._crossfade()
        if Input.is_action_just_pressed("dialogue"):
            self._begin_dialogue()
        self._update_status()

    def _crossfade(self):
        outgoing = self._active
        incoming = self._b if outgoing is self._a else self._a
        # One call arms both halves in the same frame: `incoming` fades in
        # (starting playback if it was stopped, from the top of its file) while
        # `outgoing` fades out and stops at the deadline. The default
        # equal_power curve keeps the summed loudness steady; pressing SPACE
        # mid-fade simply retargets both ramps.
        outgoing.crossfade(incoming, CROSSFADE_SECONDS)
        self._active = incoming

    def _begin_dialogue(self):
        if self._dialogue:
            return
        self.start_coroutine(self._dialogue_moment(self._active))

    def _dialogue_moment(self, music: AudioPlayer):
        # fade_to's gain is a multiplier on volume_db and it holds its target,
        # so the duck coexists with any volume the user has set on the player.
        self._dialogue = '"The music dips while I speak, then swells back."'
        music.fade_to(DUCK_GAIN, DUCK_FADE_SECONDS)
        yield from wait(DIALOGUE_SECONDS)
        music.fade_to(1.0, RESTORE_FADE_SECONDS)
        self._dialogue = ""

    def _update_status(self):
        def describe(player: AudioPlayer, label: str) -> str:
            marker = "->" if player is self._active else "  "
            state = "playing" if player.is_playing() else "stopped"
            # get_playback_position works for streaming players too: the
            # backend accounts for the frames it has decoded and mixed. On a
            # looping stream it keeps counting up rather than wrapping.
            return f"{marker} {label}: {state} at {player.get_playback_position():.1f}s"

        self._status.text = (
            "Music Streaming\n\n"
            f"{describe(self._a, 'Track A (warm arpeggio)')}\n"
            f"{describe(self._b, 'Track B (bright reed)')}\n\n"
            f"Dialogue: {self._dialogue or '(none)'}\n\n"
            "SPACE: crossfade to the other track\n"
            "D: dialogue line (duck to 25%, then restore)\n"
            "ESC: quit"
        )


def _selftest() -> bool:
    """Headless: drive the demo on a SceneTree against a recording backend.

    NullAudioBackend deliberately cannot stream, so the test installs a stub
    that satisfies the streaming facet (open_stream / feed_audio_chunk) and
    records every volume the engine pushes. The fades then become observable
    without a sound card: the crossfade midpoint, the duck floor and the
    restore all appear in the pushed dB values.
    """
    from simvx.core import SceneTree

    class RecordingBackend:
        """Streaming-capable, device-free backend: pure bookkeeping."""

        def __init__(self):
            self._next = 1
            self.ring_frames = 24000
            self.fed: dict[int, bytearray] = {}
            self.volumes: dict[int, list[float]] = {}

        def open_stream(self, *, volume_db=0.0, bus="Master", buffer_seconds=0.5, loop=False, stream=None) -> int:
            cid = self._next
            self._next += 1
            self.fed[cid] = bytearray()
            self.volumes[cid] = [volume_db]
            return cid

        def feed_audio_chunk(self, channel_id, chunk) -> None:
            self.fed[channel_id].extend(chunk)

        def stream_format(self) -> tuple[int, int]:
            return 48000, 2

        def frames_available(self, channel_id) -> int:
            return max(0, self.ring_frames - len(self.fed.get(channel_id, b"")) // 4)

        def get_playback_position(self, channel_id) -> float:
            # Pretend everything fed has played: 48 kHz stereo int16 frames.
            return len(self.fed.get(channel_id, b"")) / (48000 * 4)

        def update_audio_2d(self, channel_id, volume_db, pan) -> None:
            self.volumes.setdefault(channel_id, []).append(volume_db)

        def stop_audio(self, channel_id) -> None:
            self.fed.pop(channel_id, None)

        def sync_bus_layout(self, layout) -> None:
            pass

        def list_capabilities(self):
            return frozenset()

        def shutdown(self) -> None:
            pass

    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}")

    # The rendered files must be real, distinct, streamable WAVs.
    path_a, path_b = _write_tracks()
    with wave.open(str(path_a), "rb") as wav_a, wave.open(str(path_b), "rb") as wav_b:
        params_ok = all(
            handle.getnchannels() == 2 and handle.getsampwidth() == 2 and handle.getframerate() == SAMPLE_RATE
            for handle in (wav_a, wav_b)
        )
        seconds_a = wav_a.getnframes() / SAMPLE_RATE
        distinct = wav_a.readframes(wav_a.getnframes()) != wav_b.readframes(wav_b.getnframes())
    check("tracks are 16-bit stereo WAVs at the declared rate", params_ok, f"track A runs {seconds_a:.1f}s")
    check("the two tracks are audibly different files", distinct, "payloads differ")

    tree = SceneTree()
    backend = RecordingBackend()
    tree.install_audio_backend(backend)
    demo = MusicDemo(name="MusicDemo")
    tree.set_root(demo)
    dt = 1.0 / 60.0

    for _ in range(12):
        tree.tick(dt)
    fed = len(backend.fed.get(1, b""))
    check(
        "track A streams: playing, with PCM chunks fed to the backend",
        demo._a.is_playing() and fed > 0,
        f"{fed} bytes fed to channel 1",
    )
    check("track B waits silently", not demo._b.is_playing(), "not playing")

    # Crossfade: both audible at the midpoint, handover complete at the deadline.
    demo._crossfade()
    for _ in range(60):  # 1.0 s into a 2.0 s crossfade
        tree.tick(dt)
    va, vb = backend.volumes[1][-1], backend.volumes[2][-1]
    check(
        "mid-crossfade both tracks are playing",
        demo._a.is_playing() and demo._b.is_playing(),
        f"A at {va:.1f} dB, B at {vb:.1f} dB",
    )
    check(
        "the equal-power midpoint holds both near -3 dB",
        -6.0 < va < -1.0 and -6.0 < vb < -1.0,
        f"A {va:.2f} dB, B {vb:.2f} dB",
    )
    for _ in range(90):  # well past the 2.0 s deadline
        tree.tick(dt)
    check(
        "the handover completed: A stopped, B looping on",
        not demo._a.is_playing() and demo._b.is_playing() and demo._active is demo._b,
        "A stopped, B playing",
    )

    # Ducking: fade_to(0.25, ...) lands on -12 dB, the restore returns to 0 dB.
    duck_db = 20.0 * math.log10(DUCK_GAIN)
    demo._begin_dialogue()
    for _ in range(36):  # 0.6 s: the 0.3 s duck has landed, the line still runs
        tree.tick(dt)
    floor = backend.volumes[2][-1]
    check("the duck landed on a quarter amplitude", abs(floor - duck_db) < 0.3, f"{floor:.2f} dB vs {duck_db:.2f} dB")
    for _ in range(int((DIALOGUE_SECONDS + RESTORE_FADE_SECONDS + 0.5) * 60)):
        tree.tick(dt)
    restored = backend.volumes[2][-1]
    check(
        "the music swelled back to full volume",
        abs(restored) < 0.3 and demo._b.is_playing() and demo._dialogue == "",
        f"{restored:.2f} dB, dialogue over",
    )

    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="Music Streaming", width=960, height=540).run(MusicDemo())