Python SDK reference

The Python SDK is the tatolab.stream package, installed from PyPI as tatolab-stream. Everything a stream or a node imports comes from it. It’s fully typed, and it imports and type-checks without the runtime, so your editor, type checker and tests work on any machine.

from tatolab.stream import StreamBuilder, VideoFrame, node, stream

The package’s own source is the definitive reference: each class and protocol documents itself in sdk/tatolab-stream/tatolab/stream/.

Streams

@stream

@stream
def main(stream_builder: StreamBuilder) -> None:
    """What this stream does."""

Marks a function as a stream. The function runs once when the stream loads, receives a StreamBuilder, and describes the graph. Its name is the stream’s name, made URL-safe. Its docstring documents it in your code; tatolab run shows the first line when a file holds several streams, and agents don’t see it. Use @stream bare, without arguments.

StreamBuilder

Member
add(node_class, *, name=None, config=None) -> NodeReference	Adds a node. Pass the class, not an instance. Without name, the node is named after its class in lowercase, and a second node of the same class gets -2, and so on. A name you give is made URL-safe but never suffixed, so two names that come out the same are an error. config is a dict of the node’s settings.
connect(source, destination)	Links an output (node.output("port"), or remote_output(...)) to an input (node.input("port"))
expose(output, level=Exposure.PRIVATE)	Lets an output be read from outside the stream. Each output can be exposed once.
remote_output(address) -> NodeOutputPortReferenceByAddress	Names another stream’s exposed output on this machine: "<stream>/<node>/<port>"
name	The stream’s name

NodeReference

Returned by add. .output("port") and .input("port") name its ports for connect and expose.

Exposure

Exposure.PRIVATE: readable by other streams and agents on this machine. Exposure.PUBLIC: also readable from other machines, by URL.

compile_stream_to_graph(stream_function, *, name=None) -> dict

Runs a stream function on a fresh builder and returns the graph it describes, as the runtime would load it. Useful in tests: assert on the nodes, links and exposures without a runtime.

Declaring nodes

@node

@node
class Detector: ...

@node(execution="continuous", interval_ms=1000, scheduling="high", description="Polls the sensor")
class SensorReader: ...

Argument
execution	"reactive" (run when input arrives; the default for a node with inputs), "continuous" (run every interval_ms), or "manual" (the node drives itself). A node without inputs must choose continuous or manual.
interval_ms	For continuous, the milliseconds between runs. The first run is immediate, and missed ticks are skipped, not queued.
scheduling	"high" asks the operating system to raise the node’s priority, and "realtime" asks for real-time scheduling. If the system refuses, the node runs at normal priority and the log says so. Most nodes don’t need either.
description	The node’s description in the catalog; defaults to the class docstring

A node class must live in an importable module, not in stream.py and not inside a function.

@node.input and @node.output

@node.input(delivery_profile="newest", description="Frames to analyze")
def video(self) -> VideoFrame: ...

@node.output(description="One bag per analyzed frame")
def detections(self) -> dict: ...

Argument
name	The port’s name; defaults to the method’s name
delivery_profile	Inputs only, required: "newest" or "ordered"
audio_window	Inputs only, optional: an AudioWindowContract (needs "ordered")
description	Shown to agents in the catalog

Port names are made URL-safe, and every port on a node, input or output, needs a different name. The method is never called; its return annotation is documentation.

Neither delivery profile guarantees delivery. Under sustained load both drop bags at the reader’s port, and a link never blocks its producer.

AudioWindowContract

AudioWindowContract(sample_rate=16_000, dtype="f32", window_size=480)

The audio format an input wants; the runtime converts and cuts the audio to match. All arguments are keywords:

Argument
sample_rate	Required. Samples per second.
dtype	Required. "f32" or "i16".
window_size	Required. Samples per channel in each window.
hop	Samples between the starts of two windows. Defaults to window_size, for back-to-back windows; smaller makes overlapping windows; larger is an error.
channels	Defaults to the source’s own count. Set it only when your code needs a particular count, such as a model trained on mono.

See Audio.

Settings

A node’s settings are the class named by the annotation on its __init__’s config parameter: a dataclass, a TypedDict or a pydantic model. Annotated[type, "description"] on a field describes it in the catalog. A node whose __init__ takes only self has no settings.

Lifecycle methods

All optional:

Method	Called	Context
__init__(self, config)	Once, when the node’s process starts	none
setup(ctx)	Once, before data flows	RuntimeContextFullAccess
start(ctx)	manual nodes only, when the node should begin	RuntimeContextFullAccess
process(ctx)	On input (reactive) or on each tick (continuous); never for manual	RuntimeContextLimitedAccess
stop(ctx)	When the node stops	RuntimeContextFullAccess
teardown(ctx)	Once, at the end	RuntimeContextFullAccess

Contexts

Both context types carry:

Member
inputs	The node’s inputs, a LinkInputDataReader
outputs	The node’s outputs, a LinkOutputDataWriter
config	The node’s settings, as a dict
time	The machine’s monotonic clock, in nanoseconds
gpu_limited_access	GPU operations that don’t allocate (GpuContextLimitedAccess)
node_id	The node’s ID, as graph shows it
runtime_id	The ID of the runtime running the node

RuntimeContextFullAccess, given to setup, start, stop and teardown, adds gpu_full_access (GpuContextFullAccess), which can create GPU resources. process gets RuntimeContextLimitedAccess, which can’t.

Ports at run time

ctx.inputs (LinkInputDataReader)

