# 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 does not ship
`ffmpeg` yet, so the node runs only where `ffmpeg` is on the `PATH`, such as a
[remote worker](workers.md).

## 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)