Build a Custom Data Graph
| Field | Value |
|---|---|
| Difficulty | Intermediate |
| Estimated Read Time | 15-20 minutes |
| Labels | graph, traversal, metadata |
Chapter 003 built an anonymous Input → Output graph and drove it with positional run() calls. Real orchestration — fan-out, fan-in, per-stream routing — needs to address endpoints by name, not by position. This chapter introduces that named-endpoint surface on the smallest possible graph, so you can see the naming and wiring mechanics in isolation before the multistream and embedded-model chapters build on them.
The public Graph is the application composition surface: you add(...) nodes, connect(...) named endpoints, build() once into a reusable Run, then push("image", ...) and pull("out", ...) by name. By the end you will have pushed one tensor Sample through a named graph and confirmed its stream_id, frame_id, and pts_ns came out unchanged — proof the runtime preserves metadata end to end.
Walkthrough
Compose the graph
Add two nodes. Input("image") declares a push endpoint named image; Output("out") declares a pull endpoint named out. The names are the contract — they are exactly the strings you will pass to push(...) and pull(...) later. Naming endpoints (rather than relying on add-order) is what makes larger graphs with multiple inputs or outputs unambiguous to drive.
Nodes come from simaai::neat::nodes::Input("image") and nodes::Output("out").
// `Graph` is the public composition type. Input("image") declares the name
// used by Run::push("image", ...). Output("out") declares the name used by
// Run::pull("out", ...).
simaai::neat::Graph graph;
graph.add(simaai::neat::nodes::Input("image"));
graph.add(simaai::neat::nodes::Output("out"));
Wire the endpoints
connect("image", "out") declares the edge: frames pushed to image flow to out. With only two nodes this is the entire topology, but connect(...) is the same call you would use to build branches and merges in a larger graph. We then print graph.describe() to dump the composed topology — a quick sanity check that the graph is wired the way you intended before building.
graph.connect("image", "out");
std::cout << graph.describe() << "\n";
Build and push a sample
build() (with no priming sample needed here) materializes the description into a runnable Run. We then construct one deterministic tensor Sample — an 8×8×3 RGB image carrying a known stream_id, frame_id, and pts_ns — and push(...) it to the image endpoint by name. The sample's metadata is what we will check on the other side.
push(...) returns a bool; on failure we surface run.last_error(). The sample is built by make_sample().
simaai::neat::Run run = graph.build();
if (!run.push("image", make_sample())) {
throw std::runtime_error("push failed: " + run.last_error());
}
Pull the output and verify metadata
pull("out", ...) retrieves the result from the named output endpoint with a timeout, after which we close() the run. Because there is no transform between input and output, a correct pipeline returns the same logical sample — so reading back stream_id, frame_id, and pts_ns and seeing the values we pushed confirms the runtime preserved per-sample metadata through traversal. That guarantee is what lets downstream stages trust frame identity and timestamps.
auto out = run.pull("out", /*timeout_ms=*/2000);
run.close();
Run
Run it and you should see the graph description followed by the round-tripped metadata. Run the Python and C++ (prebuilt) commands from the Neat install root (the directory that contains share/ and lib/); run the build from source commands from the repo root. This chapter needs no model archive.
C++ (prebuilt):
./lib/sima-neat/tutorials/tutorial_013_build_a_custom_data_graph
C++ (build from source):
./build.sh --target tutorial_013_build_a_custom_data_graph
./build/tutorials-standalone/tutorial_013_build_a_custom_data_graph
Expected output (preceded by the graph.describe() dump):
stream=graph frame=42 pts_ns=123456789
[OK] 013_build_a_custom_data_graph
(The Python build prints stream_id=graph frame_id=42 pts_ns=123456789.) To integrate this chapter's C++ source into your own project with a custom CMakeLists.txt (no extras folder required), see How to Run Tutorials on the landing page.
Full source
Show the complete C++ and Python programs
// Compose a minimal public Neat Graph: named Input -> named Output.
//
// Usage:
// tutorial_013_build_a_custom_data_graph
#include "neat.h"
#include <cstdint>
#include <iostream>
#include <stdexcept>
#include <utility>
#include <vector>
namespace {
std::vector<int64_t> contiguous_strides_bytes(const std::vector<int64_t>& shape,
int64_t elem_bytes) {
std::vector<int64_t> strides(shape.size(), 0);
int64_t stride = elem_bytes;
for (int i = static_cast<int>(shape.size()) - 1; i >= 0; --i) {
strides[static_cast<size_t>(i)] = stride;
stride *= shape[static_cast<size_t>(i)];
}
return strides;
}
simaai::neat::Sample make_sample() {
const int w = 8;
const int h = 8;
const int c = 3;
const std::size_t bytes = static_cast<std::size_t>(w) * h * c;
simaai::neat::Tensor t;
t.device = {simaai::neat::DeviceType::CPU, 0};
t.dtype = simaai::neat::TensorDType::UInt8;
t.layout = simaai::neat::TensorLayout::HWC;
t.shape = {h, w, c};
t.semantic.image = simaai::neat::ImageSpec{simaai::neat::ImageSpec::PixelFormat::RGB, ""};
t.storage = simaai::neat::make_cpu_owned_storage(bytes);
t.strides_bytes = contiguous_strides_bytes(t.shape, 1);
t.read_only = false;
{
auto map = t.map(simaai::neat::MapMode::Write);
auto* p = static_cast<std::uint8_t*>(map.data);
for (std::size_t i = 0; i < bytes; ++i)
p[i] = static_cast<std::uint8_t>(i % 255);
}
t.read_only = true;
simaai::neat::Sample s;
s.kind = simaai::neat::SampleKind::Tensor;
s.tensor = std::move(t);
s.stream_id = "graph";
s.frame_id = 42;
s.pts_ns = 123456789;
return s;
}
} // namespace
int main() {
try {
// CORE LOGIC
// `Graph` is the public composition type. Input("image") declares the name
// used by Run::push("image", ...). Output("out") declares the name used by
// Run::pull("out", ...).
simaai::neat::Graph graph;
graph.add(simaai::neat::nodes::Input("image"));
graph.add(simaai::neat::nodes::Output("out"));
graph.connect("image", "out");
std::cout << graph.describe() << "\n";
simaai::neat::Run run = graph.build();
if (!run.push("image", make_sample())) {
throw std::runtime_error("push failed: " + run.last_error());
}
auto out = run.pull("out", /*timeout_ms=*/2000);
run.close();
if (!out.has_value())
throw std::runtime_error("graph produced no output");
std::cout << "stream=" << out->stream_id << " frame=" << out->frame_id
<< " pts_ns=" << out->pts_ns << "\n";
std::cout << "[OK] 013_build_a_custom_data_graph\n";
return 0;
} catch (const std::exception& e) {
std::cerr << "[FAIL] " << e.what() << "\n";
return 1;
}
}