5. Optional: show motion on the picture

This step is optional. It shows how a stream processes video on the GPU, steered by your Python code: a GPU effect that tints the picture red while the detector reports motion. If your streams don’t work with video, skip to Keep it running.

The effect needs two inputs, the camera’s frames and the detector’s reports, and it passes a value from Python into the GPU program on every frame.

Write the effect

Create nodes/motion_tint.py:

nodes/motion_tint.py

"""Tints the picture red while the motion detector reports motion."""

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

# `dials.strength` is set from Python on every frame: 1.0 while there's
# motion, 0.0 otherwise. The tint pulses gently, using the elapsed time.
TINT_GLSL = """
vec4 effect(vec4 source, ivec2 at) {
    vec3 red = vec3(1.0, 0.1, 0.1);
    float pulse = 0.3 + 0.1 * sin(streamlib_elapsed_seconds * 6.0);
    return vec4(mix(source.rgb, red, dials.strength * pulse), source.a);
}
"""


@node
class MotionTint:
    """Passes the camera through, tinted red while there's motion."""

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

    @node.input(delivery_profile="newest")
    def motion(self) -> dict: ...

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

    def __init__(self) -> None:
        self.moving = False

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.tint = GlslPixelEffect.compile(
            ctx.gpu_full_access, effect_glsl=TINT_GLSL, dials={"strength": "float"}
        )

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        # Remember the latest motion report; the picture keeps flowing either way.
        motion = ctx.inputs.read("motion")
        if motion is not None:
            self.moving = motion["moving"]

        frame = ctx.inputs.read("video", into=VideoFrame)
        if frame is None:
            return
        strength = 1.0 if self.moving else 0.0
        ctx.outputs.write(
            "tinted_video",
            self.tint.apply_to_frame(ctx.gpu_limited_access, frame, dials={"strength": strength}),
        )

New ideas in this node:

The GLSL function. A pixel effect is one function, vec4 effect(vec4 source, ivec2 at), that returns the output color for one pixel. source is that pixel’s color, with each channel from 0.0 to 1.0, and at is its position. GlslPixelEffect wraps the function in everything else a GPU program needs and runs it over every pixel. mix(a, b, t) blends from a to b by t, so this one blends the picture toward red by up to 40%.

Dials. dials={"strength": "float"} declares a value the shader reads as dials.strength, and apply_to_frame(..., dials={"strength": ...}) sets it for each frame. That’s how Python logic steers a GPU effect without rebuilding it. Dials can be float, int, vec2 or vec4.

Built-in helpers. streamlib_elapsed_seconds is the time since the effect first ran, which makes the tint pulse. You also get streamlib_extent (the frame’s width and height), streamlib_source_at(ivec2) to read another pixel, and streamlib_source_uv(vec2) to sample at normalized coordinates.

setup(ctx) runs once, before the first frame. Compiling the shader is the kind of work that belongs there. Notice the two context types: setup gets RuntimeContextFullAccess, which can create GPU resources; process gets RuntimeContextLimitedAccess, which can use them but not create new ones. That keeps expensive allocations out of the per-frame path.

Two inputs. process runs when either input has something new. It first takes the latest motion report, if there is one, and remembers it; then it takes the latest frame and draws it. Both inputs are newest, because only the current state matters. A frame without a fresh report is drawn with the last one.

A new output frame. apply_to_frame writes into a new frame of the effect’s own and returns its bag, which goes straight to the output. The camera’s frame isn’t modified, which matters because the detector reads the same frame.

Wire it in

Put the tint between the camera and the window, and connect the detector’s output to the tint’s second input:

stream.py

"""A motion watcher: the camera in a window, tinted red while something moves."""

from tatolab.stream import CameraSource, DisplayWindow, StreamBuilder, stream

from nodes.motion_detector import MotionDetector
from nodes.motion_tint import MotionTint


@stream
def motion_watcher(stream_builder: StreamBuilder) -> None:
    """Watches the camera, tints it red while something moves, and shares the camera and the reports."""
    camera = stream_builder.add(CameraSource, name="camera")
    detector = stream_builder.add(MotionDetector, name="detector", config={"sensitivity": 0.02})
    window = stream_builder.add(DisplayWindow, config={"title": "Motion watcher"})
    tint = stream_builder.add(MotionTint, name="tint")

    stream_builder.connect(camera.output("video"), detector.input("video"))
    stream_builder.connect(camera.output("video"), tint.input("video"))
    stream_builder.connect(detector.output("motion"), tint.input("motion"))
    stream_builder.connect(tint.output("tinted_video"), window.input("video"))

    stream_builder.expose(camera.output("video"))
    stream_builder.expose(detector.output("motion"))

The window now reads the tint’s output instead of the camera’s. The exposed outputs don’t change, so motion_log keeps reading the reports without noticing.

Save, and wave at the camera. The picture pulses red while you move and clears when you stop.

Make it yours

• Change red to another color, like vec3(0.1, 0.4, 1.0) for blue.
• Change 6.0 in the sin(...) to make the pulse faster or slower, or replace pulse with a constant for a steady tint.
• Tint only where things change: the detector would need to send a mask, which is a good exercise once you’ve read GPU effects.

If something goes wrong

• The window is black and the terminal shows a GLSL error. The message names the line of your GLSL that failed. Fix it and save. Common mistakes are a missing semicolon, and returning a vec3 where effect must return a vec4.
• The picture never turns red. Check that detector.output("motion") is connected to tint.input("motion"), and that the terminal shows motion started when you move.
• An error says node `tint` has no input port `video` . Two methods with the same name are one method in Python, so an output method also called video replaces the input. Give every port its own name; that’s why the output is tinted_video.

Next

Next you’ll keep both streams running after you close the terminals: Keep it running.