nirs4all.pipeline.bundle.loader module

Bundle Loader - Load and predict from exported bundles.

This module provides the BundleLoader class for loading legacy Python prediction bundles (.n4a format) and running predictions without the original workspace. This replay path is deprecated; DAG-ML/Core archives use their native replay sessions and never execute BundleLoader model artifacts.

The loader supports:

• Loading bundle metadata and structure

• Extracting artifacts on-demand

• Building artifact provider for prediction

• Resolving bundles for PredictionResolver

• Full support for branching pipelines

• Full support for meta-models (stacking)

• CV ensemble prediction with fold weights

Example

>>> from nirs4all.pipeline.bundle import BundleLoader
>>>
>>> loader = BundleLoader("model.n4a")
>>> print(loader.metadata.pipeline_uid)
>>> y_pred = loader.predict(X_new)
>>>
>>> # Or via PredictionResolver
>>> resolver = PredictionResolver(workspace_path)
>>> resolved = resolver.resolve("model.n4a")  # Automatically uses BundleLoader

exception nirs4all.pipeline.bundle.loader.BundleArtifactLoadError

Bases: RuntimeError

An indexed Python artifact could not be read or deserialized.

An unavailable or damaged fitted transformer must never be treated as an absent optional step: continuing could silently change model predictions.

class nirs4all.pipeline.bundle.loader.BundleArtifactProvider(bundle_path: str | Path, artifact_index: dict[str, str], fold_weights: dict[int, float] | None = None, step_info: dict[int, dict[str, Any]] | None = None, trace: ExecutionTrace | None = None)

Bases: ArtifactProvider

Artifact provider for bundles.

Provides artifacts from a loaded bundle, with lazy loading and caching. Supports branching pipelines and meta-models.

Variables:

• bundle_path – Path to the bundle file

• artifact_cache – Cache of loaded artifacts

• artifact_index – Index mapping step/fold to artifact filenames

• fold_weights – Fold weights for CV ensemble

• step_info – Information about each step (operator_type, branch_path, etc.)

get_artifact(step_index: int, fold_id: int | str | None = None) → Any | None

Get a single artifact for a step.

Parameters:

• step_index – 1-based step index

• fold_id – Optional fold ID for fold-specific artifacts

Returns:

Artifact object or None if not found

get_artifacts_for_step(step_index: int, branch_path: list[int] | None = None, branch_id: int | None = None, source_index: int | None = None, substep_index: int | None = None) → list[tuple[str, Any]]

Get all artifacts for a step.

Parameters:

• step_index – 1-based step index

• branch_path – Optional branch path filter

Returns:

List of (artifact_id, artifact_object) tuples

get_fold_artifacts(step_index: int, branch_path: list[int] | None = None) → list[tuple[int, Any]]

Get all fold-specific artifacts for a step.

Parameters:

• step_index – 1-based step index

• branch_path – Optional branch path filter

Returns:

List of (fold_id, artifact_object) tuples, sorted by fold_id

get_fold_weights() → dict[int, float]

Get fold weights for CV ensemble.

Returns:

Fold weights dictionary

get_meta_model_sources(step_index: int) → list[tuple[int, str]]

Get source model info for a meta-model step.

Parameters:

step_index – Step index of the meta-model

Returns:

List of (source_step_index, artifact_id) tuples

get_step_operator_type(step_index: int) → str | None

Get the operator type for a step.

Parameters:

step_index – 1-based step index

Returns:

Operator type string or None

has_artifacts_for_step(step_index: int) → bool

Check if artifacts exist for a step.

Parameters:

step_index – 1-based step index

Returns:

True if artifacts are available for this step

class nirs4all.pipeline.bundle.loader.BundleLoader(bundle_path: str | Path)

Bases: object

Load and use deprecated legacy Python prediction bundles.

Provides functionality for loading .n4a bundles, extracting metadata, and running predictions. DAG-ML/Core archive replay does not use this class.

Warning

Historical Python bundles contain joblib/pickle artifacts. Loading or predicting can execute producer-controlled Python code; only use a trusted producer and delivery channel. Checksums do not establish trust. Use validated Core Archive V2/N4MM for the non-pickle portable profile.

Variables:

• 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

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]

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

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]

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]]

Get information about steps in the bundle.

Returns:

List of step info dictionaries

has_partitioner_routing() → bool

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]

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

property named_outputs: tuple[str, ...]

Stable output binding IDs of a multi-output archive, empty for ordinary bundles.

predict(X: Any, branch_path: list[int] | None = None) → ndarray

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 a numpy array. For bundles carrying a relational replay manifest, a RawMultiSourceDataset staging object is also accepted and materialized through the saved representation plan before prediction.

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

Returns:

Predictions as numpy array

predict_output(name: str, X: Any) → ndarray

Replay exactly one explicitly named output from a multi-output archive.

predict_outputs(X: Any) → dict[str, ndarray]

Replay all independent outputs as a name-to-prediction mapping.

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

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

Convert bundle to ResolvedPrediction for use with Predictor.

Returns:

ResolvedPrediction instance

class nirs4all.pipeline.bundle.loader.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>, relation_replay_manifest: dict[str, ~typing.Any]=<factory>, retrain_lineage: dict[str, ~typing.Any]=<factory>)

Bases: object

Metadata for a prediction bundle.

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

Variables:

• bundle_format_version (str) – Version of the bundle format

• nirs4all_version (str) – Version of nirs4all that created the bundle

• created_at (str) – ISO timestamp of bundle creation

• pipeline_uid (str) – UID of the source pipeline

• source_type (str) – Type of source that was exported

• model_step_index (int | None) – Index of the model step

• fold_strategy (str) – Strategy for combining fold predictions

• preprocessing_chain (str) – Summary of preprocessing steps

• trace_id (str | None) – ID of the execution trace (if available)

• original_manifest (dict[str, Any]) – Subset of original manifest metadata

• partitioner_routing (dict[str, Any]) – Routing info for metadata partitioner branches

• relation_replay_manifest (dict[str, Any]) – Reference to the optional N9 relation replay manifest embedded in the bundle.

• retrain_lineage (dict[str, Any]) – Additive API-004 provenance for a natively retrained artifact, empty for ordinary prediction bundles.

bundle_format_version: str = '1.0'

created_at: str = ''

fold_strategy: str = 'weighted_average'

classmethod from_dict(data: dict[str, Any]) → BundleMetadata

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 = ''

relation_replay_manifest: dict[str, Any]

retrain_lineage: dict[str, Any]

source_type: str = ''

trace_id: str | None = None

nirs4all.pipeline.bundle.loader.load_bundle(bundle_path: str | Path) → BundleLoader

Convenience function to load a bundle.

Parameters:

bundle_path – Path to the .n4a bundle file

Returns:

BundleLoader instance