nirs4all.pipeline.execution.preprocessing module

Fold-local fitting and replay of learned legacy feature preprocessing.

class nirs4all.pipeline.execution.preprocessing.FoldPreprocessedModel(model: Any, preprocessing: PreprocessingReplay)[source]

Bases: object

Persist a model together with its fold’s fitted preprocessing.

model: Any

predict(X: ndarray, **kwargs: Any) → Any[source]

predict_proba(X: ndarray, **kwargs: Any) → Any[source]

preprocessing: PreprocessingReplay

class nirs4all.pipeline.execution.preprocessing.FoldPreprocessing(dataset: SpectroDataset, replay: PreprocessingReplay, active: bool = False, unsupported_reason: str | None = None)[source]

Bases: object

Keep raw features and stage specifications until a fold is fitted.

active: bool = False

classmethod capture(dataset: SpectroDataset, context: ExecutionContext) → FoldPreprocessing[source]

dataset: SpectroDataset

prepare_fold(dataset: SpectroDataset, model_context: ExecutionContext, train_ids: Any) → PreprocessingReplay[source]

raw_features(ids: Any, layout: str = '2d') → ndarray[source]

replay: PreprocessingReplay

unsupported_reason: str | None = None

class nirs4all.pipeline.execution.preprocessing.PreprocessingReplay(shapes: list[tuple[int, int]], processings: list[list[str]], headers: list[list[str]], units: list[str], context: ~nirs4all.pipeline.config.context.ExecutionContext, input_step: int = 0, steps: list[tuple[int, ~typing.Any]] = <factory>, artifacts: dict[int, list[tuple[str, ~typing.Any]]] = <factory>, source_artifacts: dict[tuple[int, int], list[tuple[str, ~typing.Any]]] = <factory>)[source]

Bases: object

Replay fitted controllers using feature schemas, without training rows.

artifacts: dict[int, list[tuple[str, Any]]]

classmethod capture(dataset: SpectroDataset, context: ExecutionContext) → PreprocessingReplay[source]

context: ExecutionContext

headers: list[list[str]]

input_step: int = 0

processings: list[list[str]]

shapes: list[tuple[int, int]]

source_artifacts: dict[tuple[int, int], list[tuple[str, Any]]]

property start_step: int

steps: list[tuple[int, Any]]

transform(X: ndarray) → ndarray[source]

units: list[str]

nirs4all.pipeline.execution.preprocessing.feature_step(step: Any) → tuple[bool, bool][source]: Identify feature stages and target-dependent stages, including containers.

nirs4all.pipeline.execution.preprocessing.observe_feature_step(step: Any, dataset: SpectroDataset, context: ExecutionContext, runtime: Any) → bool[source]: Capture stages even inside branches; return whether replay owns this stage.