diff --git a/CHANGELOG.md b/CHANGELOG.md index 690d0a9..eb1bd9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,27 @@ those changes. ## [Unreleased] +## [1.57.0] - 2026-07-16 + +### Added + +- Batch preprocessing transformers (sklearn `BaseEstimator` / `TransformerMixin` + facades over the existing free functions, so a batch workflow can be a + pipeline): `batch.BatchScaler` (range scaling with `inverse_transform`), + `batch.ResampleAligner` (linear resampling to a common length), + `batch.DTWAligner` (iterative weighted DTW alignment, learning a reference and + weights and aligning new batches), and `batch.BatchFeatureExtractor` (a + batch-data dictionary to a batch-by-feature matrix for a PLS-to-quality model). +- `batch.features.f_rupture` is now implemented (previously a stub): it detects + the first changepoint in a tag's trajectory per batch and phase using the + optional `ruptures` library. + +### Changed + +- The `batch` optional extra now provides `ruptures` (used by `f_rupture`) + instead of `openpyxl` and `scikit-image`, which were declared but imported + nowhere. + ## [1.56.0] - 2026-07-16 ### Added @@ -2557,7 +2578,8 @@ this entry records them together. - Reworked the README with a sharper value proposition and a "Why not scikit-learn?" comparison table. -[Unreleased]: https://github.com/kgdunn/process-improve/compare/v1.56.0...HEAD +[Unreleased]: https://github.com/kgdunn/process-improve/compare/v1.57.0...HEAD +[1.57.0]: https://github.com/kgdunn/process-improve/compare/v1.56.0...v1.57.0 [1.56.0]: https://github.com/kgdunn/process-improve/compare/v1.55.0...v1.56.0 [1.55.0]: https://github.com/kgdunn/process-improve/compare/v1.54.0...v1.55.0 [1.54.0]: https://github.com/kgdunn/process-improve/compare/v1.53.0...v1.54.0 diff --git a/CITATION.cff b/CITATION.cff index 353ba3c..ddbee1c 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -12,7 +12,7 @@ authors: repository-code: "https://github.com/kgdunn/process-improve" url: "https://kgdunn.github.io/process-improve/" license: MIT -version: 1.56.0 +version: 1.57.0 date-released: "2026-07-16" keywords: - chemometrics diff --git a/docs/api/batch.rst b/docs/api/batch.rst index 8e93d6a..a4ea984 100644 --- a/docs/api/batch.rst +++ b/docs/api/batch.rst @@ -13,6 +13,10 @@ Batch Data Analysis :members: :show-inheritance: +.. automodule:: process_improve.batch._transformers + :members: + :show-inheritance: + .. automodule:: process_improve.batch._batch_plots :members: :show-inheritance: diff --git a/pyproject.toml b/pyproject.toml index a27bbed..1e2cab8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "process-improve" -version = "1.56.0" +version = "1.57.0" description = 'Designed Experiments; Latent Variables (PCA, PLS, multivariate methods with missing data); Process Monitoring; Batch data analysis.' readme = "README.md" license = "MIT" @@ -61,11 +61,10 @@ expt = [ "pyDOE3>=1.0", ] -# Batch process data analysis: image-IO + Excel readers used by the -# batch alignment / fixture loaders. +# Batch process data analysis: changepoint detection for the f_rupture +# landmark feature (process_improve.batch.features.f_rupture). batch = [ - "openpyxl>=3.1.5", - "scikit-image>=0.25.2", + "ruptures>=1.1.9", ] # MCP server entry-point. @@ -94,12 +93,11 @@ all = [ "matplotlib>=3.10.8", "mcp>=1.0", "numba>=0.63.1", - "openpyxl>=3.1.5", "plotly>=6.5.2", "pulp>=2.8", "pyDOE3>=1.0", "ridgeplot>=0.5.0", - "scikit-image>=0.25.2", + "ruptures>=1.1.9", "seaborn>=0.13.2", ] diff --git a/src/process_improve/_extras.py b/src/process_improve/_extras.py index e0da1ea..f16a26a 100644 --- a/src/process_improve/_extras.py +++ b/src/process_improve/_extras.py @@ -6,7 +6,7 @@ pip install 'process-improve[plotting]' # matplotlib + plotly + seaborn + ridgeplot pip install 'process-improve[expt]' # pyDOE3 (DOE / experiments helpers) - pip install 'process-improve[batch]' # scikit-image + openpyxl + pip install 'process-improve[batch]' # ruptures (changepoint detection) pip install 'process-improve[mcp]' # mcp pip install 'process-improve[fast]' # numba (JIT) pip install 'process-improve[ilp]' # pulp (OMARS design generator) diff --git a/src/process_improve/batch/__init__.py b/src/process_improve/batch/__init__.py index 4ed3a28..825fac0 100644 --- a/src/process_improve/batch/__init__.py +++ b/src/process_improve/batch/__init__.py @@ -7,6 +7,12 @@ online_monitoring_plot, time_varying_loading_plot, ) +from process_improve.batch._transformers import ( + BatchFeatureExtractor, + BatchScaler, + DTWAligner, + ResampleAligner, +) from process_improve.batch.data_input import ( check_valid_batch_dict, dict_to_melted, @@ -50,8 +56,12 @@ __all__ = [ # Modelling and monitoring + "BatchFeatureExtractor", "BatchMonitor", "BatchPCA", + "BatchScaler", + "DTWAligner", + "ResampleAligner", # Preprocessing/alignment "batch_dtw", # Data input/conversion diff --git a/src/process_improve/batch/_transformers.py b/src/process_improve/batch/_transformers.py new file mode 100644 index 0000000..c37916b --- /dev/null +++ b/src/process_improve/batch/_transformers.py @@ -0,0 +1,340 @@ +# (c) Kevin Dunn, 2010-2026. MIT License. Based on own private work over the years. +"""sklearn-style transformer facades for the batch preprocessing pipeline. + +These wrap the standalone alignment, scaling, and feature-extraction functions +in :mod:`process_improve.batch.preprocessing` and +:mod:`process_improve.batch.features` in the ``BaseEstimator`` / +``TransformerMixin`` API, so a batch workflow can be expressed as an sklearn +pipeline and the learned state (reference batch, weights, scaling ranges) is +carried on a fitted estimator. The free functions remain the implementation +layer and are unchanged. +""" + +from __future__ import annotations + +import typing + +import numpy as np +import pandas as pd +from sklearn.base import BaseEstimator, TransformerMixin + +from .features import ( + f_agemax, + f_agemin, + f_count, + f_iqr, + f_last, + f_max, + f_mean, + f_median, + f_min, + f_robust_mad, + f_std, + f_sum, +) +from .preprocessing import ( + align_with_path, + apply_scaling, + batch_dtw, + determine_scaling, + dtw_core, + find_reference_batch, + resample_to_reference, + reverse_scaling, +) + +if typing.TYPE_CHECKING: + from collections.abc import Callable, Hashable + + +def _str_keyed(batches: dict) -> dict: + """Cast a batch dict to the ``dict[str, DataFrame]`` the wrapped functions annotate. + + The alignment / scaling free functions annotate their batch-key type as + ``str``, but they operate on any hashable key (the bundled datasets use + integer batch ids). This cast keeps mypy satisfied without changing the + runtime behaviour. + """ + return typing.cast("dict[str, pd.DataFrame]", batches) + +# The feature extractors that share the uniform (data, tags, batch_col, +# phase_col) signature, so they can be dispatched by name. +_UNIFORM_FEATURES: dict[str, Callable] = { + "mean": f_mean, + "median": f_median, + "std": f_std, + "iqr": f_iqr, + "robust_mad": f_robust_mad, + "sum": f_sum, + "min": f_min, + "max": f_max, + "agemin": f_agemin, + "agemax": f_agemax, + "last": f_last, + "count": f_count, +} + + +class BatchScaler(TransformerMixin, BaseEstimator): + """Range-scale batch trajectories, learning the ranges from a training set. + + Wraps :func:`process_improve.batch.determine_scaling` (fit) and + :func:`process_improve.batch.apply_scaling` / + :func:`process_improve.batch.reverse_scaling` (transform / + inverse_transform). Each tag is scaled to a robust range so that no single + high-variance tag dominates a downstream alignment or model. + + Parameters + ---------- + columns_to_align : list, optional + Tags to scale; defaults to every column of the first batch. + robust : bool, default=True + Use a robust range (98th minus 2nd percentile) rather than the raw + (max minus min) range. + + Attributes (after fitting) + -------------------------- + scale_df_ : pd.DataFrame + The learned per-tag range and offset. + """ + + def __init__(self, columns_to_align: list | None = None, *, robust: bool = True) -> None: + self.columns_to_align = columns_to_align + self.robust = robust + + def fit(self, X: dict[Hashable, pd.DataFrame], y: object = None) -> BatchScaler: # noqa: ARG002 + """Learn the per-tag scaling ranges from the batches ``X``.""" + self.scale_df_ = determine_scaling( + batches=_str_keyed(X), columns_to_align=self.columns_to_align, settings={"robust": self.robust} + ) + return self + + def transform(self, X: dict[Hashable, pd.DataFrame]) -> dict: + """Scale the batches using the learned ranges.""" + return apply_scaling(_str_keyed(X), self.scale_df_, self.columns_to_align) + + def inverse_transform(self, X: dict[Hashable, pd.DataFrame]) -> dict: + """Undo the scaling, returning the batches to their original units.""" + return reverse_scaling(_str_keyed(X), self.scale_df_, self.columns_to_align) + + +class ResampleAligner(TransformerMixin, BaseEstimator): + """Align batches to a common length by resampling (linear time warping). + + Wraps :func:`process_improve.batch.resample_to_reference`: every batch is + linearly resampled onto the sample grid of a reference batch, so all + batches end up with the same number of samples. This is the simplest + alignment and assumes the duration difference is spread evenly through the + batch. + + Parameters + ---------- + columns_to_align : list + Tags to align. + reference_batch : Hashable, optional + Key of the batch whose length is the target. If omitted, the most + central batch is chosen with + :func:`process_improve.batch.find_reference_batch` at fit time. + + Attributes (after fitting) + -------------------------- + reference_batch_ : Hashable + The reference batch key actually used. + """ + + def __init__(self, columns_to_align: list, reference_batch: Hashable | None = None) -> None: + self.columns_to_align = columns_to_align + self.reference_batch = reference_batch + + reference_batch_: Hashable + + def fit(self, X: dict[Hashable, pd.DataFrame], y: object = None) -> ResampleAligner: # noqa: ARG002 + """Choose (or record) the reference batch.""" + if self.reference_batch is None: + reference = find_reference_batch(_str_keyed(X), columns_to_align=list(self.columns_to_align)) + self.reference_batch_ = reference[0] if isinstance(reference, list) else reference + else: + self.reference_batch_ = self.reference_batch + self._reference_length = X[self.reference_batch_].shape[0] + return self + + def transform(self, X: dict[Hashable, pd.DataFrame]) -> dict: + """Resample every batch onto the reference length.""" + # resample_to_reference needs the reference batch present in the dict; + # supply a target of the learned length via any batch of that length, + # falling back to the stored reference when it is present. + reference = self.reference_batch_ if self.reference_batch_ in X else next(iter(X)) + resampled = resample_to_reference( + _str_keyed(X), list(self.columns_to_align), reference_batch=typing.cast("str", reference) + ) + if resampled[reference].shape[0] != self._reference_length: + # Force the learned length by resampling against a synthetic index. + resampled = {k: _resample_frame(v, self._reference_length) for k, v in resampled.items()} + return resampled + + +def _resample_frame(frame: pd.DataFrame, n_samples: int) -> pd.DataFrame: + """Linearly resample every column of ``frame`` to ``n_samples`` rows.""" + source = np.linspace(0.0, 1.0, frame.shape[0]) + target = np.linspace(0.0, 1.0, n_samples) + return pd.DataFrame( + {column: np.interp(target, source, frame[column].to_numpy()) for column in frame.columns} + ) + + +class DTWAligner(TransformerMixin, BaseEstimator): + """Align batches with iterative weighted dynamic time warping (DTW). + + Wraps :func:`process_improve.batch.batch_dtw` (Kassidas et al.): learns a + reference trajectory and per-tag weights from the training batches, warping + every batch nonlinearly onto the reference. New batches are aligned to the + learned reference with the learned weights. + + Parameters + ---------- + columns_to_align : list + Tags to align. + reference_batch : Hashable, optional + Key of the initial reference batch. If omitted, it is chosen with + :func:`process_improve.batch.find_reference_batch` at fit time. + settings : dict, optional + Extra settings forwarded to :func:`process_improve.batch.batch_dtw`. + + Attributes (after fitting) + -------------------------- + aligned_batches_ : dict + The training batches after alignment. + reference_batch_ : Hashable + The initial reference batch key. + scale_df_ : pd.DataFrame + The scaling ranges used internally by the DTW. + weights_ : np.ndarray + The final per-tag weight vector. + """ + + def __init__( + self, + columns_to_align: list, + reference_batch: Hashable | None = None, + settings: dict | None = None, + ) -> None: + self.columns_to_align = columns_to_align + self.reference_batch = reference_batch + self.settings = settings + + reference_batch_: Hashable + + def fit(self, X: dict[Hashable, pd.DataFrame], y: object = None) -> DTWAligner: # noqa: ARG002 + """Run iterative weighted DTW and store the reference and weights.""" + columns = list(self.columns_to_align) + if self.reference_batch is None: + reference = find_reference_batch(_str_keyed(X), columns_to_align=columns, settings=self.settings) + self.reference_batch_ = reference[0] if isinstance(reference, list) else reference + else: + self.reference_batch_ = self.reference_batch + + result = batch_dtw( + _str_keyed(X), + columns_to_align=columns, + reference_batch=typing.cast("str", self.reference_batch_), + settings=self.settings, + ) + self.aligned_batches_ = result["aligned_batch_dfdict"] + self.scale_df_ = result["scale_df"] + self.weights_ = result["weight_history"].iloc[-1].to_numpy() + self._reference = result["last_average_batch"] + self._reference_scaled = apply_scaling({"_ref_": self._reference}, self.scale_df_, columns)["_ref_"] + self._n_reference_rows = self.aligned_batches_[next(iter(self.aligned_batches_))].shape[0] + return self + + def transform(self, X: dict[Hashable, pd.DataFrame]) -> dict: + """Align batches to the learned reference. + + Training batches return their stored aligned version; any other batch + is warped to the learned reference using the learned weights. + """ + columns = list(self.columns_to_align) + weight_matrix = np.diag(self.weights_) + out = {} + for batch_id, batch in X.items(): + if batch_id in self.aligned_batches_: + out[batch_id] = self.aligned_batches_[batch_id] + continue + scaled = apply_scaling(_str_keyed({batch_id: batch}), self.scale_df_, columns)[batch_id] + result = dtw_core(scaled[columns], self._reference_scaled[columns], weight_matrix) + synced = align_with_path(result.md_path, batch) + out[batch_id] = _resample_frame( + synced[[c for c in synced.columns if c in columns]], self._n_reference_rows + ) + return out + + +class BatchFeatureExtractor(TransformerMixin, BaseEstimator): + """Extract per-batch landmark features into a batch-by-feature matrix. + + Wraps the feature extractors in :mod:`process_improve.batch.features`, + turning a batch-data dictionary into a single ``(n_batches, n_features)`` + table (one row per batch, one column per tag-and-feature) suitable as the X + block of a PLS-to-quality model. + + Parameters + ---------- + features : sequence of str, optional + Which features to compute; defaults to ``("mean", "std", "min", + "max", "last")``. Each must be one of the uniform-signature extractors: + mean, median, std, iqr, robust_mad, sum, min, max, agemin, agemax, + last, count. + tags : list of str, optional + Which tags to extract features from; defaults to every tag. + phase_col : str, optional + Column identifying the phase within a batch, passed through to the + extractors. + + Attributes (after fitting) + -------------------------- + feature_names_out_ : list of str + The generated ``"{tag}_{feature}"`` column names. + """ + + def __init__( + self, + features: typing.Sequence[str] = ("mean", "std", "min", "max", "last"), + tags: list[str] | None = None, + phase_col: str | None = None, + ) -> None: + self.features = features + self.tags = tags + self.phase_col = phase_col + + def _extract(self, X: dict[Hashable, pd.DataFrame]) -> pd.DataFrame: + """Compute and horizontally concatenate the requested feature blocks.""" + unknown = [name for name in self.features if name not in _UNIFORM_FEATURES] + if unknown: + raise ValueError( + f"Unknown feature(s) {unknown}; choose from {sorted(_UNIFORM_FEATURES)}." + ) + # Melt directly (allowing unequal batch lengths, unlike dict_to_melted, + # which requires aligned batches): landmark features summarise each + # batch regardless of its duration. + frames: list[pd.DataFrame] = [] + for batch_id, batch in X.items(): + frame = batch.copy() + frame["batch_id"] = np.full(len(batch), batch_id, dtype=object) + frames.append(frame) + melted = pd.concat(frames, ignore_index=True) + blocks = [] + for name in self.features: + block = _UNIFORM_FEATURES[name](melted, tags=self.tags, batch_col="batch_id", phase_col=self.phase_col) + blocks.append(block.reset_index(drop=True) if self.phase_col is None else block) + combined = pd.concat(blocks, axis=1) + combined.index = pd.Index(list(X.keys()), name="batch_id") + return combined + + def fit(self, X: dict[Hashable, pd.DataFrame], y: object = None) -> BatchFeatureExtractor: # noqa: ARG002 + """Record the generated feature-column names.""" + self.feature_names_out_ = list(self._extract(X).columns) + return self + + def transform(self, X: dict[Hashable, pd.DataFrame]) -> pd.DataFrame: + """Return the ``(n_batches, n_features)`` feature matrix for ``X``.""" + return self._extract(X) diff --git a/src/process_improve/batch/features.py b/src/process_improve/batch/features.py index 5f2c607..8591f76 100644 --- a/src/process_improve/batch/features.py +++ b/src/process_improve/batch/features.py @@ -336,43 +336,95 @@ def f_area( # Breakpoint detection: rupture / breakpoint within a particular tag. # ------------------------------------------ -def f_rupture( +def f_rupture( # noqa: PLR0913 data: pd.DataFrame, columns: list[str] | None = None, batch_col: str | None = None, phase_col: str | None = None, -) -> None: - """ - Feature: rupture. - - The breakpoint in a given tag in ``columns`` (usually it is 1 tag), - for each unique batch in the ``batch_col`` indicator column, and - within each unique phase, per batch, of the ``phase_col`` column. + penalty: float = 10.0, + model: str = "rbf", +) -> pd.DataFrame: + """Detect the first changepoint (rupture) in a single tag's trajectory. + + Fits a PELT changepoint model to the tag's signal in each batch (and + phase) and returns the location of the first detected changepoint, i.e. + the sample index at which the trajectory's statistical behaviour first + shifts. This is a data-driven landmark feature: unlike a fixed threshold + crossing, it locates a break in level or variance without the analyst + specifying where to look. + + Requires the optional ``ruptures`` library, part of the ``batch`` extra + (``pip install 'process-improve[batch]'``). + + Parameters + ---------- + data : pd.DataFrame + Batch data in melted (long) form. + columns : list of str + Exactly one tag (column) name to detect the changepoint in. + batch_col : str, optional + Column identifying the batch. If omitted, all rows are treated as a + single batch. + phase_col : str, optional + Column identifying the phase within a batch. If omitted, each batch is + treated as a single phase. + penalty : float, default=10.0 + PELT penalty term; larger values detect fewer changepoints. + model : str, default="rbf" + Cost model passed to ``ruptures`` (for example ``"rbf"``, ``"l2"``, + ``"l1"``). + + Returns + ------- + pd.DataFrame + One row per (batch, phase) group, with a single column + ``"{tag}_rupture"`` giving the first changepoint sample index within + that group, or ``NaN`` when no changepoint is detected. + + See Also + -------- + f_crossing : threshold-crossing landmark at a fixed level. + f_elbow : elbow / knee-point landmark. + + References + ---------- + Truong, C., Oudre, L. and Vayatis, N., "Selective review of offline + change point detection methods", Signal Processing, 167, 2020. See also + https://github.com/deepcharles/ruptures. """ - # Handle phase detection based on 1 column for now. if columns is None or len(columns) != 1: - raise NotImplementedError( - f"Phase detection currently supports a single column only; got {columns!r}." - ) + raise ValueError(f"f_rupture supports a single column only; got {columns!r}.") - # TODO: see https://github.com/deepcharles/ruptures - - # base_name = "rupture" - # prepared = _prepare_data(data, columns, batch_col, phase_col) - # feature_columns = prepared['columns'] - # grouper = prepared['data'] - - # import ruptures as rpt - # import matplotlib.pyplot as plt - # output = pd.DataFrame() - # for batch_id, subset in grouper: - # signal = subset[columns[0]].values - # algo = rpt.Pelt(model="rbf").fit(signal) - # result = algo.predict(pen=100) - # print(result) - # plt.show() - # fig = rpt.display(signal, result, computed_chg_pts=result) - # fig.save(batchid) + # Deferred import so that importing this module does not require the + # optional ``ruptures`` dependency; only f_rupture needs it. + try: + import ruptures as rpt # noqa: PLC0415 + except ImportError as exc: + from .._extras import require_extra # noqa: PLC0415 + + raise require_extra("ruptures", "batch") from exc + + column = columns[0] + base_name = "rupture" + prepared, _tags, _output, _df_out = _prepare_data(data, columns, batch_col, phase_col) + f_name = f"{column}_{base_name}" + + locations = {} + for group_id, subset in prepared: + signal = np.asarray(subset[column].to_numpy(), dtype=float) + if signal.size < 2 or not np.isfinite(signal).all(): + locations[group_id] = np.nan + continue + algo = rpt.Pelt(model=model).fit(signal) + breakpoints = algo.predict(pen=penalty) + # ruptures always returns the signal length as the final breakpoint; + # the first genuine changepoint (if any) precedes it. + interior = [bp for bp in breakpoints if bp < signal.size] + locations[group_id] = float(interior[0]) if interior else np.nan + + output = pd.Series(locations, name=f_name).to_frame() + output.index = output.index.set_names(prepared.keys) + return output # Extreme features diff --git a/tests/batch/test_transformers.py b/tests/batch/test_transformers.py new file mode 100644 index 0000000..6fd51b2 --- /dev/null +++ b/tests/batch/test_transformers.py @@ -0,0 +1,124 @@ +"""Tests for the batch sklearn transformer facades and f_rupture.""" + +import numpy as np +import pandas as pd +import pytest + +from process_improve.batch._transformers import ( + BatchFeatureExtractor, + BatchScaler, + DTWAligner, + ResampleAligner, +) +from process_improve.batch.datasets import load_nylon + + +@pytest.fixture +def nylon() -> dict: + """Load the nylon batch dictionary.""" + return load_nylon() + + +@pytest.fixture +def nylon_tags(nylon: dict) -> list: + """Return the nylon tag names.""" + return list(next(iter(nylon.values())).columns) + + +def test_batch_scaler_round_trip(nylon: dict, nylon_tags: list) -> None: + """Apply then reverse scaling returns the original trajectories.""" + scaler = BatchScaler(nylon_tags).fit(nylon) + scaled = scaler.transform(nylon) + restored = scaler.inverse_transform(scaled) + np.testing.assert_allclose(nylon[1][nylon_tags].to_numpy(), restored[1][nylon_tags].to_numpy(), atol=1e-6) + + +def test_batch_scaler_learns_ranges(nylon: dict, nylon_tags: list) -> None: + """The fitted scaler exposes a per-tag range table.""" + scaler = BatchScaler(nylon_tags).fit(nylon) + assert scaler.scale_df_.shape[0] == len(nylon_tags) + + +def test_resample_aligner_equal_lengths(nylon: dict, nylon_tags: list) -> None: + """Resampling gives every batch the reference length.""" + aligner = ResampleAligner(nylon_tags, reference_batch=1).fit(nylon) + aligned = aligner.transform(nylon) + assert len({v.shape[0] for v in aligned.values()}) == 1 + assert aligner.reference_batch_ == 1 + assert next(iter(aligned.values())).shape[0] == nylon[1].shape[0] + + +def test_resample_aligner_auto_reference(nylon: dict, nylon_tags: list) -> None: + """With no reference, one is chosen and recorded.""" + aligner = ResampleAligner(nylon_tags).fit(nylon) + assert aligner.reference_batch_ in nylon + + +def test_dtw_aligner_train_and_new(nylon: dict, nylon_tags: list) -> None: + """DTW aligns the training set to equal length and a new batch to the reference.""" + keys = list(nylon) + train = {k: nylon[k] for k in keys[:6]} + held_key = keys[7] + aligner = DTWAligner(nylon_tags, settings={"show_progress": False}).fit(train) + aligned = aligner.transform(train) + assert len({v.shape[0] for v in aligned.values()}) == 1 + new_out = aligner.transform({held_key: nylon[held_key]}) + assert new_out[held_key].shape[0] == aligner._n_reference_rows + + +def test_feature_extractor_matrix_shape(nylon: dict, nylon_tags: list) -> None: + """The feature matrix is (n_batches, n_tags * n_features), batch-indexed.""" + extractor = BatchFeatureExtractor(features=("mean", "std", "last"), tags=nylon_tags).fit(nylon) + matrix = extractor.transform(nylon) + assert matrix.shape == (57, len(nylon_tags) * 3) + assert matrix.index.name == "batch_id" + assert "Tag01_mean" in matrix.columns + assert extractor.feature_names_out_ == list(matrix.columns) + + +def test_feature_extractor_feeds_pls(nylon: dict, nylon_tags: list) -> None: + """The feature matrix is a valid X block for a PLS-to-quality model.""" + from process_improve.multivariate import PLS, MCUVScaler + + features = BatchFeatureExtractor(features=("mean", "std"), tags=nylon_tags).fit_transform(nylon) + y = pd.Series({b: float(nylon[b]["Tag01"].mean()) for b in nylon}, name="q").to_frame() + x_scaled = MCUVScaler().fit_transform(features) + y_scaled = MCUVScaler().fit_transform(y) + model = PLS(n_components=2).fit(x_scaled, y_scaled) + assert model.r2_cumulative_.iloc[-1] > 0.5 + + +def test_feature_extractor_unknown_feature(nylon: dict) -> None: + """An unknown feature name is rejected.""" + with pytest.raises(ValueError, match="Unknown feature"): + BatchFeatureExtractor(features=("mean", "banana")).fit(nylon) + + +def test_f_rupture_detects_step() -> None: + """f_rupture locates a step change in each batch's trajectory.""" + pytest.importorskip("ruptures") + from process_improve.batch.features import f_rupture + + def make(step_at: int) -> pd.DataFrame: + rng = np.random.default_rng(0) + x = np.concatenate([np.zeros(step_at), np.full(30 - step_at, 5.0)]) + 0.01 * rng.standard_normal(30) + return pd.DataFrame({"x": x}) + + melted = pd.concat( + [make(10).assign(batch_id=1), make(20).assign(batch_id=2)], ignore_index=True + ) + out = f_rupture(melted, columns=["x"], batch_col="batch_id", penalty=5.0) + assert list(out.columns) == ["x_rupture"] + values = out["x_rupture"].to_numpy() + assert values[0] == pytest.approx(10, abs=2) + assert values[1] == pytest.approx(20, abs=2) + + +def test_f_rupture_requires_single_column() -> None: + """f_rupture rejects a multi-column request.""" + pytest.importorskip("ruptures") + from process_improve.batch.features import f_rupture + + df = pd.DataFrame({"x": [1.0, 2.0], "y": [3.0, 4.0], "batch_id": [1, 1]}) + with pytest.raises(ValueError, match="single column"): + f_rupture(df, columns=["x", "y"], batch_col="batch_id")