Record and encode

Compress video and audio with hardware encoders, record any number of tracks into one MP4, and work with the encoded stream in Python.

Recording takes two kinds of built-in nodes: an encoder that compresses frames or audio, and Mp4Sink, which writes what the encoders produce into an MP4 file. Video encoding uses the GPU’s hardware encoder, so recording costs Python nothing per frame.

Record the camera

from pathlib import Path

from tatolab.stream import CameraSource, H264Encoder, Mp4Sink, StreamBuilder, stream

# Built-in nodes run inside the runtime, not in the project folder,
# so give the recorder a full path.
RECORDING = Path(__file__).resolve().parent / "camera.mp4"


@stream
def camera_recorder(stream_builder: StreamBuilder) -> None:
    """Records the camera to camera.mp4."""
    camera = stream_builder.add(CameraSource)
    encoder = stream_builder.add(H264Encoder)
    recorder = stream_builder.add(Mp4Sink, config={"path": str(RECORDING)})
    stream_builder.connect(camera.output("video"), encoder.input("video"))
    stream_builder.connect(encoder.output("encoded_video"), recorder.input("tracks"))

Run it with tatolab run, stop it with Ctrl-C, and open camera.mp4 in a video player. mpv and VLC play it as is; Ubuntu’s default player may ask for H.264 codecs first.

One file, many tracks

Every link into Mp4Sink’s tracks input becomes one track in the file. The sink works out each track’s kind from what arrives: H.264 or H.265 makes a video track, Opus an audio track.

Camera and microphone together:

from pathlib import Path

from tatolab.stream import (
    CameraSource,
    H264Encoder,
    MicrophoneSource,
    Mp4Sink,
    OpusEncoder,
    StreamBuilder,
    stream,
)

RECORDING = Path(__file__).resolve().parent / "session.mp4"


@stream
def session_recorder(stream_builder: StreamBuilder) -> None:
    """Records the camera and the microphone into one file."""
    camera = stream_builder.add(CameraSource)
    video_encoder = stream_builder.add(H264Encoder, config={"bitrate_bps": 8_000_000})
    microphone = stream_builder.add(MicrophoneSource)
    audio_encoder = stream_builder.add(OpusEncoder)
    recorder = stream_builder.add(Mp4Sink, config={"path": str(RECORDING)})

    stream_builder.connect(camera.output("video"), video_encoder.input("video"))
    stream_builder.connect(video_encoder.output("encoded_video"), recorder.input("tracks"))
    stream_builder.connect(microphone.output("audio"), audio_encoder.input("audio"))
    stream_builder.connect(audio_encoder.output("encoded_audio"), recorder.input("tracks"))

OpusEncoder takes the microphone at whatever rate and channel count it opened with; the runtime converts it to what Opus needs. To record two cameras, add two CameraSource nodes with different device_ids, give each its own encoder, and connect both encoders to the same tracks input. You get one file with two video tracks.

Each track is named after the link it came from, <producer node ID>/<port>, as tatolab graph shows it.

What the file looks like

It’s a fragmented MP4. After a short header, the file grows in fragments. A fragment closes at each video keyframe (or once a second when there’s no video), so a recording plays up to its last closed fragment even if the process dies. A clean stop also writes out the open fragment. The file isn’t synced to disk as it’s written, so a power cut can lose more.
The header waits for every track. The sink writes the file’s header once every track has delivered its first keyframe or audio packet. If one track never produces anything (a dead microphone, say), the log names it once a second. Once the other tracks have queued as much as the sink holds, it gives up on the silent one, by name, and records the rest.
Each load starts the file over. The file is created, or emptied, each time the stream loads. See Recordings and restarts.
One problem track doesn’t stop the others. If a track changes in a way MP4 can’t represent mid-file, such as a new resolution’s parameter sets, that track stops with a message naming it, and the rest keep recording.

Recordings and restarts

Mp4Sink creates its file, or empties it, every time its stream loads. For an attached stream (tatolab run), that’s each time you start it, so a path with a timestamp in it gives every run its own file:

from datetime import datetime

RECORDING = Path(__file__).resolve().parent / f"camera-{datetime.now():%Y%m%d-%H%M%S}.mp4"

