"""The silent (no-op) audio backend.
``NullAudioBackend`` is selected when no audio device is available (sandboxed
CI, headless containers). Calls return valid channel IDs and behave
consistently, but nothing is rendered to a device.
"""
from __future__ import annotations
import threading
import weakref
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from ..audio_protocol import Capability
from ._shared import (
_DEFAULT_CHANNELS,
_DEFAULT_SAMPLE_RATE,
_alloc_channel_id,
_register_atexit_shutdown,
)
if TYPE_CHECKING:
from ..audio import AudioClip
from ..audio_bus import AudioBusLayout
from ..scene_tree import SceneTree
def _clamp_pitch(pitch: float) -> float:
"""Clamp a pitch the way ``MiniaudioBackend`` clamps it before ``ma_sound``."""
return max(0.1, float(pitch))
@dataclass(slots=True)
class _NullVoice:
"""One silent channel, when it finishes, and how fast it is playing.
``ends_at`` is a reading of the backend's clock; ``None`` means the
channel runs until ``stop_audio`` (a loop, a paused channel, or a clip
whose length could not be read). ``remaining`` holds the time left while
the channel is paused. ``pitch`` is the resampler ratio in force, kept so
a mid-playback change can rescale whichever of the two is live.
"""
ends_at: float | None = None
remaining: float | None = None
pitch: float = 1.0
[docs]
class NullAudioBackend:
"""Silent backend implementing :class:`AudioPlaybackBackend` + :class:`AudioBusBackend`.
Selected automatically by :func:`make_backend` when neither the native
extension nor the legacy ``miniaudio`` path can start (e.g. no audio
device on the host, ALSA/Pulse not running, the ``miniaudio`` Python
package not installed, or a sandboxed environment with no compiler).
The engine, scene tree, and :class:`AudioPlayer` nodes all
operate normally: calls return valid channel IDs and
:meth:`is_channel_active` behaves consistently, but nothing is
rendered to a device.
A non-looping channel reports itself finished once the clip's duration
has elapsed, so ``queue_free_on_end`` players reap on a machine with no
audio just as they do on one with a sound card. The elapsed time is
read from the tree this backend was installed on, not from a wall
clock: a headless run that ticks faster than real time ends its sounds
after the same number of ticks either way, a paused tree holds its
sounds where they are, and a time-scaled one stretches them with
everything else. Only that tree's ticks count, so another tree built
in the same process (an editor's edited scene, which nothing ticks)
neither ends these sounds nor stalls them. That is the one
divergence from a real device, whose playback cursor is driven by the
audio hardware and knows nothing of the tree; it buys a deterministic
silent backend for CI and headless capture, which is what this backend
exists for. Tests substitute ``_clock`` to drive it by hand.
Clip length comes from the PCM a synthetic clip carries or, for a
file-backed one, from the container's own metadata: WAV, OGG, MP3 and
FLAC alike, since reading a length needs a decoder but no device. A clip
whose length cannot be read keeps its channel until an explicit
``stop_audio``: that covers a clip with no file behind it, an
unrecognised or corrupt container, and the case where ``miniaudio``
itself is missing rather than merely deviceless.
A change of pitch mid-playback moves the end with it, because the
resampler it stands for changes how long the rest of the clip takes.
NullBackend does NOT implement :class:`AudioStreamingBackend`. Code
paths that need streaming (procedural synth via
:class:`AudioSynth.attach_to`, AudioWorklet feeds, etc.) must check
``isinstance(backend, AudioStreamingBackend)`` and raise / warn-once
when it's absent. Advertised capabilities are narrowed to
``{Capability.PLAY_BASIC}`` so effect modules don't try to
instantiate native nodes on the null path.
"""
def __init__(
self,
sample_rate: int = _DEFAULT_SAMPLE_RATE,
nchannels: int = _DEFAULT_CHANNELS,
):
# The device format this backend stands in for. Nothing is rendered
# here, so they are recorded rather than mixed with; ``make_backend``
# hands the same pair to whichever backend it picks.
self._sample_rate = sample_rate
self._nchannels = nchannels
self._lock = threading.Lock()
self._channels: dict[int, _NullVoice] = {}
self._paused: set[int] = set()
# The tree that installed this backend, held weakly so the backend
# does not keep a finished scene's whole node graph alive.
self._tree: weakref.ref[SceneTree] | None = None
# Time source for channel-end bookkeeping: that tree's time, so a
# headless run ends its sounds on ticks rather than on how fast the
# machine got through them. Swappable so tests can drive it by hand.
self._clock: Callable[[], float] = self._tree_time
# Symmetric atexit registration: no audio thread to leak here, but
# keeps shutdown semantics consistent across all three backends so
# tests can iterate ``_atexit_backends`` and assert every active
# backend gets joined at interpreter exit.
_register_atexit_shutdown(self)
[docs]
def bind_tree(self, tree: SceneTree) -> None:
"""Adopt *tree*'s clock; called by ``SceneTree.install_audio_backend``.
This backend's half of
:class:`~simvx.core.audio_protocol.TreeBoundAudioBackend`, which is
what installing isinstance-checks. Channel ends are then measured
against the ticks of the tree whose sounds they are, which is the only
tree that can stop or reap them.
"""
self._tree = weakref.ref(tree)
def _tree_time(self) -> float:
"""Seconds elapsed on the installing tree, or ``0.0`` when there is none.
The reading is :attr:`SceneTree.now`: it advances by each tick's
delta after any ``time_scale`` scaling and freezes while the tree is
paused or an inert overlay is open. A backend that no tree has
installed reads a constant
``0.0``, so a channel started on it holds until an explicit
``stop_audio`` or until a tree installs the backend and ticks.
"""
ref = self._tree
tree = ref() if ref is not None else None
return tree.now if tree is not None else 0.0
def _alloc(self, ends_at: float | None, pitch: float) -> int:
cid = _alloc_channel_id()
with self._lock:
# Sweep the channels whose clip has run out. ``is_channel_active``
# drops the ones somebody polls, but a fire-and-forget
# ``play_audio`` is never polled by anyone, so without this a long
# headless run of one-shot sounds grows the table for ever.
now = self._clock()
for dead in [c for c, v in self._channels.items() if v.ends_at is not None and now >= v.ends_at]:
del self._channels[dead]
self._paused.discard(dead)
self._channels[cid] = _NullVoice(ends_at=ends_at, pitch=pitch)
return cid
def _repitch(self, channel_id: int, pitch: float) -> None:
"""Move a channel's end to where the new resampler ratio puts it.
A device plays the rest of the clip through its resampler, so raising
the pitch mid-playback brings the end nearer and lowering it pushes
the end away. Whichever of ``ends_at`` / ``remaining`` is live is
scaled by the ratio of the old rate to the new one; a loop or a clip
of unknown length has no end to move and only records the new rate.
"""
pitch = _clamp_pitch(pitch)
with self._lock:
voice = self._channels.get(channel_id)
if voice is None or pitch == voice.pitch:
return
ratio = voice.pitch / pitch
if voice.remaining is not None:
voice.remaining *= ratio
elif voice.ends_at is not None:
now = self._clock()
voice.ends_at = now + max(0.0, voice.ends_at - now) * ratio
voice.pitch = pitch
[docs]
def play_audio(
self,
stream: AudioClip,
*,
mode: str = "non_positional",
position: Any = None,
volume_db: float = 0.0,
pitch: float = 1.0,
loop: bool = False,
bus: str = "Master",
max_distance: float = 100.0,
from_position: float = 0.0,
pan: float = 0.0,
gain_db: float = 0.0,
) -> int | None:
# NullBackend is silent regardless of any kwarg; accept the full
# signature so AudioPlaybackBackend Protocol parity holds. Volume,
# pan and bus have nothing to act on, but loop / from_position /
# pitch decide when the channel ends, so they are honoured.
rate = _clamp_pitch(pitch)
# No other backend requires a stream to carry a duration, so this one
# must not either: it is the headless fallback, and the point of it is
# that node lifecycle code runs unchanged. A stream without one plays
# forever here, which is what an unknown length already means below.
duration = getattr(stream, "duration", None)
if loop or duration is None:
return self._alloc(None, rate)
# Pitch is the resampler ratio, so it scales how long the clip takes
# to play out.
left = max(0.0, duration - max(0.0, from_position)) / rate
return self._alloc(self._clock() + left, rate)
[docs]
def stop_audio(self, channel_id: int) -> None:
with self._lock:
self._channels.pop(channel_id, None)
self._paused.discard(channel_id)
[docs]
def pause_audio(self, channel_id: int) -> None:
with self._lock:
voice = self._channels.get(channel_id)
if voice is None:
return
self._paused.add(channel_id)
if voice.ends_at is not None:
# Freeze the countdown: a paused channel stays active and
# resumes with the same amount of clip left.
voice.remaining = max(0.0, voice.ends_at - self._clock())
voice.ends_at = None
[docs]
def resume_audio(self, channel_id: int) -> None:
with self._lock:
self._paused.discard(channel_id)
voice = self._channels.get(channel_id)
if voice is not None and voice.remaining is not None:
voice.ends_at = self._clock() + voice.remaining
voice.remaining = None
[docs]
def update_audio_2d(self, channel_id: int, volume_db: float, pan: float) -> None:
return
[docs]
def update_audio_3d(self, channel_id: int, volume_db: float, pan: float, pitch: float) -> None:
# Volume and pan have nothing to act on here, but the Doppler-shifted
# pitch decides when the channel ends, exactly as it does on a device.
self._repitch(channel_id, pitch)
[docs]
def set_pitch(self, channel_id: int, pitch: float) -> None:
self._repitch(channel_id, pitch)
# Listener pose endpoints: Null backend has no spatializer.
[docs]
def set_listener_position(self, x: float, y: float, z: float) -> None:
return
[docs]
def set_listener_velocity(self, x: float, y: float, z: float) -> None:
return
[docs]
def set_listener_direction(self, x: float, y: float, z: float) -> None:
return
[docs]
def set_listener_world_up(self, x: float, y: float, z: float) -> None:
return
[docs]
def get_playback_position(self, channel_id: int) -> float:
return 0.0
[docs]
def is_channel_active(self, channel_id: int) -> bool:
with self._lock:
voice = self._channels.get(channel_id)
if voice is None:
return False
if voice.ends_at is None or self._clock() < voice.ends_at:
return True
# The clip has run its course. Drop the voice here so a long
# headless run of one-shot sounds doesn't accumulate them: a
# player that polls its channel is under no obligation to stop
# one that has ended by itself.
del self._channels[channel_id]
self._paused.discard(channel_id)
return False
[docs]
def shutdown(self) -> None:
with self._lock:
self._channels.clear()
self._paused.clear()
[docs]
def list_capabilities(self) -> frozenset[Capability]:
# The null backend supports only the unconditional "play.basic"
# advertisement: every call succeeds (silently) and returns a
# valid channel id. Spatial / streaming / effect capabilities are
# deliberately omitted so callers that gate on them raise
# AudioCapabilityError or skip work rather than producing the
# illusion of a working pipeline.
return frozenset({Capability.PLAY_BASIC})
[docs]
def sync_bus_layout(self, layout: AudioBusLayout) -> None:
return