2. Write your first node

In this step you’ll write MotionDetector, a node that compares the camera’s frames a few times a second and reports how much of the picture changed. Its reports are ordinary data, which in the next step you’ll share with another stream.

Write the node

Create nodes/motion_detector.py:

nodes/motion_detector.py

"""Measures how much of the picture changes, a few times a second."""

from dataclasses import dataclass

import numpy
from tatolab.stream import RuntimeContextLimitedAccess, VideoFrame, log, node


@dataclass
class MotionDetectorConfig:
    # How much of the picture has to change to count as motion: 0.02 is 2%.
    sensitivity: float = 0.02
    # How often to compare frames. Each check reads one frame on the CPU.
    checks_per_second: int = 10


@node
class MotionDetector:
    """Reports how much of the picture changed, and whether that counts as motion."""

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

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

    def __init__(self, config: MotionDetectorConfig) -> None:
        self.sensitivity = config.sensitivity
        self.check_interval_ns = 1_000_000_000 // config.checks_per_second
        self.next_check_at_ns = 0
        self.previous: numpy.ndarray | None = None
        self.moving = False

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        frame = ctx.inputs.read("video", into=VideoFrame)
        if frame is None or ctx.time < self.next_check_at_ns:
            return
        self.next_check_at_ns = ctx.time + self.check_interval_ns

        # A small grayscale copy of the frame: every 8th pixel in each
        # direction, averaged over red, green and blue.
        with frame.cpu() as pixels:
            gray = pixels[::8, ::8, :3].mean(axis=2)

        if self.previous is None or self.previous.shape != gray.shape:
            self.previous = gray
            return
        # The share of those pixels whose brightness moved by more than 25 (of 255).
        changed = float((numpy.abs(gray - self.previous) > 25).mean())
        self.previous = gray
        moving = changed > self.sensitivity

        ctx.outputs.write(
            "motion",
            {"changed": round(changed, 4), "moving": moving, "timestamp_ns": frame.timestamp_ns},
        )
        if moving != self.moving:
            log.info("motion started" if moving else "motion stopped", changed=round(changed, 4))
            self.moving = moving

Here’s what each part does.

@node turns the class into a node. It registers nothing; it records what the class declares, so the runtime can read it when a stream uses the class.

The ports. @node.input and @node.output mark methods that declare ports; the methods are never called, which is why their bodies are .... The port takes the method’s name, so this node has an input called video and an output called motion. The return annotations (-> VideoFrame, -> dict) are for you and your editor. The runtime doesn’t check them.

delivery_profile="newest". Every input must say how it reads. newest means “give me the most recent frame, and skip any I didn’t get to”. That’s what a detector wants: if a check runs late, there’s no point catching up on old frames. The other choice, ordered, delivers bags in the order they were sent, and drops them (counted) only if the reader falls far behind; logs and recorders use it. (A bag is one piece of data on a link: a frame, or one of this node’s reports.)

The config. The node’s settings are a dataclass, named by the annotation on __init__’s config parameter. When the node starts, the runtime builds the dataclass from the config dict you pass in stream.py. Anything you leave out takes its default, and a misspelled key raises an error instead of being quietly ignored.

process(ctx) is called whenever new data arrives on an input. It reads the latest frame and returns early if there isn’t one, or if it isn’t time for the next check yet. ctx.time is the machine’s monotonic clock in nanoseconds; comparing against it is how a node does something “ten times a second” without sleeping.

frame.cpu() hands you the frame’s pixels as a numpy array, valid inside the with block. Reading a frame on the CPU is the slow way to get at pixels, so the node does it only ten times a second. Inside the block, [::8, ::8, :3] picks every 8th pixel without the alpha channel, and .mean(axis=2) copies that into a small new array that stays valid after the block ends.

ctx.outputs.write("motion", {...}) sends a bag to everything connected to the motion output. A bag is a dict; there’s no schema to declare.

log.info(...) writes a log record with fields. It shows up in the terminal running tatolab dev, and in tatolab logs. The node logs only when motion starts or stops, so the terminal stays readable.

The class lives in its own module under nodes/, never in stream.py. Each Python node runs in its own process, which imports the class by its module path: nodes.motion_detector:MotionDetector.

Wire it in

Now replace the whole of stream.py with this. The window shows the camera as it is, and the detector reads the same camera. The expose line is gone for now, since nothing reads the effect’s output any more; step 3 exposes the outputs worth sharing.

stream.py

"""A motion watcher: the camera in a window, with a motion detector watching."""

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

from nodes.motion_detector import MotionDetector


@stream
def motion_watcher(stream_builder: StreamBuilder) -> None:
    """Watches the camera for motion."""
    camera = stream_builder.add(CameraSource)
    detector = stream_builder.add(MotionDetector, config={"sensitivity": 0.02})
    window = stream_builder.add(DisplayWindow, config={"title": "Motion watcher"})

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

Delete nodes/brightness_meter.py and nodes/inverting_effect.py; nothing uses them now.

Save. tatolab dev loads the new version. Sit still for a moment, then wave a hand in front of the camera. The terminal shows a motion started record, with changed= saying how much of the picture changed, and a motion stopped record once you’re still again.

The camera’s output now feeds two nodes, the window and the detector. It isn’t split between them, and neither can slow the other down: if the detector stalls, the window keeps showing the latest frame.

Tune it

If your room flickers or your camera is noisy, the detector may report motion that isn’t there. Raise sensitivity in stream.py, say to 0.05, and save. To check more or less often, add "checks_per_second": 5. The stream reloads with the new settings each time.

Try a typo, like "sensitivty": 0.05, and save. When the detector starts, the terminal shows an error naming the unexpected keyword sensitivty, so you find the mistake at once instead of wondering why your setting does nothing. Fix it and save again.

If something goes wrong

• An error says the node is defined in the entry file. The class must be in a module of its own, like nodes/motion_detector.py, and imported into stream.py.
• An error mentions delivery_profile. Every @node.input needs one: newest or ordered.
• Nothing is logged when you move. Check that the camera’s output is connected to detector.input("video"), and try a lower sensitivity, like 0.01.

Next

The detector reports motion, but only its own stream can read the reports. Next you’ll share them: Share the motion reports.