Build a Custom Data Graph

Field	Value
Difficulty	Intermediate
Estimated Read Time	15-20 minutes
Labels	graph, traversal, metadata

Concept

Build the smallest useful Neat graph — one pipeline node wired to one stage node — then push a sample through and verify metadata survives traversal. This is the baseline before hybrid or multistream graph tutorials.

A graph is an explicit DAG of nodes you build programmatically, separate from the pipeline/session abstraction. You add nodes with graph.add(...), wire them with graph.connect(...), and run the whole thing via GraphSession.

APIs introduced

• pyneat.graph.Graph() — the graph container.
• graph.add(node) — add a node; returns an ID.
• graph.connect(src_id, dst_id) — wire outputs to inputs.
• pyneat.graph.nodes.pipeline_node(inner_node, name) — wrap a regular pipeline node for use inside a graph.
• pyneat.graph.nodes.stamp_frame_id(name) — a stage node that tags samples with frame identity.
• pyneat.graph.GraphSession(graph).build() — materialize the graph into a runnable.

When to use this

• Custom orchestration where pipeline/session semantics don't fit (fan-out, fan-in, per-stream routing).
• Multistream scheduling (see chapter 014).
• Embedding model execution as one stage of a larger flow (see chapter 013).

Prerequisites Chapter 003 (Session basics).

References

• Graph
• Session

Learning Process

1. Build a minimal graph and push one deterministic tensor sample.
2. Run preferred pipeline+stage composition, with stage-only fallback when needed.
3. Pull graph output and validate stream/frame metadata stamping.

Run

Python:

python3 share/sima-neat/tutorials/012_build_a_custom_data_graph/build_a_custom_data_graph.py

C++ (prebuilt):

./lib/sima-neat/tutorials/tutorial_012_build_a_custom_data_graph

C++ (build from source):

./build.sh --target tutorial_012_build_a_custom_data_graph
./build/tutorials-standalone/tutorial_012_build_a_custom_data_graph

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.

Code

tutorials/012_build_a_custom_data_graph/build_a_custom_data_graph.cpp

// Compose a minimal Neat Graph: PipelineNode -> StampFrameIdNode, push one sample, pull output.
//
// Usage:
//   tutorial_012_build_a_custom_data_graph

#include "neat/graph.h"
#include "neat/nodes.h"
#include "neat/session.h"

#include <cstdint>
#include <iostream>
#include <memory>
#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 = -1;
  return s;
}

} // namespace

int main() {
  try {
    using namespace simaai::neat::graph;

    // CORE LOGIC
    // A Graph is a DAG of nodes. Here: a PipelineNode (wraps a classic gst node)
    // feeds a StampFrameIdNode, which assigns a monotonic frame id.
    Graph g;

    const NodeId pipe = g.add(
        std::make_shared<nodes::PipelineNode>(simaai::neat::nodes::VideoConvert(), "convert"));
    const NodeId stamp = g.add(nodes::StampFrameIdNode("stamp"));
    g.connect(pipe, stamp);

    std::cout << GraphPrinter::to_text(g) << "\n";

    GraphRun run = GraphSession(std::move(g)).build();
    run.push(pipe, make_sample());
    auto out = run.pull(stamp, /*timeout_ms=*/2000);
    run.stop();

    if (!out.has_value())
      throw std::runtime_error("graph produced no output");
    std::cout << "stream=" << out->stream_id << " frame=" << out->frame_id << "\n";
    std::cout << "[OK] 012_build_a_custom_data_graph\n";
    return 0;
  } catch (const std::exception& e) {
    std::cerr << "[FAIL] " << e.what() << "\n";
    return 1;
  }
}

Source

• C++
• Python
• README