"""The native (``ma_engine``) production audio backend.
``MiniaudioBackend`` mixes, spatializes, and runs DSP entirely in native C via
the ``simvx.core._native.miniaudio_engine`` CFFI wrapper. Python only pushes
parameter updates. Falls back to the legacy / null backends via
``make_backend`` when the compiled extension is unavailable.
"""
from __future__ import annotations
import threading
from typing import TYPE_CHECKING, Any
import numpy as np
from .._native import miniaudio_engine as _me
from ..audio_errors import (
AudioCapabilityError,
AudioError,
InvalidStreamError,
raise_or_warn,
)
from ..audio_protocol import Capability
from ._effects import _NativeEffectChain
from ._shared import (
_DEFAULT_CHANNELS,
_DEFAULT_SAMPLE_RATE,
_DESKTOP_CAPABILITIES,
_NATIVE_EFFECT_CAPABILITIES,
_alloc_channel_id,
_db_to_linear,
_register_atexit_shutdown,
_SoundEntry,
_warn_if_no_vorbis,
)
if TYPE_CHECKING:
from ..audio import AudioClip
from ..audio_bus import AudioBus, AudioBusLayout
[docs]
class MiniaudioBackend:
"""Native-mixed audio backend on top of ``ma_engine``.
All mixing, spatialization, and DSP runs in native C: Python only
pushes parameter updates. Target latency: ``buffersize_msec=20`` (vs
the legacy backend's 100 ms).
The native extension must be built once after install:
simvx build-audio
If the extension is missing, the constructor raises
``MiniaudioEngineUnavailable``. Use ``make_backend()`` for an
automatic fallback to ``_LegacyMiniaudioBackend`` with a one-time
warning.
"""
def __init__(
self,
sample_rate: int = _DEFAULT_SAMPLE_RATE,
nchannels: int = _DEFAULT_CHANNELS,
):
self._sample_rate = sample_rate
self._nchannels = nchannels
self._lock = threading.Lock()
self._sounds: dict[int, _SoundEntry] = {}
self._groups: dict[str, _me.SoundGroup] = {}
# Per-bus native effect chains (one per group). Built lazily on
# first sync_bus_layout that sees a non-empty effects list.
self._effect_chains: dict[str, _NativeEffectChain] = {}
# Snapshot of (own gain, routing parent) last pushed to each native
# group, so sync_bus_layout is cheap to re-call every frame and only
# writes on change. The gain is the bus's own, since the group graph
# applies its ancestors; solo is the one layout-wide input to it, so
# a solo elsewhere still has to invalidate this entry.
self._bus_snapshot: dict[str, tuple[float, str]] = {}
_warn_if_no_vorbis()
self._engine = _me.Engine(sample_rate=sample_rate, channels=nchannels, device=True)
# Seed default groups so unfamiliar bus names play at master gain.
self._ensure_default_groups()
# Belt-and-braces: register an atexit hook so a `sys.exit(0)` from
# anywhere (test runners, demos that bypass App.quit()) still
# shuts down the engine and joins miniaudio's audio thread.
# Captured by weakref so the hook doesn't pin the backend alive.
_register_atexit_shutdown(self)
# -- internal helpers -----------------------------------------------------
def _ensure_default_groups(self) -> None:
from ..audio_bus import AudioBusLayout
layout = AudioBusLayout.get_default()
self.sync_bus_layout(layout)
def _ensure_group(self, bus_name: str, layout: AudioBusLayout | None = None) -> _me.SoundGroup:
"""Return the SoundGroup for ``bus_name``, lazily creating if absent.
Bus names must match ``layout`` (the active layout when it is not
given) exactly: lookup is case-sensitive. Unknown buses raise
:class:`UnknownBusError` so typos surface at the call site rather
than playing on a phantom root-attached group.
A group created here is routed straight away, so a sound played on
a fresh bus reaches its parent's gain and effects on its first
frame rather than on the next :meth:`sync_bus_layout`.
"""
group = self._groups.get(bus_name)
if group is not None:
return group
# Validate against the layout being synced before allocating native state.
if layout is None:
from ..audio_bus import AudioBusLayout
layout = AudioBusLayout.get_default()
bus = layout.get_bus(bus_name) # raises UnknownBusError on miss
try:
group = _me.SoundGroup(self._engine)
except Exception as exc:
raise AudioError(f"Failed to create native group for bus {bus_name!r}") from exc
self._groups[bus_name] = group
self._route_group(bus, layout)
return group
def _destination_node(self, parent_name: str, layout: AudioBusLayout) -> Any:
"""The node a bus's output attaches to: a parent group, or the endpoint."""
if not parent_name:
return _me.engine_endpoint(self._engine)
return _me.sound_group_as_node(self._ensure_group(parent_name, layout))
def _route_group(self, bus: AudioBus, layout: AudioBusLayout) -> None:
"""Wire ``bus``'s group (or its effect chain's tail) into its parent."""
destination = self._destination_node(layout._routing_parent(bus), layout)
chain = self._effect_chains.get(bus.name)
try:
if chain is not None:
chain.set_destination(destination)
else:
_me.attach_node(_me.sound_group_as_node(self._groups[bus.name]), 0, destination, 0)
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.sync_bus_layout.route_failed",
message=f"MiniaudioBackend: failed to route bus {bus.name!r} to its send_to target",
)
def _make_sound(
self,
stream: AudioClip,
*,
bus: str,
spatial: bool,
pitch_enabled: bool,
) -> tuple[Any, Any] | tuple[None, None]:
"""Build a ``_me.Sound`` for `stream`. Returns (sound, keeper).
`keeper` pins an ``AudioBuffer`` when the source is an ndarray;
``None`` when loaded from a file path.
"""
group = self._ensure_group(bus)
backend_data = getattr(stream, "backend_data", None)
if isinstance(backend_data, np.ndarray):
try:
buf = _me.AudioBuffer(
backend_data,
sample_rate=self._sample_rate,
channels=self._nchannels,
)
sound = _me.Sound.from_buffer(
self._engine,
buf,
group=group,
spatial=spatial,
pitch_enabled=pitch_enabled,
)
return sound, buf
except Exception as exc:
raise InvalidStreamError(f"Failed to play ndarray source: {exc}") from exc
path = getattr(stream, "path", None)
if path:
try:
sound = _me.Sound.from_file(
self._engine,
path,
group=group,
spatial=spatial,
pitch_enabled=pitch_enabled,
)
return sound, None
except Exception as exc:
raise InvalidStreamError(f"Failed to play file {path!r}: {exc}") from exc
return None, None
def _reap_finished(self) -> None:
"""Release the sounds that have played themselves out.
A one-shot channel nobody polls is never handed back: whoever
started it may have forgotten it, so without this sweep every
fire-and-forget sound keeps its ``ma_sound`` (and the PCM behind it)
for the life of the backend. Run when a new sound is allocated, so
the table is bounded by the voices actually in flight.
Paused channels are alive by definition. Streaming channels are
exempt because a starved stream reports itself at the end when its
producer has merely fallen behind, and pulling it down then would
turn an underrun into a dead channel.
"""
with self._lock:
entries = list(self._sounds.items())
for cid, entry in entries:
if entry.paused or entry.streaming:
continue
try:
finished = entry.sound.at_end() or not entry.sound.is_playing()
except Exception:
# A sound that cannot answer is left for stop_audio to deal with.
continue
if finished:
self.stop_audio(cid)
# -- public interface -----------------------------------------------------
[docs]
def play_audio(
self,
stream: AudioClip,
*,
mode: str = "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:
# 2D / 3D players compute ``pan`` + ``volume_db`` Python-side and
# forward them here so the native spatializer applies them before
# the first buffer (matches web backend behaviour exactly). HRTF
# is intentionally not used for the 2D path: wrong shape for 2D
# scenes, so the native ``spatial`` flag stays False; ``mode`` is
# accepted for Protocol parity but the native ma_engine path does
# not currently branch on it.
del mode, position, max_distance
self._reap_finished()
sound, keeper = self._make_sound(stream, bus=bus, spatial=False, pitch_enabled=True)
if sound is None:
return None
sound.set_volume(_db_to_linear(volume_db + gain_db))
sound.set_pitch(max(0.1, float(pitch)))
sound.set_looping(bool(loop))
# Apply pan *before* ``sound.start()`` so the audio thread starts
# mixing with the correct pan on its very first buffer. Without
# this, spatial players play centred for one buffer (~20 ms on the
# native path) before the per-frame ``update_audio_2d/3d`` lands.
sound.set_pan(max(-1.0, min(1.0, float(pan))))
if from_position > 0.0:
try:
sound.seek_to_frame(int(float(from_position) * self._sample_rate))
except Exception as exc:
sound.shutdown()
raise AudioError(f"seek_to_frame failed for from_position={from_position}: {exc}") from exc
try:
sound.start()
except Exception as exc:
sound.shutdown()
raise AudioError(f"sound.start failed: {exc}") from exc
cid = _alloc_channel_id()
with self._lock:
self._sounds[cid] = _SoundEntry(sound=sound, bus=bus, keeper=keeper)
return cid
[docs]
def stop_audio(self, channel_id: int) -> None:
with self._lock:
entry = self._sounds.pop(channel_id, None)
if entry is None:
return
try:
entry.sound.stop()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.stop_audio.sound_stop_failed",
message="MiniaudioBackend.stop_audio: sound.stop failed",
)
entry.sound.shutdown()
[docs]
def pause_audio(self, channel_id: int) -> None:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None:
return
try:
entry.sound.stop()
entry.paused = True
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.pause_audio.sound_stop_failed",
message="MiniaudioBackend.pause_audio: sound.stop failed",
)
[docs]
def resume_audio(self, channel_id: int) -> None:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None or not entry.paused:
return
try:
entry.sound.start()
entry.paused = False
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.resume_audio.sound_start_failed",
message="MiniaudioBackend.resume_audio: sound.start failed",
)
[docs]
def update_audio_2d(self, channel_id: int, volume_db: float, pan: float) -> None:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None:
return
entry.sound.set_volume(_db_to_linear(volume_db))
entry.sound.set_pan(max(-1.0, min(1.0, float(pan))))
[docs]
def update_audio_3d(self, channel_id: int, volume_db: float, pan: float, pitch: float) -> None:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None:
return
entry.sound.set_volume(_db_to_linear(volume_db))
entry.sound.set_pan(max(-1.0, min(1.0, float(pan))))
entry.sound.set_pitch(max(0.1, float(pitch)))
[docs]
def set_pitch(self, channel_id: int, pitch: float) -> None:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None:
return
if entry.streaming:
# Native streaming sounds are built with ``pitch_enabled=False``
# (the resampler would read past what the producer has written);
# surface that as a typed capability error rather than silently
# dropping the change.
raise AudioCapabilityError(
"pitch.streaming",
backend="MiniaudioBackend",
advertised=self.list_capabilities(),
remediation=(
"Pitch modulation is not supported on streaming sounds. "
"Use AudioPlayer with stream_mode='memory' for "
"pitch control."
),
)
entry.sound.set_pitch(max(0.1, float(pitch)))
[docs]
def get_playback_position(self, channel_id: int) -> float:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is not None and isinstance(entry.keeper, _me.StreamSource):
# A chunk-fed ring exposes no cursor (ma_pcm_rb's data source
# leaves it unimplemented, so ``cursor_frames`` would raise
# MA_NOT_IMPLEMENTED). Derive the position instead: frames the
# producer has fed minus frames still queued in the ring is
# exactly what the audio thread has consumed. During underrun
# the ring pads with silence, so the position holds at the end
# of the fed data rather than counting the padding.
played = entry.frames_fed - entry.keeper.available_read_frames
return max(0, played) / float(self._sample_rate)
if entry is None:
return 0.0
return entry.sound.cursor_frames() / float(self._sample_rate)
[docs]
def is_channel_active(self, channel_id: int) -> bool:
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None:
return False
if entry.paused:
return True
if entry.draining and isinstance(entry.keeper, _me.StreamSource):
# A ring-backed sound pads its underruns with silence rather than
# ending, so ``is_playing`` stays True for ever once the producer
# stops. Having been told the last chunk is in, the frames still
# queued are the whole of what is left to hear.
return entry.keeper.available_read_frames > 0
return entry.sound.is_playing() and not entry.sound.at_end()
[docs]
def end_of_stream(self, channel_id: int) -> None:
"""Mark a chunk-fed channel as having had its last chunk fed.
This backend's half of
:class:`~simvx.core.audio_protocol.AudioStreamDraining`. The sound
keeps playing out the ring; :meth:`is_channel_active` reports it
finished once the ring is empty. A no-op for a channel whose audio
the native decoder owns, which has no ring and ends by itself.
"""
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None or not isinstance(entry.keeper, _me.StreamSource):
return
entry.draining = True
[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.
``buffer_seconds`` (default 0.5 s) trades latency for underrun
tolerance. At 48 kHz stereo s16 = ~96 KB. Drop to 0.1 s for
low-latency interactive synthesis; bump to 1-2 s for music
streams under heavy CPU load. Producer pushes int16 stereo
bytes via ``feed_audio_chunk``; underrun produces silence
(no glitch). Only applies to the chunk-fed PCM path: when
``stream`` is a compressed container, miniaudio's
``ma_sound_init_from_file(..., MA_SOUND_FLAG_STREAM)`` owns the
buffering internally.
Container routing (when ``stream`` is provided):
* ``"wav"`` / ``"pcm"`` (synthetic) / ``None``: open an
``ma_pcm_rb``-backed sound. Caller feeds raw int16 stereo bytes.
* ``"ogg"`` / ``"mp3"`` / ``"flac"``: open the file directly via
``Sound.from_file(stream=True)``. Subsequent
:meth:`feed_audio_chunk` calls are no-ops for this channel.
* ``"unknown"``: raise :class:`InvalidStreamError`.
``loop`` applies to the compressed path only, where this backend owns
the decoder and so knows where the file starts. A ring-backed channel
holds nothing but the frames its producer has just written, so its
producer is the one that rewinds.
Pitch is intentionally not a parameter: streaming sounds are
built with ``pitch_enabled=False`` so the resampler can't read
past what the producer has written. ``set_pitch`` against the
returned channel raises :class:`AudioCapabilityError`.
"""
group = self._ensure_group(bus)
container = stream.container if stream is not None else "wav"
if container in ("ogg", "mp3", "flac"):
path = stream.path if stream is not None else ""
if not path:
raise InvalidStreamError(
f"Native streaming open: compressed container {container!r} "
"requires a file path on the AudioClip"
)
try:
sound = _me.Sound.from_file(self._engine, path, group=group, stream=True, pitch_enabled=False)
except Exception as exc:
raise AudioError(f"open_stream failed for {container} file {path!r}: {exc}") from exc
sound.set_volume(_db_to_linear(volume_db))
sound.set_looping(bool(loop))
try:
sound.start()
except Exception as exc:
sound.shutdown()
raise AudioError(f"open_stream sound.start failed: {exc}") from exc
cid = _alloc_channel_id()
with self._lock:
# ``keeper`` is None: there's no Python-side rb to pin. The
# ma_sound itself owns the decoder + buffering. ``streaming``
# stays True so ``set_pitch`` raises the same
# ``AudioCapabilityError`` as the chunk-fed path.
self._sounds[cid] = _SoundEntry(sound=sound, bus=bus, keeper=None, streaming=True)
return cid
if container == "unknown":
path = stream.path if stream is not None else "<no-path>"
raise InvalidStreamError(
f"Native streaming open: could not detect audio container for {path!r}. "
"Expected WAV/OGG/MP3/FLAC header."
)
# container in {"wav", "pcm"} (or stream is None: legacy callers
# without a stream object get the chunk-fed PCM ring path).
try:
rb = _me.StreamSource(
self._engine,
sample_rate=self._sample_rate,
channels=self._nchannels,
buffer_seconds=buffer_seconds,
format="s16",
)
sound = _me.Sound.from_stream(self._engine, rb, group=group)
except Exception as exc:
raise AudioError(f"open_stream failed: {exc}") from exc
sound.set_volume(_db_to_linear(volume_db))
try:
sound.start()
except Exception as exc:
sound.shutdown()
rb.shutdown()
raise AudioError(f"open_stream sound.start failed: {exc}") from exc
cid = _alloc_channel_id()
with self._lock:
self._sounds[cid] = _SoundEntry(sound=sound, bus=bus, keeper=rb, streaming=True)
return cid
[docs]
def frames_available(self, channel_id: int) -> int:
"""Frames the channel's ring can take right now.
``0`` for an unknown channel and for a compressed channel the native
decoder owns, neither of which has a ring to write into.
"""
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None or not isinstance(entry.keeper, _me.StreamSource):
return 0
return int(entry.keeper.available_write_frames)
[docs]
def feed_audio_chunk(self, channel_id: int, chunk: bytes) -> None:
"""Push raw int16 stereo bytes into the channel's ring buffer.
Misaligned chunks (length not a multiple of bytes-per-frame)
have the trailing partial frame trimmed. When the ring is full
the excess is silently dropped: call :meth:`frames_available`
first to write only what fits.
"""
if not chunk:
return
with self._lock:
entry = self._sounds.get(channel_id)
if entry is None or not isinstance(entry.keeper, _me.StreamSource):
return
try:
# Write and count under the one lock so a concurrent
# position query never sees frames queued in the ring that
# the fed-frames tally does not yet cover (which would read
# as the position jumping backwards).
entry.frames_fed += entry.keeper.write(chunk)
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.feed_audio_chunk.ring_write_failed",
message="MiniaudioBackend.feed_audio_chunk: ring write failed",
)
# ------------------------------------------------------------------
# 3D listener: drives the native ma_engine spatializer.
# ------------------------------------------------------------------
[docs]
def set_listener_position(self, x: float, y: float, z: float) -> None:
"""Forward listener position to the native ma_engine spatializer."""
self._engine.set_listener_position(float(x), float(y), float(z))
[docs]
def set_listener_velocity(self, x: float, y: float, z: float) -> None:
"""Forward listener velocity (m/s) to the native engine for Doppler."""
self._engine.set_listener_velocity(float(x), float(y), float(z))
[docs]
def set_listener_direction(self, x: float, y: float, z: float) -> None:
"""Forward listener forward vector to the native engine."""
self._engine.set_listener_direction(float(x), float(y), float(z))
[docs]
def set_listener_world_up(self, x: float, y: float, z: float) -> None:
"""Forward listener world-up vector to the native engine."""
self._engine.set_listener_world_up(float(x), float(y), float(z))
[docs]
def shutdown(self) -> None:
with self._lock:
entries = list(self._sounds.values())
self._sounds.clear()
chains = list(self._effect_chains.values())
self._effect_chains.clear()
groups = list(self._groups.values())
self._groups.clear()
for entry in entries:
try:
entry.sound.shutdown()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.shutdown.sound_cleanup_failed",
message="MiniaudioBackend.shutdown: sound cleanup failed",
)
for chain in chains:
try:
chain.shutdown()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.shutdown.effect_chain_cleanup_failed",
message="MiniaudioBackend.shutdown: effect chain cleanup failed",
)
for group in groups:
try:
group.shutdown()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.shutdown.group_cleanup_failed",
message="MiniaudioBackend.shutdown: group cleanup failed",
)
try:
self._engine.shutdown()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.shutdown.engine_cleanup_failed",
message="MiniaudioBackend.shutdown: engine cleanup failed",
)
[docs]
def list_capabilities(self) -> frozenset[Capability]:
return _DESKTOP_CAPABILITIES | _NATIVE_EFFECT_CAPABILITIES
[docs]
def sync_bus_layout(self, layout: AudioBusLayout) -> None:
"""Reconcile native ``ma_sound_group`` gains, routing and effect chains against `layout`.
The native groups mirror the layout's ``send_to`` graph: each group
attaches, through its own effect chain when it has one, to the
group of the bus it routes into. miniaudio therefore multiplies the
routing chain itself and a bus effect processes every bus routed
into it, which is what ``send_to`` means and what the browser lane
does. Each group carries only its bus's own gain.
Cheap and idempotent: pushes only on change. Called by the
audio server every frame; safe to call manually.
"""
for bus in layout.buses:
group = self._ensure_group(bus.name, layout)
if group is None:
continue
# Own gain, not the chain-effective one: the graph applies every
# ancestor's volume and mute, so pre-multiplying would apply
# them twice. Solo is layout-wide and appears in no bus's own
# state, so it has to be part of the snapshot to be noticed.
gain = layout._own_linear_volume(bus)
parent = layout._routing_parent(bus)
previous = self._bus_snapshot.get(bus.name)
if previous is None or previous[0] != gain:
try:
group.set_volume(gain)
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.sync_bus_layout.set_volume_failed",
message=f"MiniaudioBackend.sync_bus_layout: set_volume failed for bus {bus.name!r}",
)
if previous is not None and previous[1] != parent:
# A re-parent at runtime: re-attach, never rebuild. Sounds
# hold their group handle, so recreating it would silence
# everything playing on the bus.
self._route_group(bus, layout)
self._bus_snapshot[bus.name] = (gain, parent)
# Effect chain reconciliation. Chain rebuilds only on signature
# change; cheap if the effects list is unchanged.
chain = self._effect_chains.get(bus.name)
if chain is None:
# Only allocate a chain object when there's at least one
# effect to install (avoids per-bus overhead when buses
# have no effects).
if not bus.effects:
continue
chain = _NativeEffectChain(self._engine, group, self._destination_node(parent, layout))
self._effect_chains[bus.name] = chain
try:
chain.reconcile(bus.effects)
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.sync_bus_layout.effect_chain_rebuild_failed",
message=f"MiniaudioBackend.sync_bus_layout: effect chain rebuild failed for bus {bus.name!r}",
)