--- title: "Write a node" description: "Everything a Python node can declare and do, with working examples of a filter, a logger, a timed source and a source driven by its own thread." type: "docs-page" url: "https://tatolab.com/docs/build/nodes" site: "Tatolab" section: "Build" --- # Write a node > Everything a Python node can declare and do, with working examples of a filter, a logger, a timed source and a source driven by its own thread. A node is a Python class decorated with `@node`. It declares ports, optionally takes settings, and does its work in a few methods the runtime calls. If you'd rather learn by building, the [tutorial](https://tatolab.com/docs/tutorial/first-node) walks through a first node step by step. ## The shape of a node Here's a complete node that appends every motion report it receives to a CSV file. It shows most of what a node can do: ```python title="nodes/motion_csv_logger.py" """Appends every motion report to a CSV file.""" import csv from dataclasses import dataclass from typing import Annotated, TypedDict from tatolab.stream import RuntimeContextFullAccess, RuntimeContextLimitedAccess, log, node class MotionReport(TypedDict): changed: float moving: bool timestamp_ns: int @dataclass class MotionCsvLoggerConfig: path: Annotated[str, "The CSV file to append to, relative to the project folder."] = "motion.csv" @node class MotionCsvLogger: """Appends every motion report it receives to a CSV file.""" @node.input(delivery_profile="ordered", description="Motion reports, in the order they were made") def motion(self) -> MotionReport: ... def __init__(self, config: MotionCsvLoggerConfig) -> None: self.path = config.path def setup(self, ctx: RuntimeContextFullAccess) -> None: self.file = open(self.path, "a", newline="") self.writer = csv.writer(self.file) def process(self, ctx: RuntimeContextLimitedAccess) -> None: # An ordered input can have several reports waiting; take them all. while (report := ctx.inputs.read("motion", into=MotionReport)) is not None: self.writer.writerow([report["timestamp_ns"], report["changed"], report["moving"]]) self.file.flush() def teardown(self, ctx: RuntimeContextFullAccess) -> None: self.file.close() log.info("motion log closed", path=self.path) ``` Use it in a stream like any other node: ```python logger = stream_builder.add(MotionCsvLogger, config={"path": "motion.csv"}) stream_builder.connect(detector.output("motion"), logger.input("motion")) ``` ## Where nodes live Put each node class in a module of its own, such as `nodes/motion_csv_logger.py`, and import it into `stream.py`. A node class can't be defined in `stream.py` itself, or inside a function: every Python node runs in its own process, and that process imports the class by its module path (here, `nodes.motion_csv_logger:MotionCsvLogger`). The runtime refuses a class it couldn't import that way, with a message saying how to fix it. Your node's process starts in the project folder, with your project's `.venv`, so it can import anything you've installed with `uv add` and open files relative to the project. ## Ports Ports are declared by decorating methods. The methods are never called, so their body is `...`. ```python @node.input(delivery_profile="newest") def video(self) -> VideoFrame: ... @node.output(description="Detections for each frame") def detections(self) -> dict: ... ``` - **The name** is the method's name, unless you pass one: `@node.input("video", ...)`. Every port on a node, input or output, needs its own name, so an effect can't have both an input and an output called `video`. A common pattern is `video` in and `processed_video` out, or the scaffold's `video_from_upstream` and `video_to_downstream`. - **Names are made URL-safe:** lowercased, with anything other than letters, digits, `-`, `.`, `_` and `~` turned into `-`. A method called `Video` declares the port `video`. - **The return annotation** documents what the port carries, for you and your editor. The runtime never checks it. - **`description`** is optional. Agents read it in the node catalog. ### Delivery profiles Every input must choose how it reads. There's no default, because the right answer depends on what the node does: | | `newest` | `ordered` | |---|---|---| | What a read returns | The most recent bag; older ones are skipped | The next bag, in the order they were sent | | Falling behind | Skips to the present | Drops bags if it falls far behind, and counts them | | Typical nodes | Effects, detectors, windows, dashboards | Encoders, recorders, loggers, audio processing | Neither profile ever slows the node that sends the bags. A reader that can't keep up loses bags at its own port, and `tatolab graph` shows how many. ## Settings A node's settings are one class, named by the annotation on the `config` parameter of `__init__`. It can be a dataclass, a `TypedDict` or a pydantic model: ```python @dataclass class ThermalAlarmConfig: threshold_celsius: Annotated[float, "Temperature that raises the alarm."] = 60.0 zone: Annotated[str, "Name of the area this camera watches."] = "line 1" ``` - The runtime builds the class from the `config` dict you pass to `stream_builder.add`. Fields you leave out take their defaults. - A dataclass refuses keys it doesn't know, so a typo raises an error when the node starts instead of being ignored. A pydantic model ignores unknown keys unless you set `extra="forbid"`, and a `TypedDict` takes whatever it's given. - `Annotated[type, "description"]` describes a setting. The runtime publishes each node's settings, with their types, defaults and descriptions, in its node catalog, which is how an agent learns what it can configure. - A node whose `__init__` takes only `self` has no settings, and refuses any `config` you pass it. Settings are fixed for the life of the node. To change one, edit `stream.py`; `tatolab dev` reloads the stream when you save. ## What the runtime calls All of these are optional; define the ones you need. | Method | When it runs | Context | |---|---|---| | `__init__(self, config)` | Once, when the node's process starts | none | | `setup(self, ctx)` | Once, before any data flows. Load models, compile shaders, open files and devices here. | full | | `start(self, ctx)` | `manual` nodes only: when the node should begin. Start your own thread here. | full | | `process(self, ctx)` | Whenever there's work: new input, or the next tick of a `continuous` node | limited | | `stop(self, ctx)` | When the node stops. Stop what `start` started. | full | | `teardown(self, ctx)` | Once, at the end. Close files and release what `setup` acquired. | full | The **full** context (`RuntimeContextFullAccess`) can create GPU resources, through `ctx.gpu_full_access`. The **limited** one (`RuntimeContextLimitedAccess`), which `process` gets, can use GPU resources but not create them. That keeps allocations out of the per-frame path. Both carry: - `ctx.inputs` and `ctx.outputs`: the node's ports. - `ctx.config`: the node's settings, as a dict. - `ctx.time`: the machine's monotonic clock, in nanoseconds. - `ctx.node_id`: the node's ID, as `tatolab graph` shows it. Keep `setup` reasonably quick. If it has to download model weights, download them once beforehand, so a node doesn't spend its start-up time on the network. ## When `process` runs `@node` takes an `execution` mode: - **`reactive`** (the default for a node with inputs): `process` runs when a bag arrives on any input. - **`continuous`**: `process` runs once straight away, then every `interval_ms` milliseconds. If a run takes longer than the interval, the ticks it missed are skipped, not queued. Use it for sources that poll something, or produce on a schedule. - **`manual`**: the runtime never calls `process`. The node drives itself, usually from a thread it starts in `start`. Use it for sources that are pushed data, like a socket or a device callback. A node with no inputs is a source, and has to choose `continuous` or `manual`; there'd be nothing to react to. You can also ask for a scheduling priority. `scheduling="high"` asks the operating system to raise the node's priority, and `"realtime"` asks for real-time scheduling, for a node that must keep a steady pace, such as one producing audio. If the system refuses, the node runs at normal priority and the log says so. Most nodes need neither. ### A timed source This one polls a sensor's JSON endpoint once a second: ```python title="nodes/http_sensor.py" """Reads a JSON sensor over HTTP once a second.""" import json import urllib.request from dataclasses import dataclass from tatolab.stream import RuntimeContextLimitedAccess, log, node @dataclass class HttpSensorConfig: url: str = "http://192.168.1.50/reading.json" timeout_seconds: float = 0.5 @node(execution="continuous", interval_ms=1000) class HttpSensor: """Polls a sensor's JSON endpoint once a second and publishes each reading.""" @node.output() def reading(self) -> dict: ... def __init__(self, config: HttpSensorConfig) -> None: self.url = config.url self.timeout = config.timeout_seconds def process(self, ctx: RuntimeContextLimitedAccess) -> None: try: with urllib.request.urlopen(self.url, timeout=self.timeout) as response: reading = json.load(response) except (OSError, ValueError) as error: log.warn("sensor unreachable", url=self.url, error=str(error)) return ctx.outputs.write("reading", {**reading, "timestamp_ns": ctx.time}) ``` ### A source with its own thread This one publishes every UDP datagram that arrives on a port. It's `manual`: `start` hands the node's outputs to a thread, and `stop` ends it. ```python title="nodes/udp_listener.py" """Publishes UDP datagrams as they arrive.""" import socket import threading from dataclasses import dataclass from tatolab.stream import LinkOutputDataWriter, RuntimeContextFullAccess, monotonic_now_ns, node @dataclass class UdpListenerConfig: host: str = "127.0.0.1" port: int = 9000 @node(execution="manual") class UdpListener: """Publishes every UDP datagram that arrives on a port.""" @node.output() def datagrams(self) -> dict: ... def __init__(self, config: UdpListenerConfig) -> None: self.address = (config.host, config.port) self.stopping = threading.Event() self.thread: threading.Thread | None = None def start(self, ctx: RuntimeContextFullAccess) -> None: outputs = ctx.outputs self.thread = threading.Thread(target=self.listen, args=(outputs,), daemon=True) self.thread.start() def listen(self, outputs: LinkOutputDataWriter) -> None: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.bind(self.address) sock.settimeout(0.2) # wake up regularly to check whether to stop while not self.stopping.is_set(): try: data, sender = sock.recvfrom(65535) except TimeoutError: continue outputs.write( "datagrams", {"data": data, "sender": sender[0], "timestamp_ns": monotonic_now_ns()}, ) def stop(self, ctx: RuntimeContextFullAccess) -> None: self.stopping.set() if self.thread is not None: self.thread.join(timeout=1.0) ``` Keep `stop` prompt: the runtime gives a node a few seconds to wind down, so join with a timeout, as here, rather than waiting forever. ## Reading inputs ```python bag = ctx.inputs.read("motion") # a dict, or None report = ctx.inputs.read("motion", into=MotionReport) # typed bag, timestamp_ns = ctx.inputs.read_with_timestamp("motion") ``` - `read` returns `None` when there's nothing new. Always check. - `into=` sets how strict the read is. A `TypedDict` costs nothing: you get the dict, with type hints. A dataclass or pydantic model checks the bag and raises if it doesn't fit. Use `VideoFrame`, `AudioBlock`, `EncodedVideoFrame` and `EncodedAudioPacket` for the runtime's own bag types. - On an `ordered` input, several bags can be waiting. Read in a loop until `read` returns `None`, as the CSV logger does. ### Several links into one input Any number of links can connect to one input. To tell them apart, read with the link's name: ```python while (received := ctx.inputs.read_from_inbound_link("readings")) is not None: bag, link_name = received # "/" self.latest[link_name] = bag ``` `ctx.inputs.inbound_link_names("readings")` lists the links in `setup`, before any data arrives. A link's name contains the sender's node ID, which changes each time the stream loads, so learn the names there rather than writing them into your code. Bags from one link arrive in that link's order; bags from different links can interleave in any order, so compare their timestamps if order matters. ## Writing outputs ```python ctx.outputs.write("detections", {"labels": ["person"], "count": 1}, timestamp_ns=frame.timestamp_ns) ``` A bag is a dict whose values msgpack can encode: Python ints and floats, strings, booleans, `None`, lists, dicts and `bytes`. Numpy numbers need converting with `int()` or `float()`. For a numpy array, send `array.tobytes()` along with its `shape` and `dtype`, or `array.tolist()` if it's small. Keep bags small; a video bag only names its frame, which stays on the GPU. Every reader of the output gets the bag. If nothing is connected, the write goes nowhere, and that's fine. Every bag carries a timestamp. When you derive a bag from a frame or an audio block, pass the source's stamp with `timestamp_ns=`, as above, so downstream nodes can line things up; it's what `read_with_timestamp` returns. Leave it out and the stamp is the moment of the write. ## Logging ```python from tatolab.stream import log log.info("person entered", zone="door", confidence=0.91) log.warn("camera is slow", fps=4) log.error("model failed to load", path=self.weights) ``` Each record carries your message and fields, plus the node and stream it came from. You see them in the terminal running `tatolab run` or `tatolab dev`, and with `tatolab logs`. Anything a node prints with `print()` is captured too, but `log` keeps the fields separate and filterable. ## Time Use `ctx.time`, or `monotonic_now_ns()` outside `process`, for anything time-based. It's the same monotonic clock every node and built-in uses, so timestamps are comparable across the whole stream. To do something "five times a second", compare `ctx.time` against the next due time, as the [tutorial's detector](https://tatolab.com/docs/tutorial/first-node) does. Don't `sleep` in `process`: a `continuous` node with `interval_ms` is the way to run on a schedule. ## Test your logic `tatolab-stream` imports without the runtime, so your node modules import in a plain test too. Keep the interesting logic in ordinary functions, and test those with pytest: ```python title="nodes/motion_math.py" import numpy def changed_fraction(previous: numpy.ndarray, current: numpy.ndarray, threshold: float = 25) -> float: """The share of pixels whose brightness moved by more than `threshold`.""" return float((numpy.abs(current - previous) > threshold).mean()) ``` ```python title="tests/test_motion_math.py" import numpy from nodes.motion_math import changed_fraction def test_a_still_picture_has_not_changed(): frame = numpy.full((90, 160), 100.0) assert changed_fraction(frame, frame) == 0.0 def test_a_moved_block_is_counted(): before = numpy.full((90, 160), 100.0) after = before.copy() after[0:45, 0:80] = 200.0 assert changed_fraction(before, after) == 0.25 ``` Run them from the project folder with `uv run --with pytest python -m pytest`. (`python -m` puts the project folder on the import path, so `from nodes…` works; plain `pytest` can't find `nodes`.) The node class stays a thin wrapper that reads, calls your functions, and writes. ## Related - [GPU effects](https://tatolab.com/docs/build/gpu-effects): nodes that work on pixels. - [Use your model](https://tatolab.com/docs/build/models): nodes that run inference. - [Python SDK reference](https://tatolab.com/docs/reference/python-api): every class and method.