Skip to content

Real-Time Audio Recipes

Practical recipes for real-time audio processing, built on coremusic.audio.streaming: AudioInputStream pulls blocks from the input device, AudioOutputStream pushes blocks to the output device, and AudioProcessor wires the two together through a function of yours.

Every callback below runs on the audio thread. Allocate nothing, block on nothing, and return promptly - anything slower becomes an audible dropout, counted in overruns and underruns.

Record Audio from Input Device

Capture from the default input device, one block at a time.

import time

from coremusic.audio.streaming import AudioInputStream

captured = []


def collect(audio_data, frame_count):
    """Runs on the audio thread for every captured block - keep it short."""
    captured.append(audio_data)


stream = AudioInputStream(channels=2, sample_rate=44100.0, buffer_size=512)
stream.add_callback(collect)

try:
    stream.start()
except RuntimeError as e:
    # macOS refuses input until the app has microphone permission
    print(e)
    raise SystemExit(0) from None

time.sleep(0.5)
stream.stop()

print(f"Captured {len(captured)} blocks, {stream.overruns} overruns")

Play Audio with Low Latency

Generate samples on demand. The buffer size sets the latency.

import math
import struct
import time

from coremusic.audio.streaming import AudioOutputStream


def make_sine(freq, sample_rate, channels, gain=0.2):
    """Return generator(frame_count) -> interleaved float32 bytes."""
    phase = 0.0
    step = 2.0 * math.pi * freq / sample_rate

    def generate(frame_count):
        nonlocal phase
        samples = []
        for _ in range(frame_count):
            samples.extend([math.sin(phase) * gain] * channels)
            phase = (phase + step) % (2.0 * math.pi)
        return struct.pack(f"<{len(samples)}f", *samples)

    return generate


# A small buffer means low latency; too small and the generator cannot keep up
stream = AudioOutputStream(channels=2, sample_rate=44100.0, buffer_size=256)
stream.set_generator(make_sine(440.0, 44100.0, channels=2))
stream.start()

print(f"Latency: {stream.latency * 1000:.1f}ms")
time.sleep(0.5)

stream.stop()

Monitor Audio Levels

Track peak and RMS of the incoming signal.

import time

import numpy as np

from coremusic.audio.streaming import AudioInputStream


class AudioLevelMonitor:
    """Real-time audio level monitoring"""

    def __init__(self, sample_rate=44100.0):
        self.stream = AudioInputStream(channels=2, sample_rate=sample_rate)
        self.peak_level = 0.0
        self.rms_level = 0.0
        self.stream.add_callback(self._measure)

    def _measure(self, audio_data, frame_count):
        if frame_count == 0:
            return
        samples = np.asarray(audio_data, dtype=np.float32)
        self.peak_level = float(np.max(np.abs(samples)))
        self.rms_level = float(np.sqrt(np.mean(samples ** 2)))

    def start(self):
        self.stream.start()

    def stop(self):
        self.stream.stop()

    def get_levels(self):
        """Get current levels in dB"""
        peak_db = 20 * np.log10(self.peak_level) if self.peak_level > 0 else -100
        rms_db = 20 * np.log10(self.rms_level) if self.rms_level > 0 else -100
        return {"peak_db": peak_db, "rms_db": rms_db}


monitor = AudioLevelMonitor()
try:
    monitor.start()
except RuntimeError as e:
    print(e)
    raise SystemExit(0) from None

for _ in range(5):
    levels = monitor.get_levels()
    print(f"Peak: {levels['peak_db']:.1f} dB, RMS: {levels['rms_db']:.1f} dB")
    time.sleep(0.1)

monitor.stop()

Process Input to Output

Read the input, process it, and play the result.

import time

from coremusic.audio.streaming import AudioProcessor


def gain(audio_data):
    """Called for every block: return the processed block."""
    return audio_data * 0.5


processor = AudioProcessor(gain, channels=2, sample_rate=44100.0, buffer_size=512)

try:
    processor.start()
except RuntimeError as e:
    print(e)
    raise SystemExit(0) from None

print(f"Round-trip latency: {processor.latency * 1000:.1f}ms")
time.sleep(0.5)
processor.stop()

print(f"Overruns: {processor.overruns}, underruns: {processor.underruns}")

See Also