diff --git a/CHANGELOG.md b/CHANGELOG.md index eb1bd9fd..b0449836 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,18 @@ those changes. ## [Unreleased] +## [1.58.0] - 2026-07-16 + +### Added + +- `batch.BatchPLS`: batchwise-unfolded (multiway) PLS relating the unfolded + ``[Z | X]`` batch matrix to a final-quality block ``Y`` (one row per batch), + built by composition over `multivariate.PLS`. Predicts the quality of a + completed batch and exposes time-varying X-weights (reshapeable to a + variable-by-time grid, and plottable with `time_varying_loading_plot`), plus + batch-level scores, SPE and Hotelling's T2. This is the batch regression / + prediction counterpart to `BatchPCA`. + ## [1.57.0] - 2026-07-16 ### Added @@ -2578,7 +2590,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.57.0...HEAD +[Unreleased]: https://github.com/kgdunn/process-improve/compare/v1.58.0...HEAD +[1.58.0]: https://github.com/kgdunn/process-improve/compare/v1.57.0...v1.58.0 [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 diff --git a/CITATION.cff b/CITATION.cff index ddbee1c4..6a95f897 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.57.0 +version: 1.58.0 date-released: "2026-07-16" keywords: - chemometrics diff --git a/docs/api/batch.rst b/docs/api/batch.rst index a4ea9848..b4c40f4b 100644 --- a/docs/api/batch.rst +++ b/docs/api/batch.rst @@ -13,6 +13,10 @@ Batch Data Analysis :members: :show-inheritance: +.. autoclass:: process_improve.batch.BatchPLS + :members: + :show-inheritance: + .. automodule:: process_improve.batch._transformers :members: :show-inheritance: diff --git a/pyproject.toml b/pyproject.toml index 1e2cab8c..b82d0f2b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "process-improve" -version = "1.57.0" +version = "1.58.0" description = 'Designed Experiments; Latent Variables (PCA, PLS, multivariate methods with missing data); Process Monitoring; Batch data analysis.' readme = "README.md" license = "MIT" diff --git a/src/process_improve/batch/__init__.py b/src/process_improve/batch/__init__.py index 825fac0b..9f92afd3 100644 --- a/src/process_improve/batch/__init__.py +++ b/src/process_improve/batch/__init__.py @@ -7,6 +7,7 @@ online_monitoring_plot, time_varying_loading_plot, ) +from process_improve.batch._batch_pls import BatchPLS from process_improve.batch._transformers import ( BatchFeatureExtractor, BatchScaler, @@ -59,6 +60,7 @@ "BatchFeatureExtractor", "BatchMonitor", "BatchPCA", + "BatchPLS", "BatchScaler", "DTWAligner", "ResampleAligner", diff --git a/src/process_improve/batch/_batch_pls.py b/src/process_improve/batch/_batch_pls.py new file mode 100644 index 00000000..6111d393 --- /dev/null +++ b/src/process_improve/batch/_batch_pls.py @@ -0,0 +1,255 @@ +# (c) Kevin Dunn, 2010-2026. MIT License. Based on own private work over the years. +"""Batchwise-unfolded (multiway) PLS relating batch trajectories to final quality. + +Mirrors :class:`process_improve.batch.BatchPCA`, but regresses the unfolded +``[Z | X]`` matrix onto a final-quality block ``Y`` (one row per batch) with the +existing :class:`process_improve.multivariate.PLS`. This is the batch +regression / prediction model: it captures how the initial conditions and the +time-varying trajectories drive the final product quality, and predicts the +quality of a completed batch from its data. + +See Wold, Kettaneh-Wold, MacGregor and Dunn, "Batch Process Modeling and +MSPC", Comprehensive Chemometrics, Elsevier, 2009. +""" + +from __future__ import annotations + +import typing + +import pandas as pd +from sklearn.base import BaseEstimator, RegressorMixin +from sklearn.utils import Bunch +from sklearn.utils.validation import check_is_fitted + +from ..multivariate._pls import PLS +from .data_input import check_valid_batch_dict, dict_to_wide + +if typing.TYPE_CHECKING: + from collections.abc import Hashable + + +class BatchPLS(RegressorMixin, BaseEstimator): + """Batchwise-unfolded PLS from batch trajectories (and Z) to final quality Y. + + Unfolds an aligned batch-data dictionary batchwise (one row per batch), + optionally joins an initial-conditions (Z) block onto that row, and fits a + :class:`process_improve.multivariate.PLS` model against a batch-indexed + quality block ``Y``. The fitted model relates the initial conditions and + the time-varying trajectory deviations to the final quality, and predicts + the quality of a completed batch. + + The batches must be aligned before fitting (every batch the same number of + samples; see :func:`process_improve.batch.resample_to_reference` and + :func:`process_improve.batch.batch_dtw`) with no missing values. + + Parameters + ---------- + n_components : int + Number of PLS components. + scale : bool, default=True + Mean-centre and unit-variance scale each column before fitting (the + centring removes the average trajectory). Passed to the underlying PLS. + group_by_batch : bool, default=False + Ordering of the unfolded column index, passed to + :func:`process_improve.batch.dict_to_wide`. + + Attributes (after fitting) + -------------------------- + x_weights_ : pd.DataFrame of shape (n_unfolded_features, n_components) + X-block weights (w*), indexed by the 2-level unfolded column index so + the trajectory part reshapes to a (tag, time) grid. + loadings_ : pd.DataFrame + Alias of ``x_weights_``, so + :func:`process_improve.batch.time_varying_loading_plot` can plot the + time-varying weights. + beta_coefficients_ : pd.DataFrame + Regression coefficients from the unfolded X to Y. + r2_cumulative_ : pd.Series + Cumulative R2 after each component. + rmse_ : pd.Series or pd.DataFrame + Root-mean-square error of the fit. + scores_, spe_, hotellings_t2_ : pd.DataFrame + Batch-level scores and diagnostics. + n_batches_, n_tags_, n_timesteps_, n_initial_conditions_ : int + Problem dimensions. + batch_ids_, tag_names_, initial_condition_names_, time_index_ : list + Labels for the batches, tags, initial conditions, and time samples. + + Examples + -------- + >>> from process_improve.batch import BatchPLS, load_dryer, resample_to_reference + >>> import pandas as pd + >>> batches = load_dryer() + >>> tags = [c for c in next(iter(batches.values())).columns if c != "ClockTime"] + >>> aligned = resample_to_reference(batches, columns_to_align=tags, reference_batch=1) + >>> quality = pd.DataFrame({"final": [float(b["DryerTemp"].iloc[-1]) for b in aligned.values()]}, + ... index=list(aligned.keys())) + >>> model = BatchPLS(n_components=2).fit(aligned, quality) # doctest: +SKIP + >>> model.predict(aligned).y_hat.shape # doctest: +SKIP + (71, 1) + + See Also + -------- + process_improve.batch.BatchPCA : the unsupervised (monitoring) counterpart. + process_improve.multivariate.PLS : the underlying estimator. + """ + + _parameter_constraints: typing.ClassVar = { + "n_components": [int, None], + "scale": [bool], + "group_by_batch": [bool], + } + + def __init__(self, n_components: int, *, scale: bool = True, group_by_batch: bool = False) -> None: + self.n_components = n_components + self.scale = scale + self.group_by_batch = group_by_batch + + def _unfold( + self, + batches: dict[Hashable, pd.DataFrame], + initial_conditions: pd.DataFrame | None, + ) -> pd.DataFrame: + """Unfold the batches batchwise and join the initial-conditions block. + + Trajectory columns keep their ``(tag, sequence)`` labels; initial- + condition columns are labelled ``(name, "")`` (or ``("", name)`` when + ``group_by_batch`` is set) since they carry no time axis. + """ + check_valid_batch_dict(batches, no_nan=True) + wide = dict_to_wide(batches, group_by_batch=self.group_by_batch) + if initial_conditions is None: + return wide + if not isinstance(initial_conditions, pd.DataFrame): + raise TypeError( + "initial_conditions must be a pandas DataFrame indexed by batch identifier; " + f"got {type(initial_conditions).__name__}." + ) + if set(initial_conditions.index) != set(wide.index): + raise ValueError("initial_conditions must have exactly one row per batch, indexed by batch id.") + z_wide = initial_conditions.reindex(wide.index) + if z_wide.select_dtypes(include="number").shape[1] != z_wide.shape[1]: + raise ValueError("All initial_conditions columns must be numeric.") + if z_wide.isna().to_numpy().sum() > 0: + raise ValueError("No missing values allowed in initial_conditions.") + tuples = [("", name) if self.group_by_batch else (name, "") for name in z_wide.columns] + z_wide.columns = pd.MultiIndex.from_tuples(tuples, names=wide.columns.names) + return pd.concat([z_wide, wide], axis=1) + + def fit( + self, + X: dict[Hashable, pd.DataFrame], + Y: pd.DataFrame, + *, + initial_conditions: pd.DataFrame | None = None, + ) -> BatchPLS: + """Fit the batchwise-unfolded PLS model against the quality block ``Y``. + + Parameters + ---------- + X : dict[Hashable, pd.DataFrame] + Standard batch-data dictionary of aligned batches. + Y : pd.DataFrame + Final-quality block: one row per batch (indexed by the same batch + identifiers as ``X``), one column per quality variable. + initial_conditions : pd.DataFrame, optional + The Z block: one row per batch, joined onto the unfolded row. + + Returns + ------- + self : BatchPLS + """ + wide = self._unfold(X, initial_conditions) + if not isinstance(Y, pd.DataFrame): + raise TypeError(f"Y must be a pandas DataFrame indexed by batch id; got {type(Y).__name__}.") + if set(Y.index) != set(wide.index): + raise ValueError("Y must have exactly one row per batch, indexed by the same batch ids as X.") + y_aligned = Y.reindex(wide.index) + + self._pls = PLS(n_components=self.n_components, scale=self.scale).fit(wide, y_aligned) + + self.feature_columns_ = wide.columns + # PLS flattens the feature index; re-attach the 2-level unfolded index + # so the trajectory weights reshape to a (tag, time) grid. + self.x_weights_ = pd.DataFrame( + self._pls.x_weights_.to_numpy(), index=wide.columns, columns=self._pls.x_weights_.columns + ) + self.loadings_ = self.x_weights_ + self.beta_coefficients_ = self._pls.beta_coefficients_ + self.r2_cumulative_ = self._pls.r2_cumulative_ + self.rmse_ = self._pls.rmse_ + self.scores_ = self._pls.scores_ + self.spe_ = self._pls.spe_ + self.hotellings_t2_ = self._pls.hotellings_t2_ + + first_batch = X[next(iter(X.keys()))] + self.batch_ids_ = list(wide.index) + self.n_batches_ = len(self.batch_ids_) + self.tag_names_ = list(first_batch.columns) + self.n_tags_ = len(self.tag_names_) + self.n_timesteps_ = int(first_batch.shape[0]) + self.time_index_ = list(range(self.n_timesteps_)) + self.target_names_ = list(Y.columns) + if initial_conditions is None: + self.initial_condition_names_ = [] + self.n_initial_conditions_ = 0 + else: + self.initial_condition_names_ = list(initial_conditions.columns) + self.n_initial_conditions_ = len(self.initial_condition_names_) + self.n_samples_ = self._pls.n_samples_ + return self + + def _scaled_wide( + self, batches: dict[Hashable, pd.DataFrame], initial_conditions: pd.DataFrame | None + ) -> pd.DataFrame: + """Unfold new batches and check they match the training column layout.""" + check_is_fitted(self, "x_weights_") + wide = self._unfold(batches, initial_conditions) + if list(wide.columns) != list(self.feature_columns_): + raise ValueError( + "The new batches do not unfold to the training column layout. Align new batches to the " + "training length and pass the same tags and initial-condition columns." + ) + return wide + + def predict( + self, + X: dict[Hashable, pd.DataFrame], + *, + initial_conditions: pd.DataFrame | None = None, + ) -> Bunch: + """Predict the final quality of completed batches. + + Parameters + ---------- + X : dict[Hashable, pd.DataFrame] + Standard batch-data dictionary of aligned batches with the same + tags and length as the training data. + initial_conditions : pd.DataFrame, optional + The Z block for the new batches; required if the model was fitted + with one. + + Returns + ------- + result : sklearn.utils.Bunch + With keys ``y_hat`` (predicted quality, one row per batch), + ``scores``, ``hotellings_t2`` and ``spe`` (batch diagnostics). + """ + wide = self._scaled_wide(X, initial_conditions) + diagnostics = self._pls.diagnose(wide) + order = list(X.keys()) + return Bunch( + y_hat=self._pls.predict(wide).reindex(order), + scores=diagnostics.scores.reindex(order), + hotellings_t2=diagnostics.hotellings_t2.reindex(order), + spe=diagnostics.spe.reindex(order), + ) + + def transform( + self, + X: dict[Hashable, pd.DataFrame], + *, + initial_conditions: pd.DataFrame | None = None, + ) -> pd.DataFrame: + """Return the batch-level PLS scores for ``X`` (in the input batch order).""" + return self._pls.transform(self._scaled_wide(X, initial_conditions)).reindex(list(X.keys())) diff --git a/tests/batch/test_batch_pls.py b/tests/batch/test_batch_pls.py new file mode 100644 index 00000000..8c4aa218 --- /dev/null +++ b/tests/batch/test_batch_pls.py @@ -0,0 +1,109 @@ +"""Tests for BatchPLS: batchwise-unfolded PLS from trajectories to final quality.""" + +import numpy as np +import pandas as pd +import pytest + +from process_improve.batch._batch_pls import BatchPLS +from process_improve.batch.datasets import load_dryer +from process_improve.batch.preprocessing import resample_to_reference + + +@pytest.fixture +def aligned_dryer() -> dict: + """Dryer batches resampled to a common length (ClockTime dropped).""" + batches = load_dryer() + tags = [c for c in next(iter(batches.values())).columns if c != "ClockTime"] + trimmed = {k: v[tags] for k, v in batches.items()} + return resample_to_reference(trimmed, columns_to_align=tags, reference_batch=1) + + +@pytest.fixture +def dryer_quality(aligned_dryer: dict) -> pd.DataFrame: + """Return a synthetic quality block driven by the mean dryer temperature.""" + return pd.DataFrame( + {"final": [float(b["DryerTemp"].mean()) for b in aligned_dryer.values()]}, + index=list(aligned_dryer.keys()), + ) + + +def test_fit_predict_on_dryer(aligned_dryer: dict, dryer_quality: pd.DataFrame) -> None: + """The model fits and predicts the quality of the training batches.""" + model = BatchPLS(n_components=3).fit(aligned_dryer, dryer_quality) + assert model.scores_.shape == (len(aligned_dryer), 3) + assert isinstance(model.x_weights_.index, pd.MultiIndex) + assert model.r2_cumulative_.iloc[-1] > 0.7 + result = model.predict(aligned_dryer) + assert result.y_hat.shape == (len(aligned_dryer), 1) + # Align predictions to the quality index by label before comparing (the + # unfolded row order need not match the input dict's insertion order). + y_hat = result.y_hat.reindex(dryer_quality.index).to_numpy().ravel() + corr = np.corrcoef(y_hat, dryer_quality["final"].to_numpy())[0, 1] + assert corr > 0.8 + + +def test_weights_reshape_to_grid(aligned_dryer: dict, dryer_quality: pd.DataFrame) -> None: + """The trajectory weights reshape to a (tag, time) grid.""" + model = BatchPLS(n_components=2).fit(aligned_dryer, dryer_quality) + grid = model.x_weights_.iloc[:, 0].unstack(level="sequence") # noqa: PD010 + assert grid.shape == (model.n_tags_, model.n_timesteps_) + + +def test_initial_conditions_join(aligned_dryer: dict, dryer_quality: pd.DataFrame) -> None: + """The Z block is appended to each unfolded row.""" + z = pd.DataFrame({"charge": [float(i) for i in range(len(aligned_dryer))]}, index=list(aligned_dryer.keys())) + model = BatchPLS(n_components=2).fit(aligned_dryer, dryer_quality, initial_conditions=z) + assert model.n_initial_conditions_ == 1 + assert model.x_weights_.shape[0] == model.n_tags_ * model.n_timesteps_ + 1 + result = model.predict(aligned_dryer, initial_conditions=z) + assert result.y_hat.shape[0] == len(aligned_dryer) + + +def test_synthetic_recovers_signal() -> None: + """A batch whose quality is a known function of its trajectories is predicted well.""" + rng = np.random.default_rng(0) + n_batches, n_timesteps = 30, 25 + t = np.linspace(0, 1, n_timesteps) + batches = {} + quality = [] + for i in range(n_batches): + level = rng.uniform(-1, 1) + # A carries a batch-specific level offset (so its deviation from the + # mean trajectory encodes `level`); B is a shared ramp with noise. + a = level + 0.5 * t + 0.02 * rng.standard_normal(n_timesteps) + b = t + 0.02 * rng.standard_normal(n_timesteps) + batches[f"b{i}"] = pd.DataFrame({"A": a, "B": b}) + quality.append(3.0 * level + 0.5) + y = pd.DataFrame({"q": quality}, index=list(batches.keys())) + model = BatchPLS(n_components=2).fit(batches, y) + pred = model.predict(batches).y_hat.reindex(y.index) + corr = np.corrcoef(pred.to_numpy().ravel(), y["q"].to_numpy())[0, 1] + assert corr > 0.9 + + +def test_y_must_be_dataframe(aligned_dryer: dict) -> None: + """A non-DataFrame Y is rejected.""" + with pytest.raises(TypeError, match="DataFrame"): + BatchPLS(n_components=2).fit(aligned_dryer, [1, 2, 3]) + + +def test_y_index_must_match(aligned_dryer: dict) -> None: + """A Y whose index does not match the batches is rejected.""" + y = pd.DataFrame({"q": [1.0, 2.0]}, index=["x", "y"]) + with pytest.raises(ValueError, match="one row per batch"): + BatchPLS(n_components=2).fit(aligned_dryer, y) + + +def test_predict_wrong_length_raises(aligned_dryer: dict, dryer_quality: pd.DataFrame) -> None: + """Predicting on batches of a different length is rejected.""" + model = BatchPLS(n_components=2).fit(aligned_dryer, dryer_quality) + shorter = {k: v.iloc[:50].reset_index(drop=True) for k, v in aligned_dryer.items()} + with pytest.raises(ValueError, match="training column layout"): + model.predict(shorter) + + +def test_transform_returns_scores(aligned_dryer: dict, dryer_quality: pd.DataFrame) -> None: + """Transform returns the batch-level scores.""" + model = BatchPLS(n_components=2).fit(aligned_dryer, dryer_quality) + scores = model.transform(aligned_dryer) + assert scores.shape == (len(aligned_dryer), 2)