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 / 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.