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

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:

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:

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:

"""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:

"""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,
            )

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 (uv add faster-whisper) and transcribes five seconds at a time:

"""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 and WebRTC.

The microphone-reverb-speaker and audio-mixer-demo examples go further: a full reverb, and a mixer combining three sources at different rates.