"""Audio effects: typed DSP nodes that attach to ``AudioBus``.
Effects are pure-Python descriptors; backends synchronise them into their
native graph during :meth:`AudioBackend.sync_bus_layout`. The Python layer
owns the *what* (parameters, ordering, capability requirements); the
backend owns the *how* (ma_engine ``ma_node_graph`` for desktop, Web Audio
``AudioNode`` chains for the browser).
Authoring::
from simvx.core import AudioBusLayout
from simvx.core.audio_effect import ReverbEffect, ParametricEQ, EQBand
layout = AudioBusLayout.get_default()
music = layout.get_bus("Music")
music.add_effect(ReverbEffect(room_size=0.7, wet=0.3))
music.add_effect(ParametricEQ(bands=[
EQBand(type="lowshelf", freq=120.0, gain_db=-3.0),
EQBand(type="peaking", freq=2500.0, q=1.4, gain_db=+2.0),
]))
Effects chain in declaration order (`source → bus_effects[0] → ... →
bus_effects[N-1] → send_to`).
Capability gating: each effect declares ``required_capability``.
Backends skip effects whose capability they don't advertise; the audio
server logs a single warning per unsupported effect / backend pair.
Effects are applied on both backends. The native desktop backend builds a
per-bus ``ma_*_node`` chain (built-in ``ma_lpf/hpf/bpf/notch/delay`` plus
custom DSP nodes for freeverb reverb, parametric EQ, soft-clip distortion,
and the compressor); the web backend maps each effect to a Web Audio node.
Effect Properties also round-trip through the inspector and scene serializer.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, ClassVar
from .audio_errors import InvalidStreamError
from .audio_protocol import Capability
from .descriptors import Property
__all__ = [
"AudioEffect",
"GainEffect",
"LowPassFilter",
"HighPassFilter",
"BandPassFilter",
"NotchFilter",
"DelayEffect",
"ReverbEffect",
"EQBand",
"ParametricEQ",
"SoftClipEffect",
"CompressorEffect",
]
# ---------------------------------------------------------------------------
# Effect registry: populated automatically by ``AudioEffect.__init_subclass__``.
#
# The key is the lowercased ``effect_type`` tag (also the value emitted under
# ``to_dict()["type"]`` and the case used by the JS bridge's switch in
# ``audio_bridge.js``). Using the wire-format tag as the registry key keeps
# serialization symmetric end-to-end.
# ---------------------------------------------------------------------------
_EFFECT_REGISTRY: dict[str, type[AudioEffect]] = {}
def _derive_effect_type(cls_name: str) -> str:
"""Mirror ``AudioEffect.effect_type`` at the class level (no instance)."""
for suffix in ("Effect", "Filter"):
if cls_name.endswith(suffix):
cls_name = cls_name[: -len(suffix)]
break
return cls_name.lower()
# ---------------------------------------------------------------------------
# Base
# ---------------------------------------------------------------------------
[docs]
class AudioEffect:
"""Base class for all audio effects.
Concrete subclasses declare typed `Property` descriptors for their
parameters; backends read parameter values via attribute access
(``effect.cutoff_hz``, ``effect.wet``, …).
`required_capability` is the :class:`Capability` a backend must
advertise to materialise this effect; backends that don't advertise
it skip the effect with a single warning per (effect, backend) pair.
`None` means "every backend supports this". Compares equal to the
underlying ``str`` for existing call sites that pattern-match on
the wire-format tag.
"""
required_capability: ClassVar[Capability | None] = None
# Effects are mutable plain objects so the inspector can poke
# Property descriptors directly. They're not Node subclasses: they
# live on an AudioBus's effects list and have no scene-tree
# lifecycle.
[docs]
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
# Internal helpers (leading underscore) and the dataclass base
# ``_BiquadEffect`` shouldn't appear under their own tag: only the
# concrete subclasses get registered. ``_BiquadEffect`` itself has no
# use as a dispatch target because LP/HP/BP/notch each derive from it
# and override the implicit type tag via their class names.
if cls.__name__.startswith("_"):
return
tag = _derive_effect_type(cls.__name__)
_EFFECT_REGISTRY[tag] = cls
def __init__(self, *, enabled: bool = True):
self.enabled = enabled
[docs]
def __repr__(self) -> str:
params = ", ".join(
f"{p.name}={getattr(self, p.name)!r}"
for p in self.__class__.__dict__.values()
if isinstance(p, Property) and p.name not in (None, "enabled")
)
state = "" if self.enabled else " [disabled]"
return f"{type(self).__name__}({params}){state}"
[docs]
def get_property_values(self) -> dict[str, Any]:
"""Return a dict of (Property name → current value) for serialization."""
out: dict[str, Any] = {}
for attr in dir(type(self)):
descriptor = getattr(type(self), attr, None)
if isinstance(descriptor, Property):
out[attr] = getattr(self, attr)
out["enabled"] = self.enabled
return out
[docs]
@property
def effect_type(self) -> str:
"""Lowercased class name with the ``Effect`` / ``Filter`` suffix stripped.
Used as the canonical type tag in backend wire protocols
(``reverb``, ``lowpass``, ``compressor``, …). Matches the
switch-cases on the JS bridge in `audio_bridge.js`.
"""
cls = type(self).__name__
for suffix in ("Effect", "Filter"):
if cls.endswith(suffix):
cls = cls[: -len(suffix)]
break
return cls.lower()
[docs]
def to_dict(self) -> dict[str, Any]:
"""Serialize for cross-backend transport (JS bridge, IPC, save files).
The format is deliberately flat: backends pattern-match on
``type`` and read ``params`` directly. Subclasses with nested
types (e.g. ``ParametricEQ`` with ``EQBand``) override this to
flatten the nesting.
"""
params = self.get_property_values()
params.pop("enabled", None)
return {
"type": self.effect_type,
"enabled": self.enabled,
"params": params,
}
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> AudioEffect:
"""Reconstruct an effect from its :meth:`to_dict` payload.
Dispatches on ``data["type"]`` via the module-level
:data:`_EFFECT_REGISTRY`, then forwards ``params`` plus the
``enabled`` flag to the target class's constructor. Subclasses
with nested types (e.g. :class:`ParametricEQ`) override this to
rebuild their nested structure from the flattened payload.
Raises:
InvalidStreamError: ``data["type"]`` isn't a known effect tag.
The message lists the registered tags so the caller can
spot the typo or register the missing effect class.
"""
tag = data.get("type")
if not isinstance(tag, str):
raise InvalidStreamError(
f"Audio effect payload missing 'type' key: {data!r}. "
f"Expected one of {sorted(_EFFECT_REGISTRY)}."
)
target = _EFFECT_REGISTRY.get(tag)
if target is None:
raise InvalidStreamError(
f"Unknown audio effect type: {tag!r}. "
f"Registered types: {sorted(_EFFECT_REGISTRY)}."
)
# Delegating to a subclass override keeps nested-type reconstruction
# working (e.g. ``ParametricEQ`` rebuilds ``EQBand`` from the
# flattened payload). Checking ``__dict__`` rather than comparing
# bound classmethods avoids recursion through the base impl.
if "from_dict" in target.__dict__:
return target.from_dict(data)
return target._construct_from_params(data)
@classmethod
def _construct_from_params(cls, data: dict[str, Any]) -> AudioEffect:
"""Default flat-payload constructor used by :meth:`from_dict`.
Pulls ``data["params"]`` and ``data["enabled"]`` and calls
``cls(**params, enabled=...)``. Effects with nested types should
override :meth:`from_dict` (not this helper) to rebuild the
nested structure before delegating here.
"""
params = dict(data.get("params") or {})
enabled = bool(data.get("enabled", True))
return cls(**params, enabled=enabled)
# ---------------------------------------------------------------------------
# Volume / fades
# ---------------------------------------------------------------------------
[docs]
class GainEffect(AudioEffect):
"""Static gain stage. Useful for trim before downstream effects."""
required_capability = Capability.EFFECT_GAIN
volume_db = Property(0.0, range=(-80.0, 24.0), hint="Gain in dB applied to the bus output.")
def __init__(self, *, volume_db: float = 0.0, enabled: bool = True):
super().__init__(enabled=enabled)
self.volume_db = volume_db
# ---------------------------------------------------------------------------
# Biquad filters (LP / HP / BP / notch): backed by ma_*pf_node
# ---------------------------------------------------------------------------
class _BiquadEffect(AudioEffect):
"""Shared base for LP/HP/BP/notch biquads.
`q` is the resonance (~0.707 is "flat", higher values peak at the
cutoff). Backends map this to their native biquad APIs.
"""
cutoff_hz = Property(1000.0, range=(20.0, 20000.0), hint="Cutoff frequency in Hz.")
q = Property(0.707, range=(0.1, 20.0), hint="Filter resonance (Q factor).")
def __init__(self, *, cutoff_hz: float = 1000.0, q: float = 0.707, enabled: bool = True):
super().__init__(enabled=enabled)
self.cutoff_hz = cutoff_hz
self.q = q
[docs]
class LowPassFilter(_BiquadEffect):
"""2nd-order low-pass biquad. Attenuates frequencies above ``cutoff_hz``."""
required_capability = Capability.EFFECT_FILTER_BIQUAD
[docs]
class HighPassFilter(_BiquadEffect):
"""2nd-order high-pass biquad. Attenuates frequencies below ``cutoff_hz``."""
required_capability = Capability.EFFECT_FILTER_BIQUAD
[docs]
class BandPassFilter(_BiquadEffect):
"""2nd-order band-pass biquad. Centered on ``cutoff_hz`` with bandwidth from ``q``."""
required_capability = Capability.EFFECT_FILTER_BIQUAD
[docs]
class NotchFilter(_BiquadEffect):
"""2nd-order notch biquad. Removes a narrow band around ``cutoff_hz``."""
required_capability = Capability.EFFECT_FILTER_BIQUAD
# ---------------------------------------------------------------------------
# Delay
# ---------------------------------------------------------------------------
[docs]
class DelayEffect(AudioEffect):
"""Single-tap delay with feedback. Wet/dry mix exposed for parallel sends."""
required_capability = Capability.EFFECT_DELAY
time_seconds = Property(0.25, range=(0.0, 5.0), hint="Delay time in seconds.")
feedback = Property(0.4, range=(0.0, 0.99), hint="Delayed signal fed back into the input.")
wet = Property(0.5, range=(0.0, 1.0), hint="Wet (delayed) signal level.")
dry = Property(0.5, range=(0.0, 1.0), hint="Dry (passthrough) signal level.")
def __init__(
self,
*,
time_seconds: float = 0.25,
feedback: float = 0.4,
wet: float = 0.5,
dry: float = 0.5,
enabled: bool = True,
):
super().__init__(enabled=enabled)
self.time_seconds = time_seconds
self.feedback = feedback
self.wet = wet
self.dry = dry
# ---------------------------------------------------------------------------
# Freeverb-style reverb
# ---------------------------------------------------------------------------
[docs]
class ReverbEffect(AudioEffect):
"""Schroeder-style algorithmic reverb (Jezar's freeverb model).
Same parameters across desktop and web. Desktop runs the freeverb
algorithm in a custom ``ma_node``; web runs a ``ConvolverNode`` with a
baked impulse response tuned to match.
"""
required_capability = Capability.EFFECT_REVERB
room_size = Property(0.5, range=(0.0, 1.0), hint="Reverb tail length (0 = tight, 1 = cathedral).")
damping = Property(0.5, range=(0.0, 1.0), hint="High-frequency damping in the reverb tail.")
wet = Property(0.3, range=(0.0, 1.0), hint="Reverberant signal level.")
dry = Property(0.7, range=(0.0, 1.0), hint="Dry (direct) signal level.")
width = Property(1.0, range=(0.0, 1.0), hint="Stereo width of the reverb tail.")
freeze = Property(False, hint="Hold the reverb tail indefinitely (infinite sustain).")
def __init__(
self,
*,
room_size: float = 0.5,
damping: float = 0.5,
wet: float = 0.3,
dry: float = 0.7,
width: float = 1.0,
freeze: bool = False,
enabled: bool = True,
):
super().__init__(enabled=enabled)
self.room_size = room_size
self.damping = damping
self.wet = wet
self.dry = dry
self.width = width
self.freeze = freeze
# ---------------------------------------------------------------------------
# Parametric EQ
# ---------------------------------------------------------------------------
[docs]
@dataclass
class EQBand:
"""One band of a `ParametricEQ`.
`type` selects the shape: ``"peaking"`` (centered bell),
``"lowshelf"`` (low-frequency cut/boost), ``"highshelf"``
(high-frequency cut/boost).
`gain_db` is the boost (positive) or cut (negative) at the band's
centre. Peaking bands also use `q` for bandwidth; shelves don't.
"""
type: str = "peaking" # "peaking" | "lowshelf" | "highshelf"
freq: float = 1000.0
q: float = 1.0
gain_db: float = 0.0
[docs]
class ParametricEQ(AudioEffect):
"""Multi-band parametric equalizer. Each band is an `EQBand`.
Default is a 3-band EQ (low shelf, mid peaking, high shelf) at flat
settings: add or modify bands to taste.
"""
required_capability = Capability.EFFECT_PARAMETRIC_EQ
def __init__(
self,
*,
bands: list[EQBand] | None = None,
enabled: bool = True,
):
super().__init__(enabled=enabled)
if bands is None:
bands = [
EQBand(type="lowshelf", freq=120.0, gain_db=0.0),
EQBand(type="peaking", freq=1000.0, q=1.0, gain_db=0.0),
EQBand(type="highshelf", freq=8000.0, gain_db=0.0),
]
self.bands = bands
[docs]
@property
def effect_type(self) -> str:
return "parametriceq"
[docs]
def to_dict(self) -> dict[str, Any]:
# Flatten EQBand into plain dicts so JS doesn't need to know about
# the dataclass type.
return {
"type": self.effect_type,
"enabled": self.enabled,
"params": {
"bands": [
{"type": b.type, "freq": b.freq, "q": b.q, "gain_db": b.gain_db}
for b in self.bands
],
},
}
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ParametricEQ:
"""Rebuild bands from the flattened :meth:`to_dict` payload."""
params = dict(data.get("params") or {})
raw_bands = params.pop("bands", [])
bands = [
EQBand(
type=str(b.get("type", "peaking")),
freq=float(b.get("freq", 1000.0)),
q=float(b.get("q", 1.0)),
gain_db=float(b.get("gain_db", 0.0)),
)
for b in raw_bands
]
return cls(bands=bands, enabled=bool(data.get("enabled", True)))
# ---------------------------------------------------------------------------
# Soft-clip distortion
# ---------------------------------------------------------------------------
[docs]
class SoftClipEffect(AudioEffect):
"""Symmetric tanh-curve soft clipping.
`drive` increases input gain before the curve (more saturation);
`output_gain` trims the output. ``tanh(x*drive) / tanh(drive)``
normalises so the curve passes through (1, 1) regardless of drive.
"""
required_capability = Capability.EFFECT_SOFTCLIP
drive = Property(2.0, range=(0.1, 20.0), hint="Pre-curve input gain (higher = more saturation).")
output_gain = Property(1.0, range=(0.0, 4.0), hint="Post-curve output trim (linear).")
def __init__(self, *, drive: float = 2.0, output_gain: float = 1.0, enabled: bool = True):
super().__init__(enabled=enabled)
self.drive = drive
self.output_gain = output_gain
# ---------------------------------------------------------------------------
# Compressor
# ---------------------------------------------------------------------------
[docs]
class CompressorEffect(AudioEffect):
"""Feed-forward compressor with soft knee.
Web maps this to `DynamicsCompressorNode` (browser-native).
Desktop runs a custom envelope-follower + gain-reduction node.
"""
required_capability = Capability.EFFECT_COMPRESSOR
threshold_db = Property(-24.0, range=(-80.0, 0.0), hint="Level above which compression engages.")
ratio = Property(4.0, range=(1.0, 20.0), hint="Input-to-output ratio above threshold.")
attack_ms = Property(5.0, range=(0.1, 200.0), hint="Envelope attack time in milliseconds.")
release_ms = Property(100.0, range=(10.0, 2000.0), hint="Envelope release time in milliseconds.")
knee_db = Property(6.0, range=(0.0, 24.0), hint="Width of the soft-knee transition in dB.")
makeup_db = Property(0.0, range=(0.0, 24.0), hint="Output gain after compression.")
def __init__(
self,
*,
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,
enabled: bool = True,
):
super().__init__(enabled=enabled)
self.threshold_db = threshold_db
self.ratio = ratio
self.attack_ms = attack_ms
self.release_ms = release_ms
self.knee_db = knee_db
self.makeup_db = makeup_db