nirs4all.pipeline.resolver module

Prediction Resolver - Unified resolution of prediction sources.

This module provides the PredictionResolver class which normalizes any prediction source (dict, folder, Run, artifact_id, bundle) to the components needed for prediction replay:

• Minimal pipeline (subset of steps)

• Artifact map (artifacts keyed by step index)

• Execution trace (for deterministic replay)

• Fold strategy (for CV ensemble averaging)

The resolver is controller-agnostic: it doesn’t know about specific controller types, but provides the artifacts and trace needed by any controller to replay a prediction.

Supported Source Types:

• prediction dict: Most common, from a previous run’s Predictions object

• folder path: Path to a pipeline directory containing manifest.yaml

• Run object: Best prediction from a Run

• artifact_id: Direct artifact reference (e.g., “0001:4:all”)

• bundle: Exported prediction bundle (.n4a file)

• trace_id: Execution trace reference (e.g., “trace:xyz789”)

Example

>>> resolver = PredictionResolver(workspace_path)
>>> resolved = resolver.resolve(best_prediction)
>>> # Use resolved.minimal_pipeline, resolved.artifact_provider
>>> y_pred = execute_prediction(resolved, new_data)

class nirs4all.pipeline.resolver.FoldStrategy(value)

Bases: str, Enum

Strategy for combining fold predictions in CV ensembles.

AVERAGE

Simple average of fold predictions

WEIGHTED_AVERAGE

Weighted average using fold weights

SINGLE

Use a single fold’s prediction

AVERAGE = 'average'

SINGLE = 'single'

WEIGHTED_AVERAGE = 'weighted_average'

class nirs4all.pipeline.resolver.PredictionResolver(workspace_path: str | Path, runs_dir: str | Path | None = None)

Bases: object

Resolves any prediction source to executable components.

This class provides a unified interface for resolving prediction sources to the components needed for prediction replay, regardless of the source type (dict, folder, run, artifact_id, bundle).

The resolver is designed to be controller-agnostic: it doesn’t know about specific controller types, but provides artifacts and trace that any controller can use.

workspace_path

Root workspace directory

runs_dir

Directory containing run outputs

Example

>>> resolver = PredictionResolver(workspace_path)
>>> resolved = resolver.resolve(best_prediction)
>>> # Execute using resolved components
>>> provider = resolved.artifact_provider
>>> artifacts = provider.get_artifacts_for_step(step_index=1)

resolve(source: Dict[str, Any] | str | Path | Any, verbose: int = 0) → ResolvedPrediction

Resolve any prediction source to executable components.

Detects the source type and delegates to the appropriate resolver.

Parameters:

• source – Prediction source (dict, folder path, Run, artifact_id, bundle)

• verbose – Verbosity level for logging

Returns:

ResolvedPrediction with all components for replay

Raises:

• ValueError – If source type cannot be determined or resolved

• FileNotFoundError – If referenced files/directories don’t exist

class nirs4all.pipeline.resolver.ResolvedPrediction(source_type: ~nirs4all.pipeline.resolver.SourceType = SourceType.UNKNOWN, minimal_pipeline: ~typing.List[~typing.Any] = <factory>, artifact_provider: ~nirs4all.pipeline.config.context.ArtifactProvider | None = None, trace: ~nirs4all.pipeline.trace.execution_trace.ExecutionTrace | None = None, fold_strategy: ~nirs4all.pipeline.resolver.FoldStrategy = FoldStrategy.WEIGHTED_AVERAGE, fold_weights: ~typing.Dict[int, float] = <factory>, model_step_index: int | None = None, target_model: ~typing.Dict[str, ~typing.Any] = <factory>, pipeline_uid: str = '', run_dir: ~pathlib.Path | None = None, manifest: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Normalized prediction source ready for execution.

Contains all components needed to replay a prediction: - minimal_pipeline: Subset of steps needed for this prediction - artifact_provider: Provider for artifacts by step index - trace: Execution trace for deterministic replay - fold_strategy: How to combine fold predictions (for CV) - fold_weights: Per-fold weights (for weighted average)

source_type

Type of the original source

Type:

nirs4all.pipeline.resolver.SourceType

minimal_pipeline

List of pipeline steps needed for replay

Type:

List[Any]

artifact_provider

Provider for step artifacts

Type:

nirs4all.pipeline.config.context.ArtifactProvider | None

trace

ExecutionTrace if available

Type:

nirs4all.pipeline.trace.execution_trace.ExecutionTrace | None

fold_strategy

Strategy for combining folds

Type:

nirs4all.pipeline.resolver.FoldStrategy

fold_weights

Per-fold weights for weighted averaging

Type:

Dict[int, float]

model_step_index

Index of the model step

Type:

int | None

target_model

Target model metadata for filtering

Type:

Dict[str, Any]

pipeline_uid

Pipeline unique identifier

Type:

str

run_dir

Path to run directory

Type:

pathlib.Path | None

manifest

Full manifest dictionary (for metadata access)

Type:

Dict[str, Any]

artifact_provider: ArtifactProvider | None = None

fold_strategy: FoldStrategy = 'weighted_average'

fold_weights: Dict[int, float]

get_preprocessing_chain() → str

Get the preprocessing chain summary.

Returns:

Preprocessing chain string (e.g., “SNV>SG>MinMax”) or empty

has_fold_artifacts() → bool

Check if fold-specific artifacts are available.

Returns:

True if this is a CV ensemble with multiple folds

has_trace() → bool

Check if execution trace is available.

Returns:

True if trace is available for deterministic replay

manifest: Dict[str, Any]

minimal_pipeline: List[Any]

model_step_index: int | None = None

pipeline_uid: str = ''

run_dir: Path | None = None

source_type: SourceType = 'unknown'

target_model: Dict[str, Any]

trace: ExecutionTrace | None = None

class nirs4all.pipeline.resolver.SourceType(value)

Bases: str, Enum

Type of prediction source.

PREDICTION

Dictionary from Predictions object

FOLDER

Path to pipeline folder

RUN

Run object (best prediction from run)

ARTIFACT_ID

Direct artifact reference string

BUNDLE

Exported .n4a bundle file

TRACE_ID

Execution trace reference

MODEL_FILE

Direct model file (.joblib, .pkl, .h5, .pt, etc.)

UNKNOWN

Unrecognized source type

ARTIFACT_ID = 'artifact_id'

BUNDLE = 'bundle'

FOLDER = 'folder'

MODEL_FILE = 'model_file'

PREDICTION = 'prediction'

RUN = 'run'

TRACE_ID = 'trace_id'

UNKNOWN = 'unknown'