nirs4all.pipeline.config.context module

Context classes for pipeline execution.

This module provides typed context components that replace the Dict[str, Any] context pattern used throughout the pipeline system. It separates three distinct concerns:

1. DataSelector: Immutable data selection parameters for dataset.x() and dataset.y()

2. PipelineState: Mutable pipeline state that evolves through transformations

3. StepMetadata: Metadata for controller coordination and step tracking

4. ExecutionContext: Composite context with custom data extensibility

5. ArtifactProvider: Interface for providing artifacts during prediction replay

The separation enables: - Type safety throughout the codebase - Clear interfaces between components - Better testability - Explicit controller communication - Future extensibility via custom dict

Example

>>> selector = DataSelector(partition="train", processing=[["raw"]])
>>> state = PipelineState(y_processing="numeric")
>>> metadata = StepMetadata(keyword="transform")
>>> context = ExecutionContext(selector=selector, state=state, metadata=metadata)
>>> new_context = context.with_partition("test")

class nirs4all.pipeline.config.context.BestChainEntry(model_name: str, avg_val_score: float, branch_steps: list, best_params: dict, metric: str, pipeline_id: str = '', config_name: str = '')

Bases: object

Best preprocessing chain for a model, accumulated during CV execution.

During cross-validation, BranchController tracks the best-performing preprocessing chain per unique model. After all variants finish, the refit pass uses these entries directly instead of querying the store.

Variables:

• model_name (str) – Human-readable model name (e.g. "PLS", "Ridge").

• avg_val_score (float) – Average validation score across CV folds.

• branch_steps (list) – Expanded branch steps (preprocessing + model steps).

• best_params (dict) – Best hyperparameters from finetuning (Optuna).

• metric (str) – Metric name used for scoring (e.g. "rmse").

• pipeline_id (str) – Source CV pipeline ID for the winning chain.

• config_name (str) – Source CV config name for the winning chain.

avg_val_score: float

best_params: dict

branch_steps: list

config_name: str = ''

metric: str

model_name: str

pipeline_id: str = ''

class nirs4all.pipeline.config.context.DataSelector(partition: str | None = 'all', processing: list[list[str]] = <factory>, layout: str = '2d', concat_source: bool = True, fold_id: int | None = None, include_augmented: bool = False, y: str | None = None, branch_id: int | None = None, branch_path: list[int] = <factory>, branch_name: str | None = None, tag_filters: dict[str, ~typing.Any]=<factory>, _extra: dict[str, ~typing.Any]=<factory>)

Bases: MutableMapping

Mutable data selection parameters for dataset operations.

This class replaces the dict-based Selector pattern used by dataset.x() and dataset.y(). It implements the MutableMapping protocol, so it can be used as a dictionary. It supports arbitrary keys via an internal dict to allow flexibility.

Variables:

• partition (str | None) – Data partition to select (“train”, “test”, “all”, “val”)

• processing (list[list[str]]) – List of processing chains (one per data source)

• layout (str) – Data layout for X retrieval (“2d”, “3d”, “4d”)

• concat_source (bool) – Whether to concatenate multiple sources

• fold_id (int | None) – Optional fold identifier for cross-validation

• include_augmented (bool) – Whether to include augmented samples

• y (str | None) – Optional target processing version (e.g. “numeric”, “scaled”)

• branch_id (int | None) – Optional branch identifier for pipeline branching (0-indexed)

• branch_path (list[int]) – List of branch indices for nested branching (e.g., [0, 2] for branch 2 inside branch 0). Empty list means pre-branch/shared artifacts.

• branch_name (str | None) – Optional human-readable branch name for tracking

• tag_filters (dict[str, Any]) – Dict of tag column names to filter conditions. Used to filter samples based on tag values computed by TagController.

Example

>>> selector = DataSelector(partition="train", processing=[["raw"]])
>>> selector["y"] = "scaled"  # Direct modification
>>> selector["custom_key"] = "value"  # Arbitrary keys supported
>>> print(selector["partition"])

__delitem__(key: str) → None

Delete extra key or set field to None.

__getitem__(key: str) → Any

Get field value or extra key.

__iter__() → Iterator[str]

Iterate over non-None fields and extra keys.

__len__() → int

Count of non-None fields and extra keys.

__setitem__(key: str, value: Any) → None

Set field value or extra key.

branch_id: int | None = None

branch_name: str | None = None

branch_path: list[int]

concat_source: bool = True

copy() → DataSelector

Create a deep copy of the selector.

fold_id: int | None = None

include_augmented: bool = False

layout: str = '2d'

partition: str | None = 'all'

processing: list[list[str]]

tag_filters: dict[str, Any]

with_augmented(include_augmented: bool) → DataSelector

Create new selector with updated include_augmented flag.

Parameters:

include_augmented – Whether to include augmented samples

Returns:

New DataSelector with updated include_augmented

with_branch(branch_id: int | None = None, branch_name: str | None = None, branch_path: list[int] | None = None) → DataSelector

Create new selector with updated branch information.

Parameters:

• branch_id – Branch identifier (0-indexed).

• branch_name – Human-readable branch name

• branch_path – List of branch indices for nested branching

Returns:

New DataSelector with updated branch info

with_fold(fold_id: int | None) → DataSelector

Create new selector with updated fold_id.

Parameters:

fold_id – New fold identifier

Returns:

New DataSelector with updated fold_id

with_layout(layout: str) → DataSelector

