Source code for simvx.core.audio

"""
Audio system — background music, UI sounds, and 3D spatial audio.

This module provides:
- AudioStream: Audio resource (WAV/OGG data)
- AudioStreamPlayer: Background music/UI sounds
- AudioStreamPlayer2D: 2D positional audio with panning
- AudioStreamPlayer3D: 3D spatial audio with attenuation

Public API:
    from simvx.core import AudioStream, AudioStreamPlayer, AudioStreamPlayer2D, AudioStreamPlayer3D

    # Background music
    player = AudioStreamPlayer(stream="audio://music/theme.ogg", autoplay=True)

    # 2D sound effect
    sfx = AudioStreamPlayer2D(stream="audio://sfx/explosion.wav", position=Vec2(100, 50))
    sfx.play()

    # 3D spatial audio
    engine = AudioStreamPlayer3D(stream="audio://sfx/engine.ogg", position=Vec3(10, 0, 5))
    engine.play()
"""


from __future__ import annotations

from __future__ import annotations

import logging
from typing import Any

from .node import Node
from .nodes_2d.node2d import Node2D
from .nodes_3d.node3d import Node3D
from .descriptors import Property
from .math.types import Vec2, Vec3, clamp

log = logging.getLogger(__name__)

__all__ = [
    "AudioStream",
    "AudioStreamPlayer",
    "AudioStreamPlayer2D",
    "AudioStreamPlayer3D",
    "AudioListener",
]


# ============================================================================
# AudioStream — Audio resource
# ============================================================================


