Getting started

DataFlowArc runs data-processing pipelines over MQTT. A pipeline is a set of nodes; each node is one function block (FB) from the library with its own configuration. Nodes talk to each other only through MQTT topics, so every connection on the canvas is a topic subscription. What the messages contain is up to the function blocks: numbers, JSON, binary data.

1. Create a pipeline on the dashboard (you must be signed in).
2. Open it, pick a broker in the top bar, and drag function blocks from the library onto the canvas.
3. Connect the broker or another node to a node's input. Click a line to edit the topics.
4. Press Sync to write the topics into the node configurations, then Start all.

The canvas

• Mouse wheel scrolls the canvas, Shift + wheel scrolls sideways, Ctrl + wheel zooms.
• Library: search by name, description, category or tag (several words must all match), sort by name, most used or newest. Hide it with ; the thin strip brings it back.
• Select a library tile and press Delete (or the small ×) to remove an unused function block.
• Double-click a node for its settings: topics, configuration, code (editing code creates a new version Name_V2), and logs.
• Audit switches the monitoring sidecar on or off per node and chooses which topics are watched.
• A node changed after it was started is marked out of sync: it still runs with the old configuration until you restart it.

Line and node status

Every line is judged on its own topics: a node fed by two sources has two lines, and one source read by several nodes has a line per reader, each with its own colour.

Function block package format

A function block is imported as a .zip of one folder (or of the files directly):

MyBlock/
├── my_block.py            the script DataFlowArc starts (with --config_private and --config_public)
├── helpers.py             any other modules, data files, subfolders…
├── requirements.txt       libraries the block needs (installed in its own virtual environment)
└── public_config.json     configuration template, copied into each node

Import it with Import in the function block library (admins). You can give it a name, a library group and search tags there; group and tags can be changed later in the block's details.

What DataFlowArc adds itself

On import DataFlowArc writes model_manifest.json into the block. Do not put one in the zip: it is ignored, and it is left out when a block is exported. The node wrapper reads it to know how to start the block, and every node records which block (and version) it is based on.

public_config.json: input and output sections

The template marks its input and output sections by name: a template postfix (_TEMPLATE, _TEMPL, _TMPL, _TPL, any case) plus in or out in the name, e.g. MQTT_IN_TEMPLATE, Input_tmpl, mqtt-out-templ. When a node is added, the template section becomes the node's real section without the postfix (MQTT_OUT_TEMPLATE → MQTT_OUT), and that name is stored in the node's metadata. Your script reads the real name.

Sections without a postfix still work: a section with in/out in its name is used as it is, and otherwise the content decides (TopicsToSubscribe = input, ModifySubtopics = output). The private config (broker host, port, login) comes from the broker chosen for the pipeline; never put credentials in a block.

Topics and ModifySubtopics

DataFlowArc does not prescribe a topic layout; any MQTT topics work. A node's output topic is made from the input topic, level by level (levels are the parts between /). ModifySubtopics has one entry per level: "" keeps the input level, text replaces it. The first empty level becomes the node id, so every node's output is easy to find and never collides with its input:

input   demo/building1/room2/temperature
modify  ["", "", "", "temperature_f"]
output  demo_Minimal/building1/room2/temperature_f

When you connect a node to another node, Sync applies the source node's ModifySubtopics to the topics it subscribes to, so the reader subscribes to exactly what the source publishes. In the line editor <ch> stands for “any value at this level” and becomes + in the configuration.

Simplest function block

The library block Example_Minimal (group Examples) scales a value: out = in × Gain + Offset. It accepts a plain number or JSON with a "value" field, so it works on the demo temperatures (Celsius → Fahrenheit) and on most simple sensor feeds. This is all of it:

import argparse, json, os
import paho.mqtt.client as mqtt

NODE_ID = os.getenv("NODE_NAME", "Example_Minimal")

def out_topic(topic, modify):
    parts = topic.split("/")
    return "/".join(NODE_ID if (i == 0 and m == "") else (m or p)
                    for i, (p, m) in enumerate(zip(parts, modify + [""] * len(parts))))

def client_for(creds):
    c = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    if creds.get("userId"):
        c.username_pw_set(creds["userId"], creds.get("password"))
    c.connect_async(creds["host"], int(creds["port"]))
    return c

