281 lines
9.4 KiB
Markdown
281 lines
9.4 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 Docker image ships `ffmpeg`;
|
||
a [remote worker](workers.md) 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:
|
||
|
||
```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 installed with `uv pip sync` and
|
||
versioned with the flows. 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\npandas\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)
|