app/docs/code/nodes.md

Writing node code

A Function node runs Python in a worker process. Write its source in the canvas, or declare functions in your project with @node and fluksio sync.

def process(temperature, setpoint=21.0):
    return {"heat": temperature < setpoint}

Declare temperature as an input port, heat as a Boolean output port, and setpoint as a setting. Now you've got a tiny thermostat.

The rules

• Define process. Without it, the worker accepts exactly one public function defined in the source. Multiple candidates produce an error.
• Arguments arrive by name from input ports and settings. A setting cannot share a name with a port. See Where values come from.
• Return a dictionary keyed by output port. Undeclared keys and values of the wrong type fail validation. With @node(provides=Port("heat", "bool")), the SDK also accepts a bare value and wraps it for that single port.
• Use JSON-compatible values. For files, use artifacts. For persistent state, use flow state; worker globals are not a reliable store.

Inside a node, import fluksio exposes reporting and artifact helpers. It does not expose the engine's internal modules. Other imports come from the node's Python environment; see Packages.

Producing values over time

Yield dictionaries to publish outputs before the node finishes:

import math


def process(lr, steps):
    loss = 1.0
    for step in range(steps):
        loss = math.exp(-lr * (step + 1))
        yield {"loss": loss}
    return {"final_loss": loss}

Declare loss as a streaming float port:

{"name": "loss", "dtype": "float", "stream": true}

Each yield is type-checked and published immediately. The final return is the node's result; without a return, the last yield is the result. A run keeps numeric emissions as metric series and waits for the result before running downstream nodes. A live flow can also trigger downstream nodes from emissions.

fluksio.emit

Use emit when a callback produces the value:

import fluksio


def on_epoch_end(epoch, logs):
    fluksio.emit(loss=logs["loss"])

Pass this callback to your training framework. It publishes through the same declared ports as yield. Outside a node, emit does nothing, so ordinary tests can still call your code.

Bytes: artifacts

Save files or bytes with save_artifact, then return the reference through an artifact port:

import fluksio


def process():
    return {
        "report": fluksio.save_artifact(
            b"Experiment complete\n", "report.txt", "text/plain"
        )
    }

Another node can read it with fluksio.load_artifact(report), which returns a local file path. These two helpers require a running node; for tests, keep the underlying calculation in ordinary Python functions.

Artifacts are stored by their SHA-256 digest, so identical bytes share storage. Use fluksio artifacts to list and download a run's files.

Media

image, audio and video ports carry artifact references with a matching media type. For example, save JPEG bytes with media_type="image/jpeg" and return the reference through an image port. See Payload types.

For transient frames, save_artifact(..., volatile=True) uses an in-memory ring and streams bytes to connected screens. New frames eventually replace old ones. A volatile artifact returned as a run result is copied to persistent storage; emitted frames alone are not recorded as run artifacts.

Making clips playable on an LVGL panel

A wall panel that renders dashboards natively, such as an ESP32-P4 running openpanel, decodes only H.264 Constrained Baseline (in software, about 29 fps at 640×360) and MJPEG. It refuses clips straight from a phone or camera, which are usually Main or High profile at 1080p or more. Re-encode those with a node that takes a video input port clip and has a video output port panel_clip:

import os
import subprocess
import tempfile

import fluksio

# H.264 Constrained Baseline, at most 640x360 and 15 fps, mono AAC.
PANEL = [
    "-c:v", "libx264", "-profile:v", "baseline", "-level:v", "3.1",
    "-pix_fmt", "yuv420p", "-fpsmax", "15",
    "-vf", "scale=w=640:h=360:force_original_aspect_ratio=decrease:force_divisible_by=2",
    "-preset", "veryfast", "-crf", "23", "-g", "30",
    "-c:a", "aac", "-ac", "1", "-b:a", "64k", "-movflags", "+faststart",
]


