"""Backend-agnostic audio interface.
The audio system talks to three thin structural :class:`Protocol` types:
* :class:`AudioPlaybackBackend`: start / stop / pause / live-update sounds.
* :class:`AudioStreamingBackend`: chunk-fed PCM streaming (for procedural
synthesis, AudioWorklet, etc.).
* :class:`AudioBusBackend`: bus-layout reconciliation and capability
advertisement.
A fourth, one-member facet, :class:`TreeBoundAudioBackend`, is implemented only
by a backend that has no device clock of its own and paces itself by the engine.
A concrete backend implements whichever facets it can. The union
:class:`AudioBackend` keeps the legacy "everything in one Protocol" surface
for callers that genuinely need all three (e.g. ``MiniaudioBackend``,
``WebAudioBackend``); narrower callers should depend on the smallest
Protocol they actually use and reach the backend via
:attr:`SceneTree.audio_playback` / :attr:`audio_streaming` /
:attr:`audio_buses` so the type checker enforces the facet boundary.
Capabilities are a :class:`enum.StrEnum` so typos are caught at import.
Existing string callers keep working: ``Capability.PLAY_BASIC`` compares
equal to ``"play.basic"`` and round-trips through ``frozenset`` literally.
Channels are opaque integer handles allocated by the backend. The backend
owns the mapping from handle to native voice; the Python layer treats them
as bookmarks.
"""
from __future__ import annotations
from enum import StrEnum
from typing import TYPE_CHECKING, Any, Literal, Protocol, runtime_checkable
if TYPE_CHECKING:
from .audio import AudioClip
from .audio_bus import AudioBusLayout
from .scene_tree import SceneTree
__all__ = [
"AudioBackend",
"AudioBusBackend",
"AudioPlaybackBackend",
"AudioStreamDraining",
"AudioStreamPacing",
"AudioStreamingBackend",
"CAPABILITIES_CORE",
"Capability",
"PlayMode",
"TreeBoundAudioBackend",
]
[docs]
class Capability(StrEnum):
"""Capabilities a backend can advertise via :meth:`AudioBusBackend.list_capabilities`.
Members are strings so ``Capability.PLAY_BASIC in caps`` works against
either a ``frozenset[Capability]`` or a legacy ``frozenset[str]``. New
capability flags must be declared here: string literals in backend code
bypass the typo check and are banned.
"""
PLAY_BASIC = "play.basic"
PLAY_2D = "play.2d"
PLAY_3D = "play.3d"
SPATIAL_HRTF = "spatial.hrtf"
SPATIAL_DOPPLER = "spatial.doppler"
STREAMING = "streaming"
# The three compressed-container flags mean "this backend has a streaming
# decoder of its own", so a caller can hand it the file and let it do the
# work. A backend without one still streams the container: the caller
# decodes to PCM and feeds it chunk by chunk through the ring.
STREAMING_WAV = "streaming.wav"
STREAMING_OGG = "streaming.ogg"
STREAMING_MP3 = "streaming.mp3"
STREAMING_FLAC = "streaming.flac"
EFFECT_GAIN = "effect.gain"
EFFECT_FILTER_BIQUAD = "effect.filter_biquad"
EFFECT_PARAMETRIC_EQ = "effect.parametric_eq"
EFFECT_DELAY = "effect.delay"
EFFECT_REVERB = "effect.reverb"
EFFECT_COMPRESSOR = "effect.compressor"
EFFECT_SOFTCLIP = "effect.softclip"
CAPABILITIES_CORE: frozenset[Capability] = frozenset(
{
Capability.PLAY_BASIC,
Capability.PLAY_2D,
Capability.PLAY_3D,
Capability.STREAMING,
Capability.STREAMING_WAV,
}
)
"""Capabilities every backend that participates in playback + streaming advertises.
The Null backend deliberately omits the streaming members because it does
not implement :class:`AudioStreamingBackend`; see ``NullAudioBackend`` for
its narrowed capability set.
"""
PlayMode = Literal["non_positional", "2d", "3d"]
"""Mode kwarg for :meth:`AudioPlaybackBackend.play_audio`.
The Node-side dispatcher (``_AudioPlaybackMixin._play_common``) selects
the mode from the player class; backends branch on it internally so the
spatial / non-spatial code paths share one entry point.
"""
[docs]
@runtime_checkable
class AudioPlaybackBackend(Protocol):
"""Backend facet for starting and controlling individual sounds.
The unified :meth:`play_audio` collapses the historical
``play_audio`` / ``play_audio_2d`` / ``play_audio_3d`` trio into one
entry point parameterised by ``mode``. Backends branch on ``mode``
internally; spatial kwargs (``position``, ``max_distance``) are
ignored when ``mode == "non_positional"``.
Channel lifetime. A channel id is owned by whoever started it and is
given back with :meth:`stop_audio`, which must be safe to call on a
channel that has already ended or already been released. A backend
may not assume every channel is released: a caller can forget one, so
a backend that holds per-channel resources reclaims the channels that
have played out by itself rather than keeping them until shutdown.
:meth:`is_channel_active` must answer ``False`` for both a finished
channel and an unknown id, so the two are indistinguishable to the
caller and reclaiming early is always safe.
"""
[docs]
def play_audio(
self,
stream: AudioClip,
*,
mode: PlayMode = "non_positional",
position: Any = None,
volume_db: float = 0.0,
pitch: float = 1.0,
loop: bool = False,
bus: str = "Master",
max_distance: float = 100.0,
from_position: float = 0.0,
pan: float = 0.0,
gain_db: float = 0.0,
) -> int | None:
"""Play ``stream``. Returns the channel id or ``None`` on failure.
``mode`` selects the spatial path:
* ``"non_positional"``: background music / UI sounds. ``position``
+ ``max_distance`` are ignored; ``bus`` defaults to ``"Master"``.
* ``"2d"``: Node-side spatialization computes ``pan`` + ``volume_db``
from the listener; the backend just applies them before the
first audio buffer is rendered. ``bus`` defaults to ``"SFX"``.
* ``"3d"``: same as 2D, plus :meth:`update_audio_3d` per frame for
Doppler. The Node pre-computes initial pitch and forwards it.
``from_position`` seeks the playback cursor (in seconds) before the
first sample is rendered. ``pan`` (``[-1, 1]``, ``0`` = centre) is
applied *before* ``sound.start()`` so spatial players' first audio
buffer is correctly panned.
"""
...
[docs]
def stop_audio(self, channel_id: int) -> None: ...
[docs]
def pause_audio(self, channel_id: int) -> None: ...
[docs]
def resume_audio(self, channel_id: int) -> None: ...
[docs]
def update_audio_2d(self, channel_id: int, volume_db: float, pan: float) -> None: ...
[docs]
def update_audio_3d(self, channel_id: int, volume_db: float, pan: float, pitch: float) -> None: ...
[docs]
def set_pitch(self, channel_id: int, pitch: float) -> None: ...
# Listener pose: drives spatializers that compute pan / attenuation /
# Doppler natively (ma_engine on desktop, Web Audio PannerNode on a
# future web backend). Every shipping backend implements these as
# either a real engine call or an accepted-and-stored no-op so the
# AudioListener3D Node can push pose every frame without isinstance
# gates.
[docs]
def set_listener_position(self, x: float, y: float, z: float) -> None: ...
[docs]
def set_listener_velocity(self, x: float, y: float, z: float) -> None: ...
[docs]
def set_listener_direction(self, x: float, y: float, z: float) -> None: ...
[docs]
def set_listener_world_up(self, x: float, y: float, z: float) -> None: ...
[docs]
def get_playback_position(self, channel_id: int) -> float: ...
[docs]
def is_channel_active(self, channel_id: int) -> bool: ...
[docs]
def shutdown(self) -> None: ...
[docs]
@runtime_checkable
class AudioStreamingBackend(Protocol):
"""Backend facet for chunk-fed PCM streaming.
Procedural synthesis (``AudioSynth.attach_to``), AudioWorklet feeds,
and compressed-container streaming via native decoders all flow through
these two methods. Backends that can't stream (the Null backend, any
no-device test stub) deliberately do NOT implement this Protocol:
callers must ``isinstance(backend, AudioStreamingBackend)`` and raise
:class:`AudioCapabilityError` (or warn-once and skip) when the facet
is absent.
"""
[docs]
def open_stream(
self,
*,
volume_db: float = 0.0,
bus: str = "Master",
buffer_seconds: float = 0.5,
loop: bool = False,
stream: AudioClip | None = None,
) -> int:
"""Open a streaming channel for chunk-fed PCM playback.
If ``stream`` is provided and its ``container`` is a compressed
format (``"ogg"``/``"mp3"``/``"flac"``), backends with a native
streaming decoder open the file directly and ignore subsequent
:meth:`feed_audio_chunk` calls. For ``"wav"`` or synthetic
``"pcm"`` streams, the caller is expected to feed raw int16 stereo
bytes via :meth:`feed_audio_chunk`. Backends that cannot decode
the requested container raise :class:`AudioCapabilityError`.
``loop`` restarts the channel at the beginning when it plays out.
It applies only when the backend owns the decoder: a chunk-fed
channel loops by having its producer rewind and feed the file
again, which is the only place that knows where the start is.
"""
...
[docs]
def feed_audio_chunk(self, channel_id: int, chunk: bytes) -> None: ...
[docs]
@runtime_checkable
class AudioStreamPacing(Protocol):
"""Backend facet that lets a chunk-fed producer size its writes.
A producer with no idea how much room is left writes as fast as its
caller ticks, which reads a file to its end in a fraction of the time
the audio takes to play. These two members let it write exactly what
fits, so the feed runs at the speed of the sound.
Kept out of :class:`AudioStreamingBackend` deliberately. That Protocol is
runtime-checked to decide whether streaming is available at all, so
widening it would take streaming away from every backend that has not
implemented pacing yet. Producers isinstance-check this facet on its own
and fall back to pacing themselves against the frame clock.
"""
[docs]
def frames_available(self, channel_id: int) -> int:
"""Frames ``channel_id`` can accept right now; ``0`` when it is full.
Also ``0`` for a channel whose audio the backend decodes itself:
such a channel has no ring, so there is nothing for a producer to
write into it.
"""
...
[docs]
@runtime_checkable
class AudioStreamDraining(Protocol):
"""Backend facet that lets a chunk-fed producer say it has finished.
:class:`AudioStreamingBackend` gives a producer no way to end a stream.
The only signal available is :meth:`AudioPlaybackBackend.stop_audio`,
which tears the channel down at once and throws away everything the
producer has written but the device has not yet played: up to a whole
buffer of the end of the sound. This facet is how the producer says
"that was the last chunk" instead, and lets the backend keep the channel
alive until what is already queued has been heard.
Kept out of :class:`AudioStreamingBackend` for the same reason as
:class:`AudioStreamPacing`: that Protocol is runtime-checked to decide
whether streaming is available at all, so a backend that has not
implemented draining yet would lose streaming entirely. Producers
isinstance-check this facet on its own and fall back to the immediate
stop, which is what they did before it existed.
"""
[docs]
def end_of_stream(self, channel_id: int) -> None:
"""Declare that no further :meth:`feed_audio_chunk` call is coming.
The channel keeps playing what it already holds.
:meth:`AudioPlaybackBackend.is_channel_active` answers ``False`` for
it once that has played out, which is the producer's cue to stop it.
A no-op for a channel the backend decodes itself, for an unknown id,
and on a second call.
"""
...
[docs]
@runtime_checkable
class AudioBusBackend(Protocol):
"""Backend facet for bus-layout reconciliation and capability discovery."""
[docs]
def sync_bus_layout(self, layout: AudioBusLayout) -> None:
"""Reconcile the backend's native bus / effect graph against ``layout``.
Cheap and idempotent: backends keep their own
``dict[bus_name, native_handle]`` and add/remove/update only what
changed since the last call. Called automatically by the audio
server when :class:`AudioBusLayout` or any bus's ``effects`` list
changes; manual call is safe.
"""
...
[docs]
def list_capabilities(self) -> frozenset[Capability]:
"""Capability flags this backend supports.
Effect modules and Nodes call this before attempting to materialise
a native feature; absent capabilities raise
:class:`AudioCapabilityError` (strict) or warn-once (lenient).
"""
...
[docs]
@runtime_checkable
class TreeBoundAudioBackend(Protocol):
"""Backend facet for a backend whose clock is a :class:`SceneTree`, not a device.
A backend driven by real audio hardware knows how much time has passed
because the device tells it. A backend with no device -- the silent
:class:`~simvx.core.audio_backend.NullAudioBackend` is the only shipped one
-- has to be told, and several trees can be alive at once, so it must be
told *which* tree's ticks are its own.
:meth:`SceneTree.install_audio_backend` isinstance-checks this facet and
calls :meth:`bind_tree` on the backends that declare it. A backend that does
not implement it is simply never bound; nothing else about installation
changes.
"""
[docs]
def bind_tree(self, tree: SceneTree) -> None:
"""Adopt *tree* as the clock this backend measures its playback against.
Called once per install, with the installing tree. A re-install onto a
different tree calls it again with the new one. The backend is expected
to hold the tree weakly: a finished scene's node graph must not be kept
alive by the audio backend that outlived it.
"""
...
[docs]
@runtime_checkable
class AudioBackend(AudioPlaybackBackend, AudioStreamingBackend, AudioBusBackend, Protocol):
"""Union Protocol: a backend that implements all three facets.
``MiniaudioBackend`` (and its legacy fallback) and ``WebAudioBackend``
satisfy this. ``NullAudioBackend`` does NOT (it doesn't implement
:class:`AudioStreamingBackend`): callers that need streaming should
isinstance-check :class:`AudioStreamingBackend` directly via
:attr:`SceneTree.audio_streaming`.
"""