---
title: "Audio"
description: "Capture from a microphone, process audio in Python in exactly the format you ask for, play it back, and line it up with video."
type: "docs-page"
url: "https://tatolab.com/docs/build/audio"
site: "Tatolab"
section: "Build"
---

# Audio

> Capture from a microphone, process audio in Python in exactly the format you ask for, play it back, and line it up with video.

Audio in Tatolab works like video: built-in nodes handle the devices, your nodes do the work in between, and everything shares one clock. The difference is that audio samples travel inside the bag, so a Python node gets them as a numpy array directly.

## The built-in audio nodes

| Node | Does | Ports |
|---|---|---|
| `MicrophoneSource` | Captures from an input device | out: `audio` |
| `SpeakerSink` | Plays to an output device, converting the format to suit it | in: `audio` (ordered) |
| `OpusEncoder` | Compresses audio with Opus, for recording or WebRTC | in: `audio`, out: `encoded_audio` |
| `OpusDecoder` | Decompresses Opus back into audio | in: `encoded_audio`, out: `audio` |

Both device nodes use the default device unless you set `device_id`, and follow the system default as it changes. On a machine with no audio devices at all, such as a server or a container, `MicrophoneSource` produces silence and `SpeakerSink` discards what it's given, so the same stream still runs.

## Hear yourself

```python title="stream.py"
from tatolab.stream import MicrophoneSource, SpeakerSink, StreamBuilder, stream


@stream
def monitor(stream_builder: StreamBuilder) -> None:
    """The microphone, played back on the speakers."""
    microphone = stream_builder.add(MicrophoneSource)
    speaker = stream_builder.add(SpeakerSink)
    stream_builder.connect(microphone.output("audio"), speaker.input("audio"))
```

Wear headphones: with speakers, the microphone hears the speakers and the sound feeds back. On a Mac, the first run asks for microphone access; audio starts once you allow it.

## Audio bags

An audio bag is a block of samples. Read it with `into=AudioBlock`:

```python
block = ctx.inputs.read("audio", into=AudioBlock)
```

| Field | Meaning |
|---|---|
| `block.samples` | A numpy array, `sample_count` × `channels`, read-only |
| `block.sample_rate` | Samples per second, per channel |
| `block.channels` | Number of channels |
| `block.sample_count` | Samples per channel in this block |
| `block.dtype` | `"f32"` (from -1.0 to 1.0) or `"i16"` |
| `block.first_sample_timestamp_ns` | When the first sample was captured, on the machine's monotonic clock |

`block.samples` is a view of the bag's own bytes, so reading it costs no copy. Use `numpy.array(block.samples)` when you need a copy you can change.

## Ask for the format you want

Microphones open at whatever rate and channel count the device likes: 44.1 kHz stereo on one machine, 48 kHz mono on the next. Instead of handling every case, an audio input can declare a **window contract**, and the runtime converts to it:

```python
from tatolab.stream import AudioWindowContract

SPEECH_WINDOW = AudioWindowContract(sample_rate=16_000, channels=1, dtype="f32", window_size=16_000)
```

With that on an input, every block the node reads is exactly one second of 16 kHz mono float audio, whatever the microphone opened at. The runtime resamples, converts the sample type, mixes channels, and cuts the audio into windows.

- `window_size` counts samples per channel. 480 samples at 48 kHz is 10 ms.
- `hop` sets how far each window moves on. Leave it out for back-to-back windows; set it lower than `window_size` for overlapping ones.
- `channels` is optional. Leave it out and the blocks keep the source's own channel count; set it when your code needs a particular count, like a model trained on mono.
- A window contract needs an `ordered` input: skipping blocks would leave holes in the audio.

## Example: a level meter

This node measures loudness in dBFS ten times a second and publishes it:

