One camera, many destinations

You want your camera to look a certain way, then go to several places: a video call, OBS, a recording at better quality than the call. Usually that means a chain of apps, one applying the look, another capturing its window or a virtual camera, a third recording, and each one change away from breaking the others. And only one of them can open the camera.

With Tatolab, one stream owns the camera and the look, and offers the result to everything else: as a camera that other apps can pick (on Linux), and as an exposed output other streams can read. Recording is a separate stream you start when you want it.

my_camera:          CameraSource ─▶ Grade ─┬─▶ VirtualCameraSink ("My Camera") ─▶ Zoom, Meet, OBS
                                           ├─▶ DisplayWindow (preview)
                                           └─▶ graded_video (exposed)
session_recording:  graded_video ─▶ H264Encoder ─┐
                    MicrophoneSource ─▶ OpusEncoder ┴─▶ Mp4Sink ─▶ session-….mp4

The streams

stream.py

from datetime import datetime
from pathlib import Path

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

from nodes.grade import Grade

FOLDER = Path(__file__).resolve().parent


@stream
def my_camera(stream_builder: StreamBuilder) -> None:
    """My webcam with a warm grade: a camera for calls, and an output other streams can record."""
    camera = stream_builder.add(CameraSource, config={"max_width": 1280, "max_height": 720})
    grade = stream_builder.add(Grade, name="grade", config={"warmth": 0.08, "contrast": 1.05})
    call_camera = stream_builder.add(VirtualCameraSink, config={"name": "My Camera"})
    preview = stream_builder.add(DisplayWindow, config={"title": "Preview", "width": 640, "height": 360})

    stream_builder.connect(camera.output("video"), grade.input("video"))
    stream_builder.connect(grade.output("graded_video"), call_camera.input("video"))
    stream_builder.connect(grade.output("graded_video"), preview.input("video"))
    stream_builder.expose(grade.output("graded_video"))


@stream
def session_recording(stream_builder: StreamBuilder) -> None:
    """Records the graded camera and the microphone, to a new file each time it starts."""
    path = FOLDER / f"session-{datetime.now():%Y%m%d-%H%M%S}.mp4"
    video_encoder = stream_builder.add(H264Encoder, config={"bitrate_bps": 8_000_000})
    microphone = stream_builder.add(MicrophoneSource)
    audio_encoder = stream_builder.add(OpusEncoder)
    recording = stream_builder.add(Mp4Sink, config={"path": str(path)})

    stream_builder.connect(stream_builder.remote_output("my_camera/grade/graded_video"), video_encoder.input("video"))
    stream_builder.connect(video_encoder.output("encoded_video"), recording.input("tracks"))
    stream_builder.connect(microphone.output("audio"), audio_encoder.input("audio"))
    stream_builder.connect(audio_encoder.output("encoded_audio"), recording.input("tracks"))

The grade is a pixel effect: one GLSL function, with two values passed in from Python as dials:

nodes/grade.py

"""A simple color grade: warmth and contrast."""

from dataclasses import dataclass
from typing import Annotated

from tatolab.stream import (
    GlslPixelEffect,
    RuntimeContextFullAccess,
    RuntimeContextLimitedAccess,
    VideoFrame,
    node,
)

GRADE_GLSL = """
vec4 effect(vec4 source, ivec2 at) {
    // Contrast around middle gray, then warm the image by trading blue for red.
    vec3 c = (source.rgb - 0.5) * dials.contrast + 0.5;
    c += vec3(dials.warmth, dials.warmth * 0.4, -dials.warmth);
    return vec4(clamp(c, 0.0, 1.0), source.a);
}
"""


@dataclass
class GradeConfig:
    warmth: Annotated[float, "Shift toward orange (positive) or blue (negative); about -0.3 to 0.3."] = 0.0
    contrast: Annotated[float, "1.0 leaves contrast alone; above 1.0 adds it."] = 1.0


@node
class Grade:
    """Warms or cools the picture, and adjusts its contrast."""

    @node.input(delivery_profile="newest")
    def video(self) -> VideoFrame: ...

    @node.output()
    def graded_video(self) -> VideoFrame: ...

    def __init__(self, config: GradeConfig) -> None:
        self.dials = {"warmth": config.warmth, "contrast": config.contrast}

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.effect = GlslPixelEffect.compile(
            ctx.gpu_full_access, effect_glsl=GRADE_GLSL, dials={"warmth": "float", "contrast": "float"}
        )

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        frame = ctx.inputs.read("video", into=VideoFrame)
        if frame is None:
            return
        ctx.outputs.write("graded_video", self.effect.apply_to_frame(ctx.gpu_limited_access, frame, dials=self.dials))

Use it

1. Set up the virtual camera once, on Linux: Virtual camera explains the one command.
2. Tune the look with tatolab dev stream.py:my_camera, next to the preview window. Each save reloads the stream, and the camera briefly disappears from other apps while it does, so tune before a call, not during one.
3. Keep it running with tatolab run -d stream.py:my_camera, and choose My Camera in your video app’s camera settings. While it runs, the webcam is in use and its light stays on; tatolab stop my_camera frees it.
4. Record a session with tatolab run stream.py:session_recording, and Ctrl-C when you’re done. Each run writes a new file. Recording in its own stream means the camera stream can stay kept without a recorder that starts its file over after every restart.

The looks in GPU effects mix in the same way: sepia, pixelate, edge detection, a vignette.

Today’s limits

• Virtual cameras are Linux only. On a Mac, the processed picture can’t reach Zoom, Meet or OBS yet; you can still preview it and record it.
• USB webcam frame rates on Linux. CameraSource reads uncompressed video only, so most USB 2.0 webcams deliver HD at 5 to 10 frames per second; their faster modes are MJPEG, which isn’t read yet. The example asks for 720p; check yours with v4l2-ctl --list-formats-ext, lower the limits, or use a USB 3 camera or an HDMI capture device.
• No RTMP, SRT or NDI. For a live service that accepts WHIP, the tatolab-webrtc package’s WhipPublisher can send H.264 and Opus to it from another stream reading graded_video; it doesn’t reconnect if the session drops. Other transports need a node you write.
• No switcher, overlays with text, or background blur built in. Those are nodes you write: a switcher is a node with several inputs that forwards one; text and logos need an image drawn in a node and composited on the GPU, as the camera-python-effects example does with Skia; blur needs a segmentation model.
• The recording’s audio is Opus, which some editing apps don’t import. Convert it with ffmpeg for editing.

Related guides

• GPU effects: write your own look.
• Virtual camera: setup and troubleshooting.
• Share between streams: how session_recording reads my_camera.