def process(clip):
    source = fluksio.load_artifact(clip)
    with tempfile.TemporaryDirectory() as tmp:
        target = os.path.join(tmp, "panel.mp4")
        subprocess.run(
            ["ffmpeg", "-nostdin", "-loglevel", "error", "-i", source, *PANEL, target],
            check=True,
            timeout=600,
        )
        name = os.path.splitext(clip.get("name") or "clip")[0] + "-panel.mp4"
        return {"panel_clip": fluksio.save_artifact(target, name, "video/mp4")}

Use it only for clips headed for such a panel. Browsers play the originals, and the result plays there too, just smaller. The Docker image ships ffmpeg; a remote worker needs it on its PATH.

Printing

print and fluksio.logger.info(...) are captured in the node logs. The worker retains the first 16,384 characters per call. Use logs to debug and output ports for numbers you want to plot or compare.

Errors

Exceptions appear on the node, with a short message and a traceback. In a live flow, one failed execution does not stop the whole flow. In a batch run, node failure makes the run fail. The canvas keeps the last error visible after recovery until you acknowledge it.

Timeouts

timeout is the allowed number of seconds without an emission. A yield or emit resets the deadline; initial compilation and imports are excluded.

There is no timeout by default. Set one on a node where silence means stuck, or use FLOW_NODE_TIMEOUT for the instance default. timeout=0 explicitly disables it for a node. A worker disappearing fails its active call without waiting for that deadline.

Running a node somewhere else

Set a device label to select an attached worker:

{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}

require is the default and waits for a worker with that label. prefer uses the local engine when none is attached. Code is compiled on the selected machine, which needs the packages your node imports. See Remote workers for setup and resource requests.

Sharing code between flows

Promote a node's source to the shared library from its panel. Other flows can reference that source, and edits apply to all of them. Shared sources live in _lib/ in the instance's flow repository.

For code-defined flows, reuse the same decorated function. use(fn, ...) changes its wiring or settings for one flow:

use(fit, wire={"dataset": "augmented"}, epochs=3)

Packages

The startup banner shows the interpreter used for local nodes:

Environment	How to add packages
Fluksio installed in an existing venv	Use pip or uv in that venv, then fluksio sync to refresh workers. Modules is read-only.
Managed node venv	Edit the manifest in Modules and apply it.
Remote worker	Install packages in the interpreter selected by --python.

With NODE_VENV=auto, Fluksio adopts its existing venv unless this instance already has a managed one. Without an existing venv, it creates a managed environment. NODE_VENV=managed always requests one; the Docker image uses this setting.

In managed mode, the Modules manifest is installed with uv pip sync and versioned with the flows. Applying it refreshes local workers. The API accepts the same manifest:

curl -X POST "$FLUKSIO/modules/apply" -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"requirements": "numpy>=2\npandas\n"}'

Here $FLUKSIO is the API base URL ending in /api/v1, and $TOKEN is an access token; see The HTTP API.

Your own code as a package

fluksio sync generates an import shim pointing at your local project. A managed environment can also install your project from a Modules manifest:

-e /home/you/my-research
numpy>=2

Then a canvas node can import ordinary helpers from it:

from myresearch.training import train_model


def process(lr, epochs):
    return {"score": train_model(lr, epochs)}

Editable installs expose changed files, but warm local workers still hold imports in memory. Apply the Modules manifest or run fluksio sync after edits. Attached workers start a process per call and read the files afresh.

Paths are resolved on the machine running the node. A remote worker or container needs its own copy or installation of your project; syncing a shim does not ship the package.

A worked example

Getting started: data science contains a complete runnable flow. The public repository also includes examples/myresearch/, with separate preparation, training and evaluation modules. From the repository root:

fluksio sync examples/myresearch
fluksio run train --lr 0.05 --wait

Start an instance first, as shown in the getting started guide.

See also

• Where a node's values come from
• Keeping state in a flow
• Runs: pipelines that finish
• Node types