stroblmeandClaude Opus 5.5
50e6c033f1
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
408 lines
14 KiB
Markdown
408 lines
14 KiB
Markdown
# 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`](cli.md#fluksio-sync).
|
||
|
||
```python
|
||
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](../concepts/values.md).
|
||
- **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](../concepts/state.md); 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](#packages).
|
||
|
||
## Producing values over time
|
||
|
||
Yield dictionaries to publish outputs before the node finishes:
|
||
|
||
```python
|
||
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:
|
||
|
||
```json
|
||
{"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](../concepts/runs.md) 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:
|
||
|
||
```python
|
||
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:
|
||
|
||
```python
|
||
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`](cli.md#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](../reference/payload-types.md#image-audio-video).
|
||
|
||
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`:
|
||
|
||
```python
|
||
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 node needs `ffmpeg` on the
|
||
`PATH`. The Docker image includes it only when built with
|
||
[`WITH_FFMPEG=true`](../reference/configuration.md#image-options) in `.env`.
|
||
Otherwise run the node on a [remote worker](workers.md) that has it.
|
||
|
||
### 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](https://github.com/SYSTRAN/faster-whisper) turns speech into
|
||
text and [piper](https://github.com/OHF-Voice/piper1-gpl) turns text into
|
||
speech.
|
||
|
||
The listening node has an `audio` input port `audio` and a `str` output port
|
||
`text`:
|
||
|
||
```python
|
||
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`:
|
||
|
||
```python
|
||
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](https://huggingface.co/rhasspy/piper-voices),
|
||
such as `de_DE-thorsten-medium`.
|
||
|
||
To wire the loop in a flow `home`:
|
||
|
||
1. Declare a [flow input](../concepts/flows.md#flow-inputs) `heard` of type
|
||
`audio` with origin `panel`.
|
||
2. Under the panel's [I/O](../interface/dashboards.md#the-devices-own-io), wire
|
||
`mic` as `in` to `home.heard` and `speaker` as `out` to `home.reply`.
|
||
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](#packages). In a managed environment, list them in **Modules**:
|
||
|
||
```text
|
||
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](workers.md) starts a process per call and
|
||
loads the model each time.
|
||
|
||
What it costs, measured on a 2016 server CPU (Xeon E5-2683 v4):
|
||
|
||
| | Size | Memory | 3 s of speech, one core | All cores |
|
||
|---|---|---|---|---|
|
||
| Packages | 150 MB download, 460 MB installed | | | |
|
||
| Whisper `tiny` | 78 MB | 300 MB | 1.6 s | 0.7 s |
|
||
| Whisper `base` | 148 MB | 375 MB | 2.9 s | 1.2 s |
|
||
| piper `en_US-lessac-medium` | 63 MB | 250 MB | 0.6 s | 0.3 s |
|
||
|
||
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 device label to select an attached worker:
|
||
|
||
```json
|
||
{"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](workers.md) 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:
|
||
|
||
```python
|
||
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 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:
|
||
|
||
```sh
|
||
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](api.md).
|
||
|
||
### 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:
|
||
|
||
```text
|
||
-e /home/you/my-research
|
||
numpy>=2
|
||
```
|
||
|
||
Then a canvas node can import ordinary helpers from it:
|
||
|
||
```python
|
||
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](../getting-started/data-science.md) contains a
|
||
complete runnable flow. The public repository also includes
|
||
`examples/myresearch/`, with separate preparation, training and evaluation
|
||
modules. From the repository root:
|
||
|
||
```sh
|
||
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](../concepts/values.md)
|
||
- [Keeping state in a flow](../concepts/state.md)
|
||
- [Runs: pipelines that finish](../concepts/runs.md)
|
||
- [Node types](../reference/node-types.md)
|