"""
Audio stream/data layer: :class:`AudioClip` resource + container detection.
Private leaf module behind the :mod:`simvx.core.audio` facade. Holds the
audio *data* concerns: the :class:`AudioClip` resource handle, the
header-probe container detection helpers, the chunk-at-a-time decoder that
feeds streaming playback, the clip-length probe, and the sample-rate/channel
constants and type aliases that belong to this layer. The playback mixin and
the player nodes live in sibling ``_audio_playback`` / ``_audio_players``
modules.
"""
from __future__ import annotations
import logging
import math
import os
from functools import lru_cache
from importlib.resources.abc import Traversable
from typing import TYPE_CHECKING, Any, Literal, Union
import numpy as np
from . import _audio_decoder
from .audio_errors import InvalidStreamError
if TYPE_CHECKING:
from .resource import Resource
# Sample rate / channel count must match the audio backend.
_SAMPLE_RATE = 44100
_NCHANNELS = 2
# The format a chunk-fed stream is decoded to when the backend does not
# report one of its own. Every shipped backend opens its device here.
_DEFAULT_STREAM_FORMAT = (48000, 2)
AudioSource = Union[str, os.PathLike, "Resource", Traversable]
# Container tags assigned by :func:`_detect_container_from_path` /
# :func:`_detect_container_from_bytes`. ``"pcm"`` is reserved for synthetic
# streams whose ``backend_data`` is already a decoded ndarray (see
# :meth:`AudioClip.tone` / :meth:`AudioClip.from_pcm`); decoders must
# never run on those. ``"unknown"`` is the strict-failure path: the
# backend raises rather than feeding the bytes through a decoder that will
# almost certainly misinterpret them.
AudioContainer = Literal["wav", "ogg", "mp3", "flac", "pcm", "unknown"]
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Container detection: header probe used by AudioClip + streaming open
# ---------------------------------------------------------------------------
def _detect_container_from_bytes(head: bytes) -> AudioContainer:
"""Identify the audio container from the first up-to-12 bytes of a file.
Returns one of ``"wav" / "ogg" / "mp3" / "flac" / "unknown"``. Used both
on :class:`AudioClip` construction (file-backed sources) and inside
the streaming open path so backends can route the file through the right
decoder rather than feeding raw container bytes to a PCM ring.
"""
if len(head) < 4:
return "unknown"
if head[:4] == b"RIFF" and len(head) >= 12 and head[8:12] == b"WAVE":
return "wav"
if head[:4] == b"OggS":
return "ogg"
if head[:4] == b"fLaC":
return "flac"
if head[:3] == b"ID3":
return "mp3"
# MPEG audio frame sync: byte 0 = 0xFF, byte 1 top 3 bits = 0b111.
if head[0] == 0xFF and (head[1] & 0xE0) == 0xE0:
return "mp3"
return "unknown"
def _detect_container_from_path(path: str) -> AudioContainer:
"""Open *path* and probe the first 12 bytes. Returns ``"unknown"`` on read failure.
The fallback to ``"unknown"`` (rather than guessing from the extension)
is deliberate: content probing avoids silent extension-based
misdetection. If the file doesn't exist or can't be read, leave the
diagnosis to the streaming open path which already emits a typed signal.
"""
if not path:
return "unknown"
try:
with open(path, "rb") as f:
head = f.read(12)
except OSError:
return "unknown"
return _detect_container_from_bytes(head)
# ============================================================================
# AudioClip: Audio resource
# ============================================================================
[docs]
class AudioClip:
"""Audio resource (WAV/OGG file or synthetic PCM).
This is a lightweight handle to audio data. Actual decoding is deferred
to the backend (miniaudio, SDL3, web audio).
Accepts any of:
- :class:`str` / :class:`os.PathLike` -- a filesystem audio file.
- :class:`~simvx.core.Resource` -- audio inside a Python package.
- :class:`importlib.resources.abc.Traversable` -- the raw return of
``importlib.resources.files(pkg) / name``.
Use :meth:`tone` for procedural sine-wave tones and :meth:`from_pcm` to
wrap pre-rendered PCM data.
Attributes:
source: Original spec the stream was constructed from -- a string,
:class:`pathlib.Path`, :class:`Resource`, or :class:`Traversable`.
Preserved verbatim so scene serialisation can round-trip it.
path: Resolved filesystem path string used by the backend (empty
string for synthetic streams that have no file).
backend_data: Backend-specific audio data (PCM ndarray, channel
id, etc). Set automatically when decoded; may also be set by
:meth:`from_pcm` / :meth:`tone`.
container: Detected container format -- one of ``"wav"``, ``"ogg"``,
``"mp3"``, ``"flac"``, ``"pcm"`` (synthetic) or ``"unknown"``.
Probed from the file header at construction time; the streaming
open path uses it to pick the right decoder.
duration: How long the clip plays for in seconds, or ``None`` when
that cannot be read.
"""
__slots__ = (
"source",
"_path",
"backend_data",
"sample_rate",
"channels",
"_container",
"_decoded_format",
)
def __init__(self, source: AudioSource):
from .resource import Resource
if isinstance(source, Resource):
self.source: Any = source
self._path: str = str(source.path)
elif isinstance(source, Traversable) and not isinstance(source, (str, os.PathLike)):
self.source = source
from .asset_resolver import _resolve_traversable
self._path = str(_resolve_traversable(source))
elif isinstance(source, (str, os.PathLike)):
spec = os.fspath(source)
if not isinstance(spec, str):
raise TypeError(f"AudioClip path must decode to str, got {type(spec).__name__}")
if spec == "":
# The legacy ``AudioClip("")`` sentinel is gone. Use the
# explicit :meth:`AudioClip.empty` classmethod for synthetic
# streams with no source.
raise InvalidStreamError(
"AudioClip() does not accept the empty string. "
"Use AudioClip.empty() for a synthetic stream with no source, "
"or AudioClip.from_pcm(samples, sample_rate=..., channels=...) "
"to wrap pre-rendered PCM."
)
self.source = source
self._path = spec
else:
raise TypeError(
"AudioClip accepts str | os.PathLike | Resource | Traversable, " f"got {type(source).__name__}"
)
self.backend_data: Any = None # Backend-specific data
# Set by :meth:`from_pcm`; ``None`` for file-backed streams (the
# decoder reads the rate/channel count from the container header).
self.sample_rate: int | None = None
self.channels: int | None = None
# ``(rate, channels)`` of the PCM a backend decoded into
# ``backend_data``, so a backend asking for a different format decodes
# the file again instead of being handed samples it would mix at the
# wrong pitch. ``None`` while nothing has decoded this clip, and for a
# buffer the caller supplied, which has no file to decode again.
self._decoded_format: tuple[int, int] | None = None
# File-backed: probe the header. Synthetic streams (from_pcm / tone)
# overwrite this to ``"pcm"`` after the bypass-init in those constructors.
self._container: AudioContainer = _detect_container_from_path(self._path)
[docs]
@property
def path(self) -> str:
"""Resolved filesystem path string used by the backend.
Synthetic streams (from_pcm / tone / empty) carry their assigned
``name`` here. The dual-field design (separate ``source`` + ``path``
slots both holding the same string for synthetic streams) was
collapsed during the audio refactor; ``path`` is now a derived
attribute backed by ``_path``.
"""
return self._path
[docs]
@property
def container(self) -> AudioContainer:
"""Detected container format. See :data:`AudioContainer` for the value set."""
return self._container
[docs]
@property
def duration(self) -> float | None:
"""Seconds of audio in this clip, or ``None`` when its length cannot be read.
Two sources answer it. A clip carrying its own decoded PCM
(:meth:`from_pcm`, :meth:`tone`, anything baked by ``AudioSynth``)
is measured from the buffer and the rate and channel count it
declares. A file-backed one is measured from the container's
metadata, read through the same decoder that plays the file, so no
audio device is opened and a compressed format needs no second
parser. File lengths are cached per ``(path, size, mtime_ns)``:
asking twice costs one ``stat``, and rewriting the file is seen.
``None`` is "unknown", never "zero". It is the answer for
:meth:`empty`, for a file that is missing, truncated, mislabelled
or in a codec this build of ``miniaudio`` lacks, and for a buffer
assigned to ``backend_data`` by hand, which declares no format to
divide by.
"""
pcm = _pcm_duration(self)
if pcm is not None:
return pcm
if not self._path or not _audio_decoder.has_duration_probe(self._container):
return None
try:
info = os.stat(self._path)
except OSError:
return None
return _probe_file_duration(self._path, self._container, (info.st_size, info.st_mtime_ns))
[docs]
def __repr__(self):
return f"AudioClip({self._path!r})"
# ------------------------------------------------------------------
# Constructors for synthetic streams
# ------------------------------------------------------------------
[docs]
@classmethod
def tone(
cls,
freq_hz: float,
*,
duration: float = 1.0,
volume: float = 0.3,
sample_rate: int = _SAMPLE_RATE,
) -> AudioClip:
"""Generate a sine-wave tone at *freq_hz* with a short fade-in/out.
The resulting stream has its PCM data baked into ``backend_data`` so
the audio backend skips file decoding entirely.
"""
if freq_hz <= 0:
raise ValueError(f"tone freq_hz must be > 0, got {freq_hz}")
if duration <= 0:
raise ValueError(f"tone duration must be > 0, got {duration}")
n_frames = int(sample_rate * duration)
if n_frames == 0:
raise ValueError(f"tone duration must cover at least one frame at {sample_rate} Hz, got {duration}")
t = np.linspace(0.0, duration, n_frames, dtype=np.float32)
fade_frames = min(int(sample_rate * 0.02), n_frames // 4)
envelope = np.ones(n_frames, dtype=np.float32)
if fade_frames > 0:
envelope[:fade_frames] = np.linspace(0.0, 1.0, fade_frames, dtype=np.float32)
envelope[-fade_frames:] = np.linspace(1.0, 0.0, fade_frames, dtype=np.float32)
mono = (np.sin(2 * math.pi * freq_hz * t) * volume * envelope).astype(np.float32)
stereo = np.empty(n_frames * _NCHANNELS, dtype=np.float32)
stereo[0::_NCHANNELS] = mono
stereo[1::_NCHANNELS] = mono
return cls.from_pcm(
stereo,
sample_rate=sample_rate,
channels=_NCHANNELS,
name=f"tone_{int(freq_hz)}Hz",
)
[docs]
@classmethod
def from_pcm(
cls,
samples: np.ndarray,
*,
sample_rate: int,
channels: int,
name: str = "pcm",
) -> AudioClip:
"""Wrap a pre-rendered PCM buffer as an AudioClip.
Args:
samples: float32 ndarray. For stereo, interleaved (channels
first within each frame); for mono, a 1-D array.
sample_rate: PCM sample rate in Hz. Required: playing a
44.1 kHz buffer on a 48 kHz backend produces wrong-pitch
audio if this is omitted.
channels: 1 (mono) or 2 (stereo). Required for the same
reason: a mono buffer played as stereo gives
left-channel-only sound.
name: Descriptive label used in :meth:`__repr__` and as the
stream's ``path``. The backend ignores it when
``backend_data`` is set.
Raises:
InvalidStreamError: ``sample_rate <= 0``, ``channels`` is not
1 or 2, the buffer holds no whole frame, or its length is
not a multiple of ``channels``.
TypeError: ``samples`` isn't a numpy ndarray.
"""
if not isinstance(samples, np.ndarray):
raise TypeError(f"from_pcm samples must be a numpy ndarray, got {type(samples).__name__}")
if not isinstance(sample_rate, int) or sample_rate <= 0:
raise InvalidStreamError(
f"AudioClip.from_pcm requires sample_rate>0: got samples shape={samples.shape!r} "
f"sample_rate={sample_rate!r}. Pass the buffer's actual rate (e.g. 44100, 48000)."
)
if channels not in (1, 2):
raise InvalidStreamError(
f"AudioClip.from_pcm requires channels in {{1, 2}}: got samples shape={samples.shape!r} "
f"channels={channels!r}. Mono=1, interleaved stereo=2."
)
frames, leftover = divmod(samples.size, channels)
if frames == 0:
raise InvalidStreamError(
f"AudioClip.from_pcm requires at least one whole frame: got samples shape={samples.shape!r} "
f"channels={channels!r}, which is {frames} frames. A clip with no frames is silence the "
"backend cannot play; check the duration the buffer was rendered at."
)
if leftover:
raise InvalidStreamError(
f"AudioClip.from_pcm requires a whole number of frames: got samples shape={samples.shape!r} "
f"channels={channels!r}, which leaves {leftover} sample(s) over. Interleaved stereo needs an "
"even length; pass the mono buffer with channels=1 instead."
)
stream = cls.__new__(cls)
stream.source = name
stream._path = name
stream.backend_data = samples.astype(np.float32, copy=False)
stream.sample_rate = sample_rate
stream.channels = channels
stream._decoded_format = None
# Synthetic streams hold already-decoded float32 PCM. Tag explicitly
# so the streaming open path never tries to decode the (non-existent)
# file behind ``stream._path``.
stream._container = "pcm"
return stream
[docs]
@classmethod
def empty(cls, *, name: str = "empty") -> AudioClip:
"""Return a synthetic stream with no audio data.
Used internally by :meth:`AudioSynth.bake` (before it overwrites
the synthetic frame buffer) and by null-backend tests that need a
placeholder stream object without touching the filesystem. The
returned stream carries ``container="pcm"`` and ``backend_data=None``:
playing it through a real backend is undefined.
Replaces the legacy ``AudioClip("")`` sentinel.
"""
stream = cls.__new__(cls)
stream.source = name
stream._path = name
stream.backend_data = None
stream.sample_rate = None
stream.channels = None
stream._container = "pcm"
stream._decoded_format = None
return stream
# ============================================================================
# Streaming decode: engine-format PCM pulled from a file a chunk at a time
# ============================================================================
class _StreamDecoder:
"""Reads an audio file as interleaved signed-16-bit PCM, a chunk at a time.
Streaming playback hands a backend raw bytes and the backend plays them
at its own rate and channel count, so those bytes have to be in that
format exactly. This wraps ``miniaudio``'s streaming decoder to guarantee
it: a 24-bit mono 22.05 kHz WAV, a 32-bit float WAV, an OGG, an MP3 and a
FLAC all come out as the same interleaved int16 at *sample_rate* /
*channels*. Reading the container's bytes directly, as the player used to,
plays anything that is not already in the output format as noise.
Args:
path: Filesystem path to the audio file.
sample_rate: Output rate in Hz, the backend's own.
channels: Output channel count, the backend's own.
from_position: Where to start, in seconds.
Raises:
InvalidStreamError: the file cannot be opened or decoded, or the
installed ``miniaudio`` has no decoder (the web runtime stubs it).
"""
# ``miniaudio``'s generator allocates ``max(frames_to_read, 16384)`` frames
# and refuses a read larger than that, so this is the ceiling on one read.
_MAX_FRAMES_PER_READ = 16384
def __init__(self, path: str, *, sample_rate: int, channels: int, from_position: float = 0.0):
self._path = path
self._sample_rate = int(sample_rate)
self._channels = int(channels)
self._from_position = max(0.0, float(from_position))
self._generator: Any = None
self._open(self._from_position)
def _open(self, from_position: float) -> None:
try:
self._generator = _audio_decoder.stream_file(
self._path,
sample_rate=self._sample_rate,
channels=self._channels,
frames_to_read=self._MAX_FRAMES_PER_READ,
seek_frame=int(from_position * self._sample_rate),
)
except _audio_decoder.DecoderUnavailableError as exc:
raise InvalidStreamError(
f"Cannot stream {self._path!r}: {exc} Streaming playback needs a real "
"decoder; load the clip with stream_mode='memory' instead."
) from exc
except _audio_decoder.DecodeFailedError as exc:
raise InvalidStreamError(str(exc)) from exc
@property
def sample_rate(self) -> int:
return self._sample_rate
@property
def channels(self) -> int:
return self._channels
def read(self, frames: int) -> bytes:
"""Return up to *frames* frames of PCM, or ``b""`` at the end of the file."""
if self._generator is None or frames <= 0:
return b""
try:
samples = self._generator.send(min(int(frames), self._MAX_FRAMES_PER_READ))
except StopIteration:
self.close()
return b""
except Exception as exc:
# Broad for the same reason as ``_open``: a decoder failure part
# way through a file ends the stream rather than escaping into the
# frame loop that was pumping it.
log.warning("Decoding %r stopped part way through: %s", self._path, exc)
self.close()
return b""
# ``samples`` is an ``array.array`` of interleaved int16, so its length
# counts samples rather than frames; only the bytes leave here.
chunk: bytes = samples.tobytes()
return chunk
def restart(self) -> None:
"""Rewind to the start of the file so a looping player can play it again.
Returns to the beginning, not to the ``from_position`` the first pass
started at: a loop plays the whole file.
"""
self.close()
self._open(0.0)
def close(self) -> None:
"""Release the decoder. Safe to call more than once."""
generator, self._generator = self._generator, None
if generator is not None:
generator.close()
# ============================================================================
# Clip length: how long a clip plays for
# ============================================================================
@lru_cache(maxsize=256)
def _probe_file_duration(path: str, container: AudioContainer, _fingerprint: tuple[int, int]) -> float | None:
"""Seconds of audio in the file at *path*, or ``None`` if it cannot be read.
Reads the container's metadata through the same decoder that plays the
file, so a compressed format needs no second parser here and no audio
device. Anything unreadable (a truncated or mislabelled file, a codec
this build of ``miniaudio`` lacks) yields ``None``.
*_fingerprint* is the file's ``(size, mtime_ns)``. It takes no part in
the probe and exists only to key the cache, so rewriting the file
re-reads it instead of serving the previous length.
"""
return _audio_decoder.file_duration(path, container)
def _pcm_duration(clip: AudioClip) -> float | None:
"""Seconds of audio in a clip carrying its own decoded PCM, else ``None``.
Covers :meth:`AudioClip.from_pcm`, :meth:`AudioClip.tone` and anything
baked by ``AudioSynth``, each of which declares the buffer's rate and
channel count. A buffer assigned to ``backend_data`` without them cannot
be divided into seconds: it answers ``None`` rather than assuming a
format, so a file-backed clip whose decoded samples are cached there
falls through to its container metadata instead.
"""
samples = clip.backend_data
if samples is None:
return None
rate = clip.sample_rate
channels = clip.channels
if rate is None or channels is None or rate <= 0 or channels <= 0:
return None
# ``size`` counts samples whatever the array's shape; ``len`` covers any
# plain sequence a caller has stashed there.
count = getattr(samples, "size", None)
if count is None:
try:
count = len(samples)
except TypeError:
return None
return (count // channels) / rate