nirs4all.pipeline.bundle package

Submodules

Module contents

Bundle module for nirs4all pipeline exports.

This module provides functionality for exporting trained pipelines as standalone prediction bundles that can be used for deployment, sharing, or archival.

Supported Formats:

  • .n4a: Full bundle (ZIP archive with artifacts and metadata)

  • .n4a.py: Portable Python script with embedded artifacts (base64)

Key Components:

  • BundleGenerator: Creates prediction bundles from trained pipelines

  • BundleLoader: Loads bundles for prediction or inspection

  • BundleMetadata: Bundle metadata and manifest structure

  • PortablePredictor: Standalone predictor from .n4a.py bundle

Usage (Export):
>>> from nirs4all.pipeline import PipelineRunner
>>>
>>> runner = PipelineRunner()
>>> predictions, _ = runner.run(pipeline, dataset)
>>> best_pred = predictions.top(n=1)[0]
>>>
>>> # Export to .n4a bundle
>>> runner.export(best_pred, "exports/wheat_model.n4a")
>>>
>>> # Export to portable Python script
>>> runner.export(best_pred, "exports/wheat_model.n4a.py", format='n4a.py')
Usage (Predict from bundle):
>>> from nirs4all.pipeline import PipelineRunner
>>>
>>> runner = PipelineRunner()
>>> y_pred, preds = runner.predict("exports/wheat_model.n4a", X_new)
>>>
>>> # Or using standalone script (no nirs4all dependency)
>>> # python wheat_model.n4a.py input.csv output.csv
Design Principles:

  1. Self-Contained: Bundle includes all artifacts and metadata

  2. Portable: Minimal dependencies for prediction

  3. Versioned: Bundle format is versioned for compatibility

  4. Deterministic: Same bundle -> same predictions

class nirs4all.pipeline.bundle.BundleFormat(value)[source]

Bases: str, Enum

Supported bundle export formats.

N4A

Full ZIP bundle with all artifacts and metadata

N4A_PY

Portable Python script with embedded artifacts

N4A = 'n4a'
N4A_PY = 'n4a.py'
class nirs4all.pipeline.bundle.BundleGenerator(workspace_path: str | Path, verbose: int = 0)[source]

Bases: object

Generate standalone prediction bundles from trained pipelines.

This class exports trained pipelines to bundle formats that can be used for deployment, sharing, or archival without requiring the original workspace or full nirs4all installation.

workspace_path

Path to the workspace root

resolver

PredictionResolver for resolving prediction sources

verbose

Verbosity level for logging

Example

>>> generator = BundleGenerator(workspace_path)
>>> generator.export(best_prediction, "model.n4a")
>>>
>>> # Export to portable script
>>> generator.export(best_prediction, "model.n4a.py", format="n4a.py")
export(source: Dict[str, Any] | str | Path, output_path: str | Path, format: str | BundleFormat = BundleFormat.N4A, include_metadata: bool = True, compress: bool = True) → Path[source]

Export a prediction source to a bundle.

Parameters:

  • source – Prediction source (prediction dict, folder path, etc.)

  • output_path – Path for the output bundle

  • format – Bundle format (‘n4a’ or ‘n4a.py’)

  • include_metadata – Whether to include full metadata in bundle

  • compress – Whether to compress artifacts (for .n4a format)

Returns:

Path to the created bundle

Raises:
class nirs4all.pipeline.bundle.BundleLoader(bundle_path: str | Path)[source]

Bases: object

Load and use prediction bundles.

Provides functionality for loading .n4a bundles, extracting metadata, and running predictions.

bundle_path

Path to the bundle file

metadata

Bundle metadata

Type:

nirs4all.pipeline.bundle.loader.BundleMetadata | None

trace

Execution trace (if available)

Type:

nirs4all.pipeline.trace.execution_trace.ExecutionTrace | None

pipeline_config

Pipeline configuration

Type:

Dict[str, Any]

fold_weights

Fold weights for CV ensemble

Type:

Dict[int, float]

artifact_provider

Provider for artifacts

Type:

nirs4all.pipeline.bundle.loader.BundleArtifactProvider | None

Example

>>> loader = BundleLoader("model.n4a")
>>> print(f"Pipeline: {loader.metadata.pipeline_uid}")
>>> print(f"Preprocessing: {loader.metadata.preprocessing_chain}")
>>> y_pred = loader.predict(X_new)
artifact_provider: BundleArtifactProvider | None
fold_weights: Dict[int, float]
get_chain_for_artifact(artifact_key: str) → OperatorChain | None[source]

Get the operator chain for an artifact from the bundle.

Parameters:

artifact_key – Artifact key (e.g., “step_1”, “step_4_fold0”)

Returns:

OperatorChain for the artifact or None if not found

get_merged_chains(import_context_chain: OperatorChain, step_offset: int = 0) → Dict[str, OperatorChain][source]

Get all artifact chains merged with an import context chain.

Used when importing a bundle into another pipeline. Each artifact’s chain is prefixed with the import context chain.

