--- title: "2. Write your first node" description: "Write a motion detector in Python and numpy, give it a setting, wire it into the stream, and watch it report motion." type: "docs-page" url: "https://tatolab.com/docs/tutorial/first-node" site: "Tatolab" section: "Tutorial" --- # 2. Write your first node > Write a motion detector in Python and numpy, give it a setting, wire it into the stream, and watch it report motion. 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`: ```python title="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. ```python title="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](https://tatolab.com/docs/tutorial/share).