macloop

🎙️ Build programmable macOS app audio capture and speech-to-text pipelines in Python without routing your whole machine through a virtual driver.

macloop is a Python-first macOS audio capture toolkit backed by a real-time Rust engine. It lets you capture microphones, system audio, or individual applications, route one stream into multiple consumers, apply processors, and feed the results into sinks such as ASR (speech-to-text), transcription, and WAV recording from one in-process API.

If you are looking for a BlackHole alternative for programmable Python workflows, macloop focuses on in-process app audio capture and recording/transcription pipelines instead of system-wide virtual-device routing.

Why it is easy to adopt

• Supports Python 3.9+
• Install with a standard command: pip install macloop
• No extra user-facing setup beyond normal installation and macOS permissions

✨ Why macloop?

Virtual devices such as BlackHole are useful when you need a system-wide virtual audio device. macloop targets a different workflow: programmable capture pipelines inside an application.

Typical use cases:

• Capture audio from a single macOS application such as Zoom
• Stream audio into ASR or speech-to-text systems in Python
• Record and transcribe meetings from one pipeline
• Replace manual BlackHole-style routing with an in-process Python API

macloop vs virtual-driver workflows

Why this matters: if your goal is “capture, transform, split, and consume audio in Python”, macloop removes a lot of the manual patch-bay work.

🧱 Tech Stack

🧩 What You Can Build

macloop is designed as a modular pipeline:

Source -> Processor(s) -> Route(s) -> Sink(s)

Examples:

• Record a meeting to WAV while streaming microphone chunks to an ASR engine.
• Capture only Zoom audio instead of the entire system mix.
• Split one microphone stream into separate routes for transcription, monitoring, and archival recording.
• Build deterministic tests with a synthetic source before touching real devices.

Current building blocks

🚀 Installation

1. Create a virtual environment

python -m venv .venv
source .venv/bin/activate

2. Upgrade packaging tools

python -m pip install --upgrade pip

3. Install macloop

pip install macloop

Requirements

• macOS
• Python 3.9+

That is the full user-facing requirement for installation: create a normal Python environment and run pip install macloop.

▶️ Quick Start

The example below creates a small audio graph:

• capture the microphone
• apply a gain processor
• split the stream into two routes
• record one route to WAV
• send the other route to an ASR sink

import macloop


with macloop.AudioEngine() as engine:
    mic = engine.create_stream(
        macloop.MicrophoneSource,
        device_id=None,
        vpio_enabled=True,
    )

    engine.add_processor(
        stream=mic,
        processor=macloop.GainProcessor(gain=1.2),
    )

    mic_for_asr = engine.route("mic_for_asr", stream=mic)
    mic_for_wav = engine.route("mic_for_wav", stream=mic)

    wav_sink = macloop.WavSink(route=mic_for_wav, file="out/mic.wav")
    asr_sink = macloop.AsrSink(
        routes=[mic_for_asr],
        chunk_frames=320,
        sample_rate=16_000,
        channels=1,
        sample_format="f32",
    )

    for chunk in asr_sink.chunks():
        print(chunk.route_id, chunk.frames, chunk.samples.dtype)
        break

    asr_sink.close()
    wav_sink.close()

AudioChunk.samples is a NumPy array:

• np.float32 for sample_format="f32"
• np.int16 for sample_format="i16"

🎧 Real Example: Record And Transcribe A Meeting

This is the workflow macloop is built for: one pipeline, multiple outputs.

import macloop


def find_zoom_pids() -> list[int]:
    pids = []
    for app in macloop.AppAudioSource.list_applications():
        if "zoom" in app["name"].lower():
            pids.append(int(app["pid"]))
    if not pids:
        raise RuntimeError("Zoom is not running")
    return pids


with macloop.AudioEngine() as engine:
    mic = engine.create_stream(macloop.MicrophoneSource, vpio_enabled=True)
    zoom = engine.create_stream(macloop.AppAudioSource, pids=find_zoom_pids())

    mic_for_asr = engine.route("mic_for_asr", stream=mic)
    zoom_for_asr = engine.route("zoom_for_asr", stream=zoom)
    mic_for_wav = engine.route("mic_for_wav", stream=mic)
    zoom_for_wav = engine.route("zoom_for_wav", stream=zoom)

    wav_sink = macloop.WavSink(
        routes=[mic_for_wav, zoom_for_wav],
        file="out/meeting.wav",
    )

    asr_sink = macloop.AsrSink(
        routes=[mic_for_asr, zoom_for_asr],
        chunk_frames=320,
        sample_rate=16_000,
        channels=1,
        sample_format="f32",
    )

    # Long-running pipeline: keep consuming until your app decides to stop.
    for chunk in asr_sink.chunks():
        print(chunk.route_id, chunk.frames)
        # Send chunk.samples into your ASR engine here.

Notes:

• AsrSink emits independent chunks per route.
• WavSink can mix several routes into one file.
• If mix_gain is not provided, WavSink uses 1 / N by default.

⚡ Asyncio

AsrSink also supports async consumption:

import asyncio
import macloop


async def main() -> None:
    with macloop.AudioEngine() as engine:
        mic = engine.create_stream(macloop.MicrophoneSource, vpio_enabled=True)
        mic_for_asr = engine.route(stream=mic)

        with macloop.AsrSink(
            routes=[mic_for_asr],
            chunk_frames=320,
            sample_rate=16_000,
            channels=1,
            sample_format="f32",
        ) as asr_sink:
            async for chunk in asr_sink.chunks_async():
                print(chunk.route_id, chunk.frames)
                break


asyncio.run(main())

🔎 Device Discovery

Microphones

import macloop

for mic in macloop.MicrophoneSource.list_devices():
    print(mic["id"], mic["name"], mic["is_default"])

Displays

import macloop

for display in macloop.SystemAudioSource.list_displays():
    print(display["id"], display["name"], display["width"], display["height"])

Applications

import macloop

for app in macloop.AppAudioSource.list_applications():
    print(app["pid"], app["name"], app["bundle_id"])

If engine.create_stream(macloop.SystemAudioSource, ...) is called without an explicit display_id, macloop uses the first available display.

engine.create_stream(macloop.AppAudioSource, ...) requires explicit pids. Use AppAudioSource.list_applications() to choose one or more target applications first.

🛠️ Example Scripts

The scripts below live in this repository, so run them from a source checkout.

Record microphone audio to WAV

python examples/write_to_wav.py --seconds 5 --output out/mic.wav

Stream microphone audio into Sherpa ONNX

uv run --with sherpa-onnx --with huggingface_hub --reinstall-package macloop \
  python examples/sherpa_asr_demo.py --seconds 5

📊 Telemetry

macloop exposes metrics at different levels of the pipeline:

• engine.stats() for per-stream real-time pipeline and processor metrics
• asr_sink.stats() for per-route ASR sink metrics
• wav_sink.stats() for WAV writer metrics

This makes it possible to inspect latency and drops at the node level instead of relying only on a single average number.

🗺️ Roadmap

• Add more built-in processors beyond GainProcessor
• Add zero-copy / lease-release delivery for Python
• Add richer pipeline examples for meeting bots and voice agents
• Add WebRTC AEC in a future iteration, with a routing model that can handle capture and reference streams cleanly

📄 License

MIT