A kept stream is different. The runtime records the graph once, at tatolab run -d, path included, and loads that same graph after every restart, reboot and crash. Each reload empties the file, so a kept recorder overwrites its last recording every time the machine restarts. To keep what you record:

Record from a separate, attached stream. Keep the stream that opens the camera, expose its output, and record from a second stream that reads it with remote_output, loaded with tatolab run for as long as you want to record. See Share between streams.
Or copy recordings off before anything restarts the runtime, if the recorder has to be kept.

Encoder settings

H264Encoder and H265Encoder take the same settings, all optional:

Setting	Default	Meaning
`bitrate_bps`	a fixed quantizer, with no target bitrate	Target bitrate, in bits per second
`keyframe_interval_seconds`	`2`	Seconds between keyframes. Shorter makes seeking finer and files bigger.
`fps`	`60`	Frame rate to assume when frames don’t say
`width`, `height`	from the first frame	Expected size. A guard, not a resize: a different size logs a warning and is encoded as it is.
`effort_level`	the codec’s default	How hard the hardware encoder works. Linux only; refused on macOS.

The encoder starts from the first frame’s size and starts again if the size changes. OpusEncoder takes bitrate_bps and application: "audio" (the default, for fidelity), "voip" (speech), or "lowdelay".

What the hardware has to have

Encoding and decoding use the GPU’s video hardware: Vulkan Video on Linux, VideoToolbox on a Mac.

No hardware encoder: the encoder logs the encoder session could not be minted; every later frame is discarded once, keeps running, and produces nothing. The rest of the stream carries on, and the recording has no video track.
No hardware decoder: H264Decoder and H265Decoder fail when the stream starts, and tatolab graph shows them in an error state.

Some GPUs encode H.265 but not H.264, or the other way round, so if one fails, try the other.

Decode and display

Decoders turn an encoded stream back into frames, for a preview of what was recorded or for video that arrives already encoded:

stream_builder.connect(encoder.output("encoded_video"), decoder.input("encoded_video"))
stream_builder.connect(decoder.output("video"), window.input("video"))

The decoder must match the encoder: H264Encoder with H264Decoder, H265Encoder with H265Decoder.

Encoded video in Python

A node can read encoded video too. Each bag is one access unit (one compressed frame), cast with EncodedVideoFrame. This node writes the raw H.264 stream to a file, which tools like ffmpeg and VLC can read:

"""Writes an H.264 stream to a raw .h264 file, starting at a keyframe."""

from dataclasses import dataclass

from tatolab.stream import (
    EncodedVideoFrame,
    RuntimeContextFullAccess,
    RuntimeContextLimitedAccess,
    log,
    node,
)


@dataclass
class H264FileWriterConfig:
    path: str = "capture.h264"


@node
class H264FileWriter:
    """Appends H.264 access units to a file, beginning at a keyframe."""

    @node.input(delivery_profile="ordered")
    def encoded_video(self) -> EncodedVideoFrame: ...

    def __init__(self, config: H264FileWriterConfig) -> None:
        self.path = config.path
        self.writing = False
        self.last_sequence_index: int | None = None

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.file = open(self.path, "wb")

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (frame := ctx.inputs.read("encoded_video", into=EncodedVideoFrame)) is not None:
            # sequence_index goes up by exactly one per frame; a jump means frames
            # were lost, so stop and wait for the next keyframe.
            if self.last_sequence_index is not None and frame.sequence_index != self.last_sequence_index + 1:
                if self.writing:
                    log.warn("frames lost; waiting for the next keyframe")
                self.writing = False
            self.last_sequence_index = frame.sequence_index
            if not self.writing and not frame.is_sync_point:
                continue
            self.writing = True
            self.file.write(frame.annex_b_access_unit_bytes)

    def teardown(self, ctx: RuntimeContextFullAccess) -> None:
        self.file.close()

This one is a Python node, so it runs in the project folder and a relative path is fine. Two rules apply to every reader of an encoded stream: start at a keyframe (is_sync_point), and after a gap, throw frames away until the next keyframe. Writing on through a gap produces a stream that decodes to garbage.

Audio works the same way with EncodedAudioPacket, whose opus_packet_bytes holds one Opus packet.

camera-audio-recorder and camera-codec-roundtrip, in the repository.
WebRTC: send encoded video to a WHIP endpoint instead of a file.
Built-in nodes: every setting of every codec.