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
(``_LegacyMiniaudioBackend``, ``NullAudioBackend`` and the accelerators'
backends, such as ``simvx.audio.miniaudio``'s ``MiniaudioBackend``) 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 ..audio_errors import InvalidStreamError, 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


# Capabilities advertised by the *legacy* (pure-Python) desktop backend.
# No 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})


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

    A file that does not exist raises ``FileNotFoundError`` and one the
    decoder cannot read raises ``InvalidStreamError``, each naming the path.
    With no usable decoder installed at all, which is a fact about the
    environment rather than the file, it logs a warning and returns None.
    """
    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 or getattr(stream, "container", None) == "pcm":
        # No file behind it: an empty synthetic clip has nothing to decode.
        return None

    if not os.path.isfile(path):
        raise FileNotFoundError(f"no audio file at {path!r}")
    try:
        samples = _audio_decoder.decode_file(path, sample_rate=sample_rate, channels=nchannels)
    except _audio_decoder.DecoderUnavailableError as exc:
        log.warning("audio_backend: failed to decode %r: %s", path, exc)
        return None
    except _audio_decoder.DecodeFailedError as exc:
        raise InvalidStreamError(str(exc)) from exc

    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


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


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"