Classic Pipeline flow

The classic Pipeline flow exposes each step directly: initialize, subscribe to output topics, publish an input tensor, run, and receive raw output tensors. It provides full control over topic wiring and tensor decoding. For most integrations, SimplifiedPipeline covers the same task with less code.

Concepts referenced below (pipelines, nodes, topics, tensors) are described in Pipeline fundamentals.

Running an exported pipeline

Python

from denkflow import Pipeline, ImageTensor

pipeline = Pipeline.from_denkflow("path/to/model.denkflow", pat="YOUR-PAT")
pipeline.initialize()

receiver = pipeline.subscribe("bounding_box_filter_node/filtered_bounding_boxes")

pipeline.publish_image_tensor("/image", ImageTensor.from_file("path/to/image.jpg"))
pipeline.run()

objects = receiver.receive_bounding_box_tensor().to_objects(0.5)

for obj in objects:
    print(obj.class_label.name, obj.confidence)

Output topic names vary by export. They are discovered after initialize() with pipeline.get_topics() (see Pipeline fundamentals).

C#

using DenkFlow;

using var license = HubLicenseSource.FromPat("YOUR-PAT");
using var pipeline =
    Pipeline.FromDenkflow("path/to/model.denkflow", license);
pipeline.Initialize();

using var receiver = pipeline.Subscribe(
    "bounding_box_filter_node/filtered_bounding_boxes");
using var image = ImageTensor.FromFile("path/to/image.jpg");

pipeline.Publish("/image", image);
pipeline.Run(TimeSpan.FromSeconds(8));

using var tensor = receiver.ReceiveBoundingBoxTensor();
var objects = tensor.ToObjects(0.5f);

foreach (var obj in objects)
    Console.WriteLine($"{obj.ClassLabel.Name}: {obj.Confidence}");

Pipeline.FromDenkflow accepts an explicit LicenseSource; it does not have Python's PAT shorthand parameters. Publish clones the native image tensor, so image remains valid after publishing and is disposed by the caller.

Output topic names vary by export. Discover them after Initialize() with pipeline.GetTopics().

C / C++

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "denkflow.h"

static void handle_error(enum DenkflowResult code, const char* fn) {
    printf("%s: %d", fn, (int)code);
    if (code != DenkflowResult_Ok) {
        char* buffer = (char*)malloc(DENKFLOW_ERROR_BUFFER_SIZE);
        denkflow_get_last_error(buffer);
        printf(" (%s)\n", buffer);
        free(buffer);
        exit(EXIT_FAILURE);
    }
    printf("\n");
}

int main(void) {
    DenkflowPipeline* pipeline = NULL;
    DenkflowInitializedPipeline* initialized_pipeline = NULL;
    DenkflowHubLicenseSource* license_source = NULL;
    DenkflowImageTensor* image_tensor = NULL;
    DenkflowReceiverTensor* receiver = NULL;
    DenkflowBoundingBoxTensor* tensor = NULL;
    DenkflowBoundingBoxResults* results = NULL;

    const char* pat = "YOUR-PAT";
    const char* denkflow_path = "path/to/model.denkflow";
    const char* image_path = "path/to/image.jpg";

    handle_error(
        denkflow_hub_license_source_from_pat(&license_source, pat, NULL, NULL),
        "denkflow_hub_license_source_from_pat"
    );
    handle_error(
        denkflow_pipeline_from_denkflow(&pipeline, denkflow_path, (void*)license_source),
        "denkflow_pipeline_from_denkflow"
    );
    handle_error(denkflow_initialize_pipeline(&initialized_pipeline, &pipeline), "denkflow_initialize_pipeline");
    handle_error(
        denkflow_initialized_pipeline_subscribe(
            &receiver,
            initialized_pipeline,
            "bounding_box_filter_node/filtered_bounding_boxes"
        ),
        "denkflow_initialized_pipeline_subscribe"
    );
    handle_error(denkflow_image_tensor_from_file(&image_tensor, image_path), "denkflow_image_tensor_from_file");
    handle_error(
        denkflow_initialized_pipeline_publish_tensor(initialized_pipeline, "/image", (void **)&image_tensor),
        "denkflow_initialized_pipeline_publish_tensor"
    );
    handle_error(denkflow_initialized_pipeline_run(initialized_pipeline, 8000), "denkflow_initialized_pipeline_run");
    handle_error(denkflow_receiver_receive_bounding_box_tensor(&tensor, receiver), "denkflow_receiver_receive_bounding_box_tensor");
    handle_error(denkflow_bounding_box_tensor_to_objects(&results, tensor, 0.5f), "denkflow_bounding_box_tensor_to_objects");

    for (int i = 0; i < results->bounding_boxes_length; ++i) {
        printf("%s: %f\n", results->bounding_boxes[i].class_label.name, results->bounding_boxes[i].confidence);
    }

    denkflow_free_object((void**)&results);
    denkflow_free_object((void**)&tensor);
    denkflow_free_object((void**)&receiver);
    denkflow_free_object((void**)&initialized_pipeline);
    denkflow_free_object((void**)&license_source);
    return 0;
}

When to use this flow

The classic flow is the basis for custom pipeline construction from .denkmodel files and for reading intermediate topics that SimplifiedPipeline does not expose. In-memory input formats are covered in Creating ImageTensors.