Parameters:

  • import_context_chain – Chain from the importing pipeline context

  • step_offset – Step offset to apply to bundle steps

Returns:

Dict mapping artifact keys to merged chains

get_partitioner_routing(step_index: int | None = None) → Dict[str, Any] | None[source]

Get partitioner routing info for a specific step or all steps.

Parameters:

step_index – Specific step index, or None for all

Returns:

Routing info dict or None

get_required_metadata_columns() → List[str][source]

Get the metadata columns required for prediction routing.

Returns:

List of column names needed for routing, empty if no routing needed.

get_step_info() → List[Dict[str, Any]][source]

Get information about steps in the bundle.

Returns:

List of step info dictionaries

has_partitioner_routing() → bool[source]

Check if the bundle has metadata partitioner routing info.

Returns:

True if the bundle contains partitioner routing configuration.

import_artifacts_to_registry(registry: ArtifactRegistry, import_context_chain: OperatorChain | None = None, step_offset: int = 0, new_pipeline_id: str | None = None) → Dict[str, str][source]

Import bundle artifacts into an artifact registry.

Registers all artifacts from this bundle into the target registry, optionally merging with an import context chain for proper V3 tracking.

Parameters:

  • registry – Target ArtifactRegistry to import into

  • import_context_chain – Optional chain from import context to prefix

  • step_offset – Step offset for imported artifacts

  • new_pipeline_id – New pipeline ID for imported artifacts

Returns:

Dict mapping original artifact keys to new artifact IDs

metadata: BundleMetadata | None
pipeline_config: Dict[str, Any]
predict(X: ndarray, branch_path: List[int] | None = None) → ndarray[source]

Run prediction on input data.

Applies all preprocessing steps and model(s) from the bundle. Supports branching pipelines, meta-models (stacking), and CV ensembles.

Parameters:

  • X – Input features as numpy array

  • branch_path – Optional branch path filter for multi-branch pipelines

Returns:

Predictions as numpy array

predict_with_metadata(X: ndarray, metadata: Dict[str, ndarray], fallback_branch: int | None = None) → ndarray[source]

Run prediction with metadata-based sample routing.

For bundles with metadata partitioner branches, this method routes each sample to the appropriate branch based on its metadata value. Each sample is processed by the transformers and models from its matching branch.

Parameters:

  • X – Input features as numpy array (n_samples, n_features)

  • metadata – Dict mapping column names to value arrays. Must include the column used for partitioning during training.

  • fallback_branch – Optional branch ID to use for samples with unknown metadata values. If None, raises error for unknowns.

Returns:

Predictions as numpy array

Raises:

ValueError – If required metadata column is missing or samples have unknown values without fallback.

Example

>>> loader = BundleLoader("model.n4a")
>>> X_new = np.random.randn(100, 500)
>>> metadata = {"site": np.array(["A"]*50 + ["B"]*50)}
>>> y_pred = loader.predict_with_metadata(X_new, metadata)
to_resolved_prediction() → Any[source]

Convert bundle to ResolvedPrediction for use with Predictor.

Returns:

ResolvedPrediction instance

trace: ExecutionTrace | None
class nirs4all.pipeline.bundle.BundleMetadata(bundle_format_version: str = '1.0', nirs4all_version: str = '', created_at: str = '', pipeline_uid: str = '', source_type: str = '', model_step_index: int | None = None, fold_strategy: str = 'weighted_average', preprocessing_chain: str = '', trace_id: str | None = None, original_manifest: Dict[str, ~typing.Any]=<factory>, partitioner_routing: Dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Metadata for a prediction bundle.

Contains information about the bundle format, source, and contents.

bundle_format_version

Version of the bundle format

Type:

str

nirs4all_version

Version of nirs4all that created the bundle

Type:

str

created_at

ISO timestamp of bundle creation

Type:

str

pipeline_uid

UID of the source pipeline

Type:

str

source_type

Type of source that was exported

Type:

str

model_step_index

Index of the model step

Type:

int | None

fold_strategy

Strategy for combining fold predictions

Type:

str

preprocessing_chain

Summary of preprocessing steps

Type:

str

trace_id

ID of the execution trace (if available)

Type:

str | None

original_manifest

Subset of original manifest metadata

Type:

Dict[str, Any]

partitioner_routing

Routing info for metadata partitioner branches

Type:

Dict[str, Any]

bundle_format_version: str = '1.0'
created_at: str = ''
fold_strategy: str = 'weighted_average'
classmethod from_dict(data: Dict[str, Any]) → BundleMetadata[source]

Create BundleMetadata from dictionary.

Parameters:

data – Dictionary from bundle manifest.json

Returns:

BundleMetadata instance

model_step_index: int | None = None
nirs4all_version: str = ''
original_manifest: Dict[str, Any]
partitioner_routing: Dict[str, Any]
pipeline_uid: str = ''
preprocessing_chain: str = ''
source_type: str = ''
trace_id: str | None = None