Source code for simvx.core.audio_backend._shared

"""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 .._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 doesn't carry a
# decoder: ``open_stream`` raises :class:`AudioCapabilityError` for them
# rather than feeding container bytes through ``np.frombuffer(dtype=int16)``
# and producing 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.

    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.
    """
    import miniaudio

    if stream.backend_data is not None and isinstance(stream.backend_data, np.ndarray):
        return stream.backend_data

    path = stream.path
    if not path:
        return None

    try:
        decoded = miniaudio.decode_file(
            path,
            output_format=miniaudio.SampleFormat.SIGNED16,
            nchannels=nchannels,
            sample_rate=sample_rate,
        )
    except (miniaudio.DecodeError, FileNotFoundError, OSError) as exc:
        log.warning("audio_backend: failed to decode %r: %s", path, exc)
        return None

    raw = np.frombuffer(decoded.samples, dtype=np.int16).astype(np.float32) / 32768.0
    stream.backend_data = raw
    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)


# 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
    # 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)


_NATIVE_REBUILD_HINT = (
    "Run `uv pip install --reinstall -e packages/core` after installing a "
    "C compiler to enable low-latency audio, or `uv run --with setuptools "
    "simvx build-audio` to rebuild without reinstalling."
)


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"