"""
Audio stream/data layer: :class:`AudioStream` resource + container detection.
Private leaf module behind the :mod:`simvx.core.audio` facade. Holds the
audio *data* concerns: the :class:`AudioStream` resource handle, the
header-probe container detection helpers, the WAV ``data`` chunk seek
helper, 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
import struct
from importlib.resources.abc import Traversable
from typing import TYPE_CHECKING, Any, BinaryIO, Literal, Union
import numpy as np
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
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:`AudioStream.tone` / :meth:`AudioStream.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 AudioStream + 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:`AudioStream` 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)
# ============================================================================
# AudioStream: Audio resource
# ============================================================================
[docs]
class AudioStream:
"""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.
"""
__slots__ = ("source", "_path", "backend_data", "sample_rate", "channels", "_container")
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"AudioStream path must decode to str, got {type(spec).__name__}")
if spec == "":
# The legacy ``AudioStream("")`` sentinel is gone. Use the
# explicit :meth:`AudioStream.empty` classmethod for synthetic
# streams with no source.
raise InvalidStreamError(
"AudioStream() does not accept the empty string. "
"Use AudioStream.empty() for a synthetic stream with no source, "
"or AudioStream.from_pcm(samples, sample_rate=..., channels=...) "
"to wrap pre-rendered PCM."
)
self.source = source
self._path = spec
else:
raise TypeError(
"AudioStream 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
# 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]
def __repr__(self):
return f"AudioStream({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,
) -> AudioStream:
"""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)
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",
) -> AudioStream:
"""Wrap a pre-rendered PCM buffer as an AudioStream.
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`` or ``channels`` is
not 1 or 2.
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"AudioStream.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"AudioStream.from_pcm requires channels in {{1, 2}}: got samples shape={samples.shape!r} "
f"channels={channels!r}. Mono=1, interleaved stereo=2."
)
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
# 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") -> AudioStream:
"""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 ``AudioStream("")`` 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"
return stream
# ============================================================================
# WAV header parsing: locate the `data` chunk for streaming playback
# ============================================================================
def _seek_wav_data_chunk(f: BinaryIO) -> int | None:
"""Position *f* at the first PCM sample of a RIFF/WAVE file.
Returns the byte offset of the data chunk's payload (so callers can
rewind here when looping), or ``None`` if the file is not a valid
WAVE container or has no ``data`` chunk. The file cursor is left
pointing at the first sample on success and unspecified on failure.
Streams the header chunk-by-chunk rather than assuming the legacy
fixed 44-byte layout: real files may carry ``fmt`` extensions,
``LIST``/``INFO`` metadata, ``bext``, etc. before ``data``.
"""
try:
riff_header = f.read(12)
if len(riff_header) < 12 or riff_header[:4] != b"RIFF" or riff_header[8:12] != b"WAVE":
return None
while True:
header = f.read(8)
if len(header) < 8:
return None
chunk_id, chunk_size = struct.unpack("<4sI", header)
if chunk_id == b"data":
return f.tell()
# Chunks are word-aligned: odd sizes have a single pad byte.
skip = chunk_size + (chunk_size & 1)
f.seek(skip, os.SEEK_CUR)
except (OSError, struct.error):
return None