```python title="nodes/level_meter.py"
"""Measures loudness ten times a second."""

import math

import numpy
from tatolab.stream import AudioBlock, AudioWindowContract, RuntimeContextLimitedAccess, node

# 100 ms of 48 kHz mono audio per block.
LEVEL_WINDOW = AudioWindowContract(sample_rate=48_000, channels=1, dtype="f32", window_size=4_800)


@node
class LevelMeter:
    """Publishes the audio's loudness in dBFS, ten times a second."""

    @node.input(delivery_profile="ordered", audio_window=LEVEL_WINDOW)
    def audio(self) -> AudioBlock: ...

    @node.output()
    def level(self) -> dict: ...

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (block := ctx.inputs.read("audio", into=AudioBlock)) is not None:
            samples = block.samples[:, 0].astype(numpy.float64)
            rms = math.sqrt(float(numpy.mean(samples * samples)))
            dbfs = 20 * math.log10(max(rms, 1e-9))
            ctx.outputs.write(
                "level", {"dbfs": round(dbfs, 1), "timestamp_ns": block.first_sample_timestamp_ns}
            )
```

## Example: an echo

This node adds an echo and sends the result on, ready for a `SpeakerSink`:

```python title="nodes/echo.py"
"""Adds an echo to the audio passing through."""

from dataclasses import dataclass

import numpy
from tatolab.stream import AudioBlock, AudioWindowContract, RuntimeContextLimitedAccess, node

# 10 ms of 48 kHz mono audio per block.
ECHO_WINDOW = AudioWindowContract(sample_rate=48_000, channels=1, dtype="f32", window_size=480)


@dataclass
class EchoConfig:
    delay_ms: int = 250
    feedback: float = 0.4
    mix: float = 0.5


@node
class Echo:
    """Adds an echo: the sound again after a delay, fading each time it repeats."""

    @node.input(delivery_profile="ordered", audio_window=ECHO_WINDOW)
    def dry_audio(self) -> AudioBlock: ...

    @node.output()
    def echoed_audio(self) -> AudioBlock: ...

    def __init__(self, config: EchoConfig) -> None:
        # The delay must be at least one window long.
        delay_samples = max(config.delay_ms * ECHO_WINDOW.sample_rate // 1000, ECHO_WINDOW.window_size)
        self.delay_line = numpy.zeros(delay_samples, dtype=numpy.float32)
        self.position = 0
        self.feedback = config.feedback
        self.mix = config.mix

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (block := ctx.inputs.read("dry_audio", into=AudioBlock)) is not None:
            dry = block.samples[:, 0]
            positions = (self.position + numpy.arange(dry.size)) % self.delay_line.size
            delayed = self.delay_line[positions]
            self.delay_line[positions] = dry + delayed * self.feedback
            self.position = (self.position + dry.size) % self.delay_line.size

            wet = numpy.clip(dry + delayed * self.mix, -1.0, 1.0).astype("<f4")
            ctx.outputs.write(
                "echoed_audio",
                {
                    "samples": wet.tobytes(),
                    "sample_rate": block.sample_rate,
                    "channels": 1,
                    "sample_count": block.sample_count,
                    "dtype": "f32",
                    "first_sample_timestamp_ns": block.first_sample_timestamp_ns,
                },
                block.first_sample_timestamp_ns,
            )
```

```python title="stream.py"
from tatolab.stream import MicrophoneSource, SpeakerSink, StreamBuilder, stream

from nodes.echo import Echo


@stream
def echo(stream_builder: StreamBuilder) -> None:
    """The microphone, with an echo, on the speakers."""
    microphone = stream_builder.add(MicrophoneSource)
    effect = stream_builder.add(Echo, config={"delay_ms": 300})
    speaker = stream_builder.add(SpeakerSink)
    stream_builder.connect(microphone.output("audio"), effect.input("dry_audio"))
    stream_builder.connect(effect.output("echoed_audio"), speaker.input("audio"))
```

`SpeakerSink` converts whatever it receives to the output device's own rate and period, so the node doesn't need to know what the speakers want. The window size sets the effect's added latency: 480 samples is 10 ms.

## Writing audio

To send audio, write a bag with these keys, as the echo does:

| Key | Value |
|---|---|
| `samples` | The samples as bytes, interleaved, little-endian: `array.astype("<f4").tobytes()` for float |
| `sample_rate`, `channels`, `sample_count` | Integers describing the block |
| `dtype` | `"f32"` or `"i16"` |
| `first_sample_timestamp_ns` | When the first sample happened; carry it over from the input block |

