Skip to content

Effects Processing

This tutorial covers audio effects processing using AudioUnits with coremusic.

Prerequisites

  • coremusic installed and built
  • Basic Python knowledge
  • Audio files to process

Understanding AudioUnits

AudioUnits are macOS audio plugins that process audio:

  • Effects (aufx): Modify audio (reverb, delay, EQ, compression)
  • Instruments (aumu): Generate audio from MIDI
  • Generators (augn): Generate audio (test tones, noise)
  • Mixers (aumx): Mix multiple audio streams

Discovering Available Effects

List All AudioUnits

from coremusic.audio import list_available_audio_units


def list_all_audio_units():
    """List all available AudioUnits."""
    units = list_available_audio_units()

    print(f"Found {len(units)} AudioUnits:\n")

    # Group by type
    by_type = {}
    for unit in units:
        by_type.setdefault(unit['type'], []).append(unit)

    type_names = {
        'aufx': 'Effects',
        'aumu': 'Instruments',
        'augn': 'Generators',
        'aumx': 'Mixers',
        'aufc': 'Format Converters',
        'auou': 'Output Units',
    }

    for unit_type, units_list in sorted(by_type.items()):
        name = type_names.get(unit_type, unit_type)
        print(f"{name} ({unit_type}): {len(units_list)} plugins")


list_all_audio_units()

List Effects Only

from coremusic.audio import get_audiounit_names


def list_effects():
    """List only effect AudioUnits."""
    names = get_audiounit_names(filter_type='aufx')

    print("Available Effects:")
    for name in sorted(names)[:10]:
        print(f"  {name}")

    return names


effects = list_effects()

Find Specific Effect

from coremusic.audio.audiounit_host import AudioUnitHost


def find_effect(name):
    """Find an effect by name."""
    host = AudioUnitHost()
    matches = [
        plugin for plugin in host.discover_plugins(type='effect')
        if name.lower() in plugin['name'].lower()
    ]

    if not matches:
        print(f"Not found: {name}")
        return None

    plugin = matches[0]
    print(f"Found: {plugin['name']}")
    print(f"  Type: {plugin['type']}")
    print(f"  Subtype: {plugin['subtype']}")
    print(f"  Manufacturer: {plugin['manufacturer']}")
    return plugin


# Find AUDelay
delay = find_effect("AUDelay")

# Find by partial name
reverb = find_effect("Reverb")

Using the CLI

# List all plugins
coremusic plugin list

# List effects only
coremusic plugin list --type aufx

# Get plugin info
coremusic plugin info AUDelay

Creating an Effects Chain

Simple Effect Chain

from coremusic.audio import AudioEffectsChain


def create_simple_chain():
    """Create a simple effect chain."""
    chain = AudioEffectsChain()
    chain.open()

    # Add effect by name
    delay_node = chain.add_effect_by_name("AUDelay")

    # Add output
    output_node = chain.add_output()

    # Connect effect to output
    chain.connect(delay_node, output_node)

    print(f"Created chain with {chain.node_count} nodes")
    return chain


chain = create_simple_chain()
chain.dispose()

Multiple Effects Chain

from coremusic.audio import AudioEffectsChain


def create_multi_effect_chain():
    """Create chain with multiple effects."""
    chain = AudioEffectsChain()
    chain.open()

    # Add effects in series: EQ -> Compressor -> Reverb -> Output
    eq_node = chain.add_effect_by_name("AUGraphicEQ")
    comp_node = chain.add_effect_by_name("AUDynamicsProcessor")
    reverb_node = chain.add_effect_by_name("AUMatrixReverb")
    output_node = chain.add_output()

    # Connect: EQ -> Compressor -> Reverb -> Output
    chain.connect(eq_node, comp_node)
    chain.connect(comp_node, reverb_node)
    chain.connect(reverb_node, output_node)

    print("Created effects chain:")
    print("  Input -> EQ -> Compressor -> Reverb -> Output")

    return chain


chain = create_multi_effect_chain()
chain.dispose()

Using Effect Descriptors

from coremusic.audio import create_simple_effect_chain