ap = argparse.ArgumentParser()
ap.add_argument("--config_private"); ap.add_argument("--config_public")
args = ap.parse_args()
cfg, creds = json.load(open(args.config_public)), json.load(open(args.config_private))
gain, offset = cfg["Processing"]["Gain"], cfg["Processing"]["Offset"]
modify = cfg["MQTT_OUT"]["ModifySubtopics"]
sub, pub = client_for(creds["MQTT_IN"]), client_for(creds["MQTT_OUT"])

def on_connect(client, userdata, flags, rc, properties=None):
    for t in cfg["MQTT_IN"]["TopicsToSubscribe"]:       # subscribe again after every reconnect
        client.subscribe(t)

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)                      # 21.4  or  {"value": 21.4, ...}
    if isinstance(data, dict):
        data["value"] = data["value"] * gain + offset
    else:
        data = data * gain + offset
    pub.publish(out_topic(msg.topic, modify), json.dumps(data))

sub.on_connect, sub.on_message = on_connect, on_message
pub.loop_start()
sub.loop_forever()

Its public_config.json (note the _TEMPLATE output section):

{
  "MQTT_IN":           { "TopicsToSubscribe": ["demo/building1/+/temperature"] },
  "MQTT_OUT_TEMPLATE": { "ModifySubtopics": ["", "", "", "temperature_f"] },
  "Processing":        { "Gain": 1.8, "Offset": 32.0, "Unit": "degF" }
}

Function blocks that need extra libraries

Each function block gets its own virtual environment (envs/fb_<name>), built from its requirements.txt on the first start. A block can therefore use libraries that are not installed in DataFlowArc itself, and different blocks can use different versions.

Example_Rainflow uses the rainflow package (fatigue cycle counting):

paho-mqtt
numpy<2
rainflow

The first start takes a little longer while the environment is installed; the node shows Preparing and the log shows the installation. The deployment package installs the same requirements.

Plotting

Add a webplot node and connect it to the stream you want to see. Open the graph from the node () or from Graphs in the top bar; several graphs can be open side by side. No extra ports are needed: graphs go through DataFlowArc.

"Graph": {
  "ylim": "auto",          // or [-250, 250] for a fixed range
  "TimeToCover": 10        // x axis length in seconds
}

webplot draws streams in a metadata + data format: for each stream a retained JSON …/metadata topic (sampling rate, samples per message, unit) and a binary …/data topic (a 28-byte header: header length, version, seconds, nanoseconds, sample counter; then float32 samples). The demo vibration signal, DemoProcessor and Example_Rainflow use this format; a block that publishes other payloads needs its own plot block.

Example pipeline: demo broker → Example_Rainflow → webplot with "ylim": "auto" shows the largest load cycle of each chunk.

Log categories

The log viewer groups lines by the tag at the start, e.g. [MQTT_IN], [MQTT_OUT], [RAINFLOW]. Add your own tags in the viewer to filter on them. Lines containing ERROR, Exception or Traceback are marked as errors.

Rights

Who can see pipelines is chosen by an admin on the Admin page:

• Owners, editors and admins always see their pipelines. A hidden pipeline is hidden everywhere: dashboard, canvas, status, logs and graphs.
• Signed-in users can create and import pipelines; the creator is the owner.
• Starting, stopping, editing, exporting and viewing code or configuration need edit rights on that pipeline (owner, editor, or admin). Others see the canvas but not the code or configuration text.
• Admins manage users and rights, brokers, and the function block library.
• Guests without a Google account: an admin creates an access code on the Admin page (valid 1, 7 or 30 days) and sends it. The guest signs in at /auth/guest (or the small code link next to Sign in) and is then a normal user that the admin gives rights or the admin role. Withdrawing the code signs the guest out at once.

Deployment package

The ⋯ menu on the dashboard (or Deploy on the canvas) builds a zip that runs the pipeline without DataFlowArc: no UI, no monitoring sidecar, no plot nodes. Unpack it on the target and run cd pipeline && ./run_pipeline.sh; it creates the virtual environments and starts the nodes (a node that stops is restarted after 3 s, logs go to logs/). Broker credentials are not included: copy credentials.example.json to credentials.json and fill it in.