Apps API#

KonfAI Apps package stable workflows behind a simpler interface than the raw TRAIN / PREDICTION / EVALUATION commands.

Local and remote app runners#

class konfai_apps.KonfAIApp(app, download, force_update)[source]

Bases: AbstractKonfAIApp

Local runner for KonfAI applications.

This class executes inference/evaluation/uncertainty/fine-tuning locally by: - building a dataset folder structure expected by KonfAI - installing the appropriate model/config assets (HF or local directory) - invoking KonfAI predictor/evaluator/trainer functions - collecting outputs into a user-defined output folder

The public methods (infer/evaluate/uncertainty/fine_tune) are wrapped by run_distributed_app, which runs each operation in an isolated temporary workspace.

static symlink(src, dst)[source]

Create a symlink from dst pointing to src, with safe replacement.

If dst already exists: - directories are removed (unless they are symlinks) - files are unlinked

On platforms or filesystems that do not support symlinks (e.g. some Windows environments), this falls back to copying: - directories via copytree - files via copy2

Parameters:

  • src (Path) – Source file or directory.

  • dst (Path) – Destination symlink path.

Return type:

None

infer(inputs, output=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/konfai/checkouts/stable/docs/source/Output'), ensemble=0, ensemble_models=[], tta=0, mc=0, patch_size=None, batch_size=None, config_overrides=None, uncertainty=False, prediction_file='Prediction.yml', gpu=None, cpu=None, quiet=False, tmp_dir=None)[source]

Run inference locally for the given inputs.

Steps: 1. Build Dataset/ from inputs 2. Install inference assets (models/config) via self.app_repository.install_inference 3. Call konfai.predictor.predict(…) 4. Copy generated predictions into output if they exist

Return type:

None

Notes

  • Executes inside an isolated temporary workspace (via run_distributed_app).

  • GPU defaults to cuda_visible_devices().

evaluate(inputs, gt, output=PosixPath('Output'), mask=None, evaluation_file='Evaluation.yml', gpu=None, cpu=None, quiet=False, tmp_dir=None)[source]

Run evaluation locally against ground-truth.

Steps: 1. Build Dataset/ from inputs and gt 2. Ensure masks exist (provided or generated) 3. Install evaluation assets via self.app_repository.install_evaluation 4. Call konfai.evaluator.evaluate(…) 5. Copy evaluation outputs into output

Return type:

None

Notes

  • Runs inside an isolated workspace (run_distributed_app).

  • GPU defaults to cuda_visible_devices().

uncertainty(inputs, output=PosixPath('Output'), uncertainty_file='Uncertainty.yml', gpu=None, cpu=None, quiet=False, tmp_dir=None)[source]

Run uncertainty estimation locally.

Steps: 1. Validate that inputs are multi-component inference stacks 2. Install uncertainty assets via self.app_repository.install_uncertainty 3. Call evaluator with an explicit output directory (./Uncertainties) 4. Copy uncertainty results into output

Return type:

None

Notes

  • Runs inside an isolated workspace (run_distributed_app).

  • GPU defaults to cuda_visible_devices().

pipeline(inputs, gt, output=PosixPath('Output'), ensemble=0, ensemble_models=[], tta=0, mc=0, patch_size=None, batch_size=None, config_overrides=None, prediction_file='Prediction.yml', mask=None, evaluation_file='Evaluation.yml', uncertainty=True, uncertainty_file='Uncertainty.yml', gpu=None, cpu=None, quiet=False, tmp_dir=None)[source]

Run a full pipeline locally: inference → evaluation → uncertainty.

This is a convenience method that orchestrates multiple stages and organizes outputs into subfolders:

  • <output>/Predictions

  • <output>/Evaluations

  • <output>/Uncertainties

Behavior:

  • always runs inference

  • runs evaluation only if gt is provided

  • runs uncertainty only if uncertainty=True

Return type:

None

fine_tune(dataset, name='Finetune', output=PosixPath('Output'), epochs=10, it_validation=1000, models=[], gpu=None, cpu=None, quiet=False, config_file='Config.yml', lr=None, tmp_dir=None)[source]

Fine-tune one or several checkpoints of the app locally.

Return type:

None

Steps: 1. Install training assets/config and resolve the selected checkpoint(s) via

self.app_repository.install_fine_tune.

  1. Link the user dataset into ./Dataset.

  2. For each selected checkpoint: sanitize it to weights-only, write a per-model config with a distinct train_name, run konfai.trainer.train(…) in resume mode, then copy the produced checkpoint back into the output app.

The output directory is left as a clean, resolvable app bundle (app.json + config + code + fine-tuned checkpoint(s)); training artifacts (Checkpoints/Statistics) are kept out of it.

Notes

  • Runs inside an isolated workspace (run_distributed_app). The CLI passes tmp_dir=output (the workspace IS the output dir); other callers get the bundle files copied into output.

  • Fine-tuning requires a CUDA GPU when the loss relies on GPU-only components.

  • models selects which checkpoint(s) to fine-tune (default: the first available).

class konfai_apps.KonfAIAppClient(app, remote_server)[source]

Bases: AbstractKonfAIApp

Client-side helper to submit jobs to a remote KonfAI app server.

This class implements: - job submission to endpoints like /apps/{app}/{action} - streaming logs via SSE (/jobs/{job_id}/logs) - result retrieval (/jobs/{job_id}/result) - remote job termination (/jobs/{job_id}/kill)

It is intended to mirror the server’s execution model: submit → stream logs → download results → (optional) kill on interruption.

stream_logs(job_id, connect_timeout=60, read_timeout=600)[source]

Stream server-side job logs using Server-Sent Events (SSE).

This method connects to:

GET /jobs/{job_id}/logs

It prints each received SSE “data:” message to stdout. The stream ends when one of the terminal markers is received: - “__DONE__” - “__ERROR__ …”

Parameters:

  • job_id (str) – Remote job identifier returned by the server.

  • connect_timeout (int) – Max seconds to wait for the initial connection.

  • read_timeout (int) – Max seconds to wait for new bytes before considering the stream stalled.

Raises:

RuntimeError – For auth errors, forbidden access, stream stalls, or other request failures.

kill_job(job_id, timeout_s=60)[source]

Request termination of a remote job.

Sends:

POST /jobs/{job_id}/kill

If successful, prints a confirmation message.

Parameters:

  • job_id (str) – Remote job identifier.

  • timeout_s (float) – Timeout (seconds) for the kill request.

Raises:
Return type:

None

download_result(job_id, out_dir, connect_timeout=60, read_timeout=600, max_wait_s=600)[source]

Download and unpack the result archive for a remote job.

Polls:

GET /jobs/{job_id}/result

Server behavior expected: - HTTP 202: result not ready → keep polling - HTTP 200: returns a zip archive → download then unpack

The downloaded archive is saved as:

<out_dir>/result.zip

Then extracted into out_dir.

Parameters:

  • job_id (str) – Remote job identifier.

  • out_dir (Path) – Destination directory where the result is extracted.

  • connect_timeout (int) – Connection timeout for each poll attempt.

  • read_timeout (int) – Read timeout for each download attempt.

  • max_wait_s (int) – Maximum total time to wait for the result to become available.

Returns:

True if the result was successfully downloaded and extracted.

Return type:

bool

Raises:

  • TimeoutError – If the result does not become ready within max_wait_s.

  • RuntimeError – For request failures, auth issues, or download errors.

static run_remote_job(func)[source]

Decorator for KonfAIAppClient methods that submit work to the remote server.

The wrapped method is treated as an “action” endpoint. For example, wrapping infer() will call POST /apps/{self.app}/infer.

Behavior:

  1. Introspects the wrapped function signature to filter kwargs.

  2. Builds a multipart request containing file fields for inputs, ground truth, and masks, plus scalar fields for other parameters.

  3. Submits the job and retrieves a job_id.

  4. Streams logs until completion markers are received.

  5. Downloads and extracts results into the requested output directory.

  6. On SIGINT or SIGTERM, triggers cleanup and kills the remote job if it is still running.

  7. Always closes local file handles.

Signal handling#

Uses ensure_finally_on_signals() so that SIGINT or SIGTERM raises CancelProcess. This ensures the finally block runs and the remote kill request is attempted when needed.

Notes

  • The decorated methods are “declarative”: they do not implement logic themselves and typically contain only pass.

  • Output directory is taken from the wrapped method’s output argument.

returns:

Wrapped method that performs remote submission + monitoring + download.

rtype:

Callable[..., None]

Remote server helpers#

class konfai.RemoteServer(host, port, token)[source]

Bases: object

Connection settings for a remote KonfAI Apps server.

get_headers()[source]

Return the HTTP headers required to talk to the remote server.

Return type:

dict[str, str]

get_url()[source]

Return the base URL of the remote server.

Return type:

str

konfai.check_server(remote_server, timeout_s=2.0)[source]

Check whether a remote KonfAI Apps server is reachable and healthy.

Parameters:

  • remote_server (RemoteServer) – Remote server connection settings.

  • timeout_s (float) – HTTP timeout used for the health check.

Returns:

A boolean success flag and a human-readable status message.

Return type:

tuple[bool, str]

konfai.get_available_devices(remote_server=None, timeout_s=2.0)[source]

Return the available GPU indices and their display names.

Parameters:

  • remote_server (RemoteServer | None) – Remote server to query instead of the local machine.

  • timeout_s (float) – HTTP timeout used for remote requests.

Returns:

Available device indices and the corresponding device names.

Return type:

tuple[list[int], list[str]]

konfai.get_ram(remote_server=None, timeout_s=2.0)[source]

Return used and total RAM in gigabytes.

Parameters:

  • remote_server (RemoteServer | None) – Remote server to query instead of the local machine.

  • timeout_s (float) – HTTP timeout used for remote requests.

Returns:

Used RAM and total RAM in gigabytes.

Return type:

tuple[float, float]

konfai.get_vram(devices, remote_server=None, timeout_s=2.0)[source]

Return used and total VRAM in gigabytes for the selected devices.

Parameters:

  • devices (list[int]) – GPU indices to inspect.

  • remote_server (RemoteServer | None) – Remote server to query instead of the local machine.

  • timeout_s (float) – HTTP timeout used for remote requests.

Returns:

Used VRAM and total VRAM in gigabytes.

Return type:

tuple[float, float]

See also#