xpectra.pipeline

Reusable training and prediction utilities for the FTIR notebooks.

class xpectra.pipeline.PreprocessConfig(label_column='type', sample_id_column='sample_id', unknown_study='kedzierski_2019_u', labeled_studies=('jung_2018', 'kedzierski_2019', 'frond_2021', 'villegas_camacho_2024_c4'), external_studies=('openspecy', 'simple', 'awi2024'), denoising_method='wavelet', baseline_method='aspls', normalization_method='spectral_moments', interpolate_method='zero', flat_windows=((1880.0, 1900.0), (2400.0, 2700.0)), exclude_regions=((0.0, 679.0), (3201.0, 5000.0)), interpolate_regions=((2300.0, 2380.0),), wn_min=680.0, wn_max=3100.0, resolution=2.0, descending=True, combine_method='pchip', derivative_window_length=15, derivative_polyorder=3, derivative_delta=1.0, random_state=42, test_size=0.2, n_jobs=-1)

Bases: object

Notebook preprocessing defaults, exposed as CLI-overridable settings.

Parameters:

• label_column (str)

• sample_id_column (str)

• unknown_study (str)

• labeled_studies (tuple[str, ...])

• external_studies (tuple[str, ...])

• denoising_method (str)

• baseline_method (str)

• normalization_method (str)

• interpolate_method (str)

• flat_windows (tuple[tuple[float, float], ...])

• exclude_regions (tuple[tuple[float, float], ...])

• interpolate_regions (tuple[tuple[float, float], ...])

• wn_min (float)

• wn_max (float)

• resolution (float)

• descending (bool)

• combine_method (str)

• derivative_window_length (int)

• derivative_polyorder (int)

• derivative_delta (float)

• random_state (int)

• test_size (float)

• n_jobs (int)

label_column: str = 'type'

sample_id_column: str = 'sample_id'

unknown_study: str = 'kedzierski_2019_u'

labeled_studies: tuple[str, ...] = ('jung_2018', 'kedzierski_2019', 'frond_2021', 'villegas_camacho_2024_c4')

external_studies: tuple[str, ...] = ('openspecy', 'simple', 'awi2024')

denoising_method: str = 'wavelet'

baseline_method: str = 'aspls'

normalization_method: str = 'spectral_moments'

interpolate_method: str = 'zero'

flat_windows: tuple[tuple[float, float], ...] = ((1880.0, 1900.0), (2400.0, 2700.0))

exclude_regions: tuple[tuple[float, float], ...] = ((0.0, 679.0), (3201.0, 5000.0))

interpolate_regions: tuple[tuple[float, float], ...] = ((2300.0, 2380.0),)

wn_min: float = 680.0

wn_max: float = 3100.0

resolution: float = 2.0

descending: bool = True

combine_method: str = 'pchip'

derivative_window_length: int = 15

derivative_polyorder: int = 3

derivative_delta: float = 1.0

random_state: int = 42

test_size: float = 0.2

n_jobs: int = -1

to_dict()

Return type:

dict[str, Any]

xpectra.pipeline.load_artifacts(models_dir, route, model_names=None)

Parameters:

• models_dir (str | Path)

• route (str)

• model_names (Iterable[str] | None)

Return type:

list[tuple[Path, dict[str, Any]]]

xpectra.pipeline.make_prediction_route_dataframes(input_df, routes=None, config=None, input_stage='raw', force_absorbance=False, absorbance_scale_factor=None, save_features_dir=None)

Parameters:

• input_df (DataFrame)

• routes (Iterable[str] | None)

• config (PreprocessConfig | None)

• input_stage (str)

• force_absorbance (bool)

• absorbance_scale_factor (float | None)

• save_features_dir (str | Path | None)

Return type:

dict[str, DataFrame]

xpectra.pipeline.predict_csv(input_csv, output_csv, routes=None, models_dir=None, model_names=None, config=None, input_stage='raw', force_absorbance=False, absorbance_scale_factor=None, include_probabilities=False, save_features_dir=None)

Parameters:

• input_csv (str | Path)

• output_csv (str | Path)

• routes (Iterable[str] | None)

• models_dir (str | Path | None)

• model_names (Iterable[str] | None)

• config (PreprocessConfig | None)

• input_stage (str)

• force_absorbance (bool)

• absorbance_scale_factor (float | None)

• include_probabilities (bool)

• save_features_dir (str | Path | None)

Return type:

DataFrame

xpectra.pipeline.safe_name(name)

Return a stable filesystem/column-safe version of a model name.

Parameters:

name (str)

Return type:

str

xpectra.pipeline.train_routes(routes=None, processed_dir=None, models_dir=None, model_names=None, config=None, fit_on='full', evaluate=False, skip_existing=False, limit_models=None)

Parameters:

• routes (Iterable[str] | None)

• processed_dir (str | Path | None)

• models_dir (str | Path | None)

• model_names (Iterable[str] | None)

• config (PreprocessConfig | None)

• fit_on (str)

• evaluate (bool)

• skip_existing (bool)

• limit_models (int | None)

Return type:

DataFrame

Modules

build_db	Build the three clean processed route files covering all eight datasets.
config	Configuration shared by the training and prediction pipelines.
core	Training, preprocessing, and prediction helpers for FTIR model routes.