"""Python wrapper around the compiled `_simvx_miniaudio_engine` cffi extension.
Exposes a small, Pythonic surface over miniaudio's `ma_engine` / `ma_sound` /
`ma_sound_group` API. Used by `simvx.core.audio_backend.MiniaudioBackend`
to run audio mixing in native C.
Build the extension once after install:
uv run python -m simvx.core._native.miniaudio_engine_build
If the .so is missing, importing this module raises `MiniaudioEngineUnavailable`
with build instructions. Callers (the backend) catch this and fall back to
the pure-Python mixer with a one-time warning.
"""
from __future__ import annotations
import abc
import logging
import os
import sys
from pathlib import Path
from typing import Any
import numpy as np
log = logging.getLogger(__name__)
__all__ = [
"MiniaudioEngineUnavailable",
"Engine",
"Sound",
"SoundGroup",
"AudioBuffer",
"StreamSource",
"EffectNode",
"LPFEffectNode",
"HPFEffectNode",
"BPFEffectNode",
"NotchEffectNode",
"PeakEffectNode",
"LowShelfEffectNode",
"HighShelfEffectNode",
"DelayEffectNode",
"SoftClipEffectNode",
"CompressorEffectNode",
"FreeverbEffectNode",
"attach_node",
"detach_node",
"set_node_output_volume",
"engine_endpoint",
"sound_group_as_node",
"is_available",
"ma_result_message",
]
[docs]
class MiniaudioEngineUnavailable(RuntimeError):
"""The compiled `_simvx_miniaudio_engine` extension was not found.
Run `uv run python -m simvx.core._native.miniaudio_engine_build` (or
`simvx build-audio`) to compile it once.
"""
def _try_import_extension() -> tuple[Any | None, Any | None]:
"""Import the compiled extension, returning (ffi, lib) or (None, None)."""
here = Path(__file__).resolve().parent
# Side-effect: ensure the extension dir is importable. Necessary because
# cffi names the module `_simvx_miniaudio_engine` (no package prefix).
if str(here) not in sys.path:
sys.path.insert(0, str(here))
try:
import _simvx_miniaudio_engine as _ext # type: ignore[import-not-found]
except ImportError:
return None, None
return _ext.ffi, _ext.lib
_FFI, _LIB = _try_import_extension()
[docs]
def is_available() -> bool:
"""True if the native extension was built and loaded successfully."""
return _LIB is not None
[docs]
def refresh() -> bool:
"""Re-run the extension import.
Called by `audio_backend.make_backend()` after a successful auto-build,
so module-level `_FFI` / `_LIB` pick up the freshly-compiled .so without
a Python restart.
"""
global _FFI, _LIB
_FFI, _LIB = _try_import_extension()
return _LIB is not None
def _require_lib():
if _LIB is None:
raise MiniaudioEngineUnavailable(
"The miniaudio engine extension is not built. Run:\n"
" uv run python -m simvx.core._native.miniaudio_engine_build"
)
return _LIB
# `ma_result` enum values we surface in error messages. Only the most-seen
# ones are spelled out; everything else falls back to "ma_result(<n>)".
_MA_RESULT_NAMES = {
0: "MA_SUCCESS",
-1: "MA_ERROR",
-2: "MA_INVALID_ARGS",
-3: "MA_INVALID_OPERATION",
-4: "MA_OUT_OF_MEMORY",
-5: "MA_OUT_OF_RANGE",
-6: "MA_ACCESS_DENIED",
-7: "MA_DOES_NOT_EXIST",
-8: "MA_ALREADY_EXISTS",
-10: "MA_PATH_TOO_LONG",
-11: "MA_AT_END",
-12: "MA_NO_SPACE",
-13: "MA_BUSY",
-14: "MA_IO_ERROR",
-15: "MA_INTERRUPT",
-16: "MA_UNAVAILABLE",
-17: "MA_ALREADY_IN_USE",
-18: "MA_BAD_ADDRESS",
-19: "MA_BAD_SEEK",
-20: "MA_BAD_PIPE",
-21: "MA_DEADLOCK",
-22: "MA_TOO_MANY_LINKS",
-23: "MA_NOT_IMPLEMENTED",
-24: "MA_NO_MESSAGE",
-25: "MA_BAD_MESSAGE",
-26: "MA_NO_DATA_AVAILABLE",
-27: "MA_INVALID_DATA",
-28: "MA_TIMEOUT",
-29: "MA_NO_NETWORK",
-30: "MA_NOT_UNIQUE",
-31: "MA_NOT_SOCKET",
-32: "MA_NO_ADDRESS",
-33: "MA_BAD_PROTOCOL",
-34: "MA_PROTOCOL_UNAVAILABLE",
-35: "MA_PROTOCOL_NOT_SUPPORTED",
-36: "MA_PROTOCOL_FAMILY_NOT_SUPPORTED",
-100: "MA_FORMAT_NOT_SUPPORTED",
-200: "MA_DEVICE_NOT_INITIALIZED",
-201: "MA_DEVICE_ALREADY_INITIALIZED",
-300: "MA_FAILED_TO_INIT_BACKEND",
}
[docs]
def ma_result_message(code: int) -> str:
"""Human-readable miniaudio result code."""
name = _MA_RESULT_NAMES.get(code)
if name:
return f"{name}({code})"
return f"ma_result({code})"
def _check(code: int, what: str) -> None:
if code != 0:
raise RuntimeError(f"{what} failed: {ma_result_message(code)}")
# ----------------------------------------------------------------- Engine
[docs]
class Engine:
"""Owns a single `ma_engine` instance.
Pass `device=True` (default) for a real playback device, or
`device=False` to use the no-device path (read mixed PCM manually
via `read_pcm_frames`: useful for headless testing).
"""
def __init__(
self,
*,
sample_rate: int = 48000,
channels: int = 2,
device: bool = True,
) -> None:
lib = _require_lib()
self._lib = lib
self._sample_rate = sample_rate
self._channels = channels
self._engine = lib.simvx_engine_alloc()
if self._engine == _FFI.NULL:
raise MemoryError("Failed to allocate ma_engine")
try:
if device:
rc = lib.simvx_engine_init_default(self._engine, sample_rate, channels)
else:
rc = lib.simvx_engine_init_no_device(self._engine, sample_rate, channels)
_check(rc, "ma_engine_init")
except Exception:
lib.simvx_engine_free(self._engine)
self._engine = _FFI.NULL
raise
[docs]
@property
def handle(self) -> Any:
return self._engine
[docs]
@property
def sample_rate(self) -> int:
if self._engine == _FFI.NULL:
return self._sample_rate
return int(self._lib.simvx_engine_get_sample_rate(self._engine))
[docs]
@property
def channels(self) -> int:
if self._engine == _FFI.NULL:
return self._channels
return int(self._lib.simvx_engine_get_channels(self._engine))
[docs]
def set_volume(self, linear_gain: float) -> None:
_check(self._lib.simvx_engine_set_volume(self._engine, float(linear_gain)),
"ma_engine_set_volume")
[docs]
def set_listener_position(self, x: float, y: float, z: float, *, idx: int = 0) -> None:
self._lib.simvx_listener_set_position(self._engine, idx, x, y, z)
[docs]
def set_listener_velocity(self, x: float, y: float, z: float, *, idx: int = 0) -> None:
self._lib.simvx_listener_set_velocity(self._engine, idx, x, y, z)
[docs]
def set_listener_direction(self, x: float, y: float, z: float, *, idx: int = 0) -> None:
self._lib.simvx_listener_set_direction(self._engine, idx, x, y, z)
[docs]
def set_listener_world_up(self, x: float, y: float, z: float, *, idx: int = 0) -> None:
self._lib.simvx_listener_set_world_up(self._engine, idx, x, y, z)
[docs]
def read_pcm_frames(self, frame_count: int) -> np.ndarray:
"""Pull `frame_count` mixed PCM frames (float32, interleaved).
Only valid for engines built with `device=False`. Returns an
ndarray of shape `(frame_count, channels)`. The number of frames
actually read may be less than requested if the engine is idle.
"""
if self._engine == _FFI.NULL:
raise RuntimeError("Engine has been shut down")
out = np.zeros(frame_count * self._channels, dtype=np.float32)
out_ptr = _FFI.cast("float*", out.ctypes.data)
read = _FFI.new("unsigned long long*")
_check(
self._lib.simvx_engine_read_pcm_frames_f32(
self._engine, out_ptr, frame_count, read
),
"ma_engine_read_pcm_frames",
)
actual = int(read[0])
return out.reshape(-1, self._channels)[:actual]
[docs]
def shutdown(self) -> None:
if self._engine != _FFI.NULL:
self._lib.simvx_engine_uninit(self._engine)
self._lib.simvx_engine_free(self._engine)
self._engine = _FFI.NULL
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
# ----------------------------------------------------------------- SoundGroup
[docs]
class SoundGroup:
"""An `ma_sound_group`: a named bus that sounds attach to.
`parent=None` parents to the engine root. Volumes multiply through
the parent chain in native code.
"""
def __init__(
self,
engine: Engine,
*,
parent: SoundGroup | None = None,
) -> None:
self._lib = engine._lib
self._engine = engine
self._group = self._lib.simvx_sound_group_alloc()
if self._group == _FFI.NULL:
raise MemoryError("Failed to allocate ma_sound_group")
try:
parent_handle = parent._group if parent is not None else _FFI.NULL
rc = self._lib.simvx_sound_group_init(
engine.handle, self._group, 0, parent_handle
)
_check(rc, "ma_sound_group_init")
except Exception:
self._lib.simvx_sound_group_free(self._group)
self._group = _FFI.NULL
raise
[docs]
@property
def handle(self) -> Any:
return self._group
[docs]
def set_volume(self, linear_gain: float) -> None:
self._lib.simvx_sound_group_set_volume(self._group, float(linear_gain))
[docs]
def shutdown(self) -> None:
if self._group != _FFI.NULL:
self._lib.simvx_sound_group_uninit(self._group)
self._lib.simvx_sound_group_free(self._group)
self._group = _FFI.NULL
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
# ----------------------------------------------------------------- AudioBuffer
[docs]
class AudioBuffer:
"""An `ma_audio_buffer` backed by a numpy ndarray.
The ndarray must remain alive for the lifetime of the buffer: we
do not copy. Stored as a Python attribute so the GC keeps it
pinned.
"""
def __init__(
self,
samples: np.ndarray,
*,
sample_rate: int,
channels: int,
) -> None:
lib = _require_lib()
self._lib = lib
# Force layout cffi can hand to ma_audio_buffer: float32 interleaved.
arr = np.ascontiguousarray(samples.astype(np.float32, copy=False)).reshape(-1)
if arr.size % channels != 0:
raise ValueError(
f"Sample buffer length {arr.size} not divisible by channels {channels}"
)
self._samples = arr # pin against GC
frame_count = arr.size // channels
self._buffer = lib.simvx_audio_buffer_alloc()
if self._buffer == _FFI.NULL:
raise MemoryError("Failed to allocate ma_audio_buffer")
try:
data_ptr = _FFI.cast("void*", arr.ctypes.data)
rc = lib.simvx_audio_buffer_init(
self._buffer,
lib.simvx_format_f32(),
channels,
frame_count,
data_ptr,
sample_rate,
)
_check(rc, "ma_audio_buffer_init")
except Exception:
lib.simvx_audio_buffer_free(self._buffer)
self._buffer = _FFI.NULL
raise
[docs]
@property
def data_source(self) -> Any:
"""The `ma_data_source*` view of this buffer (usable by Sound)."""
return self._lib.simvx_audio_buffer_as_data_source(self._buffer)
[docs]
def shutdown(self) -> None:
if self._buffer != _FFI.NULL:
self._lib.simvx_audio_buffer_uninit(self._buffer)
self._lib.simvx_audio_buffer_free(self._buffer)
self._buffer = _FFI.NULL
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
# ----------------------------------------------------------------- StreamSource
[docs]
class StreamSource:
"""An `ma_pcm_rb` (PCM ring buffer) wrapped as a `ma_data_source`.
Used by ``MiniaudioBackend.open_stream`` for live procedural audio
(``AudioSynth.attach_to``) and disk-streamed playback. Producer
writes chunks via ``write(bytes)`` from the main thread; miniaudio's
audio thread pulls from the ring buffer's built-in data source view.
SPSC thread-safe: no locks needed.
Underrun semantics: when the ring is empty, miniaudio's read callback
pads with silence and continues. The audio thread never stalls or
glitches; consumers just hear silence until more chunks arrive.
Default format is ``s16`` to match the legacy backend's
``feed_audio_chunk(bytes)`` convention.
"""
_FORMAT_TO_BYTES: dict[str, tuple[int, int]] = {
# name → (ma_format enum value, bytes per sample)
# populated lazily by __init__ from the live cffi lib
}
def __init__(
self,
engine: Engine,
*,
sample_rate: int,
channels: int,
buffer_seconds: float = 0.5,
format: str = "s16",
):
lib = _require_lib()
self._lib = lib
if not self._FORMAT_TO_BYTES:
# Populate once at first use.
self._FORMAT_TO_BYTES.update(
{
"f32": (int(lib.simvx_format_f32()), 4),
"s16": (int(lib.simvx_format_s16()), 2),
"s32": (int(lib.simvx_format_s32()), 4),
"u8": (int(lib.simvx_format_u8()), 1),
}
)
if format not in self._FORMAT_TO_BYTES:
raise ValueError(
f"format must be one of {list(self._FORMAT_TO_BYTES)}, got {format!r}"
)
fmt_id, bytes_per_sample = self._FORMAT_TO_BYTES[format]
self._format = format
self._sample_rate = int(sample_rate)
self._channels = int(channels)
self._bytes_per_frame = bytes_per_sample * int(channels)
buffer_size_frames = max(1, int(buffer_seconds * sample_rate))
self._buffer_size_frames = buffer_size_frames
self._rb = lib.simvx_pcm_rb_alloc()
if self._rb == _FFI.NULL:
raise MemoryError("Failed to allocate ma_pcm_rb")
try:
rc = lib.simvx_pcm_rb_init(
self._rb, fmt_id, int(channels), buffer_size_frames, int(sample_rate)
)
_check(rc, "ma_pcm_rb_init")
except Exception:
lib.simvx_pcm_rb_free(self._rb)
self._rb = _FFI.NULL
raise
[docs]
@property
def data_source(self) -> Any:
"""The ``ma_data_source*`` view of this ring buffer (for `Sound.from_data_source`)."""
return self._lib.simvx_pcm_rb_as_data_source(self._rb)
[docs]
@property
def sample_rate(self) -> int:
return self._sample_rate
[docs]
@property
def channels(self) -> int:
return self._channels
[docs]
@property
def buffer_size_frames(self) -> int:
return self._buffer_size_frames
[docs]
@property
def available_write_frames(self) -> int:
return int(self._lib.simvx_pcm_rb_available_write(self._rb))
[docs]
@property
def available_read_frames(self) -> int:
return int(self._lib.simvx_pcm_rb_available_read(self._rb))
[docs]
def write(self, chunk: bytes) -> int:
"""Push `chunk` into the ring buffer. Returns frames actually written.
Frame-misaligned chunks have the trailing partial frame trimmed
rather than rejected (most consumers feed in fixed-size
sample-count buffers, so misalignment usually means the caller
produced more samples than channels divide cleanly).
Returns the number of frames written. If the ring is full,
returns less than requested: the caller can retry on the next
tick.
"""
if self._rb == _FFI.NULL:
return 0
if not chunk:
return 0
n_bytes = len(chunk) - (len(chunk) % self._bytes_per_frame)
if n_bytes == 0:
return 0
frame_count = n_bytes // self._bytes_per_frame
if n_bytes != len(chunk):
# Trim by giving cffi a sliced view (avoids a copy).
chunk = chunk[:n_bytes]
frames_written = _FFI.new("unsigned int*")
rc = self._lib.simvx_pcm_rb_write(
self._rb,
_FFI.from_buffer(chunk),
frame_count,
frames_written,
)
_check(rc, "ma_pcm_rb_write")
return int(frames_written[0])
[docs]
def reset(self) -> None:
"""Drop all queued frames (useful when restarting a stream)."""
if self._rb != _FFI.NULL:
self._lib.simvx_pcm_rb_reset(self._rb)
[docs]
def shutdown(self) -> None:
if self._rb != _FFI.NULL:
try:
self._lib.simvx_pcm_rb_uninit(self._rb)
except Exception:
pass
self._lib.simvx_pcm_rb_free(self._rb)
self._rb = _FFI.NULL
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
# ----------------------------------------------------------------- Sound
[docs]
class Sound:
"""An `ma_sound`: a single playable voice.
Construct via `Sound.from_file(engine, path)` or
`Sound.from_buffer(engine, buffer)`. Optional `group` attaches the
sound to a `SoundGroup` (bus).
"""
def __init__(
self,
engine: Engine,
*,
spatial: bool = False,
pitch_enabled: bool = True,
) -> None:
self._lib = engine._lib
self._engine = engine
self._sound = self._lib.simvx_sound_alloc()
if self._sound == _FFI.NULL:
raise MemoryError("Failed to allocate ma_sound")
self._initialised = False
self._source_keeper: Any = None # pin AudioBuffer / file path
self._spatial = spatial
self._pitch_enabled = pitch_enabled
[docs]
@classmethod
def from_file(
cls,
engine: Engine,
path: str | os.PathLike,
*,
group: SoundGroup | None = None,
spatial: bool = False,
pitch_enabled: bool = True,
decode_now: bool = True,
stream: bool = False,
) -> Sound:
sound = cls(engine, spatial=spatial, pitch_enabled=pitch_enabled)
lib = sound._lib
flags = 0
if stream:
flags |= int(lib.simvx_flag_stream())
elif decode_now:
flags |= int(lib.simvx_flag_decode())
if not spatial:
flags |= int(lib.simvx_flag_no_spatialization())
if not pitch_enabled:
flags |= int(lib.simvx_flag_no_pitch())
group_handle = group.handle if group is not None else _FFI.NULL
path_bytes = os.fspath(path).encode("utf-8")
rc = lib.simvx_sound_init_from_file(
engine.handle, sound._sound, path_bytes, flags, group_handle
)
if rc != 0:
lib.simvx_sound_free(sound._sound)
sound._sound = _FFI.NULL
raise RuntimeError(
f"ma_sound_init_from_file({path!r}) failed: {ma_result_message(rc)}"
)
sound._initialised = True
sound._source_keeper = path_bytes
return sound
[docs]
@classmethod
def from_buffer(
cls,
engine: Engine,
buffer: AudioBuffer,
*,
group: SoundGroup | None = None,
spatial: bool = False,
pitch_enabled: bool = True,
) -> Sound:
return cls.from_data_source(
engine, buffer.data_source, keeper=buffer,
group=group, spatial=spatial, pitch_enabled=pitch_enabled,
)
[docs]
@classmethod
def from_stream(
cls,
engine: Engine,
stream: StreamSource,
*,
group: SoundGroup | None = None,
spatial: bool = False,
) -> Sound:
"""Bind a `ma_sound` to a live `StreamSource` (ma_pcm_rb backed).
Pitch is disabled by default because the engine's pitch resampler
would try to read ahead into the ring buffer past what's been
written, producing audible artifacts. Stream consumers wanting
pitch control should resample Python-side before
`feed_audio_chunk`.
"""
return cls.from_data_source(
engine, stream.data_source, keeper=stream,
group=group, spatial=spatial, pitch_enabled=False,
)
[docs]
@classmethod
def from_data_source(
cls,
engine: Engine,
data_source: Any,
*,
keeper: Any = None,
group: SoundGroup | None = None,
spatial: bool = False,
pitch_enabled: bool = True,
) -> Sound:
"""Bind a `ma_sound` to any ma_data_source-compatible source.
`keeper` is pinned via the sound's `_source_keeper` so the
source's GC can't reap it while the sound is alive (essential
for ndarray-backed AudioBuffer + ring-buffer-backed StreamSource).
"""
sound = cls(engine, spatial=spatial, pitch_enabled=pitch_enabled)
lib = sound._lib
flags = 0
if not spatial:
flags |= int(lib.simvx_flag_no_spatialization())
if not pitch_enabled:
flags |= int(lib.simvx_flag_no_pitch())
group_handle = group.handle if group is not None else _FFI.NULL
rc = lib.simvx_sound_init_from_data_source(
engine.handle, sound._sound, data_source, flags, group_handle
)
if rc != 0:
lib.simvx_sound_free(sound._sound)
sound._sound = _FFI.NULL
raise RuntimeError(
f"ma_sound_init_from_data_source failed: {ma_result_message(rc)}"
)
sound._initialised = True
sound._source_keeper = keeper
return sound
[docs]
def start(self) -> None:
_check(self._lib.simvx_sound_start(self._sound), "ma_sound_start")
[docs]
def stop(self) -> None:
_check(self._lib.simvx_sound_stop(self._sound), "ma_sound_stop")
[docs]
def set_volume(self, linear_gain: float) -> None:
self._lib.simvx_sound_set_volume(self._sound, float(linear_gain))
[docs]
def set_pan(self, pan: float) -> None:
# ma_sound_set_pan expects [-1, +1]
self._lib.simvx_sound_set_pan(self._sound, float(pan))
[docs]
def set_pitch(self, pitch: float) -> None:
self._lib.simvx_sound_set_pitch(self._sound, float(pitch))
[docs]
def set_looping(self, looping: bool) -> None:
self._lib.simvx_sound_set_looping(self._sound, 1 if looping else 0)
[docs]
def set_position(self, x: float, y: float, z: float) -> None:
self._lib.simvx_sound_set_position(self._sound, x, y, z)
[docs]
def set_velocity(self, x: float, y: float, z: float) -> None:
self._lib.simvx_sound_set_velocity(self._sound, x, y, z)
[docs]
def set_spatialization_enabled(self, enabled: bool) -> None:
self._lib.simvx_sound_set_spatialization_enabled(self._sound, 1 if enabled else 0)
[docs]
def set_min_distance(self, d: float) -> None:
self._lib.simvx_sound_set_min_distance(self._sound, float(d))
[docs]
def set_max_distance(self, d: float) -> None:
self._lib.simvx_sound_set_max_distance(self._sound, float(d))
[docs]
def set_rolloff(self, r: float) -> None:
self._lib.simvx_sound_set_rolloff(self._sound, float(r))
[docs]
def set_doppler_factor(self, f: float) -> None:
self._lib.simvx_sound_set_doppler_factor(self._sound, float(f))
[docs]
def is_playing(self) -> bool:
return self._lib.simvx_sound_is_playing(self._sound) != 0
[docs]
def at_end(self) -> bool:
return self._lib.simvx_sound_at_end(self._sound) != 0
[docs]
def seek_to_frame(self, frame: int) -> None:
_check(self._lib.simvx_sound_seek_to_pcm_frame(self._sound, int(frame)),
"ma_sound_seek_to_pcm_frame")
[docs]
def cursor_frames(self) -> int:
out = _FFI.new("unsigned long long*")
_check(self._lib.simvx_sound_get_cursor_pcm_frames(self._sound, out),
"ma_sound_get_cursor_in_pcm_frames")
return int(out[0])
[docs]
def length_frames(self) -> int:
out = _FFI.new("unsigned long long*")
_check(self._lib.simvx_sound_get_length_pcm_frames(self._sound, out),
"ma_sound_get_length_in_pcm_frames")
return int(out[0])
[docs]
def shutdown(self) -> None:
if self._sound != _FFI.NULL:
if self._initialised:
self._lib.simvx_sound_uninit(self._sound)
self._initialised = False
self._lib.simvx_sound_free(self._sound)
self._sound = _FFI.NULL
self._source_keeper = None
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
# ----------------------------------------------------------------- EffectNode
[docs]
class EffectNode(abc.ABC):
"""Base wrapper for ma_*_node effect nodes attached to the engine's node graph.
Abstract: instantiating it directly (or a subclass that omits ``_init``)
raises ``TypeError``. Subclasses set ``_alloc_fn`` / ``_free_fn`` /
``_uninit_fn`` (callable cffi function handles) and implement
``_init(engine)``. The base class manages allocation, init,
output-bus connections, and cleanup.
`handle` returns the raw ``ma_node*`` (the first field is
``ma_node_base``, so the alloc'd struct pointer doubles as a node).
Engine, source, and destination nodes plug into each other via
``attach_node`` / ``detach_node`` module-level helpers.
"""
_alloc_fn = None # set by subclasses
_free_fn = None
_uninit_fn = None
def __init__(self, engine: Engine):
self._lib = _require_lib()
self._engine = engine
self._node = self._alloc_fn()
if self._node == _FFI.NULL:
raise MemoryError(f"Failed to allocate {type(self).__name__}")
try:
self._init(engine)
self._initialised = True
except Exception:
self._free_fn(self._node)
self._node = _FFI.NULL
self._initialised = False
raise
@abc.abstractmethod
def _init(self, engine: Engine) -> None:
raise NotImplementedError(
"EffectNode._init is abstract; subclass must initialise its ma_*_node via the engine."
)
[docs]
@property
def handle(self) -> Any:
"""Pointer-typed handle suitable for ``attach_node`` / ``detach_node``."""
return self._node
[docs]
def shutdown(self) -> None:
if self._node == _FFI.NULL:
return
if self._initialised:
try:
self._uninit_fn(self._node)
except Exception:
log.debug("EffectNode uninit failed", exc_info=True)
self._initialised = False
self._free_fn(self._node)
self._node = _FFI.NULL
[docs]
def __del__(self) -> None:
try:
self.shutdown()
except Exception:
pass
def _bind_methods(cls, prefix: str):
"""Wire ``_alloc_fn`` / ``_free_fn`` / ``_uninit_fn`` from the cffi lib once."""
if _LIB is None:
return
cls._alloc_fn = getattr(_LIB, f"simvx_{prefix}_alloc")
cls._free_fn = getattr(_LIB, f"simvx_{prefix}_free")
cls._uninit_fn = getattr(_LIB, f"simvx_{prefix}_uninit")
[docs]
class LPFEffectNode(EffectNode):
"""2nd-order low-pass biquad with configurable Q (resonance).
Wraps the C `simvx_biquad_lpf_node` (an ma_node subclass that runs
`ma_lpf2_process_pcm_frames` in its process callback). Unlike the
stock `ma_lpf_node`, which hardcodes Q to 0.707, this honours the Q
passed by `LowPassFilter(q=...)` so the engine's filter behaviour
matches the web backend's BiquadFilterNode.
"""
def __init__(self, engine: Engine, *, cutoff_hz: float = 1000.0, q: float = 0.707):
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_biquad_lpf_node_init(
engine.handle, self._node, self._cutoff_hz, self._q
),
"simvx_biquad_lpf_node_init",
)
[docs]
def set_params(self, cutoff_hz: float, q: float) -> None:
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
_check(
self._lib.simvx_biquad_lpf_node_update_params(
self._node, self._engine.handle, self._cutoff_hz, self._q
),
"simvx_biquad_lpf_node_update_params",
)
[docs]
class HPFEffectNode(EffectNode):
"""2nd-order high-pass biquad with configurable Q (resonance)."""
def __init__(self, engine: Engine, *, cutoff_hz: float = 1000.0, q: float = 0.707):
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_biquad_hpf_node_init(
engine.handle, self._node, self._cutoff_hz, self._q
),
"simvx_biquad_hpf_node_init",
)
[docs]
def set_params(self, cutoff_hz: float, q: float) -> None:
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
_check(
self._lib.simvx_biquad_hpf_node_update_params(
self._node, self._engine.handle, self._cutoff_hz, self._q
),
"simvx_biquad_hpf_node_update_params",
)
[docs]
class BPFEffectNode(EffectNode):
"""2nd-order band-pass biquad with configurable Q (bandwidth)."""
def __init__(self, engine: Engine, *, cutoff_hz: float = 1000.0, q: float = 0.707):
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_biquad_bpf_node_init(
engine.handle, self._node, self._cutoff_hz, self._q
),
"simvx_biquad_bpf_node_init",
)
[docs]
def set_params(self, cutoff_hz: float, q: float) -> None:
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
_check(
self._lib.simvx_biquad_bpf_node_update_params(
self._node, self._engine.handle, self._cutoff_hz, self._q
),
"simvx_biquad_bpf_node_update_params",
)
[docs]
class NotchEffectNode(EffectNode):
"""2nd-order notch biquad (band-stop)."""
def __init__(self, engine: Engine, *, cutoff_hz: float = 1000.0, q: float = 1.0):
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(self._lib.simvx_notch_node_init(engine.handle, self._node, self._cutoff_hz, self._q),
"ma_notch_node_init")
[docs]
def set_params(self, cutoff_hz: float, q: float) -> None:
self._cutoff_hz = float(cutoff_hz)
self._q = float(q)
_check(
self._lib.simvx_notch_node_reinit(self._node, self._engine.handle, self._cutoff_hz, self._q),
"ma_notch_node_reinit",
)
[docs]
class PeakEffectNode(EffectNode):
"""Peaking-EQ band biquad."""
def __init__(self, engine: Engine, *, freq: float = 1000.0, q: float = 1.0, gain_db: float = 0.0):
self._freq = float(freq)
self._q = float(q)
self._gain_db = float(gain_db)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(self._lib.simvx_peak_node_init(engine.handle, self._node, self._gain_db, self._q, self._freq),
"ma_peak_node_init")
[docs]
class LowShelfEffectNode(EffectNode):
"""Low-shelf parametric-EQ band biquad."""
def __init__(self, engine: Engine, *, freq: float = 200.0, q: float = 1.0, gain_db: float = 0.0):
self._freq = float(freq)
self._q = float(q)
self._gain_db = float(gain_db)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(self._lib.simvx_loshelf_node_init(engine.handle, self._node, self._gain_db, self._q, self._freq),
"ma_loshelf_node_init")
[docs]
class HighShelfEffectNode(EffectNode):
"""High-shelf parametric-EQ band biquad."""
def __init__(self, engine: Engine, *, freq: float = 8000.0, q: float = 1.0, gain_db: float = 0.0):
self._freq = float(freq)
self._q = float(q)
self._gain_db = float(gain_db)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(self._lib.simvx_hishelf_node_init(engine.handle, self._node, self._gain_db, self._q, self._freq),
"ma_hishelf_node_init")
[docs]
class DelayEffectNode(EffectNode):
"""Single-tap delay with feedback decay and wet/dry mix."""
def __init__(
self,
engine: Engine,
*,
delay_seconds: float = 0.25,
decay: float = 0.4,
wet: float = 0.5,
dry: float = 0.5,
):
self._delay_seconds = float(delay_seconds)
self._decay = float(decay)
self._wet = float(wet)
self._dry = float(dry)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_delay_node_init(
engine.handle, self._node, self._delay_seconds, self._decay
),
"ma_delay_node_init",
)
# Wet/dry are post-init runtime params.
self._lib.simvx_delay_node_set_wet(self._node, self._wet)
self._lib.simvx_delay_node_set_dry(self._node, self._dry)
[docs]
def set_wet(self, wet: float) -> None:
self._wet = float(wet)
self._lib.simvx_delay_node_set_wet(self._node, self._wet)
[docs]
def set_dry(self, dry: float) -> None:
self._dry = float(dry)
self._lib.simvx_delay_node_set_dry(self._node, self._dry)
[docs]
def set_decay(self, decay: float) -> None:
self._decay = float(decay)
self._lib.simvx_delay_node_set_decay(self._node, self._decay)
[docs]
class SoftClipEffectNode(EffectNode):
"""Tanh-curve symmetric soft clipping (custom DSP node)."""
def __init__(self, engine: Engine, *, drive: float = 2.0, output_gain: float = 1.0):
self._drive = float(drive)
self._output_gain = float(output_gain)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_softclip_node_init(
engine.handle, self._node, self._drive, self._output_gain
),
"simvx_softclip_node_init",
)
[docs]
def set_params(self, drive: float, output_gain: float) -> None:
self._drive = float(drive)
self._output_gain = float(output_gain)
self._lib.simvx_softclip_node_set_params(self._node, self._drive, self._output_gain)
[docs]
class CompressorEffectNode(EffectNode):
"""Feed-forward compressor with soft knee (custom DSP node)."""
def __init__(
self,
engine: Engine,
*,
threshold_db: float = -24.0,
ratio: float = 4.0,
attack_ms: float = 5.0,
release_ms: float = 100.0,
knee_db: float = 6.0,
makeup_db: float = 0.0,
):
self._threshold_db = float(threshold_db)
self._ratio = float(ratio)
self._attack_ms = float(attack_ms)
self._release_ms = float(release_ms)
self._knee_db = float(knee_db)
self._makeup_db = float(makeup_db)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_compressor_node_init(
engine.handle, self._node,
self._threshold_db, self._ratio,
self._attack_ms, self._release_ms,
self._knee_db, self._makeup_db,
),
"simvx_compressor_node_init",
)
[docs]
def set_params(
self,
*,
threshold_db: float | None = None,
ratio: float | None = None,
attack_ms: float | None = None,
release_ms: float | None = None,
knee_db: float | None = None,
makeup_db: float | None = None,
) -> None:
if threshold_db is not None: self._threshold_db = float(threshold_db)
if ratio is not None: self._ratio = float(ratio)
if attack_ms is not None: self._attack_ms = float(attack_ms)
if release_ms is not None: self._release_ms = float(release_ms)
if knee_db is not None: self._knee_db = float(knee_db)
if makeup_db is not None: self._makeup_db = float(makeup_db)
self._lib.simvx_compressor_node_set_params(
self._node, self._engine.handle,
self._threshold_db, self._ratio,
self._attack_ms, self._release_ms,
self._knee_db, self._makeup_db,
)
[docs]
class FreeverbEffectNode(EffectNode):
"""Schroeder-style algorithmic reverb (custom DSP node).
8 parallel comb filters → 4 series allpass filters per channel. Same
parameters as `WebAudioBackend`'s ConvolverNode-based reverb so the
cross-backend sound is consistent within "same vibe" tolerance.
"""
def __init__(
self,
engine: Engine,
*,
room_size: float = 0.5,
damping: float = 0.5,
wet: float = 0.3,
dry: float = 0.7,
width: float = 1.0,
freeze: bool = False,
):
self._room_size = float(room_size)
self._damping = float(damping)
self._wet = float(wet)
self._dry = float(dry)
self._width = float(width)
self._freeze = bool(freeze)
super().__init__(engine)
def _init(self, engine: Engine) -> None:
_check(
self._lib.simvx_freeverb_node_init(
engine.handle, self._node,
self._room_size, self._damping,
self._wet, self._dry, self._width,
1 if self._freeze else 0,
),
"simvx_freeverb_node_init",
)
[docs]
def set_params(
self,
*,
room_size: float | None = None,
damping: float | None = None,
wet: float | None = None,
dry: float | None = None,
width: float | None = None,
freeze: bool | None = None,
) -> None:
if room_size is not None: self._room_size = float(room_size)
if damping is not None: self._damping = float(damping)
if wet is not None: self._wet = float(wet)
if dry is not None: self._dry = float(dry)
if width is not None: self._width = float(width)
if freeze is not None: self._freeze = bool(freeze)
self._lib.simvx_freeverb_node_set_params(
self._node, self._room_size, self._damping,
self._wet, self._dry, self._width,
1 if self._freeze else 0,
)
# Wire alloc/free/uninit at module load (skipped if extension missing).
for _cls, _prefix in (
(LPFEffectNode, "biquad_lpf_node"),
(HPFEffectNode, "biquad_hpf_node"),
(BPFEffectNode, "biquad_bpf_node"),
(NotchEffectNode, "notch_node"),
(PeakEffectNode, "peak_node"),
(LowShelfEffectNode, "loshelf_node"),
(HighShelfEffectNode, "hishelf_node"),
(DelayEffectNode, "delay_node"),
(SoftClipEffectNode, "softclip_node"),
(CompressorEffectNode, "compressor_node"),
(FreeverbEffectNode, "freeverb_node"),
):
_bind_methods(_cls, _prefix)
# --- module-level node-graph helpers --------------------------------------
[docs]
def engine_endpoint(engine: Engine) -> Any:
"""Return the engine's audio destination node (chain tail attaches here)."""
return _require_lib().simvx_engine_endpoint(engine.handle)
[docs]
def sound_group_as_node(group: SoundGroup) -> Any:
"""Cast a SoundGroup to a generic node pointer for chain wiring."""
return _require_lib().simvx_sound_group_as_node(group.handle)
[docs]
def attach_node(src: Any, src_bus: int, dst: Any, dst_bus: int) -> None:
"""Route ``src`` output bus to ``dst`` input bus. Either may be NULL."""
_check(_require_lib().simvx_node_attach(src, src_bus, dst, dst_bus),
"ma_node_attach_output_bus")
[docs]
def detach_node(src: Any, src_bus: int) -> None:
"""Detach the output bus on ``src`` (no-op if not attached)."""
_check(_require_lib().simvx_node_detach(src, src_bus),
"ma_node_detach_output_bus")
[docs]
def set_node_output_volume(node: Any, bus: int, volume: float) -> None:
"""Scale the gain on an output bus (used for GainEffect: FadeEffect was
removed in the audio refactor; per-player fades live on AudioStreamPlayer
.fade_in/.fade_out)."""
_check(_require_lib().simvx_node_set_output_volume(node, bus, float(volume)),
"ma_node_set_output_bus_volume")