def create_chain_from_descriptors():
    """Create chain using explicit descriptors."""
    # Effect descriptors: (type, subtype, manufacturer)
    effects = [
        ("aufx", "dely", "appl"),  # Apple Delay
        ("aufx", "mrev", "appl"),  # Apple Matrix Reverb
    ]

    chain = create_simple_effect_chain(effects)

    print(f"Created chain with {chain.node_count} nodes")
    return chain


chain = create_chain_from_descriptors()
chain.dispose()

AudioEffectsChain builds an AUGraph, which runs live and feeds the output device. When you want to push your own blocks through the same effects and get the processed audio back, use AudioUnitChain from the plugin host:

from coremusic.audio.audiounit_host import AudioUnitChain

# AudioEffectsChain builds an AUGraph, which routes live audio to the output
# device. To push blocks of your own through the same effects and get the
# result back, use AudioUnitChain.
block = bytes(512 * 2 * 4)  # 512 stereo frames of float32

with AudioUnitChain() as chain:
    chain.add_plugin("AUDelay")
    chain.add_plugin("AUMatrixReverb")
    processed = chain.process(block, wet_dry_mix=0.8)

print(f"{len(processed)} bytes out")

Processing Audio Files

Using the CLI

# Apply effect to audio file
coremusic plugin process AUDelay input.wav -o output.wav

# Use a preset
coremusic plugin process AUDelay input.wav -o output.wav --preset "Long Delay"

# List available presets
coremusic plugin preset list AUDelay

Programmatic Processing

import numpy as np

from coremusic import capi
from coremusic.audio import AudioFile, AudioFormat, ExtendedAudioFile
from coremusic.audio.audiounit_host import AudioUnitPlugin


