"""Shared helpers, constants, and small data structures for the desktop audio backends.
Everything in this module is used by two or more of the concrete backends
(``MiniaudioBackend``, ``_LegacyMiniaudioBackend``, ``NullAudioBackend``) or by
``make_backend``. It depends on nothing else in the ``audio_backend`` package.
"""
from __future__ import annotations
import atexit
import logging
import os
import threading
import weakref
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
import numpy as np
from .. import _audio_decoder
from .._native import miniaudio_engine as _me
from ..audio_errors import raise_or_warn
from ..audio_protocol import CAPABILITIES_CORE, Capability
if TYPE_CHECKING:
from ..audio import AudioClip
log = logging.getLogger(__name__)
_DEFAULT_SAMPLE_RATE = 48000
_DEFAULT_CHANNELS = 2
# Per-channel high-water mark for the legacy backend's streaming buffer.
# Producer code that overfeeds (main thread runs faster than the audio
# period) used to grow the buffer without bound; now we drop oldest bytes
# on overflow and warn-once per channel. Override via
# ``SIMVX_AUDIO_STREAM_BUFFER_BYTES`` (positive int, bytes).
_DEFAULT_STREAM_BUFFER_MAX_BYTES = 1 << 20 # 1 MiB
def _stream_buffer_cap() -> int:
"""Resolve the legacy streaming buffer cap from env / default.
Reads ``SIMVX_AUDIO_STREAM_BUFFER_BYTES`` on every call so tests can
monkeypatch the env mid-suite. Falls back to the default if the value
is missing or unparseable.
"""
raw = os.environ.get("SIMVX_AUDIO_STREAM_BUFFER_BYTES")
if raw:
try:
cap = int(raw)
if cap > 0:
return cap
except ValueError:
pass
return _DEFAULT_STREAM_BUFFER_MAX_BYTES
# Streaming-capable compressed formats: only the native (miniaudio C
# extension) backend can decode these on-the-fly. The legacy pure-Python
# mixer would need its own streaming decoder, which doesn't exist.
_COMPRESSED_STREAMING_FORMATS: frozenset[Capability] = frozenset(
{Capability.STREAMING_OGG, Capability.STREAMING_MP3, Capability.STREAMING_FLAC}
)
# Capabilities advertised by the *native* desktop backend.
_DESKTOP_CAPABILITIES: frozenset[Capability] = (
CAPABILITIES_CORE | _COMPRESSED_STREAMING_FORMATS | frozenset({Capability.SPATIAL_DOPPLER})
)
# Capabilities advertised by the *legacy* (pure-Python) desktop backend.
# Drops the compressed-streaming formats since the mixer carries no decoder
# of its own. It still streams them: the player reads the absence of these
# flags as "decode for me" and feeds it PCM, rather than handing over a file
# whose container bytes the mixer would read as int16 and play as noise.
_LEGACY_CAPABILITIES: frozenset[Capability] = CAPABILITIES_CORE | frozenset({Capability.SPATIAL_DOPPLER})
# Effect capabilities the *native* backend additionally supports: built-in
# ma_*_node filters/delay plus custom DSP nodes (freeverb, compressor,
# soft-clip). Legacy backend doesn't advertise these: the Python mixer
# can't run them at acceptable performance.
_NATIVE_EFFECT_CAPABILITIES: frozenset[Capability] = frozenset(
{
Capability.EFFECT_GAIN,
Capability.EFFECT_FILTER_BIQUAD,
Capability.EFFECT_DELAY,
Capability.EFFECT_PARAMETRIC_EQ,
Capability.EFFECT_REVERB,
Capability.EFFECT_SOFTCLIP,
Capability.EFFECT_COMPRESSOR,
}
)
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
def _db_to_linear(db: float) -> float:
"""Convert decibels to linear gain. ``db <= -80`` is treated as silence."""
if db <= -80.0:
return 0.0
return 10.0 ** (db / 20.0)
def _decode_stream_to_ndarray(stream: AudioClip, sample_rate: int, nchannels: int) -> np.ndarray | None:
"""Decode an AudioClip into float32 interleaved samples at the given format.
Used by the legacy backend (which mixes in Python) and by the native
backend when the source is an ndarray-backed ``AudioClip`` (e.g.
``AudioClip.from_pcm`` or anything baked by ``AudioSynth``).
File-backed streams on the native path go through
``ma_sound_init_from_file`` instead.
A file decoded here is cached on the clip together with the rate and
channel count it was decoded at, so asking for a different format decodes
the file again rather than handing back samples the caller will mix at the
wrong pitch. A buffer the caller supplied (``from_pcm``, ``tone``, anything
baked by ``AudioSynth``, or one assigned to ``backend_data`` by hand) is
returned as it stands: there is no file behind it to decode again.
"""
cached = stream.backend_data
if isinstance(cached, np.ndarray):
decoded_at = getattr(stream, "_decoded_format", None)
if decoded_at is None or decoded_at == (sample_rate, nchannels):
return cached
path = stream.path
if not path:
return None
try:
samples = _audio_decoder.decode_file(path, sample_rate=sample_rate, channels=nchannels)
except (_audio_decoder.DecoderUnavailableError, _audio_decoder.DecodeFailedError) as exc:
log.warning("audio_backend: failed to decode %r: %s", path, exc)
return None
raw = np.frombuffer(samples, dtype=np.int16).astype(np.float32) / 32768.0
stream.backend_data = raw
stream._decoded_format = (sample_rate, nchannels)
return raw
@dataclass
class _Channel:
"""A single active voice in the legacy mixer."""
samples: np.ndarray # float32, interleaved stereo
cursor: int = 0
total_frames: int = 0
volume: float = 1.0
pan: float = 0.0
pitch: float = 1.0
loop: bool = False
paused: bool = False
stopped: bool = False
bus: str = "Master"
streaming: bool = False
stream_buffer: bytearray = field(default_factory=bytearray)
# Set by ``end_of_stream``: the producer has fed its last chunk, so the
# mixer stops the channel when ``stream_buffer`` runs out instead of
# underrunning into silence and waiting for a feed that never comes.
draining: bool = False
# Monotonic channel id counter (shared across legacy + native: both keep
# `dict[int, ...]` lookups so the global counter avoids collisions).
_next_id = 0
_id_lock = threading.Lock()
def _alloc_channel_id() -> int:
global _next_id
with _id_lock:
_next_id += 1
return _next_id
@dataclass
class _SoundEntry:
"""One active voice on the native backend."""
sound: Any # _me.Sound
bus: str
keeper: Any = None # AudioBuffer (pin against GC) when buffer-backed
paused: bool = False
# True for channels opened via ``open_stream`` (chunk-fed PCM). Native
# streams build their Sound with ``pitch_enabled=False`` so the resampler
# can't read past what the producer has written: that makes ``set_pitch``
# against a streaming channel a guaranteed no-op. Used to raise
# ``AudioCapabilityError`` instead of silently dropping the change.
streaming: bool = False
# Total frames the producer has fed into a chunk-fed ring (keeper is a
# ``StreamSource``). The ring's data source has no native cursor, so the
# playback position is derived as frames fed minus frames still queued.
frames_fed: int = 0
# Set by ``end_of_stream``: the producer has fed its last chunk. An
# ``ma_pcm_rb``-backed sound underruns into silence rather than ending, so
# this is the only thing that lets ``is_channel_active`` tell a chunk-fed
# channel that has played out from one that is merely waiting for a feed.
draining: bool = False
# Pre-paused position so a later `resume_audio` can restart from the cursor.
# ma_sound_start auto-resumes from the internal cursor after ma_sound_stop;
# we just need to remember whether we deliberately paused.
# One-time warning flag for a native extension built without Vorbis support
# (stale .so from before stb_vorbis was bundled). Module-level so repeated
# backend constructions don't spam the log.
_vorbis_warned = False
def _warn_if_no_vorbis() -> None:
global _vorbis_warned
if _vorbis_warned or _me.has_vorbis():
return
_vorbis_warned = True
log.warning(
"Native audio extension was built without Vorbis support: .ogg files "
"will fail to load. Rebuild it with "
"`uv run --with setuptools simvx build-audio`."
)
# Weak-set of native backends still alive at interpreter shutdown. Each
# entry's shutdown() is called via atexit so miniaudio's audio thread
# is joined cleanly even when callers bypass `App.quit()` via sys.exit
# (the long-standing pattern noted in feedback_app_quit_pattern.md).
_atexit_backends: weakref.WeakSet[Any] = weakref.WeakSet()
_atexit_registered = False
def _register_atexit_shutdown(backend) -> None:
global _atexit_registered
_atexit_backends.add(backend)
if _atexit_registered:
return
_atexit_registered = True
def _shutdown_all_atexit():
for be in list(_atexit_backends):
try:
be.shutdown()
except Exception as exc:
raise_or_warn(
exc,
key="audio.native.atexit.backend_shutdown_failed",
message="atexit: backend shutdown failed",
)
atexit.register(_shutdown_all_atexit)
# The warning is one line, because a warning nobody finishes reading is a
# warning nobody acts on. It states the two things that changed and links the
# page that explains them.
_NATIVE_HELP_URL = "https://simvx.com/docs/install.html#native-audio-extension"
# The long form belongs to the strict-mode exception instead: that path only
# fires when someone set SIMVX_ALLOW_LEGACY_AUDIO=0 to be told about exactly
# this, and an exception is read once rather than scrolled past every run.
#
# Both remedies have to work where the reader actually is. A wheel install has
# no repository to point at, and cannot rebuild in place either: the wheels
# carry the CFFI glue but not the vendored miniaudio source, which ships only
# in the sdist. So the general answer is a source reinstall, and the in-place
# rebuild is offered only to someone who already has a checkout.
_NATIVE_REBUILD_HINT = (
"With a C compiler installed, reinstall from source: "
"`pip install --force-reinstall --no-binary simvx-core simvx-core`. "
"From a source checkout, `uv run --with setuptools simvx build-audio` "
"builds it in place."
)
def _legacy_allowed() -> bool:
"""Read ``SIMVX_ALLOW_LEGACY_AUDIO`` fresh each call (tests monkeypatch it)."""
return os.environ.get("SIMVX_ALLOW_LEGACY_AUDIO", "1") != "0"