◆ FORGE Suite
GitHubMechanical Neuroimaging Lab · Univ. of Delaware
FORGE Studio

Appearance

Python Backend ​

Python modules are called via a persistent forge_dispatcher.py session. The forge_studio package provides the full API for logging, progress, convergence, and image preview.

Module Structure ​

Every Python module must export a run() function:

python
def run(config_path: str) -> list[str]:
    ...
    return [output_path]
  • config_path — path to the JSON config written by FORGE Studio.
  • Return value — list of absolute output file paths. These become the inputs for the next pipeline stage.

Minimal Example ​

python
"""Minimal FORGE Studio Python module."""
import os
import numpy as np
from forge_studio import forge_log, forge_read_config

def run(config_path: str) -> list[str]:
    config = forge_read_config(config_path)
    output_dir = config["output_dir"]

    forge_log("info", "starting")

    result = np.zeros((64, 64, 32), dtype=np.float32)  # replace with real recon
    out_path = os.path.join(output_dir, "result.npy")
    np.save(out_path, result)

    forge_log("info", "done, wrote %s", os.path.basename(out_path))
    return [out_path]

Iterative Reconstruction Example ​

python
import os
import numpy as np
from forge_studio import (
    forge_log, forge_read_config, forge_param,
    forge_progress, forge_progress_done, forge_convergence,
)

def run(config_path: str) -> list[str]:
    config     = forge_read_config(config_path)
    output_dir = config["output_dir"]
    input_path = config["input_paths"][0]

    n_iter    = forge_param(config, "n_iter",    default=100)
    step_size = forge_param(config, "step_size", default=0.01)

    forge_log("info", "starting: n_iter=%d step_size=%.4f", n_iter, step_size)

    x = np.zeros((64, 64, 32), dtype=np.complex64)
    for i in range(1, n_iter + 1):
        # --- your reconstruction update here ---
        error_norm = 1.0 / i
        penalty    = 0.5 / i

        forge_progress("iter", i, n_iter, "Iterating",
                       postfix=f"err={error_norm:.3g}")
        forge_convergence("iter", i, error_norm, penalty)

    forge_progress_done("iter")

    out_path = os.path.join(output_dir, "recon.npy")
    np.save(out_path, x)
    forge_log("info", "done")
    return [out_path]

API Reference ​

forge_read_config(config_path: str) -> dict ​

Loads and validates the step config JSON. Automatically initialises the session state so all other forge_* functions work without extra setup.

python
config = forge_read_config(config_path)

config["job_id"]       # str — job identifier
config["step_id"]      # str — step identifier
config["stage_index"]  # int — 0-based stage number
config["input_paths"]  # list[str] — input files from previous stage
config["output_dir"]   # str — directory to write outputs
config["input_format"] # str — "twix" | "ismrmrd" | "mat" | "jld2" | "npy" | "npz"
config["params"]       # dict — user-configured parameter values
# Optional acquisition context (when available):
config["channels"]     # int — number of receive channels
config["platform"]     # str — e.g. "VE11E", "XA61"

forge_param(config: dict, name: str, default=None) ​

Reads a named parameter from config["params"]. Returns default if the key is absent, empty string, or empty list.

python
n_iter = forge_param(config, "n_iter", default=100)

forge_log(level: str, message: str, *args) ​

Emits a log event visible in the Monitor tab console.

python
forge_log("info",  "Starting calibration")
forge_log("warn",  "Low SNR detected: %d channels", n_channels)
forge_log("error", "Convergence failed at iter %d", i)

Levels: "debug", "info", "warn", "error".

When running outside FORGE Studio (e.g. in a script), messages are printed to stderr.

forge_progress(bar_id, current, max_val, label, postfix="") ​

Reports progress for a named progress bar. The first call creates the bar; subsequent calls update it.

python
forge_progress("recon", i, n_iter, "Reconstructing", postfix=f"err={err:.3g}")

forge_progress_done(bar_id: str) ​

Marks a progress bar complete.

python
forge_progress_done("recon")

forge_convergence(bar_id, iter, error_norm, penalty, **kwargs) ​

Emits a metrics event plotted on the Monitor tab convergence chart.

python
forge_convergence("recon", i, error_norm, penalty, step_size=alpha)

**kwargs are additional named metrics shown in the chart.

forge_image_preview(bar_id: str, iter: int, volume: np.ndarray) ​

Emits orthogonal slice previews visible in the Monitor tab (if image preview is enabled).

python
# volume shape: (nx, ny, nz), any numeric dtype
forge_image_preview("recon", i, x)

forge_is_session() -> bool ​

Returns True when the module is running inside a FORGE Studio session. Use for conditional behaviour:

python
if forge_is_session():
    forge_log("info", "running in FORGE Studio")
else:
    print("standalone mode")

Running Outside FORGE Studio ​

All forge_* functions degrade gracefully when called outside a session:

  • forge_log → prints to stderr.
  • forge_progress → prints a text progress line to stderr.
  • forge_convergence → prints metrics to stderr.
  • forge_image_preview → no-op.

This means you can test your module directly:

bash
# Write a minimal config JSON first, then:
python3 my_module.py /tmp/test_config.json

Or use the forge-session dev tool →.

Package Location ​

The forge_studio package lives at resources/python/forge_studio/ within the app bundle. It is added to sys.path automatically by the dispatcher. You do not need to install it.

For development, install the project's Python venv:

bash
uv sync
.venv/bin/python3 -c "from forge_studio import forge_log; print('ok')"

FORGE Studio