"""
Shared audio playback control: :class:`_AudioPlaybackMixin`.
Private leaf module behind the :mod:`simvx.core.audio` facade. Provides the
play/stop/pause/fade/crossfade/queue-free state machine shared by every
player node. The player nodes live in the sibling ``_audio_players`` module;
the stream/data layer lives in ``_audio_stream``.
"""
from __future__ import annotations
import logging
import math
from typing import Any
from . import audio_errors
from ._audio_stream import AudioClip, AudioSource
from .audio_errors import (
AudioError,
AudioMutationDuringPlaybackError,
raise_or_warn,
warn_once,
)
log = logging.getLogger(__name__)
#: Quietest level the engine represents; a fade reaching it is silent.
_SILENCE_DB = -80.0
#: How a fade travels between two gains.
#:
#: ``"equal_gain"`` moves the amplitude in a straight line, which is what a
#: fader does and what a single fade in or out should sound like.
#: ``"equal_power"`` follows a quarter-turn of sine, so two players fading in
#: opposite directions sum to a steady loudness instead of dipping through the
#: middle. It is the right law for a crossfade between unrelated sounds and the
#: wrong one for a single fade, which it makes linger.
FADE_CURVES = ("equal_gain", "equal_power")
def _as_clip(source: AudioSource | AudioClip) -> AudioClip:
"""``source`` as an :class:`AudioClip`, which is what the players hold."""
return source if isinstance(source, AudioClip) else AudioClip(source)
# ============================================================================
# _AudioPlaybackMixin: Shared play/stop/pause/is_playing logic
# ============================================================================
class _AudioPlaybackMixin:
"""Mixin providing common audio playback state and backend interaction.
Subclasses call :meth:`_init_playback` from ``__init__`` and define the
``volume_db``, ``pitch_scale``, ``bus``, ``autoplay`` and ``loop``
Properties.
"""
def _init_playback(self, stream: AudioSource | AudioClip | None):
# Written past the accessor: construction is not a mutation, so the
# next_play policy has nothing to say about it, and the state the
# policy reads does not exist yet at this point.
self._stream: AudioClip | None = None if stream is None else _as_clip(stream)
self._playing: bool = False
self._paused: bool = False
self._backend_channel: Any = None
# The backend that granted ``_backend_channel``. A channel belongs to
# the backend it came from, and that backend is not always reachable
# through the tree when the channel has to go back: a node is
# unlinked from the tree before its exit hooks run, so the pair is
# kept together for the lifetime of the channel.
self._channel_backend: Any = None
# Suppresses on_change-driven backend pushes while ``set_pan_and_gain``
# writes both Properties so the dual-write lands as one backend call.
self._suppress_audio_push: bool = False
# Fade modulation applied on top of ``volume_db``, as a linear gain
# multiplier: 1.0 is the configured volume and 0.0 is silence. Held as
# a gain rather than a dB offset because a target of "silence" is then
# 0.0 whatever ``volume_db`` is doing, so raising the volume mid-fade
# cannot leave the fade landing somewhere above silence.
self._fade_gain: float = 1.0
self._fade_from_gain: float = 1.0
self._fade_to_gain: float = 1.0
# Elapsed and total rather than a per-second step: ``duration`` is a
# deadline, so a fade lands when its time is up however far it had to
# travel. Integrating a step instead would drift, and a fade with
# nowhere to travel would never arrive at all.
self._fade_elapsed: float = 0.0
self._fade_duration: float = 0.0
self._fade_curve: str = "equal_gain"
self._fade_stop_on_complete: bool = False
# One-shot reap flag: set when an *internal* stop fires for a player
# with ``queue_free_on_end=True`` (typically fade_out's terminal stop
# or a natural-end watcher path). The next ``on_update`` tick honours
# the flag and removes the node from its parent. Decoupled from the
# "poll for channel-end" path in ``_check_queue_free_on_end`` so the
# two state machines compose: fade-out into stop still reaps cleanly.
self._queue_free_on_end_pending: bool = False
@property
def stream(self) -> AudioClip | None:
"""What this player plays: an :class:`AudioClip`, or ``None``.
Accepts anything :class:`AudioClip` does -- a path, a
:class:`~simvx.core.Resource`, a ``Traversable`` -- and stores the clip
it wraps, so reading back always gives a clip.
**A new stream takes effect at the next** ``play()``, like ``bus``,
``loop``, ``autoplay``, ``stream_mode``, ``buffer_size`` and
``queue_free_on_end``. Assigning one while a channel is live cannot
change the sound already going out: a backend channel is opened around
the clip it was given. So the write follows the same policy as the rest
of that group and raises :class:`AudioMutationDuringPlaybackError` in
strict mode (the dev default), or warns once and defers otherwise.
``stop()`` first when swapping the sound of a live player::
player.stop()
player.stream = other_clip
player.play()
Assigning and then calling ``play()`` needs no ``stop()``: ``play()``
re-reads this attribute, and a player that is not playing has nothing
to interrupt.
"""
return self._stream
@stream.setter
def stream(self, value: AudioSource | AudioClip | None) -> None:
clip = None if value is None else _as_clip(value)
if clip is self._stream:
return
self._stream = clip
self._handle_next_play_mutation("stream")
def _get_backend(self):
tree = self.tree
return tree.audio_backend if tree is not None else None
def _owning_backend(self):
"""Return the backend a live channel must be handed back to.
Falls back to the tree's backend when no channel is bound, so a
player that has never played still resolves the one it would use.
"""
return self._channel_backend if self._channel_backend is not None else self._get_backend()
def _claim_channel(self, backend, channel) -> None:
"""Bind a freshly acquired ``channel`` to the ``backend`` that granted it.
Called by every player at the point it opens a channel: the handle
alone is not enough to release one, since the release can happen
after the player has lost its tree link.
"""
self._backend_channel = channel
self._channel_backend = backend if channel is not None else None
# ------------------------------------------------------------------
# Live property push: shared on_change handler
# ------------------------------------------------------------------
def _push_audio_state(self, backend, channel) -> None:
"""Push current volume/pan/pitch through ``backend`` for ``channel``.
Subclasses override this to supply spatial pan/pitch. The base
non-positional player sends pan=0.
"""
backend.update_audio_2d(channel, self._effective_volume_db(), 0.0)
def _on_volume_db_changed(self) -> None:
"""Push live volume change to the active backend channel.
No-op while not playing or before the backend wires up; tolerates
partial state because Property on_change can fire during ``__init__``.
"""
if getattr(self, "_suppress_audio_push", False):
return
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is None or channel is None:
return
self._push_audio_state(backend, channel)
# ------------------------------------------------------------------
# Property mutation policy: live / next_play
# ------------------------------------------------------------------
def _is_playing(self) -> bool:
"""Return True iff this player owns an active backend channel.
Used by ``next_play`` Property handlers to gate the strict/warn
path. Independent of ``is_playing()`` (the public accessor) which
also checks the paused flag: a paused channel is still bound to
the backend and so should be considered "playing" for the
mutation-policy check.
"""
return getattr(self, "_backend_channel", None) is not None
def _handle_next_play_mutation(self, prop_name: str) -> None:
"""Strict-raise or warn-once when a ``next_play`` Property is written mid-playback.
``next_play`` state (``stream``, ``bus``, ``loop``, ``autoplay``,
``stream_mode``, ``buffer_size``, ``queue_free_on_end``) requires
a fresh ``play()`` call to take effect. Mutating them while a
channel is active either raises :class:`AudioMutationDuringPlaybackError`
(strict mode, dev default) or logs a one-time warning and defers
the change to the next ``play()`` (non-strict).
"""
if not self._is_playing():
return
player_repr = f"{type(self).__name__}(name={self.name!r})"
if audio_errors.STRICT:
raise AudioMutationDuringPlaybackError(prop_name, player=player_repr)
warn_once(
f"audio.player.{prop_name}.mid_play",
"Property %r on %s changed mid-playback; deferred to next play(). "
"Set SIMVX_AUDIO_STRICT=1 to raise instead.",
prop_name,
player_repr,
)
def _on_pitch_scale_changed(self) -> None:
"""Push live pitch change to the active backend channel (live policy).
Mirrors :meth:`_on_volume_db_changed`. ``set_pitch`` is fed the
already-clamped Property value (clamp range ``[0.5, 2.0]`` lives
on the Property declaration), so the backend never sees an
out-of-bounds pitch.
"""
if getattr(self, "_suppress_audio_push", False):
return
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is None or channel is None:
return
set_pitch = getattr(backend, "set_pitch", None)
if set_pitch is not None:
set_pitch(channel, float(self.pitch_scale))
def _on_bus_changed(self) -> None:
self._handle_next_play_mutation("bus")
def _on_loop_changed(self) -> None:
self._handle_next_play_mutation("loop")
def _on_autoplay_changed(self) -> None:
self._handle_next_play_mutation("autoplay")
def _on_stream_mode_changed(self) -> None:
self._handle_next_play_mutation("stream_mode")
def _on_buffer_size_changed(self) -> None:
self._handle_next_play_mutation("buffer_size")
def _on_queue_free_on_end_changed(self) -> None:
self._handle_next_play_mutation("queue_free_on_end")
def _on_spatial_changed(self) -> None:
"""Live push for spatial Properties (max_distance / attenuation / doppler_scale).
These all feed into ``_push_audio_state`` (which spatial subclasses
override to compute volume/pan/pitch from the latest values). One
canonical handler avoids duplicating the same body on every
Property.
"""
if getattr(self, "_suppress_audio_push", False):
return
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is None or channel is None:
return
self._push_audio_state(backend, channel)
def stop(self):
"""Stop playback."""
self._playing = False
self._paused = False
# Clear fade so the next play() starts at full volume.
self._fade_gain = 1.0
self._fade_from_gain = 1.0
self._fade_to_gain = 1.0
self._fade_elapsed = 0.0
self._fade_duration = 0.0
self._fade_stop_on_complete = False
backend = self._owning_backend()
channel = self._backend_channel
if backend is not None and channel is not None:
# Unbind before releasing: a second stop must not offer the
# backend a channel it has already taken back.
self._backend_channel = None
self._channel_backend = None
backend.stop_audio(channel)
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 is_paused(self) -> bool:
"""Check if audio is currently paused (call ``play()`` to resume)."""
return 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
# ------------------------------------------------------------------
# Fade / crossfade
# ------------------------------------------------------------------
def _effective_volume_db(self) -> float:
"""Current playback volume in dB, including any active fade.
``volume_db`` is the user-set target and the fade is a gain multiplier
on top of it, so a fade to silence lands on silence whatever the user
does to ``volume_db`` while it runs.
"""
if self._fade_gain <= 0.0:
return _SILENCE_DB
return max(_SILENCE_DB, self.volume_db + 20.0 * math.log10(self._fade_gain))
def _faded_gain_at(self, t: float) -> float:
"""Gain part-way through the current fade, ``t`` running 0 to 1."""
a, b = self._fade_from_gain, self._fade_to_gain
if self._fade_curve == "equal_power":
# A quarter-turn of sine rather than a straight line. Two players
# crossfading under it sum to constant power, so the pair holds a
# steady loudness instead of dipping through the middle; a single
# fade under it lingers, which is why it is not the default.
if b >= a:
return a + (b - a) * math.sin(t * math.pi * 0.5)
return b + (a - b) * math.cos(t * math.pi * 0.5)
return a + (b - a) * t
def _start_fade(
self,
target_gain: float,
duration: float,
*,
stop_on_complete: bool,
curve: str = "equal_gain",
) -> None:
"""Ramp the fade gain from where it is now to ``target_gain``.
Starts from the current gain rather than from a fixed point, so calling
this during a running fade retargets that fade in place instead of
jumping. ``duration <= 0`` arrives immediately.
"""
if curve not in FADE_CURVES:
raise ValueError(f"curve must be one of {FADE_CURVES}, got {curve!r}")
self._fade_curve = curve
if duration <= 0.0:
self._fade_gain = target_gain
self._fade_duration = 0.0
self._fade_elapsed = 0.0
self._fade_stop_on_complete = False
if stop_on_complete:
self._arm_reap_and_stop()
return
self._fade_from_gain = self._fade_gain
self._fade_to_gain = target_gain
self._fade_elapsed = 0.0
self._fade_duration = duration
self._fade_stop_on_complete = stop_on_complete
def _arm_reap_and_stop(self) -> None:
"""Stop as the terminal step of a fade, letting ``queue_free_on_end`` reap."""
if getattr(self, "queue_free_on_end", False) and not getattr(self, "loop", False):
self._queue_free_on_end_pending = True
self.stop()
def _tick_fade(self, delta: float) -> bool:
"""Advance the active fade. Returns True if the audio state needs a push."""
if self._fade_duration <= 0.0:
return False
self._fade_elapsed += delta
if self._fade_elapsed >= self._fade_duration:
self._fade_gain = self._fade_to_gain
self._fade_duration = 0.0
if self._fade_stop_on_complete:
self._fade_stop_on_complete = False
self._arm_reap_and_stop()
return True
self._fade_gain = self._faded_gain_at(self._fade_elapsed / self._fade_duration)
return True
def fade_out(self, duration: float, *, curve: str = "equal_gain") -> None:
"""Ramp volume down to silence over ``duration`` seconds, then stop.
Ramps from the volume the player currently has, including any fade
already running, so calling this during another fade retargets it
rather than restarting from full volume.
``duration`` is a deadline, not a rate: playback stops that many
seconds from now however far the volume had to fall, so code that
schedules a scene change against the fade never has to inspect the
player first. A no-op for a player that is not playing.
"""
if not self._playing or self._backend_channel is None:
return
self._start_fade(0.0, duration, stop_on_complete=True, curve=curve)
def fade_in(self, duration: float, *, curve: str = "equal_gain") -> None:
"""Ramp volume up to ``volume_db`` over ``duration`` seconds.
A player that is not yet playing starts from silence. One that is
already audible ramps from where it is, so interrupting a ``fade_out``
with ``fade_in`` brings the sound back from the level it had reached
instead of cutting to silence first. A player already at full volume
has nothing to move and the call does nothing.
"""
if not self._playing:
self._fade_gain = 0.0
self.play()
self._start_fade(1.0, duration, stop_on_complete=False, curve=curve)
def fade_to(self, gain: float, duration: float, *, curve: str = "equal_gain") -> None:
"""Ramp to ``gain`` over ``duration`` seconds and hold there.
``gain`` is a multiplier on the configured ``volume_db``: ``1.0`` is
that volume, ``0.0`` is silence, ``0.25`` is a quarter of the
amplitude. It is relative because a player's output also passes
through distance attenuation and the bus chain, so no method here can
promise an absolute level.
This is the primitive the other fades are written in terms of, and the
one to reach for when ducking::
music.fade_to(0.25, 0.3) # under the cutscene
music.fade_to(1.0, 0.5) # and back
"""
self._start_fade(max(0.0, min(1.0, gain)), duration, stop_on_complete=False, curve=curve)
def _poll_natural_end(self) -> bool:
"""Notice a channel that has played out and clear the playing state.
Runs on every tick of every player, whatever ``queue_free_on_end``
says: a sound that has finished is finished, and ``is_playing()`` is
the obvious way to ask. Returns True when this call ended playback.
The channel is released through :meth:`stop` while the handle is
still here. A backend that ended the sound itself may still hold
what is behind it (the native backend keeps its ``ma_sound`` until
told to stop), so dropping the handle first would leave nothing to
release it with and the entry would outlive the node. ``stop_audio``
on a channel the backend has already dropped is a no-op everywhere,
so an ended channel is never stopped twice.
A looping sound has no natural end, and a paused one is still bound
to the backend: neither is polled.
"""
if not self._playing or self._paused or getattr(self, "loop", False):
return False
channel = self._backend_channel
if channel is None:
return False
backend = self._owning_backend()
if backend is None:
return False
is_active = getattr(backend, "is_channel_active", None)
if is_active is None or is_active(channel):
return False
# Arm the reap before stopping: ``stop`` clears the handle, so the
# queue-free path below could no longer tell a finished sound from
# one the user stopped by hand.
if getattr(self, "queue_free_on_end", False):
self._queue_free_on_end_pending = True
self.stop()
return True
def _check_queue_free_on_end(self) -> bool:
"""Poll for the end of playback, reaping the node when it is due.
Two things happen here, in this order, on every tick:
1. :meth:`_poll_natural_end` notices a channel the backend has
finished with and clears the playing state. This is what makes
``is_playing()`` answer ``False`` after a sound plays out, and it
runs whether or not ``queue_free_on_end`` is set.
2. A *pending* reap is honoured, removing this node from its parent
and returning True. The flag is set by the poll above and by
internal terminal stops (``fade_out`` ramping to silence, a
chunk-fed stream on a backend that cannot drain), each of which
calls ``stop()`` and so clears the handle the poll reads.
Loops never trigger queue-free: they only end via explicit ``stop``.
Explicit-stop paths are handled by ``_exit_tree`` so they don't
double-free.
"""
self._poll_natural_end()
if not self._queue_free_on_end_pending:
return False
self._queue_free_on_end_pending = False
if not getattr(self, "queue_free_on_end", False) or getattr(self, "loop", False):
return False
parent = getattr(self, "parent", None)
if parent is not None:
parent.remove_child(self)
return True
def pitch_modulate(self, scale: float) -> None:
"""Live pitch shift on an already-playing stream.
Sets ``pitch_scale`` and pushes the new resampler ratio to the active
backend channel. Without this, ``pitch_scale = x`` does not propagate
mid-stream (HexGL engine hum doppler regression). The base 2D player
path has no per-frame pitch update, so we route through the
backend's ``set_pitch`` method directly.
Clamped to the ``pitch_scale`` Property's [0.5, 2.0] range.
"""
clamped = max(0.5, min(2.0, float(scale)))
self.pitch_scale = clamped
backend = self._get_backend()
channel = getattr(self, "_backend_channel", None)
if backend is None or channel is None:
return
set_pitch = getattr(backend, "set_pitch", None)
if set_pitch is not None:
set_pitch(channel, clamped)
def crossfade(self, other: _AudioPlaybackMixin, duration: float, *, curve: str = "equal_power") -> None:
"""Hand playback over to ``other``: it fades in as this one fades out and stops.
Both halves are armed in the same frame and share ``duration``, and
because that is a deadline they land together even when they start from
different levels.
The default curve differs from the single fades on purpose. Two sounds
crossfading with a straight amplitude ramp are each at half amplitude
in the middle, so the pair dips; the sine law used here holds the sum
steady. Pass ``curve="equal_gain"`` for two takes of the *same* sound,
where the signals reinforce rather than add as separate sources.
``other`` is not restarted if it is already playing, so a stream
assigned to it after it started is not picked up by this call.
"""
if other is self:
raise ValueError("crossfade needs a second player; use fade_out to stop this one")
other.fade_in(duration, curve=curve)
self.fade_out(duration, curve=curve)
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
# A pending reap from a previous fade_out terminal stop is invalidated
# by an explicit re-play: the user is reusing this player, not
# cleaning it up.
self._queue_free_on_end_pending = False
if from_position == 0.0 and self._resume_if_paused():
return False
# Stop any active channel from a previous play() before starting a new
# one, so rapid re-trigger on the same player doesn't pile up
# overlapping sounds. Use multiple AudioPlayers (pool pattern) if
# you genuinely want overlapping plays of the same stream.
if self._backend_channel is not None:
backend = self._owning_backend()
if backend is not None:
try:
backend.stop_audio(self._backend_channel)
except AudioError as exc:
raise_or_warn(
exc,
key="audio.player.stop_previous_failed",
message="Failed to stop previous channel before re-trigger",
)
self._backend_channel = None
self._channel_backend = None
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()