def process_audio_with_effect(input_path, output_path, effect_name):
    """Process an audio file through an effect, block by block."""
    with AudioFile(input_path) as audio:
        samples = audio.read_as_numpy().astype(np.float32) / 32768.0
        channels = audio.format.channels_per_frame
        sample_rate = audio.format.sample_rate

    block_frames = 512
    processed_blocks = []

    with AudioUnitPlugin.from_name(effect_name, component_type="aufx") as plugin:
        for start in range(0, len(samples), block_frames):
            block = samples[start:start + block_frames]
            if len(block) < block_frames:
                # The plugin wants full blocks; pad the tail
                block = np.pad(block, ((0, block_frames - len(block)), (0, 0)))

            out = plugin.process(block.tobytes(), num_frames=block_frames)
            processed_blocks.append(np.frombuffer(out, dtype=np.float32))

    result = np.concatenate(processed_blocks)

    out_format = AudioFormat.pcm(
        sample_rate, channels=channels, bits=32, is_float=True
    )
    with ExtendedAudioFile.create(
        output_path, capi.fourchar_to_int('WAVE'), out_format
    ) as output:
        output.write(len(result) // channels, result.tobytes())

    print(f"Processed {input_path} -> {output_path}")


process_audio_with_effect("input.wav", "processed.wav", "AUDelay")

Configuring Effect Parameters

Listing Parameters

from coremusic.audio.audiounit_host import AudioUnitPlugin


def list_effect_parameters(effect_name):
    """List all parameters of an effect."""
    with AudioUnitPlugin.from_name(effect_name, component_type="aufx") as plugin:
        print(f"Parameters for {effect_name}:")
        print("-" * 50)

        for param in plugin.parameters:
            print(f"  {param.name}")
            print(f"    ID: {param.id}")
            print(f"    Range: {param.min_value} - {param.max_value}")
            print(f"    Default: {param.default_value}")
            print(f"    Value: {param.value}")
            print()


list_effect_parameters("AUDelay")

Setting Parameters

from coremusic.audio.audiounit_host import AudioUnitPlugin


def configure_delay_effect():
    """Configure delay effect parameters."""
    with AudioUnitPlugin.from_name("AUDelay", component_type="aufx") as plugin:
        # Parameters are addressed by name or by id
        plugin.set_parameter("Delay Time", 0.25)     # seconds
        plugin.set_parameter("Feedback", 50.0)       # percent
        plugin['Dry/Wet Mix'] = 30.0                 # percent

        print("Delay configured:")
        print(f"  Delay Time: {plugin.get_parameter('Delay Time').value}s")
        print(f"  Feedback: {plugin.get_parameter('Feedback').value}%")
        print(f"  Mix: {plugin.get_parameter('Dry/Wet Mix').value}%")


configure_delay_effect()

Using Presets

from coremusic.audio.audiounit_host import AudioUnitPlugin


def use_effect_preset(effect_name, preset_name):
    """Apply a factory preset to an effect."""
    with AudioUnitPlugin.from_name(effect_name, component_type="aufx") as plugin:
        presets = plugin.factory_presets

        print(f"Available presets for {effect_name}:")
        for i, preset in enumerate(presets):
            print(f"  [{i}] {preset.name}")

        for preset in presets:
            if preset_name.lower() in preset.name.lower():
                plugin.load_factory_preset(preset)
                print(f"\nApplied preset: {preset.name}")
                return

        print(f"\nPreset not found: {preset_name}")


use_effect_preset("AUMatrixReverb", "Large Hall")

Real-Time Effects Processing

import time

from coremusic.audio import AudioEffectsChain


class RealTimeEffectsProcessor:
    """Process audio in real-time with effects."""

    def __init__(self):
        self.chain = None
        self.running = False

    def setup(self, effect_names):
        """Set up effects chain."""
        self.chain = AudioEffectsChain()
        self.chain.open()

        # Add effects
        prev_node = None
        for name in effect_names:
            node = self.chain.add_effect_by_name(name)
            if node is None:
                print(f"Warning: Effect not found: {name}")
                continue

            if prev_node is not None:
                self.chain.connect(prev_node, node)
            prev_node = node

        # Add output
        output_node = self.chain.add_output()
        if prev_node:
            self.chain.connect(prev_node, output_node)

        # Initialize
        self.chain.initialize()

        print(f"Effects chain ready with {self.chain.node_count} nodes")

    def start(self):
        """Start real-time processing."""
        if self.chain:
            self.chain.start()
            self.running = True
            print("Effects processing started")

    def stop(self):
        """Stop processing."""
        if self.chain:
            self.chain.stop()
            self.running = False
            print("Effects processing stopped")

    def cleanup(self):
        """Clean up resources."""
        if self.chain:
            self.chain.dispose()
            self.chain = None


# Use the processor
processor = RealTimeEffectsProcessor()
processor.setup(["AUDelay", "AUMatrixReverb"])
processor.start()

# Let it run for a while
time.sleep(1)

processor.stop()
processor.cleanup()

Common Effect Configurations

Reverb

from coremusic.audio.audiounit_host import AudioUnitPlugin

ROOMS = {
    "small": "Small Room",
    "medium": "Medium Room",
    "large": "Large Room",
    "hall": "Large Hall",
}


def make_reverb(room_size="medium"):
    """Return an initialized reverb set to a room preset."""
    plugin = AudioUnitPlugin.from_name("AUMatrixReverb", component_type="aufx")
    plugin.instantiate().initialize()

    wanted = ROOMS.get(room_size, "Medium Room")
    for preset in plugin.factory_presets:
        if preset.name == wanted:
            plugin.load_factory_preset(preset)
            print(f"Reverb configured: {preset.name}")
            break

    return plugin


reverb = make_reverb("large")
reverb.dispose()

Delay

from coremusic.audio.audiounit_host import AudioUnitPlugin

NOTE_VALUES = {
    "1/1": 4.0,
    "1/2": 2.0,
    "1/4": 1.0,
    "1/8": 0.5,
    "1/16": 0.25,
    "1/8T": 1.0 / 3.0,  # Triplet
    "1/8D": 0.75,       # Dotted
}


def make_delay(tempo_bpm=120, note_value="1/4"):
    """Return a delay whose time matches a note value at a tempo."""
    beat_duration = 60.0 / tempo_bpm
    delay_time = beat_duration * NOTE_VALUES.get(note_value, 1.0)

    plugin = AudioUnitPlugin.from_name("AUDelay", component_type="aufx")
    plugin.instantiate().initialize()
    plugin.set_parameter("Delay Time", delay_time)

    print(f"Delay configured for {tempo_bpm} BPM:")
    print(f"  Note value: {note_value}")
    print(f"  Delay time: {delay_time:.3f}s")

    return plugin


delay = make_delay(tempo_bpm=120, note_value="1/8")
delay.dispose()

EQ

from coremusic.audio.audiounit_host import AudioUnitPlugin


def make_eq(gains_db):
    """Return an N-band EQ with the given band gains applied."""
    plugin = AudioUnitPlugin.from_name("AUNBandEQ", component_type="aufx")
    plugin.instantiate().initialize()

    # Band gains are the parameters whose names end in "gain"
    gain_params = [p for p in plugin.parameters if p.name.lower().endswith("gain")]
    # Fewer gains than bands is fine; the rest keep their current value
    for param, gain in zip(gain_params, gains_db, strict=False):
        plugin.set_parameter(param.id, gain)

    print(f"EQ configured across {len(gain_params)} bands")
    return plugin


eq = make_eq([-2, 0, 3, 2, -1])
eq.dispose()

Complete Example: Audio Processor

import sys
from pathlib import Path

import numpy as np

from coremusic import capi
from coremusic.audio import AudioFile, AudioFormat, ExtendedAudioFile
from coremusic.audio.audiounit_host import AudioUnitChain


class AudioProcessor:
    """Process audio files through a chain of effects."""

    BLOCK_FRAMES = 512

    def __init__(self):
        self.chain = None

    def setup_chain(self, effects):
        """Set up effects chain."""
        self.chain = AudioUnitChain()
        for effect in effects:
            self.chain.add_plugin(effect)

    def process_file(self, input_path, output_path):
        """Process audio file."""
        if not self.chain:
            raise RuntimeError("Chain not set up")

        print(f"Processing: {input_path}")
        print(f"Output: {output_path}")

        with AudioFile(input_path) as audio:
            samples = audio.read_as_numpy().astype(np.float32) / 32768.0
            channels = audio.format.channels_per_frame
            sample_rate = audio.format.sample_rate
            print(f"Duration: {audio.duration:.2f}s")

        blocks = []
        for start in range(0, len(samples), self.BLOCK_FRAMES):
            block = samples[start:start + self.BLOCK_FRAMES]
            if len(block) < self.BLOCK_FRAMES:
                block = np.pad(block, ((0, self.BLOCK_FRAMES - len(block)), (0, 0)))
            out = self.chain.process(block.tobytes(), num_frames=self.BLOCK_FRAMES)
            blocks.append(np.frombuffer(out, dtype=np.float32))

        result = np.concatenate(blocks)

        out_format = AudioFormat.pcm(
            sample_rate, channels=channels, bits=32, is_float=True
        )
        with ExtendedAudioFile.create(
            output_path, capi.fourchar_to_int('WAVE'), out_format
        ) as output:
            output.write(len(result) // channels, result.tobytes())

        print("Processing complete!")

    def cleanup(self):
        """Clean up resources."""
        if self.chain:
            self.chain.dispose()
            self.chain = None


def main():
    if len(sys.argv) < 3:
        print("Usage: python audio_processor.py <input.wav> <output.wav> [effects...]")
        print("Example: python audio_processor.py in.wav out.wav AUDelay AUMatrixReverb")
        sys.exit(1)

    input_file = sys.argv[1]
    output_file = sys.argv[2]
    effects = sys.argv[3:] or ["AUMatrixReverb"]

    if not Path(input_file).exists():
        print(f"Error: Input file not found: {input_file}")
        sys.exit(1)

    processor = AudioProcessor()

    try:
        print(f"Setting up effects: {', '.join(effects)}")
        processor.setup_chain(effects)
        processor.process_file(input_file, output_file)
    finally:
        processor.cleanup()


if __name__ == "__main__":
    main()

Next Steps

See Also