Pass the same timestamp as `write`'s third argument, so readers that use timestamps see it too.

## Example: speech to text

A transcriber is a good fit for a window contract: speech models want 16 kHz mono, whatever the microphone produces. This node uses [faster-whisper](https://github.com/SYSTRAN/faster-whisper) (`uv add faster-whisper`) and transcribes five seconds at a time:

```python title="nodes/transcriber.py"
"""Transcribes speech, a few seconds at a time."""

from dataclasses import dataclass

import numpy
from faster_whisper import WhisperModel
from tatolab.stream import (
    AudioBlock,
    AudioWindowContract,
    RuntimeContextFullAccess,
    RuntimeContextLimitedAccess,
    log,
    node,
)

# One second of 16 kHz mono audio per block: the format Whisper expects.
SPEECH_WINDOW = AudioWindowContract(sample_rate=16_000, channels=1, dtype="f32", window_size=16_000)


@dataclass
class TranscriberConfig:
    model_size: str = "small"
    seconds_per_chunk: int = 5
    language: str = "en"


@node
class Transcriber:
    """Transcribes speech from its input and publishes the text."""

    @node.input(delivery_profile="ordered", audio_window=SPEECH_WINDOW)
    def speech(self) -> AudioBlock: ...

    @node.output()
    def transcript(self) -> dict: ...

    def __init__(self, config: TranscriberConfig) -> None:
        self.model_size = config.model_size
        self.seconds_per_chunk = config.seconds_per_chunk
        self.language = config.language
        self.pending: list[numpy.ndarray] = []
        self.chunk_started_at_ns = 0

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.model = WhisperModel(self.model_size, device="auto")

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (block := ctx.inputs.read("speech", into=AudioBlock)) is not None:
            if not self.pending:
                self.chunk_started_at_ns = block.first_sample_timestamp_ns
            self.pending.append(numpy.array(block.samples[:, 0]))
            if len(self.pending) < self.seconds_per_chunk:
                continue
            audio = numpy.concatenate(self.pending)
            self.pending = []
            segments, _info = self.model.transcribe(audio, language=self.language)
            text = " ".join(segment.text.strip() for segment in segments)
            if text:
                log.info("heard", text=text)
                ctx.outputs.write("transcript", {"text": text, "timestamp_ns": self.chunk_started_at_ns})
```

Connect `MicrophoneSource`'s `audio` to its `speech` input. Each transcript carries the time its audio began, so you can match it to the video frames from the same moment.

Two things to plan for:

- **Download the model first.** faster-whisper downloads the model the first time it's used, and a node gets 60 seconds to import and run `setup`. Fetch it once beforehand: `uv run python -c "from faster_whisper import WhisperModel; WhisperModel('small')"`.
- **`transcribe` blocks `process`.** While it runs, only about a second of microphone audio can wait at the input, so a transcription that takes longer than that drops audio (counted in `tatolab graph`). Run the model on a GPU, choose a smaller model, or use shorter chunks.

## Choosing devices

Set `device_id` on `MicrophoneSource` or `SpeakerSink` to pick a device other than the default:

- **Linux:** the audio system's device name. On PipeWire, `<sink name>.monitor` captures whatever that output is playing, which is how a stream records its own sound.
- **macOS:** the device's Core Audio UID. Naming a device that isn't connected is refused, and the message lists the UIDs that are.

A `device_id` that can't be opened is an error, never a silent switch to another device.

## Audio with video

Every timestamp on the machine comes from the same monotonic clock: a frame's `timestamp_ns`, an audio block's `first_sample_timestamp_ns`, and `ctx.time` in any node. Lining up a sound with a frame is a subtraction. To record or publish them together, encode each and connect both to one `Mp4Sink` or `WhipPublisher`; see [Record and encode](https://tatolab.com/docs/build/recording) and [WebRTC](https://tatolab.com/docs/build/webrtc).

The [microphone-reverb-speaker](https://github.com/tatolab/streamlib/tree/main/examples/microphone-reverb-speaker) and [audio-mixer-demo](https://github.com/tatolab/streamlib/tree/main/examples/audio-mixer-demo) examples go further: a full reverb, and a mixer combining three sources at different rates.