"""
Audio player nodes: :class:`AudioPlayer` / ``2D`` / ``3D``.
Private leaf module behind the :mod:`simvx.core.audio` facade. The player
node classes compose the shared :class:`_AudioPlaybackMixin` with the node
base classes and add the non-positional / 2D / 3D playback specifics. The
shared playback state machine lives in ``_audio_playback``; the stream/data
layer in ``_audio_stream``.
"""
from __future__ import annotations
import logging
import os
from ._audio_playback import _AudioPlaybackMixin
from ._audio_stream import _DEFAULT_STREAM_FORMAT, AudioClip, AudioSource, _StreamDecoder
from .audio_errors import AudioError, InvalidStreamError
from .audio_protocol import AudioStreamDraining, AudioStreamPacing, Capability
from .descriptors import Property
from .math.types import Vec2, Vec3, clamp
from .node import Node
from .nodes_2d.node2d import Node2D
from .nodes_3d.node3d import Node3D
from .properties import Colour
from .signals import Signal
log = logging.getLogger(__name__)
# Containers a backend may hold its own streaming decoder for. When it
# advertises the flag it gets handed the file and decodes it itself; when it
# does not, the player decodes and feeds the result in. WAV is absent on
# purpose: it is the format the chunk-fed ring already speaks, so there is
# nothing to gain by handing the file over.
_NATIVE_STREAM_CAPABILITY: dict[str, Capability] = {
"ogg": Capability.STREAMING_OGG,
"mp3": Capability.STREAMING_MP3,
"flac": Capability.STREAMING_FLAC,
}
def _backend_decodes(backend, container: str) -> bool:
"""Whether *backend* has a streaming decoder of its own for *container*."""
capability = _NATIVE_STREAM_CAPABILITY.get(container)
if capability is None:
return False
list_capabilities = getattr(backend, "list_capabilities", None)
if list_capabilities is None:
return False
return capability in list_capabilities()
def _backend_stream_format(backend) -> tuple[int, int]:
"""``(sample_rate, channels)`` *backend* wants fed chunks in."""
if isinstance(backend, AudioStreamPacing):
rate, channels = backend.stream_format()
return int(rate), int(channels)
return _DEFAULT_STREAM_FORMAT
# ============================================================================
# AudioPlayer: Background music / UI sounds
# ============================================================================
[docs]
class AudioPlayer(_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 AudioPlayer2D or AudioPlayer3D 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. One of ``"Master"``, ``"Music"``, ``"SFX"``,
``"Voice"``, ``"UI"`` (case-sensitive, Godot convention).
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: Largest chunk, in bytes, fed to the backend in one frame
while streaming (default 64KB). The feed is sized to the room the
backend actually has, so this is a ceiling rather than the amount
written each frame.
"""
volume_db = Property(
0.0,
range=(-80.0, 24.0),
hint="Volume in decibels [live: pushed mid-playback]",
group="Playback",
on_change="_on_volume_db_changed",
)
pitch_scale = Property(
1.0,
range=(0.5, 2.0),
hint="Playback speed [live: pushed mid-playback via backend.set_pitch]",
group="Playback",
on_change="_on_pitch_scale_changed",
)
bus = Property(
"Master",
enum=["Master", "Music", "SFX", "Voice", "UI"],
hint="Audio bus [next_play: mid-playback writes raise in strict mode]",
group="Playback",
on_change="_on_bus_changed",
)
autoplay = Property(
False,
hint="Auto-play on ready [next_play: changing after on_ready has no effect]",
group="Playback",
on_change="_on_autoplay_changed",
)
loop = Property(
False,
hint="Loop playback [next_play: mid-playback writes raise in strict mode]",
group="Playback",
on_change="_on_loop_changed",
)
queue_free_on_end = Property(
False,
hint="Remove this node from the tree once playback finishes [next_play]",
group="Playback",
on_change="_on_queue_free_on_end_changed",
)
stream_mode = Property(
"memory",
enum=["memory", "streaming"],
hint="Load mode [next_play: mid-playback writes raise in strict mode]",
group="Playback",
on_change="_on_stream_mode_changed",
)
buffer_size = Property(
65536,
range=(4096, 524288),
hint="Largest streaming chunk fed per frame, in bytes [next_play]",
group="Playback",
on_change="_on_buffer_size_changed",
)
# Emitted when `stream_mode == "streaming"` and `play()` cannot open or parse
# the source file. Argument is the resolved path string. Games can hook this
# to fall back to a placeholder track, log to a UI, etc.
stream_open_failed = Signal(str)
def __init__(self, stream: AudioSource | AudioClip | None = None, **kwargs):
super().__init__(**kwargs)
self._init_playback(stream)
# Decoder feeding the backend while streaming. ``None`` whenever the
# backend decodes the file itself, and between plays.
self._stream_decoder: _StreamDecoder | None = None
# Format the decoder produces; resolved from the backend at play().
self._stream_rate, self._stream_channels = _DEFAULT_STREAM_FORMAT
[docs]
def on_enter_tree(self):
# Per entry, not once: ``_exit_tree`` stops the sound, so an autoplay
# player brought back into the tree has to be started again.
self._autoplay_check()
[docs]
def on_update(self, delta: float):
"""Feed audio chunks to backend in streaming mode; tick fades; auto-free."""
if self._playing and not self._paused and self._tick_fade(delta):
backend = self._get_backend()
if backend and self._backend_channel is not None:
self._push_audio_state(backend, self._backend_channel)
if self.stream_mode == "streaming" and self._playing and not self._paused:
self._process_streaming(delta)
if self._check_queue_free_on_end():
return
def _stream_feed_frames(self, backend, delta: float) -> int:
"""How many frames to feed this tick, capped by ``buffer_size``.
A backend that reports its free space gets exactly that: the ring
fills on the first tick and is topped up by whatever the audio thread
consumed since. One that does not is paced against the frame clock
instead, because feeding a fixed amount every tick regardless empties
the file in a fraction of the time it takes to play.
"""
if isinstance(backend, AudioStreamPacing):
frames = backend.frames_available(self._backend_channel)
else:
frames = int(delta * self._stream_rate)
return int(min(frames, self.buffer_size // (self._stream_channels * 2)))
def _process_streaming(self, delta: float):
"""Top up the backend's buffer with decoded PCM without overrunning it.
A no-op when the backend decodes the file itself: there is no decoder
here and, on a backend that reports its free space, no room either.
"""
decoder = self._stream_decoder
if decoder is None or not self._tree or self._backend_channel is None:
return
backend = self._get_backend()
if not backend:
return
frames = self._stream_feed_frames(backend, delta)
if frames <= 0:
return
chunk = decoder.read(frames)
if not chunk:
if self.loop:
decoder.restart()
chunk = decoder.read(frames)
if not chunk:
self._finish_fed_stream(backend)
return
backend.feed_audio_chunk(self._backend_channel, chunk)
def _finish_fed_stream(self, backend) -> None:
"""Hand over the end of a chunk-fed stream once the decoder runs dry.
The decoder is finished with either way, so it is closed here rather
than waiting for the ``stop`` that eventually follows: its file
handle has nothing left to read.
What happens to the channel depends on the backend. One that
implements :class:`AudioStreamDraining` is told the last chunk is in
and keeps playing what it already holds; the natural-end poll then
stops this player at the point the last frame is heard. One that
does not is stopped immediately, which loses whatever is still
queued, because a hard stop is the only end-of-stream signal such a
backend has.
"""
self._close_stream_decoder()
if isinstance(backend, AudioStreamDraining):
backend.end_of_stream(self._backend_channel)
return
if self.queue_free_on_end and not self.loop:
self._queue_free_on_end_pending = True
self.stop()
[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
# Retire the decoder from the previous play before anything can
# replace it. Resuming from pause returned above, so reaching here
# always means the old decoder's file is finished with.
self._close_stream_decoder()
backend = self._get_backend()
if self.stream_mode == "streaming":
self._play_streaming(backend, from_position)
return
if backend:
self._claim_channel(
backend,
backend.play_audio(
self.stream,
mode="non_positional",
volume_db=self._effective_volume_db(),
pitch=self.pitch_scale,
loop=self.loop,
bus=self.bus,
from_position=from_position,
),
)
def _close_stream_decoder(self) -> None:
"""Close and drop the decoder feeding the backend, if there is one."""
decoder = self._stream_decoder
if decoder is not None:
self._stream_decoder = None
decoder.close()
def _play_streaming(self, backend, from_position: float) -> None:
"""Open a streaming channel and, unless the backend decodes the file
itself, the decoder that will feed it.
Both happen before the channel is claimed: a failure after allocating
one would leave a silent channel attached to the bus with nothing
feeding it.
"""
path = self.stream.path if self.stream else ""
container = self.stream.container if self.stream else "unknown"
if backend is None:
self._playing = False
return
# Judge the source before the backend: an unplayable file is
# unplayable whichever backend was going to receive it.
if not self._stream_source_is_streamable(path, container):
self._playing = False
return
# Narrow to AudioStreamingBackend: NullAudioBackend (and any future
# playback-only backend) doesn't implement streaming. Surface that as
# a typed AudioCapabilityError so the user gets a clear remediation
# rather than an AttributeError.
from .audio_protocol import AudioStreamingBackend
if not isinstance(backend, AudioStreamingBackend):
from .audio_errors import AudioCapabilityError
self.stream_open_failed(path)
self._playing = False
raise AudioCapabilityError(
"streaming",
backend=type(backend).__name__,
advertised=backend.list_capabilities(),
remediation=(
"AudioPlayer(stream_mode='streaming') requires an "
"AudioStreamingBackend (open_stream / feed_audio_chunk). "
"Use stream_mode='memory' or install the native extension."
),
)
self._stream_rate, self._stream_channels = _backend_stream_format(backend)
# A backend with its own decoder for this container is handed the
# file; anything else is fed engine-format PCM decoded here.
backend_decodes = _backend_decodes(backend, container)
decoder: _StreamDecoder | None = None
if not backend_decodes:
try:
decoder = _StreamDecoder(
path,
sample_rate=self._stream_rate,
channels=self._stream_channels,
from_position=from_position,
)
except InvalidStreamError as exc:
log.warning("AudioPlayer: cannot stream %s: %s", path, exc)
self.stream_open_failed(path)
self._playing = False
return
try:
channel = backend.open_stream(
volume_db=self._effective_volume_db(),
bus=self.bus,
loop=self.loop if backend_decodes else False,
stream=self.stream if backend_decodes else None,
)
except AudioError as exc:
log.warning("AudioPlayer: backend rejected open_stream for %r: %s", path, exc)
if decoder is not None:
decoder.close()
self.stream_open_failed(path)
self._playing = False
return
self._stream_decoder = decoder
self._claim_channel(backend, channel)
if decoder is not None:
# Fill the backend's buffer before the first frame, so playback
# starts on audio rather than on the silence of an empty ring.
prefill_seconds = self.buffer_size / float(self._stream_channels * 2 * self._stream_rate)
self._process_streaming(prefill_seconds)
def _stream_source_is_streamable(self, path: str, container: str) -> bool:
"""Whether ``self.stream`` can be streamed at all, logging why not.
Emits ``stream_open_failed`` and returns False for a source with no
file behind it, a synthetic clip, or a container no decoder
recognises. Failing here rather than at the decoder keeps the
diagnosis specific.
"""
if not path:
log.warning("AudioPlayer: streaming mode requires a file path")
self.stream_open_failed(path)
return False
# Synthetic streams (from_pcm / tone) carry decoded ndarray data;
# they were never meant to flow through the chunk-fed streaming
# path. The stream player should use stream_mode="memory" for them.
if container == "pcm":
log.warning(
"AudioPlayer: stream_mode='streaming' is not supported for "
"synthetic AudioClip (from_pcm/tone). Use stream_mode='memory'. path=%s",
path,
)
self.stream_open_failed(path)
return False
if container == "unknown":
# The header probe failed to recognise the format. Fail loud here
# rather than letting a decoder guess from the file extension.
log.warning(
"AudioPlayer: unrecognised audio container for streaming source %s "
"(expected WAV/OGG/MP3/FLAC header)",
path,
)
self.stream_open_failed(path)
return False
if not os.path.isfile(path):
log.warning(
"AudioPlayer: streaming file does not exist or is not a regular file: %s",
path,
)
self.stream_open_failed(path)
return False
return True
[docs]
def stop(self):
"""Stop playback and reset position to beginning."""
self._close_stream_decoder()
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
# ============================================================================
# AudioPlayer2D: 2D positional audio
# ============================================================================
[docs]
class AudioPlayer2D(_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 [live]",
group="Playback",
on_change="_on_volume_db_changed",
)
pitch_scale = Property(
1.0,
range=(0.5, 2.0),
hint="Playback speed [live: pushed mid-playback via backend.set_pitch]",
group="Playback",
on_change="_on_pitch_scale_changed",
)
bus = Property(
"SFX",
enum=["Master", "Music", "SFX", "Voice", "UI"],
hint="Audio bus [next_play]",
group="Playback",
on_change="_on_bus_changed",
)
autoplay = Property(
False,
hint="Auto-play on ready [next_play]",
group="Playback",
on_change="_on_autoplay_changed",
)
loop = Property(
False,
hint="Loop playback [next_play]",
group="Playback",
on_change="_on_loop_changed",
)
queue_free_on_end = Property(
False,
hint="Remove this node from the tree once playback finishes [next_play]",
group="Playback",
on_change="_on_queue_free_on_end_changed",
)
max_distance = Property(
2000.0,
range=(1.0, 10000.0),
hint="Max hearing distance (pixels) [live]",
group="Spatial",
on_change="_on_spatial_changed",
)
attenuation = Property(
1.0,
range=(0.1, 4.0),
hint="Distance attenuation exponent [live]",
group="Spatial",
on_change="_on_spatial_changed",
)
pan_override = Property(
None,
hint="Override positional pan; None = use world position [live]",
group="Playback",
on_change="_on_spatial_changed",
)
gizmo_colour = Colour((0.6, 0.4, 1.0, 0.5))
[docs]
def get_gizmo_lines(self) -> list[tuple[Vec2, Vec2]]:
"""Return circle showing the audio range."""
from .gizmo 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: AudioSource | AudioClip | None = None, **kwargs):
super().__init__(**kwargs)
self._init_playback(stream)
[docs]
def on_enter_tree(self):
# Per entry, not once: ``_exit_tree`` stops the sound, so an autoplay
# player brought back into the tree has to be started again.
self._autoplay_check()
# ------------------------------------------------------------------
# Spatial helpers
# ------------------------------------------------------------------
def _attenuated_volume_db(self, distance: float) -> float:
"""Return distance-attenuated volume_db for a given listener distance.
Includes any active fade offset on top of the base ``volume_db``.
"""
if distance > self.max_distance:
return -80.0
# volume_db = base - 80 * (dist / max_dist) ^ attenuation
dist_ratio = distance / self.max_distance
return self._effective_volume_db() - 80.0 * (dist_ratio**self.attenuation)
def _compute_positional_pan(self) -> float:
"""Compute the positional stereo pan (-1=left, 0=center, 1=right)."""
tree = self.tree
listener = tree.audio_listener_2d() if tree is not None else None
# World space on both sides: a listener parented under a camera has a
# local position relative to that camera, not the ear's scene position.
listener_x = float(listener.world_position.x) if listener is not None else 0.0
dx = self.world_position.x - listener_x
return clamp(dx / self.max_distance, -1.0, 1.0)
def _push_audio_state(self, backend, channel) -> None:
"""Send distance-attenuated volume + (override or positional) pan."""
tree = self.tree
listener = tree.audio_listener_2d() if tree is not None else None
if listener is not None:
distance = (self.world_position - listener.world_position).length()
else:
distance = self.world_position.length()
volume = self._attenuated_volume_db(distance)
pan = self.pan_override if self.pan_override is not None else self._compute_positional_pan()
backend.update_audio_2d(channel, volume, pan)
[docs]
def set_pan_and_gain(self, pan: float, gain_db: float) -> None:
"""Atomically set ``pan_override`` + ``volume_db`` with one backend call.
Use when both must land on the same audio frame (e.g. fade-out before
retrigger). Setting them as separate Property assignments would fire
two backend updates.
"""
self._suppress_audio_push = True
try:
self.pan_override = pan
self.volume_db = gain_db
finally:
self._suppress_audio_push = False
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is not None and channel is not None:
self._push_audio_state(backend, channel)
[docs]
def on_update(self, delta: float):
"""Update 2D spatialization each frame; tick fades; auto-free."""
# Pending reap (from fade_out → stop) must run even after _playing
# flips to False, otherwise queue_free_on_end never fires for the
# fade-out-into-cleanup pattern. Check first so the early-return
# path below doesn't skip it.
if self._check_queue_free_on_end():
return
if not self._playing or self._paused:
return
self._tick_fade(delta)
backend = self._get_backend()
if backend and self._backend_channel is not None:
self._push_audio_state(backend, self._backend_channel)
self._check_queue_free_on_end()
[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:
# Pre-compute attenuated volume + pan from the current listener
# state so the backend can apply them *before* the first audio
# buffer is rendered. Without this, the channel plays unattenuated
# + centred for one buffer (~20-100 ms depending on backend)
# before the per-frame ``update_audio_2d`` lands: audible for
# short SFX (footsteps, gunshots) or sources spawned beyond
# max_distance. Fixes bug-audio-legacy-spatial-first-frame.
tree = self.tree
listener = tree.audio_listener_2d() if tree is not None else None
if listener is not None:
distance = (self.world_position - listener.world_position).length()
else:
distance = self.world_position.length()
initial_volume = self._attenuated_volume_db(distance)
initial_pan = self.pan_override if self.pan_override is not None else self._compute_positional_pan()
self._claim_channel(
backend,
backend.play_audio(
self.stream,
mode="2d",
position=self.world_position,
volume_db=initial_volume,
pitch=self.pitch_scale,
loop=self.loop,
bus=self.bus,
max_distance=self.max_distance,
from_position=from_position,
pan=float(initial_pan),
),
)
# ============================================================================
# AudioPlayer3D: 3D spatial audio
# ============================================================================
[docs]
class AudioPlayer3D(_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 [live]",
group="Playback",
on_change="_on_volume_db_changed",
)
pitch_scale = Property(
1.0,
range=(0.5, 2.0),
hint="Playback speed [live: pushed mid-playback via backend.set_pitch]",
group="Playback",
on_change="_on_pitch_scale_changed",
)
bus = Property(
"SFX",
enum=["Master", "Music", "SFX", "Voice", "UI"],
hint="Audio bus [next_play]",
group="Playback",
on_change="_on_bus_changed",
)
autoplay = Property(
False,
hint="Auto-play on ready [next_play]",
group="Playback",
on_change="_on_autoplay_changed",
)
loop = Property(
False,
hint="Loop playback [next_play]",
group="Playback",
on_change="_on_loop_changed",
)
queue_free_on_end = Property(
False,
hint="Remove this node from the tree once playback finishes [next_play]",
group="Playback",
on_change="_on_queue_free_on_end_changed",
)
max_distance = Property(
100.0,
range=(1.0, 1000.0),
hint="Max hearing distance [live]",
group="Spatial",
on_change="_on_spatial_changed",
)
attenuation = Property(
1.0,
range=(0.1, 4.0),
hint="Distance attenuation exponent [live]",
group="Spatial",
on_change="_on_spatial_changed",
)
doppler_scale = Property(
0.0,
range=(0.0, 4.0),
hint="Doppler effect strength [live]",
group="Spatial",
on_change="_on_spatial_changed",
)
pan_override = Property(
None,
hint="Override directional pan; None = use world position [live]",
group="Playback",
on_change="_on_spatial_changed",
)
gizmo_colour = Colour((0.6, 0.4, 1.0, 0.5))
[docs]
def get_gizmo_lines(self) -> list[tuple[Vec3, Vec3]]:
"""Return 3 circles showing the audio range sphere."""
from .gizmo 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: AudioSource | AudioClip | None = None, **kwargs):
super().__init__(**kwargs)
self._init_playback(stream)
self._prev_position: Vec3 = Vec3() # For Doppler
[docs]
def on_enter_tree(self):
# Per entry, not once. ``_exit_tree`` stops the sound, and the Doppler
# baseline has to be re-seeded or the first frame back reads as one
# enormous velocity.
self._prev_position = self.world_position
self._autoplay_check()
# ------------------------------------------------------------------
# Spatial helpers
# ------------------------------------------------------------------
def _compute_3d_state(self, delta: float) -> tuple[float, float, float]:
"""Return (volume_db, pan, pitch) for the current frame.
``delta`` drives Doppler. Pass 0.0 from on_change handlers (no
Doppler kick on a Property nudge: Doppler reapplies next frame).
``_prev_position`` is left untouched here; ``on_update()`` is the
sole owner of that state to keep Doppler velocity stable.
"""
tree = self.tree
listener = tree.audio_listener_3d() if tree is not None else None
if listener is not None:
# World space on both sides: the canonical setup parents the
# listener under a moving camera, so its local position is the
# camera-relative offset (usually zero), not where the ear is.
listener_pos = listener.world_position
listener_forward = listener.forward
listener_up = listener.up
else:
# No camera and no listener: degrade to origin/default orientation.
# Already warned by _autocreate_listener_3d.
listener_pos = Vec3()
listener_forward = Vec3(0, 0, -1)
listener_up = Vec3(0, 1, 0)
to_source = self.world_position - listener_pos
distance = to_source.length()
# Distance attenuation
if distance > self.max_distance:
volume = -80.0
else:
d = max(distance, 0.1) # avoid divide-by-zero at the listener
dist_ratio = d / self.max_distance
volume = self._effective_volume_db() - 80.0 * (dist_ratio**self.attenuation)
# Pan: positional unless caller has overridden it
if self.pan_override is not None:
pan = float(self.pan_override)
elif distance > 0.01:
to_source_norm = to_source / distance
right = listener_forward.cross(listener_up)
pan = clamp(to_source_norm.dot(right), -1.0, 1.0)
else:
pan = 0.0
# Doppler: pitch shift from radial 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 = velocity.dot(-to_source) / (distance if distance > 0.1 else 0.1)
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)
return volume, pan, pitch
def _push_audio_state(self, backend, channel) -> None:
"""Send current 3D volume/pan/pitch (without advancing Doppler velocity)."""
volume, pan, pitch = self._compute_3d_state(0.0)
backend.update_audio_3d(channel, volume, pan, pitch)
[docs]
def set_pan_and_gain(self, pan: float, gain_db: float) -> None:
"""Atomically set ``pan_override`` + ``volume_db`` with one backend call."""
self._suppress_audio_push = True
try:
self.pan_override = pan
self.volume_db = gain_db
finally:
self._suppress_audio_push = False
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is not None and channel is not None:
self._push_audio_state(backend, channel)
[docs]
def on_update(self, delta: float):
"""Update 3D spatialization each frame; tick fades; auto-free."""
# Pending reap (from fade_out → stop) must run even after _playing
# flips to False. See AudioPlayer2D.on_update for rationale.
if self._check_queue_free_on_end():
return
if not self._playing or self._paused:
return
self._tick_fade(delta)
volume, pan, pitch = self._compute_3d_state(delta)
self._prev_position = Vec3(self.world_position)
backend = self._get_backend()
if backend and self._backend_channel is not None:
backend.update_audio_3d(self._backend_channel, volume, pan, pitch)
self._check_queue_free_on_end()
[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:
# Pre-compute attenuated volume + pan + pitch from the current
# 3D listener state so the backend can apply them *before* the
# first audio buffer. Mirrors AudioPlayer2D.play. Doppler
# is intentionally skipped on the first frame (delta=0.0) so a
# stationary source isn't kicked by an undefined prev_position
# delta.
initial_volume, initial_pan, initial_pitch = self._compute_3d_state(0.0)
self._claim_channel(
backend,
backend.play_audio(
self.stream,
mode="3d",
position=self.world_position,
volume_db=initial_volume,
pitch=initial_pitch,
loop=self.loop,
bus=self.bus,
max_distance=self.max_distance,
from_position=from_position,
pan=float(initial_pan),
),
)