Create new selector with updated layout.

Parameters:

layout – New layout value

Returns:

New DataSelector with updated layout

with_partition(partition: str | None) → DataSelector

Create new selector with updated partition.

Parameters:

partition – New partition value

Returns:

New DataSelector with updated partition

with_processing(processing: list[list[str]]) → DataSelector

Create new selector with updated processing chains.

Parameters:

processing – New processing chains

Returns:

New DataSelector with updated processing

with_tag_filter(tag_name: str, condition: Any) → DataSelector

Create new selector with an additional tag filter.

Tag filters are used to filter samples based on tag column values. Multiple tag filters can be combined by calling this method multiple times.

Parameters:

• tag_name – Name of the tag column to filter on

• condition – Filter condition. Supported formats: - Boolean: True, False - exact match - Comparison string: “> 0.8”, “<= 50”, “== 1”, “!= 0” - Range string: “0..50” (inclusive), “50..” (open end), “..50” (open start) - List of values: [“a”, “b”, “c”] - matches any value in list - Callable: lambda x: x > 0.8 - custom predicate

Returns:

New DataSelector with added tag filter

Example

>>> selector = DataSelector(partition="train")
>>> selector = selector.with_tag_filter("is_outlier", False)  # Keep non-outliers
>>> selector = selector.with_tag_filter("cluster_id", [1, 2])  # Only clusters 1 and 2

y: str | None = None

class nirs4all.pipeline.config.context.LoaderArtifactProvider(loader: Any, trace: Any | None = None)

Bases: ArtifactProvider

Artifact provider backed by an ArtifactLoader.

Wraps an ArtifactLoader to provide artifacts on-demand with lazy loading and caching. Used when loading from a manifest for prediction.

Variables:

• loader – The underlying ArtifactLoader

• trace – Optional ExecutionTrace for step-to-artifact mapping

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

Get a single artifact for a step.

If trace is available, uses trace to find artifact IDs. Otherwise, uses loader’s step-based lookup.

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_artifact_by_chain(chain_path: str) → Any | None

Get artifact by V3 chain path.

Parameters:

chain_path – Full operator chain path (e.g., “s1.MinMaxScaler>s3.PLS”)

Returns:

Artifact object or None if not found

get_artifacts_for_chain_prefix(chain_prefix: str) → list[tuple[str, Any]]

Get all artifacts matching a chain path prefix.

Parameters:

chain_prefix – Chain path prefix to match

Returns:

List of (chain_path, artifact_object) tuples

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

• branch_id – Optional branch ID filter

• source_index – Optional source index filter for multi-source pipelines

• substep_index – Optional substep index filter (not used in loader provider)

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_refit_artifact(step_index: int, branch_path: list[int] | None = None) → Any | None

Get the refit (“final”) artifact for a step.

Checks the execution trace for a fold artifact keyed "final".

Parameters:

• step_index – 1-based step index.

• branch_path – Optional branch path filter.

Returns:

The refit model object, or None if not found.

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.config.context.PipelineState(y_processing: str = 'numeric', step_number: int = 0, mode: str = 'train')

Bases: object

Mutable pipeline state that evolves through execution.

This class tracks state that changes as the pipeline executes: - Y transformation state (e.g., “encoded_LabelEncoder_001”) - Current step number in execution

Unlike DataSelector, this is mutable because state must evolve.

Variables:

• y_processing (str) – Current y transformation identifier

• step_number (int) – Current step number (1-indexed)

• mode (str) – Execution mode (“train”, “predict”, “explain”)

Example

>>> state = PipelineState(y_processing="numeric")
>>> state.step_number = 2  # Mutable update
>>> state.y_processing = "encoded_LabelEncoder_001"

copy() → PipelineState

Create a deep copy of this state.

Returns:

Deep copy of PipelineState

mode: str = 'train'

step_number: int = 0

y_processing: str = 'numeric'

class nirs4all.pipeline.config.context.StepMetadata(keyword: str = '', step_id: str = '', augment_sample: bool = False, add_feature: bool = False, replace_processing: bool = False, target_samples: list[int] = <factory>, target_features: list[int] = <factory>)

Bases: object

Metadata for controller coordination and step tracking.

This class handles: - Controller coordination flags (augment_sample, add_feature, replace_processing) - Step identification (step_id, keyword) - Target specification for augmentation operations

These are ephemeral flags set/cleared between steps for controller communication.

Variables:

• keyword (str) – Step keyword (e.g., “model”, “transform”)

• step_id (str) – Step identifier (e.g., “001”, “002”)

• augment_sample (bool) – Flag for sample augmentation mode

• add_feature (bool) – Flag for feature augmentation mode

• replace_processing (bool) – Flag to replace processing chains

• target_samples (list[int]) – Target sample IDs for augmentation

• target_features (list[int]) – Target feature indices for augmentation

Example

>>> metadata = StepMetadata(keyword="transform", step_id="001")
>>> metadata.augment_sample = True
>>> metadata.target_samples = [42]

add_feature: bool = False

augment_sample: bool = False

copy() → StepMetadata

Create a deep copy of this metadata.

Returns:

Deep copy of StepMetadata

keyword: str = ''

replace_processing: bool = False

reset_ephemeral_flags() → None

Reset ephemeral flags after step execution.

Clears augment_sample, add_feature, replace_processing flags and target lists to prevent leakage between steps.

step_id: str = ''

target_features: list[int]

target_samples: list[int]