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:

Self-Contained: Bundle includes all artifacts and metadata
Portable: Minimal dependencies for prediction
Versioned: Bundle format is versioned for compatibility
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:

ValueError – If format is not supported
FileNotFoundError – If source cannot be resolved

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

trace: Execution trace (if available)

pipeline_config: Pipeline configuration

fold_weights: Fold weights for CV ensemble

artifact_provider: Provider for artifacts

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)

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

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

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: ~typing.Dict[str, ~typing.Any] = <factory>, partitioner_routing: ~typing.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