nirs4all.data.predictions module

Store-backed predictions management.

This module provides the Predictions facade for accumulating, ranking, filtering, and exporting prediction records. After the SQLite storage migration the class was rewritten to work exclusively with WorkspaceStore.

When a store is supplied (normal execution path), every call to add_prediction() buffers the record in memory and flush() writes the buffer to the database. Query methods (top, filter_predictions, get_best, …) delegate to the store’s columnar query engine.

When no store is supplied (lightweight / test usage), the class operates in a purely in-memory mode backed by a Polars DataFrame, preserving backward compatibility for code that creates Predictions() without a workspace.

class nirs4all.data.predictions.MergeReport(total_sources: int = 0, predictions_merged: int = 0, conflicts_resolved: int = 0, datasets_merged: list[str] = <factory>, errors: list[str] = <factory>)

Bases: object

Report returned by Predictions.merge_stores().

conflicts_resolved: int = 0

datasets_merged: list[str]

errors: list[str]

predictions_merged: int = 0

total_sources: int = 0

class nirs4all.data.predictions.PredictionResult

Bases: dict

Enhanced dictionary for a single prediction with convenience methods.

Extends standard dict with property accessors and methods for saving, evaluating, and summarizing predictions.

Features:

• Property accessors (id, model_name, dataset_name, etc.)

• save_to_csv() - save individual result

• eval_score() - compute metrics on-the-fly

• summary() - generate tab report

Examples

>>> result = PredictionResult({
...     "id": "abc123",
...     "dataset_name": "wheat",
...     "model_name": "PLS",
...     "y_true": [1, 2, 3],
...     "y_pred": [1.1, 2.2, 3.3]
... })
>>> result.model_name
'PLS'
>>> scores = result.eval_score(["rmse", "r2"])
>>> result.save_to_csv("results")

__repr__() → str

String representation showing key info.

__str__() → str

String representation showing key info.

property config_name: str

Get config name.

property dataset_name: str

Get dataset name.

eval_score(metrics: list[str] | None = None) → dict[str, Any]

Evaluate scores for this prediction using specified metrics.

Parameters:

metrics – List of metrics to compute (if None, returns all available metrics)

Returns:

Dictionary of metric names to scores. For aggregated results: {“train”: {…}, “val”: {…}, “test”: {…}} For single partition: {“rmse”: …, “r2”: …, …}

Examples

>>> scores = result.eval_score(["rmse", "r2", "mae"])
>>> # For aggregated: scores = {"train": {"rmse": 0.5}, "val": {...}, "test": {...}}
>>> # For single: scores = {"rmse": 0.5, "r2": 0.9}

property fold_id: str

Get fold ID.

property id: str

Get prediction ID.

property model_name: str

Get model name.

property op_counter: int

Get operation counter.

save_to_csv(path_or_file: str = 'results', filename: str | None = None) → None

Save prediction result to CSV file.

Parameters:

• path_or_file – Base path (folder) or complete file path (if ends with .csv)

• filename – Optional filename (if path_or_file is a folder)

Examples

>>> result.save_to_csv("output")  # Saves to output/{dataset}/{id}.csv
>>> result.save_to_csv("output/my_result.csv")  # Saves to output/my_result.csv
>>> result.save_to_csv("output", "my_result.csv")  # Saves to output/my_result.csv

property step_idx: int

Get pipeline step index.

summary() → str

Generate a summary tab report for this prediction.

Works with both aggregated and non-aggregated prediction results.

Returns:

Formatted string with tab report

Examples

>>> report = result.summary()
>>> print(report)

class nirs4all.data.predictions.PredictionResultsList(predictions: Sequence[dict[str, Any] | PredictionResult] | None = None)

Bases: list[PredictionResult]

List container for PredictionResult objects with batch operations.

Extends standard list with prediction-specific batch functionality.

Features:

• save() - batch CSV export

• get() - retrieve by ID

• filter() - chain filtering

• Iterator support

Examples

>>> results = PredictionResultsList([result1, result2, result3])
>>> results.save("output/predictions.csv")
>>> best = results.get("abc123")
>>> len(results)
3

__repr__() → str

String representation showing count and brief info.

get(prediction_id: str) → PredictionResult | None

Get a prediction by its ID.

Parameters:

prediction_id – The ID of the prediction to retrieve

Returns:

PredictionResult if found, None otherwise

Examples

>>> result = results.get("abc123")

save(path: str = 'results', filename: str | None = None) → None

Save all predictions to a single CSV file with structured headers.

CSV Structure:

• Line 1: dataset_name

• Line 2: model_classname + model_id

• Line 3: fold_id

• Line 4: partition

• Lines 5+: prediction data (y_true, y_pred columns)

Parameters:

• path – Base directory path (default: “results”)

• filename – Optional filename (if None, auto-generated from first prediction)

Examples

>>> results.save("output")
>>> results.save("output", "my_predictions.csv")