nodes/audio_manager.py

Part of Dungeon Explorer.

  1"""Centralised audio manager: procedural SFX via the engine AudioSynth API.
  2
  3Replaces the previous list-of-floats stub with real AudioSynth bakes
  4played through pooled AudioPlayers. Each named effect maps to an
  5oscillator (or noise) + ADSR envelope; the baked AudioClip is cached
  6the first time it plays.
  7
  8In headless mode (no audio backend) the calls remain harmless: AudioSynth
  9bakes the stream regardless, and AudioPlayer.play() resolves to a
 10no-op when no backend is attached.
 11"""
 12
 13from simvx.core import (
 14    ADSR,
 15    AudioPlayer,
 16    AudioSynth,
 17    Node2D,
 18    Oscillator,
 19)
 20
 21# ── Sound definitions ─────────────────────────────────────────────────────
 22# Each entry: (oscillator factory args, ADSR shape, total duration).
 23
 24_SFX_DEFS: dict[str, dict] = {
 25    # Combat
 26    "swing": {"type": "noise", "duration": 0.08, "attack": 0.005, "release": 0.04, "gain": 0.6},
 27    "hit": {"type": "noise", "duration": 0.10, "attack": 0.002, "release": 0.06, "gain": 0.7},
 28    "crit": {"type": "sine", "freq": 880, "duration": 0.15, "attack": 0.005, "release": 0.1, "gain": 0.7},
 29    "block": {"type": "square", "freq": 200, "duration": 0.08, "attack": 0.002, "release": 0.04, "gain": 0.5},
 30    "dodge": {"type": "noise", "duration": 0.12, "attack": 0.002, "release": 0.08, "gain": 0.4},
 31    # World
 32    "pickup": {"type": "sine", "freq": 660, "duration": 0.12, "attack": 0.005, "release": 0.08, "gain": 0.5},
 33    "chest_open": {"type": "sine", "freq": 440, "duration": 0.20, "attack": 0.01, "release": 0.15, "gain": 0.6},
 34    "level_up": {"type": "sine", "freq": 523, "duration": 0.40, "attack": 0.02, "release": 0.3, "gain": 0.7},
 35    "coin": {"type": "sine", "freq": 1047, "duration": 0.08, "attack": 0.002, "release": 0.05, "gain": 0.4},
 36    "stairs": {"type": "noise", "duration": 0.15, "attack": 0.005, "release": 0.1, "gain": 0.5},
 37    # UI
 38    "menu_nav": {"type": "sine", "freq": 440, "duration": 0.04, "attack": 0.002, "release": 0.02, "gain": 0.3},
 39    "menu_confirm": {"type": "sine", "freq": 660, "duration": 0.08, "attack": 0.005, "release": 0.05, "gain": 0.4},
 40    "menu_cancel": {"type": "sine", "freq": 330, "duration": 0.06, "attack": 0.003, "release": 0.04, "gain": 0.4},
 41    "popup": {"type": "sine", "freq": 550, "duration": 0.10, "attack": 0.005, "release": 0.07, "gain": 0.4},
 42    "error": {"type": "square", "freq": 150, "duration": 0.15, "attack": 0.003, "release": 0.1, "gain": 0.5},
 43    # Boss
 44    "boss_roar": {"type": "noise", "duration": 0.50, "attack": 0.02, "release": 0.3, "gain": 0.8},
 45    "boss_charge": {"type": "noise", "duration": 0.30, "attack": 0.01, "release": 0.2, "gain": 0.6},
 46    "boss_slam": {"type": "noise", "duration": 0.40, "attack": 0.005, "release": 0.2, "gain": 0.9},
 47    "boss_phase": {"type": "sine", "freq": 330, "duration": 0.60, "attack": 0.05, "release": 0.4, "gain": 0.7},
 48}
 49
 50
 51def _bake_sfx(key: str):
 52    """Bake the named SFX into an `AudioClip` using AudioSynth."""
 53    defn = _SFX_DEFS[key]
 54    synth = AudioSynth()
 55    wtype = defn["type"]
 56    duration = defn["duration"]
 57    gain = defn.get("gain", 0.5)
 58    envelope = ADSR(
 59        attack=defn["attack"],
 60        decay=0.01,
 61        sustain=1.0,
 62        release=defn["release"],
 63    )
 64    if wtype == "sine":
 65        synth.add(Oscillator.sine(defn["freq"]), envelope=envelope, gain=gain)
 66    elif wtype == "square":
 67        synth.add(Oscillator.square(defn["freq"]), envelope=envelope, gain=gain)
 68    elif wtype == "noise":
 69        synth.add(Oscillator.noise.white(), envelope=envelope, gain=gain)
 70    else:
 71        return None
 72    return synth.bake(duration=duration, sample_rate=44100, channels=2)
 73
 74
 75class AudioManager(Node2D):
 76    """Plays procedurally generated sound effects and ambient audio.
 77
 78    SFX are baked once via AudioSynth on first play and cached. The
 79    manager owns a pool of AudioPlayers; concurrent plays beyond
 80    the pool size are dropped (typical for arcade SFX).
 81    """
 82
 83    def __init__(self, **kwargs):
 84        super().__init__(name="AudioManager", **kwargs)
 85        self._max_concurrent = 8
 86        self._music_volume = 1.0
 87        self._sfx_volume = 1.0
 88        self._master_volume = 1.0
 89        self._current_music: str | None = None
 90        # Stream cache: key → AudioClip (baked lazily on first play).
 91        self._streams: dict[str, object] = {}
 92        # Player pool: round-robin allocation for concurrent SFX.
 93        self._players: list[AudioPlayer] = []
 94        self._next_player_idx = 0
 95
 96    def on_ready(self):
 97        # Allocate the SFX player pool.
 98        for i in range(self._max_concurrent):
 99            p = AudioPlayer(name=f"SfxPlayer_{i}", bus="SFX")
