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.

The header follows the ports¶

On the canvas, the parameters of process and the node's input ports and settings are kept in step, in both directions.

• Changing ports or settings in the node panel rewrites the def process(...) line and leaves the rest of the code as it is. Input ports come first, in their order, then the settings. A parameter that stays keeps its annotation and default.
• Saving the code reads that line back. A name the node already has keeps its kind. A new parameter becomes a setting when it has a default, which becomes its value, or when it comes after a setting. Any other new parameter becomes an input port. Renaming a single one renames the port (keeping its message) or the setting (keeping its value). Removing a parameter removes its port or setting once you leave the editor, so one deleted mid-edit and typed back keeps its message.
• Shared code, code from fluksio sync, and a header with a decorator, *args, **kwargs, a bare * or / are left alone in both directions.

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
import imageio_ffmpeg

# 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)
    ffmpeg = imageio_ffmpeg.get_ffmpeg_exe()
    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. Add imageio-ffmpeg on the Modules screen first. It brings a static ffmpeg build for Linux on x86 and ARM, so nothing has to be installed into the image.

Speech to and from a panel¶

openpanel publishes a 16 kHz mono WAV on its mic port after its wake word, and plays any audio reference sent to its speaker port (WAV, MP3 or AAC). Two Function nodes close that loop with local models. faster-whisper turns speech into text and piper turns text into speech.

The listening node has an audio input port audio and a str output port text:

from functools import cache
from pathlib import Path

from faster_whisper import WhisperModel

import fluksio

MODELS = Path.home() / ".cache" / "voice"


@cache
def _model(size):
    # Downloaded on the first call, then held by the worker between calls.
    return WhisperModel(
        size, device="cpu", compute_type="int8", download_root=str(MODELS)
    )


def process(audio, model="base", language="en"):
    # vad_filter skips silence, which Whisper would otherwise hear as "You".
    segments, _ = _model(model).transcribe(
        fluksio.load_artifact(audio), language=language or None, vad_filter=True
    )
    return {"text": " ".join(segment.text.strip() for segment in segments)}

The speaking node has a str input port text and an audio output port speech:

import io
import wave
from functools import cache
from pathlib import Path

from piper import PiperVoice
from piper.download_voices import download_voice

import fluksio

MODELS = Path.home() / ".cache" / "voice"


@cache
def _voice(name):
    # Downloaded on the first call, then held by the worker between calls.
    MODELS.mkdir(parents=True, exist_ok=True)
    download_voice(name, MODELS)
    return PiperVoice.load(MODELS / f"{name}.onnx")


def process(text, voice="en_US-lessac-medium"):
    buffer = io.BytesIO()
    with wave.open(buffer, "wb") as wav:
        _voice(voice).synthesize_wav(text, wav)
    speech = fluksio.save_artifact(buffer.getvalue(), "speech.wav", "audio/wav")
    return {"speech": speech}

model, language and voice are settings. tiny is faster than base and less accurate. An empty language lets Whisper guess it for each clip. voice takes any name from the piper voices, such as de_DE-thorsten-medium.

To wire the loop in a flow home:

1. Declare a flow input heard of type audio with origin panel.
2. Add a Device node for the panel, publishing mic to heard and following reply on speaker.
3. Bind the listening node's audio to heard and its text to said.
4. Have your own nodes read said and publish an answer as a str, such as answer.
5. Bind the speaking node's text to answer and its speech to reply.

Binding the speaking node's text straight to said makes the panel repeat what it heard, which tests the loop before there is any logic in it.

Both packages go into the node's environment, as described in Packages. In a managed environment, list them in Modules:

faster-whisper
av<19
piper-tts

av<19 holds PyAV back, since faster-whisper 1.2.1 cannot open files with PyAV 19. In your own venv or on a remote worker, pip install faster-whisper "av<19" piper-tts does the same.

The models land in ~/.cache/voice on the machine running the node. The Docker image keeps its home directory inside the container, so a recreated container downloads them again. Point MODELS at a directory under /data to keep them on the volume. An attached worker starts a process per call and loads the model each time.

What it costs, measured on a 2016 server CPU (Xeon E5-2683 v4):

Each model downloads once, on its node's first call. Loading it takes about a second for Whisper and four seconds for piper. Memory is the worker's peak while it holds the model. These figures come from piper's own speech fed back into Whisper. A room microphone is harder, on tiny most of all.

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 worker label to select an attached worker:

{"id": "train", "worker": "gpu", "worker_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:

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 resolved with uv pip compile, installed with uv pip sync and versioned with the flows. Listing a package brings its dependencies along. 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\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