[docs] class AudioStream: """Audio resource (WAV/OGG file). This is a lightweight handle to audio data. Actual loading is deferred to the backend (SDL_mixer, OpenAL, etc). Attributes: path: File path or URI (e.g., "audio://music/theme.ogg") resource_uri: Full URI if loaded via ResourceCache backend_data: Backend-specific audio data (channel, source ID, etc) """ __slots__ = ("path", "resource_uri", "backend_data") def __init__(self, path: str): self.path = path self.resource_uri: str = path self.backend_data: Any = None # Backend-specific data
[docs] def __repr__(self): return f"AudioStream({self.path!r})"
# ============================================================================ # AudioListener — Tracks camera position for 3D audio # ============================================================================
[docs] class AudioListener: """Singleton that tracks the active camera for 3D audio spatialization. The listener position is automatically updated from Camera3D or Camera2D by the backend's audio system. """ _instance: AudioListener | None = None def __init__(self): self.position_3d: Vec3 = Vec3() self.position_2d: Vec2 = Vec2() self.forward: Vec3 = Vec3(0, 0, -1) # Camera forward vector self.up: Vec3 = Vec3(0, 1, 0)
[docs] @classmethod def get(cls) -> AudioListener: if cls._instance is None: cls._instance = cls() return cls._instance
[docs] @classmethod def reset(cls): """Reset singleton (useful for tests).""" cls._instance = None
# ============================================================================ # _AudioPlaybackMixin — Shared play/stop/pause/is_playing logic # ============================================================================ class _AudioPlaybackMixin: """Mixin providing common audio playback state and backend interaction. Subclasses must set self.stream, self._playing, self._paused, self._backend_channel in __init__, and define volume_db, pitch_scale, bus, autoplay, loop Settings. """ def _init_playback(self, stream: str | AudioStream | None): self.stream: AudioStream | None = None if stream: self.stream = AudioStream(stream) if isinstance(stream, str) else stream self._playing: bool = False self._paused: bool = False self._backend_channel: Any = None def _get_backend(self): tree = getattr(self, "_tree", None) return getattr(tree, "_audio_backend", None) if tree else None def stop(self): """Stop playback.""" self._playing = False self._paused = False backend = self._get_backend() if backend and self._backend_channel is not None: backend.stop_audio(self._backend_channel) self._backend_channel = None def pause(self): """Pause playback. Call play() to resume.""" if not self._playing: return self._paused = True self._playing = False backend = self._get_backend() if backend and self._backend_channel is not None: backend.pause_audio(self._backend_channel) def is_playing(self) -> bool: """Check if audio is currently playing.""" return self._playing and not self._paused def _resume_if_paused(self) -> bool: """If paused, resume playback and return True; otherwise return False.""" if not self._paused or self._backend_channel is None: return False self._playing = True self._paused = False backend = self._get_backend() if backend: backend.resume_audio(self._backend_channel) return True def _play_common(self, from_position: float = 0.0) -> bool: """Common play() preamble. Returns True if playback should proceed.""" if not self.stream: return False if from_position == 0.0 and self._resume_if_paused(): return False self._playing = True self._paused = False return True def _autoplay_check(self): """Start playback if autoplay is enabled. Call from ready().""" if self.autoplay and self.stream: self.play() def _exit_tree(self): super()._exit_tree() self.stop() # ============================================================================ # AudioStreamPlayer — Background music / UI sounds # ============================================================================
[docs] class AudioStreamPlayer(_AudioPlaybackMixin, Node): """Non-positional audio player for background music and UI sounds. This player does not use 3D positioning — volume is constant regardless of camera position. Use AudioStreamPlayer2D or AudioStreamPlayer3D for spatial audio. Settings: volume_db: Volume in decibels (-80 to 24). 0 = full volume. pitch_scale: Playback speed multiplier (0.5 to 2.0). bus: Audio bus name ("master", "music", "sfx", etc). autoplay: Start playing when added to scene tree. loop: Loop playback when finished. stream_mode: "memory" loads entire file; "streaming" reads in chunks. buffer_size: Chunk size in bytes for streaming mode (default 64KB). """ volume_db = Property(0.0, range=(-80.0, 24.0), hint="Volume in decibels", group="Playback") pitch_scale = Property(1.0, range=(0.5, 2.0), hint="Playback speed", group="Playback") bus = Property("master", enum=["master", "music", "sfx", "ui"], hint="Audio bus", group="Playback") autoplay = Property(False, hint="Auto-play on ready", group="Playback") loop = Property(False, hint="Loop playback", group="Playback") stream_mode = Property("memory", enum=["memory", "streaming"], hint="Load mode", group="Playback") buffer_size = Property(65536, range=(4096, 524288), hint="Streaming buffer size", group="Playback") def __init__(self, stream: str | AudioStream | None = None, **kwargs): super().__init__(**kwargs) self._init_playback(stream) self._stream_file: Any = None # Open file handle for streaming mode self._stream_offset: int = 0 # Current read offset in file self._stream_data_offset: int = 0 # Byte offset past file header (e.g. WAV header)
[docs] def ready(self): self._autoplay_check()
[docs] def process(self, delta: float): """Feed audio chunks to backend in streaming mode.""" if self.stream_mode == "streaming" and self._playing and not self._paused: self._process_streaming()
def _process_streaming(self): """Read next chunk from file and feed to audio backend.""" if not self._stream_file or not self._tree: return backend = self._get_backend() if not backend: return chunk = self._stream_file.read(self.buffer_size) if not chunk: if self.loop: self._stream_file.seek(self._stream_data_offset) chunk = self._stream_file.read(self.buffer_size) if not chunk: self.stop() return backend.feed_audio_chunk(self._backend_channel, chunk)
[docs] def play(self, from_position: float = 0.0): """Start or resume playback. Args: from_position: Start position in seconds (0.0 = beginning). """ if not self._play_common(from_position): return # Open file for streaming mode if self.stream_mode == "streaming" and self.stream.path: if self._stream_file: self._stream_file.close() self._stream_file = None try: self._stream_file = open(self.stream.path, "rb") # noqa: SIM115 self._stream_data_offset = 0 header = self._stream_file.read(4) if header == b"RIFF": self._stream_file.seek(44) # Standard WAV header size self._stream_data_offset = 44 else: self._stream_file.seek(0) except OSError: log.warning("Failed to open audio stream file: %s", self.stream.path) self._stream_file = None backend = self._get_backend() if backend: if self.stream_mode == "streaming": self._backend_channel = backend.open_stream( volume_db=self.volume_db, pitch=self.pitch_scale, bus=self.bus, ) else: self._backend_channel = backend.play_audio( self.stream, volume_db=self.volume_db, pitch=self.pitch_scale, loop=self.loop, bus=self.bus, )
[docs] def stop(self): """Stop playback and reset position to beginning.""" if self._stream_file: self._stream_file.close() self._stream_file = None super().stop()
[docs] def get_playback_position(self) -> float: """Get current playback position in seconds.""" backend = self._get_backend() if backend and self._backend_channel is not None: return backend.get_playback_position(self._backend_channel) return 0.0
# ============================================================================ # AudioStreamPlayer2D — 2D positional audio # ============================================================================
[docs] class AudioStreamPlayer2D(_AudioPlaybackMixin, Node2D): """2D positional audio player with stereo panning. Audio volume and pan are calculated based on distance from the 2D listener (typically Camera2D position). Left/right panning simulates direction. Settings: volume_db: Base volume in decibels (-80 to 24). pitch_scale: Playback speed multiplier (0.5 to 2.0). bus: Audio bus name. autoplay: Start playing when added to scene tree. loop: Loop playback when finished. max_distance: Distance at which audio is inaudible (pixels). attenuation: Distance attenuation exponent (1.0 = linear, 2.0 = inverse square). """ volume_db = Property(0.0, range=(-80.0, 24.0), hint="Volume in decibels", group="Playback") pitch_scale = Property(1.0, range=(0.5, 2.0), hint="Playback speed", group="Playback") bus = Property("sfx", enum=["master", "music", "sfx", "ui"], hint="Audio bus", group="Playback") autoplay = Property(False, hint="Auto-play on ready", group="Playback") loop = Property(False, hint="Loop playback", group="Playback") max_distance = Property(2000.0, range=(1.0, 10000.0), hint="Max hearing distance (pixels)", group="Spatial") attenuation = Property(1.0, range=(0.1, 4.0), hint="Distance attenuation exponent", group="Spatial") gizmo_colour = Property((0.6, 0.4, 1.0, 0.5), hint="Editor gizmo colour")
[docs] def get_gizmo_lines(self) -> list[tuple[Vec2, Vec2]]: """Return circle showing the audio range.""" from .physics_nodes import _circle_lines_2d p = self.world_position return _circle_lines_2d(p.x, p.y, float(self.max_distance), 32)
def __init__(self, stream: str | AudioStream | None = None, **kwargs): super().__init__(**kwargs) self._init_playback(stream)
[docs] def ready(self): self._autoplay_check()
[docs] def process(self, delta: float): """Update 2D spatialization each frame.""" if not self._playing or self._paused: return # Calculate volume and pan based on listener position listener = AudioListener.get() distance = (self.world_position - listener.position_2d).length() # Distance attenuation if distance > self.max_distance: volume = -80.0 # Silent else: # Attenuation formula: volume_db = base - k * (dist / max_dist) ^ attenuation dist_ratio = distance / self.max_distance volume = self.volume_db - 80.0 * (dist_ratio**self.attenuation) # Stereo panning (-1.0 = left, 0.0 = center, 1.0 = right) dx = self.world_position.x - listener.position_2d.x pan = clamp(dx / self.max_distance, -1.0, 1.0) # Update backend backend = self._get_backend() if backend and self._backend_channel is not None: backend.update_audio_2d(self._backend_channel, volume, pan)
[docs] def play(self, from_position: float = 0.0): """Start or resume playback.""" if not self._play_common(from_position): return backend = self._get_backend() if backend: self._backend_channel = backend.play_audio_2d( self.stream, position=self.world_position, volume_db=self.volume_db, pitch=self.pitch_scale, loop=self.loop, bus=self.bus, max_distance=self.max_distance, )
# ============================================================================ # AudioStreamPlayer3D — 3D spatial audio # ============================================================================
[docs] class AudioStreamPlayer3D(_AudioPlaybackMixin, Node3D): """3D spatial audio player with distance attenuation and directional panning. Audio volume and stereo panning are calculated based on distance and direction from the 3D listener (typically Camera3D position/orientation). Settings: volume_db: Base volume in decibels (-80 to 24). pitch_scale: Playback speed multiplier (0.5 to 2.0). bus: Audio bus name. autoplay: Start playing when added to scene tree. loop: Loop playback when finished. max_distance: Distance at which audio is inaudible (world units). attenuation: Distance attenuation exponent (1.0 = linear, 2.0 = inverse square). doppler_scale: Doppler effect strength (0.0 = off, 1.0 = realistic). """ volume_db = Property(0.0, range=(-80.0, 24.0), hint="Volume in decibels", group="Playback") pitch_scale = Property(1.0, range=(0.5, 2.0), hint="Playback speed", group="Playback") bus = Property("sfx", enum=["master", "music", "sfx", "ui"], hint="Audio bus", group="Playback") autoplay = Property(False, hint="Auto-play on ready", group="Playback") loop = Property(False, hint="Loop playback", group="Playback") max_distance = Property(100.0, range=(1.0, 1000.0), hint="Max hearing distance", group="Spatial") attenuation = Property(1.0, range=(0.1, 4.0), hint="Distance attenuation exponent", group="Spatial") doppler_scale = Property(0.0, range=(0.0, 4.0), hint="Doppler effect strength", group="Spatial") gizmo_colour = Property((0.6, 0.4, 1.0, 0.5), hint="Editor gizmo colour")
[docs] def get_gizmo_lines(self) -> list[tuple[Vec3, Vec3]]: """Return 3 circles showing the audio range sphere.""" from .physics_nodes import _circle_lines_3d p = self.world_position r = float(self.max_distance) lines: list[tuple[Vec3, Vec3]] = [] lines.extend(_circle_lines_3d(p, Vec3(1, 0, 0), Vec3(0, 1, 0), r)) lines.extend(_circle_lines_3d(p, Vec3(1, 0, 0), Vec3(0, 0, 1), r)) lines.extend(_circle_lines_3d(p, Vec3(0, 1, 0), Vec3(0, 0, 1), r)) return lines
def __init__(self, stream: str | AudioStream | None = None, **kwargs): super().__init__(**kwargs) self._init_playback(stream) self._prev_position: Vec3 = Vec3() # For Doppler
[docs] def ready(self): self._prev_position = self.world_position self._autoplay_check()
[docs] def process(self, delta: float): """Update 3D spatialization each frame.""" if not self._playing or self._paused: return # Calculate volume, pan, and pitch based on listener position listener = AudioListener.get() to_source = self.world_position - listener.position_3d distance = to_source.length() # Distance attenuation if distance > self.max_distance: volume = -80.0 # Silent else: # Inverse distance law with attenuation exponent if distance < 0.1: distance = 0.1 # Avoid division by zero dist_ratio = distance / self.max_distance volume = self.volume_db - 80.0 * (dist_ratio**self.attenuation) # Stereo panning based on left/right position relative to camera if distance > 0.01: to_source_norm = to_source / distance # Get camera's right vector (cross product of up and forward) right = listener.forward.cross(listener.up) right_dot = to_source_norm.dot(right) pan = clamp(right_dot, -1.0, 1.0) else: pan = 0.0 # Doppler effect (pitch shift based on velocity) pitch = self.pitch_scale if self.doppler_scale > 0.0 and delta > 0.0: velocity = (self.world_position - self._prev_position) / delta # Velocity towards listener increases pitch (positive dot = approaching) velocity_towards = velocity.dot(-to_source) / (distance if distance > 0.1 else 0.1) # Doppler: pitch_shift = 1 / (1 - v/c) approx (1 + v/c) for small v speed_of_sound = 343.0 doppler_factor = 1.0 + (velocity_towards / speed_of_sound) * self.doppler_scale pitch = self.pitch_scale * clamp(doppler_factor, 0.5, 2.0) self._prev_position = Vec3(self.world_position) # Update backend backend = self._get_backend() if backend and self._backend_channel is not None: backend.update_audio_3d(self._backend_channel, volume, pan, pitch)
[docs] def play(self, from_position: float = 0.0): """Start or resume playback.""" if not self._play_common(from_position): return self._prev_position = self.world_position backend = self._get_backend() if backend: self._backend_channel = backend.play_audio_3d( self.stream, position=self.world_position, volume_db=self.volume_db, pitch=self.pitch_scale, loop=self.loop, bus=self.bus, max_distance=self.max_distance, )