100            self.add_child(p)
101            self._players.append(p)
102
103    # ── Volume control ────────────────────────────────────────────────
104
105    @property
106    def sfx_volume(self) -> float:
107        return self._sfx_volume
108
109    @sfx_volume.setter
110    def sfx_volume(self, value: float):
111        self._sfx_volume = max(0.0, min(1.0, value))
112
113    @property
114    def music_volume(self) -> float:
115        return self._music_volume
116
117    @music_volume.setter
118    def music_volume(self, value: float):
119        self._music_volume = max(0.0, min(1.0, value))
120
121    @property
122    def master_volume(self) -> float:
123        return self._master_volume
124
125    @master_volume.setter
126    def master_volume(self, value: float):
127        self._master_volume = max(0.0, min(1.0, value))
128
129    # ── SFX ───────────────────────────────────────────────────────────
130
131    def _ensure_stream(self, key: str):
132        if key not in self._streams:
133            stream = _bake_sfx(key)
134            if stream is None:
135                return None
136            self._streams[key] = stream
137        return self._streams[key]
138
139    def play_sfx(self, key: str):
140        """Play a sound effect by name. Cached on first use."""
141        if key not in _SFX_DEFS:
142            return
143        if not self._players:
144            return
145        stream = self._ensure_stream(key)
146        if stream is None:
147            return
148        # Round-robin pick: stops whatever was on the chosen player.
149        player = self._players[self._next_player_idx]
150        self._next_player_idx = (self._next_player_idx + 1) % len(self._players)
151        # The round-robin pick may still be finishing its last sound, and a new
152        # stream only takes effect at the next play().
153        player.stop()
154        player.stream = stream
155        # Map the manager's 0..1 sfx_volume to dB; -80 = silent.
156        if self._sfx_volume <= 0:
157            return
158        import math
159
160        player.volume_db = 20.0 * math.log10(max(self._sfx_volume * self._master_volume, 1e-4))
161        player.play()
162
163    # ── Music track selection ─────────────────────────────────────────
164    #
165    # The demo bakes its SFX procedurally and ships no music assets, so these
166    # only record which track the game asked for. Point them at an AudioPlayer
167    # with a streamed clip to add a soundtrack.
168
169    def play_music(self, key: str):
170        """Select ``key`` as the current music track."""
171        self._current_music = key
172
173    def stop_music(self):
174        """Clear the current music track."""
175        self._current_music = None