Source code for simvx.core.audio_protocol

"""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 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 AudioStream
    from .audio_bus import AudioBusLayout

__all__ = [
    "AudioBackend",
    "AudioBusBackend",
    "AudioPlaybackBackend",
    "AudioStreamingBackend",
    "CAPABILITIES_CORE",
    "Capability",
    "PlayMode",
]


[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" 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"``. """
[docs] def play_audio( self, stream: AudioStream, *, 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, stream: AudioStream | 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`. """ ...
[docs] def feed_audio_chunk(self, channel_id: int, chunk: bytes) -> None: ...
[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 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`. """