"""The one place the engine reaches the ``miniaudio`` package.
``miniaudio`` decodes audio files and, on the legacy mixer, owns the playback
device. It is a dependency of ``simvx-core`` on the desktop and absent in the
browser, where the web runtime installs a name-compatible stand-in that
decodes nothing.
Reaching it by a bare ``import miniaudio`` from inside a call path leaves no
seam. Substituting a decoder then means mutating :data:`sys.modules`, which is
process-global and outlives whatever installed the stand-in, and a stand-in
that omits a name the engine references fails at the point of reference rather
than at a checked boundary. That is how a decode failure under the browser
stand-in surfaced as ``AttributeError`` raised while handling the original
exception: the ``except`` clause named ``miniaudio.DecodeError``, which the
stand-in does not carry.
Every name the engine needs is wrapped here. A stand-in therefore has one
surface to be compatible with, an absent or partial install is reported once as
:class:`DecoderUnavailableError` rather than as whatever attribute happened to
be touched first, and a test substitutes a function on this module with
``monkeypatch.setattr`` instead of reaching for a global.
This is a leaf module: it imports nothing else from ``simvx``, so the modules
below ``audio.py`` and the modules in ``audio_backend`` can both use it without
inverting the import graph between them.
"""
from __future__ import annotations
from typing import Any
__all__ = [
"DecodeFailedError",
"DecoderUnavailableError",
"decode_file",
"decoder_available",
"file_duration",
"has_duration_probe",
"playback_device",
"stream_file",
]
[docs]
class DecoderUnavailableError(RuntimeError):
"""No usable ``miniaudio`` is installed, so nothing here can decode."""
[docs]
class DecodeFailedError(RuntimeError):
"""A usable decoder refused this particular file."""
_INSTALL_HINT = (
"`miniaudio` is a dependency of `simvx-core`, so a desktop install has it: "
"reinstall with `pip install --force-reinstall simvx-core`. In a browser "
"export the runtime decodes and mixes audio itself and this path is not used."
)
# The names an installed ``miniaudio`` must carry before the engine will treat
# it as a decoder. Two of them are types the engine evaluates as arguments and
# in ``except`` clauses, so a stand-in missing either fails at a name lookup
# rather than at a decode, which is the partial install this reports.
_DECODE_SYMBOLS = ("decode_file", "stream_file", "SampleFormat", "DecodeError")
# A decoder that answers ``None`` decodes nothing at all: the real package
# either returns samples or raises. A stand-in carrying every name but
# answering ``None`` is the browser one, so this reads as "no decoder here"
# rather than as "this file is unplayable", and the caller can say what to do
# instead.
_RETURNED_NOTHING = "The installed `miniaudio` returned nothing, so it decodes nothing."
# Container tag -> the metadata reader for it. ``miniaudio``'s own
# ``get_file_info`` dispatches on the filename extension, which the engine
# deliberately does not trust: :attr:`AudioClip.container` is probed from the
# file's header, so the reader is picked from that instead.
_DURATION_PROBES: dict[str, str] = {
"wav": "wav_get_file_info",
"ogg": "vorbis_get_file_info",
"mp3": "mp3_get_file_info",
"flac": "flac_get_file_info",
}
def _module() -> Any:
"""The ``miniaudio`` module itself.
Raises:
DecoderUnavailableError: it is not importable.
"""
try:
import miniaudio
except ImportError as exc:
raise DecoderUnavailableError(f"The audio decoder is not installed. {_INSTALL_HINT}") from exc
return miniaudio
def _symbols(*names: str) -> tuple[Any, ...]:
"""Resolve *names* off the module, naming any a stand-in does not carry.
Every call goes through the import, so a name is looked up against
whatever is installed at the time rather than against whatever was
installed the first time audio was touched.
"""
module = _module()
missing = [name for name in names if not hasattr(module, name)]
if missing:
raise DecoderUnavailableError(
f"The installed `miniaudio` cannot decode: it has no {', '.join(missing)}. {_INSTALL_HINT}"
)
return tuple(getattr(module, name) for name in names)
[docs]
def decoder_available() -> bool:
"""Whether a decoder able to turn an audio file into PCM is installed.
False both when ``miniaudio`` is absent and when what is installed is a
stand-in that carries only some of the names decoding needs.
Answered from the names alone, so it opens no file and costs one import.
A stand-in that carries every name and decodes nothing still answers True
here; it is caught at the first decode, which raises
:class:`DecoderUnavailableError` rather than reporting the file as
unplayable.
"""
try:
_symbols(*_DECODE_SYMBOLS)
except DecoderUnavailableError:
return False
return True
[docs]
def decode_file(path: str, *, sample_rate: int, channels: int) -> Any:
"""Decode the whole of *path* to interleaved signed 16-bit PCM.
Args:
path: Filesystem path to the audio file.
sample_rate: Output rate in Hz. The decoder resamples to it.
channels: Output channel count. The decoder mixes down or up to it.
Returns:
The samples, as an object supporting the buffer protocol whose items
are interleaved ``int16`` in the requested format.
Raises:
DecoderUnavailableError: no usable decoder is installed, or what is
installed decodes nothing.
DecodeFailedError: the file is missing, unreadable, or in a codec this
build has no decoder for.
"""
decode, sample_format, decode_error = _symbols("decode_file", "SampleFormat", "DecodeError")
try:
decoded = decode(
path,
output_format=sample_format.SIGNED16,
nchannels=channels,
sample_rate=sample_rate,
)
except (decode_error, FileNotFoundError, OSError) as exc:
raise DecodeFailedError(f"Cannot decode {path!r}: {exc}") from exc
if decoded is None:
raise DecoderUnavailableError(f"{_RETURNED_NOTHING} {_INSTALL_HINT}")
return decoded.samples
[docs]
def stream_file(
path: str,
*,
sample_rate: int,
channels: int,
frames_to_read: int,
seek_frame: int = 0,
) -> Any:
"""Open *path* as a generator of interleaved signed 16-bit PCM.
Args:
path: Filesystem path to the audio file.
sample_rate: Output rate in Hz.
channels: Output channel count.
frames_to_read: Frames the generator allocates for, and the ceiling on
one ``send``.
seek_frame: Frame to start at.
Returns:
A generator whose ``send(frames)`` yields up to *frames* frames as an
``array.array`` of interleaved ``int16``, and which raises
``StopIteration`` at the end of the file.
Raises:
DecoderUnavailableError: no usable decoder is installed, or what is
installed decodes nothing.
DecodeFailedError: the file cannot be opened or decoded.
"""
stream, sample_format = _symbols("stream_file", "SampleFormat")
try:
generator = stream(
path,
output_format=sample_format.SIGNED16,
nchannels=channels,
sample_rate=sample_rate,
frames_to_read=frames_to_read,
seek_frame=seek_frame,
)
except Exception as exc:
# Deliberately broad: the decoder is third-party C over a file the
# caller chose, and every way it can object (an unreadable file, a
# codec this build lacks, a header that lied) has the same answer here,
# which is that this stream cannot be played.
raise DecodeFailedError(f"Cannot stream {path!r}: {exc}") from exc
if generator is None:
raise DecoderUnavailableError(f"{_RETURNED_NOTHING} {_INSTALL_HINT}")
return generator
[docs]
def has_duration_probe(container: str) -> bool:
"""Whether :func:`file_duration` can read a length for this container tag."""
return container in _DURATION_PROBES
[docs]
def file_duration(path: str, container: str) -> float | None:
"""Seconds of audio in *path*, read from the container's own metadata.
Opens no audio device and decodes no samples. ``None`` is "unknown": the
container has no metadata reader here, no decoder is installed, or the file
is missing, truncated, mislabelled or in a codec this build lacks.
"""
probe_name = _DURATION_PROBES.get(container)
if probe_name is None:
return None
try:
(probe,) = _symbols(probe_name)
except DecoderUnavailableError:
return None
try:
duration = float(probe(path).duration)
except Exception:
# Deliberately broad, for the reason given in :func:`stream_file`, and
# with the added constraint that reading a length happens underneath
# playback: no exception from a third-party reader may escape into it.
return None
return duration if duration >= 0.0 else None
[docs]
def playback_device(*, sample_rate: int, channels: int, buffersize_msec: int) -> Any:
"""Open a signed-16-bit playback device the legacy mixer can push frames to.
Raises:
DecoderUnavailableError: ``miniaudio`` is absent, or carries no
``PlaybackDevice``. Anything the device itself objects to (no sound
server, no card, the device held by another process) is raised by
``miniaudio`` and passes through: ``make_backend`` reads that as
"this backend cannot start" and falls through to the next one.
"""
device, sample_format = _symbols("PlaybackDevice", "SampleFormat")
result: Any = device(
output_format=sample_format.SIGNED16,
nchannels=channels,
sample_rate=sample_rate,
buffersize_msec=buffersize_msec,
)
return result