Method	Returns
read(port, *, into=None)	The next bag (the latest, for newest), or None. into casts it: a TypedDict passes it through; a dataclass, pydantic model or Tatolab type validates it.
read_with_timestamp(port)	(bag, timestamp_ns), or (None, None)
read_from_inbound_link(port, *, into=None)	(bag, link_name) or None, to tell several links into one input apart. A link’s name is <sender's node ID>/<port>, which changes each time the stream loads, so learn the names with inbound_link_names in setup.
read_from_inbound_link_with_timestamp(port, *, into=None)	(bag, link_name, timestamp_ns) or None
inbound_link_names(port)	The names of the links into port; readable in setup
has_data(port)	Whether a bag is waiting

ctx.outputs (LinkOutputDataWriter)

write(port, bag, timestamp_ns=None) sends a dict to every reader of port, stamped with timestamp_ns, or with the time of the write if you leave it out. Pass the stamp of the frame or block a bag was derived from, so readers can line things up; it’s what read_with_timestamp returns. A bag holds dicts with string keys, lists and tuples, strings, bytes, Python ints and floats, booleans and None. Anything else raises TypeError, including numpy arrays and most numpy numbers: send .tolist() for a few values, .tobytes() for many, and int() or float() for a single numpy number. A bag too big for the link is dropped and counted against the port rather than raised.

Bag types

VideoFrame

Read with ctx.inputs.read(port, into=VideoFrame).

Member
surface_id, width, height, timestamp_ns	The frame’s GPU surface, size and capture time
fps, color_info, content_light, mastering_display	Optional metadata, when the producer sets it
torch.from_dlpack(frame)	The pixels as a height × width × 4 uint8 tensor, on the frame’s device
with frame.cpu() as pixels:	The pixels as a numpy array, valid inside the block: the slow path. Read-only unless the frame can take an edit.
with frame.writable() as tensor:	A GPU tensor for editing the frame in place, through torch.from_dlpack(tensor). Refused when the frame can’t take an edit.
VideoFrame.from_bag(bag)	Validate a bag as a frame

While the object exists, the producer can’t reuse the frame’s memory; letting it go releases it. Holding frames for a long time costs the producer memory, and then its own frames, but never slows another node.

AudioBlock

Member
samples	A read-only numpy array, sample_count × channels, over the block’s bytes
interleaved_sample_bytes	The same samples as raw bytes
sample_rate, channels, sample_count, dtype	The block’s format
first_sample_timestamp_ns	When the first sample was captured

To write audio, send a dict with samples (interleaved little-endian bytes), sample_rate, channels, sample_count, first_sample_timestamp_ns and dtype ("f32" if left out).

EncodedVideoFrame, EncodedAudioPacket

One compressed video frame, or one Opus packet. EncodedVideoFrame has codec, annex_b_access_unit_bytes, is_sync_point, group_index, sequence_index, width and height (the coded size, which can be padded, such as 1088 for 1080p). EncodedAudioPacket has codec, opus_packet_bytes, is_sync_point, sequence_index, sample_rate, channels, sample_count and pre_skip. A sequence_index that jumps by more than one means data was lost; wait for the next sync point.

GPU

GlslPixelEffect

Member
GlslPixelEffect.compile(gpu_full_access, effect_glsl, dials=None)	In setup. effect_glsl defines vec4 effect(vec4 source, ivec2 at); dials maps names to "float", "int", "vec2" or "vec4".
effect.apply_to_frame(gpu_limited_access, frame, dials=None)	In process. Returns the bag of a new RGBA frame. Supply every dial every time.

See GPU effects.

ModelInputTensorKernel

Member
ModelInputTensorKernel.compile(gpu_full_access, width=None, height=None, fit="stretch", pad_to_multiple_of=None, channel_order="rgb", layout="nchw", dtype="float32", scale=1/255, mean=(0, 0, 0), std=(1, 1, 1))	In setup. fit is "stretch", "letterbox" or "pad_bottom_right" (no resize, so leave out width and height); channel_order is "rgb" or "bgr"; layout is "nchw" or "nhwc"; dtype is "float32" or "float16". Each channel value x, from 0 to 255, becomes (x * scale - mean) / std.
kernel.apply_to_surface(gpu_limited_access, surface) -> ModelInputTensor	In process, with a surface from gpu_limited_access.resolve_surface(frame.surface_id)
model_input.tensor_surface	Use as with model_input.tensor_surface as tensor:, then torch.from_dlpack(tensor)
model_input.geometry.boxes_to_source(boxes)	Maps x0, y0, x1, y1 boxes from the tensor back to frame pixels

See Use your model.

GPU contexts

ctx.gpu_full_access (in setup and friends) creates resources: create_compute_kernel, create_graphics_kernel, create_ray_tracing_kernel (Linux), build_triangles_blas and build_tlas, create_window, kernel_dispatch_batch, and raw handle export for native code: export_dma_buf and export_opaque_fd on Linux, export_iosurface on macOS.

ctx.gpu_limited_access (everywhere) uses them: resolve_surface, claim_surface_against_producer_reuse, copy_surface_to_surface, and acquiring pixel buffers, textures and storage buffers, including from the node’s output pool.

These are the building blocks under GlslPixelEffect and ModelInputTensorKernel. The camera-compute-kernel and raytracing-showcase examples show them in full.

Logging

from tatolab.stream import log

log.info("message", key=value, other=1)

log.trace, log.debug, log.info, log.warn (or log.warning) and log.error each take a message and any keyword fields. Records appear in tatolab run and tatolab dev output and in tatolab logs, tagged with the node and stream.

The clock

ctx.time	Nanoseconds on the machine’s monotonic clock, inside a node
monotonic_now_ns()	The same clock, anywhere in a node’s process
start_monotonic_timer(interval_ns)	A timer that ticks on that clock without drifting. Use it in a with block; wait(timeout_ms=100) returns how many ticks fired, or 0 on timeout.

Every timestamp in a stream comes from this clock, so they compare directly.