diff --git a/CHANGELOG.md b/CHANGELOG.md index 10560f96..690d0a9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,23 @@ those changes. ## [Unreleased] +## [1.56.0] - 2026-07-16 + +### Added + +- `batch.BatchMonitor`: online (real-time) Nomikos-MacGregor monitoring for a + fitted `BatchPCA`. Builds time-varying Hotelling's T2 and SPE control limits + from good batches, and tracks a new batch sample-by-sample, flagging the + samples where each statistic exceeds its limit. +- `BatchPCA.predict_online`: projection to the model plane (PMP) for a + partially-observed batch. At each time sample the future trajectory columns + are treated as missing and the score vector is the least-squares fit of the + observed columns; initial conditions, known from the batch start, are always + observed. At full observation it matches `BatchPCA.diagnose`. +- `batch.online_monitoring_plot`: the online SPE or T2 monitoring chart, drawing + the batch trace over the time-varying limit and the mean good-batch trace with + alarm samples marked. + ## [1.55.0] - 2026-07-16 ### Added @@ -2540,7 +2557,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.55.0...HEAD +[Unreleased]: https://github.com/kgdunn/process-improve/compare/v1.56.0...HEAD +[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 [1.53.0]: https://github.com/kgdunn/process-improve/compare/v1.52.4...v1.53.0 diff --git a/CITATION.cff b/CITATION.cff index e9a06570..353ba3c4 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.55.0 +version: 1.56.0 date-released: "2026-07-16" keywords: - chemometrics diff --git a/docs/api/batch.rst b/docs/api/batch.rst index 932b7969..8e93d6a3 100644 --- a/docs/api/batch.rst +++ b/docs/api/batch.rst @@ -9,6 +9,10 @@ Batch Data Analysis :members: :show-inheritance: +.. autoclass:: process_improve.batch.BatchMonitor + :members: + :show-inheritance: + .. automodule:: process_improve.batch._batch_plots :members: :show-inheritance: diff --git a/pyproject.toml b/pyproject.toml index e7686464..a27bbed2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "process-improve" -version = "1.55.0" +version = "1.56.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 4601862a..4ed3a282 100644 --- a/src/process_improve/batch/__init__.py +++ b/src/process_improve/batch/__init__.py @@ -1,8 +1,10 @@ """Batch process data analysis: alignment, feature extraction, modelling, and visualization.""" +from process_improve.batch._batch_monitor import BatchMonitor from process_improve.batch._batch_pca import BatchPCA from process_improve.batch._batch_plots import ( contribution_at_time_plot, + online_monitoring_plot, time_varying_loading_plot, ) from process_improve.batch.data_input import ( @@ -47,7 +49,8 @@ ) __all__ = [ - # Modelling + # Modelling and monitoring + "BatchMonitor", "BatchPCA", # Preprocessing/alignment "batch_dtw", @@ -83,6 +86,7 @@ "load_nylon", "melted_to_dict", "melted_to_wide", + "online_monitoring_plot", "resample_to_reference", "time_varying_loading_plot", "wide_to_dict", diff --git a/src/process_improve/batch/_batch_monitor.py b/src/process_improve/batch/_batch_monitor.py new file mode 100644 index 00000000..1c8eb769 --- /dev/null +++ b/src/process_improve/batch/_batch_monitor.py @@ -0,0 +1,198 @@ +# (c) Kevin Dunn, 2010-2026. MIT License. Based on own private work over the years. +"""Online (real-time) monitoring of batches against a fitted BatchPCA model. + +Implements the Nomikos-MacGregor online monitoring scheme: build a batchwise +PCA model from good (common-cause) batches, then track a new batch in real time +by projecting the partially-observed trajectories at each time sample (via +:meth:`process_improve.batch.BatchPCA.predict_online`) and comparing the +resulting Hotelling's T2 and SPE against time-varying control limits. The limits +are learned from the same projection applied to the good batches, so the +statistic at each sample is compared against the good-batch spread at that same +point in the batch evolution. + +See Nomikos and MacGregor, "Multivariate SPC Charts for Monitoring Batch +Processes", Technometrics, 37, 41-59, 1995. +""" + +from __future__ import annotations + +import typing + +import numpy as np +from sklearn.base import BaseEstimator +from sklearn.utils import Bunch +from sklearn.utils.validation import check_is_fitted + +from ..multivariate._limits import spe_calculation + +if typing.TYPE_CHECKING: + import pandas as pd + + from ._batch_pca import BatchPCA + + +class BatchMonitor(BaseEstimator): + """Time-varying (online) control limits for a fitted :class:`BatchPCA` model. + + Builds a Hotelling's T2 and SPE control limit at every time sample by + passing each good batch through the model's projection-to-the-model-plane + (PMP) estimate at that sample and summarising the good-batch spread. A new + batch can then be tracked in real time: at each sample its projected T2 and + SPE are compared against the limit for that sample, flagging abnormal + behaviour while the batch is still running. + + Parameters + ---------- + model : BatchPCA + A fitted :class:`process_improve.batch.BatchPCA` model, ideally built + from good (common-cause) batches only. + conf_level : float, default=0.99 + Confidence level for the control limits. + + Attributes (after fitting) + -------------------------- + spe_limit_over_time_ : np.ndarray of shape (n_timesteps,) + SPE control limit at each time sample. + t2_limit_over_time_ : np.ndarray of shape (n_timesteps,) + Hotelling's T2 control limit at each time sample. + spe_mean_over_time_, t2_mean_over_time_ : np.ndarray of shape (n_timesteps,) + Mean good-batch SPE and T2 trace, for plotting the expected trajectory. + n_timesteps_ : int + Number of time samples per batch (from the model). + + Examples + -------- + >>> from process_improve.batch import BatchPCA, BatchMonitor, load_nylon, resample_to_reference + >>> batches = load_nylon() + >>> tags = list(next(iter(batches.values())).columns) + >>> aligned = resample_to_reference(batches, columns_to_align=tags, reference_batch=1) + >>> good = {k: v for k, v in aligned.items() if k <= 40} + >>> model = BatchPCA(n_components=3).fit(good) + >>> monitor = BatchMonitor(model, conf_level=0.99).fit(good) + >>> trace = monitor.monitor(aligned[49]) + >>> bool(trace.spe_alarm.any()) # doctest: +SKIP + True + + References + ---------- + Nomikos, P. and MacGregor, J.F., "Multivariate SPC Charts for Monitoring + Batch Processes", Technometrics, 37, 41-59, 1995. + """ + + _parameter_constraints: typing.ClassVar = { + "model": [BaseEstimator], + "conf_level": [float], + } + + def __init__(self, model: BatchPCA, *, conf_level: float = 0.99) -> None: + self.model = model + self.conf_level = conf_level + + def _trace_for_batch( + self, batch: pd.DataFrame, initial_conditions: pd.Series | pd.DataFrame | None + ) -> tuple[np.ndarray, np.ndarray]: + """Return the (T2, SPE) trace over every time sample for one batch.""" + n_timesteps = self.model.n_timesteps_ + t2 = np.empty(n_timesteps) + spe = np.empty(n_timesteps) + for k in range(1, n_timesteps + 1): + result = self.model.predict_online(batch, upto_k=k, initial_conditions=initial_conditions) + t2[k - 1] = result.hotellings_t2 + spe[k - 1] = result.spe + return t2, spe + + def fit( + self, + good_batches: dict[typing.Hashable, pd.DataFrame], + y: object = None, # noqa: ARG002 + *, + initial_conditions: pd.DataFrame | None = None, + ) -> BatchMonitor: + """Learn the time-varying control limits from good batches. + + Parameters + ---------- + good_batches : dict[Hashable, pd.DataFrame] + Standard batch-data dictionary of aligned good (common-cause) + batches, the same tags and length as the model's training data. + y : ignored + Present for sklearn Pipeline compatibility. + initial_conditions : pd.DataFrame, optional + The Z block for the good batches; required if (and only if) the + model was fitted with one. + + Returns + ------- + self : BatchMonitor + """ + check_is_fitted(self.model, "loadings_") + n_timesteps = self.model.n_timesteps_ + batch_ids = list(good_batches.keys()) + t2_matrix = np.empty((len(batch_ids), n_timesteps)) + spe_matrix = np.empty((len(batch_ids), n_timesteps)) + for row, batch_id in enumerate(batch_ids): + z = None if initial_conditions is None else initial_conditions.loc[[batch_id]] + t2_matrix[row], spe_matrix[row] = self._trace_for_batch(good_batches[batch_id], z) + + spe_limits = np.array( + [spe_calculation(spe_matrix[:, k], conf_level=self.conf_level) for k in range(n_timesteps)] + ) + # The Hotelling's T2 limit is the analytical F-based limit; it does not + # vary over time, but is stored per-sample so the online chart has a + # limit array of matching length. + t2_limit = self.model.hotellings_t2_limit(conf_level=self.conf_level) + + self.spe_limit_over_time_ = spe_limits + self.t2_limit_over_time_ = np.full(n_timesteps, t2_limit) + self.spe_mean_over_time_ = spe_matrix.mean(axis=0) + self.t2_mean_over_time_ = t2_matrix.mean(axis=0) + self.n_timesteps_ = n_timesteps + return self + + def monitor( + self, + batch: pd.DataFrame, + upto_k: int | None = None, + *, + initial_conditions: pd.Series | pd.DataFrame | None = None, + ) -> Bunch: + """Track a batch in real time against the time-varying limits. + + Parameters + ---------- + batch : pd.DataFrame + A single aligned batch to monitor (the training tags as columns). + upto_k : int, optional + Track only up to this time sample (simulating a still-running + batch). Defaults to the full batch length. + initial_conditions : pd.Series or pd.DataFrame, optional + The Z block for this batch; required if the model was fitted with + one. + + Returns + ------- + result : sklearn.utils.Bunch + With keys ``time`` (1-based sample indices), ``hotellings_t2`` and + ``spe`` (the batch's statistic traces), ``t2_limit`` and + ``spe_limit`` (the limits over the same samples), and + ``t2_alarm`` / ``spe_alarm`` (boolean arrays where the statistic + exceeds its limit). + """ + check_is_fitted(self, "spe_limit_over_time_") + end = self.n_timesteps_ if upto_k is None else int(upto_k) + if not 1 <= end <= self.n_timesteps_: + raise ValueError(f"upto_k must lie in [1, {self.n_timesteps_}]; got {upto_k}.") + + t2, spe = self._trace_for_batch(batch, initial_conditions) + t2, spe = t2[:end], spe[:end] + t2_limit = self.t2_limit_over_time_[:end] + spe_limit = self.spe_limit_over_time_[:end] + return Bunch( + time=np.arange(1, end + 1), + hotellings_t2=t2, + spe=spe, + t2_limit=t2_limit, + spe_limit=spe_limit, + t2_alarm=t2 > t2_limit, + spe_alarm=spe > spe_limit, + ) diff --git a/src/process_improve/batch/_batch_pca.py b/src/process_improve/batch/_batch_pca.py index 2be9dd43..aa107ad7 100644 --- a/src/process_improve/batch/_batch_pca.py +++ b/src/process_improve/batch/_batch_pca.py @@ -21,8 +21,10 @@ import functools import typing +import numpy as np import pandas as pd from sklearn.base import BaseEstimator, TransformerMixin +from sklearn.utils import Bunch from sklearn.utils.validation import check_is_fitted from ..multivariate._diagnostics import spe_contributions as _spe_contributions @@ -41,9 +43,6 @@ if typing.TYPE_CHECKING: from collections.abc import Callable, Hashable - import numpy as np - from sklearn.utils import Bunch - def _pca_method(fn: Callable[..., typing.Any]) -> Callable[..., typing.Any]: """Wrap a module-level ``fn(model, ...)`` as a method forwarding ``self._pca``. @@ -370,6 +369,110 @@ def diagnose( """ return self._pca.diagnose(self._scaled_wide(X, initial_conditions)) + def predict_online( + self, + batch: pd.DataFrame, + upto_k: int, + *, + initial_conditions: pd.Series | pd.DataFrame | None = None, + ) -> Bunch: + """Project a partially-complete batch via projection to the model plane (PMP). + + During a running batch, the trajectory data for the future (time + samples ``upto_k`` and later) are not yet known. This method projects + the batch onto the fitted model using only the already-observed + columns: the score vector is the least-squares fit of the observed, + centred and scaled values onto the corresponding loading rows, + ``t = pinv(P_obs) @ x_obs`` (the projection-to-the-model-plane estimate + of Nomikos and MacGregor). Initial conditions, known from the batch + start, are always part of the observed set, so they sharpen the + projection from the first sample. + + The batch is expected to be aligned to the training length; ``upto_k`` + selects how many leading time samples are treated as observed. To + compare the returned statistics against control limits, use + :class:`process_improve.batch.BatchMonitor`, which builds the + time-varying limits from good batches. + + Parameters + ---------- + batch : pd.DataFrame + A single aligned batch (``n_timesteps`` rows, the training tags as + columns). + upto_k : int + Number of leading time samples to treat as observed, in + ``1 .. n_timesteps_``. At ``upto_k == n_timesteps_`` every + trajectory column is observed and the result matches + :meth:`diagnose` for that batch. + initial_conditions : pd.Series or pd.DataFrame, optional + The Z block for this batch (required if the model was fitted with + one). A Series of the initial-condition values, or a single-row + DataFrame. + + Returns + ------- + result : sklearn.utils.Bunch + With keys ``scores`` (Series, one entry per component), + ``hotellings_t2`` (float, cumulative over all components), and + ``spe`` (float, over the observed columns). + """ + check_is_fitted(self, "loadings_") + if not 1 <= upto_k <= self.n_timesteps_: + raise ValueError(f"upto_k must lie in [1, {self.n_timesteps_}]; got {upto_k}.") + + z_frame = self._coerce_online_initial_conditions(initial_conditions) + wide = self._unfold({"_online_": batch}, z_frame) + if list(wide.columns) != list(self.feature_columns_): + raise ValueError( + "The batch does not unfold to the training column layout. Align it to the training " + f"length ({self.n_timesteps_} samples) and pass the same tags and initial conditions." + ) + + # Observed columns: initial conditions (sequence == "") plus trajectory + # columns whose time sample is strictly before upto_k. + sequence = wide.columns.get_level_values("sequence") + observed = np.array([s == "" or (isinstance(s, (int, np.integer)) and s < upto_k) for s in sequence]) + + row = wide.iloc[0].to_numpy(dtype=float) + center = self.center_.to_numpy(dtype=float) + scale = self.scale_.to_numpy(dtype=float) + x_scaled = (row - center) / scale + + loadings = self.loadings_.to_numpy(dtype=float) + p_obs = loadings[observed, :] + x_obs = x_scaled[observed] + scores = np.linalg.pinv(p_obs) @ x_obs + + s = self.scaling_factor_for_scores_.to_numpy(dtype=float) + hotellings_t2 = float(np.sum((scores / s) ** 2)) + residual_obs = x_obs - p_obs @ scores + spe = float(np.sqrt(np.sum(residual_obs**2))) + + return Bunch( + scores=pd.Series(scores, index=self.scores_.columns, name="scores"), + hotellings_t2=hotellings_t2, + spe=spe, + ) + + def _coerce_online_initial_conditions( + self, initial_conditions: pd.Series | pd.DataFrame | None + ) -> pd.DataFrame | None: + """Normalise a single batch's initial conditions to a 1-row DataFrame. + + Accepts a Series (one value per initial condition) or a single-row + DataFrame, and validates presence against how the model was fitted. + """ + if self.n_initial_conditions_ == 0: + if initial_conditions is not None: + raise ValueError("The model was fitted without initial conditions; do not pass any.") + return None + if initial_conditions is None: + raise ValueError("The model was fitted with initial conditions; they are required here.") + frame = initial_conditions.to_frame().T if isinstance(initial_conditions, pd.Series) else initial_conditions + if frame.shape[0] != 1: + raise ValueError("initial_conditions for a single batch must have exactly one row.") + return frame.set_axis(["_online_"], axis=0) + def hotellings_t2_limit(self, conf_level: float = 0.95) -> float: """Hotelling's T2 limit at the given confidence level.""" check_is_fitted(self, "loadings_") diff --git a/src/process_improve/batch/_batch_plots.py b/src/process_improve/batch/_batch_plots.py index 9346c62e..4495d601 100644 --- a/src/process_improve/batch/_batch_plots.py +++ b/src/process_improve/batch/_batch_plots.py @@ -27,9 +27,10 @@ go = _MissingExtra("plotly", "plotting") # type: ignore[assignment] -from ..visualization.themes import DEFAULT_THEME, REFERENCE_LINE_COLOR +from ..visualization.themes import DEFAULT_THEME, LIMIT_LINE_COLOR, REFERENCE_LINE_COLOR if typing.TYPE_CHECKING: + from ._batch_monitor import BatchMonitor from ._batch_pca import BatchPCA @@ -180,3 +181,77 @@ def contribution_at_time_plot( yaxis_title="Contribution", ) return fig + + +def online_monitoring_plot( + monitor: BatchMonitor, + batch: pd.DataFrame, + statistic: str = "spe", + *, + initial_conditions: pd.Series | pd.DataFrame | None = None, + fig: go.Figure | None = None, +) -> go.Figure: + """Plot a batch's online SPE or T2 trace against the time-varying limit. + + Tracks the batch through the fitted + :class:`process_improve.batch.BatchMonitor` and draws its statistic over + time overlaid on the control limit and the mean good-batch trace, with the + alarm samples marked. This is the online (real-time) monitoring chart of + Nomikos and MacGregor. + + Parameters + ---------- + monitor : BatchMonitor + A fitted :class:`process_improve.batch.BatchMonitor`. + batch : pd.DataFrame + A single aligned batch to monitor. + statistic : {"spe", "t2"}, default="spe" + Which statistic to plot. + initial_conditions : pd.Series or pd.DataFrame, optional + The Z block for this batch; required if the model was fitted with one. + fig : plotly.graph_objects.Figure, optional + Figure to draw into; a new one is created when omitted. + + Returns + ------- + plotly.graph_objects.Figure + """ + statistic = statistic.lower() + if statistic not in {"spe", "t2"}: + raise ValueError(f"statistic must be 'spe' or 't2'; got {statistic!r}.") + + result = monitor.monitor(batch, initial_conditions=initial_conditions) + time = result.time + if statistic == "spe": + trace, limit, alarm = result.spe, result.spe_limit, result.spe_alarm + mean_trace = monitor.spe_mean_over_time_[: len(time)] + label = "SPE" + else: + trace, limit, alarm = result.hotellings_t2, result.t2_limit, result.t2_alarm + mean_trace = monitor.t2_mean_over_time_[: len(time)] + label = "Hotelling's T2" + + if fig is None: + fig = go.Figure() + fig.add_trace( + go.Scatter(x=time, y=mean_trace, mode="lines", name="good-batch mean", line={"color": REFERENCE_LINE_COLOR}) + ) + fig.add_trace( + go.Scatter(x=time, y=limit, mode="lines", name=f"{int(monitor.conf_level * 100)}% limit", + line={"color": LIMIT_LINE_COLOR, "dash": "dash"}) + ) + fig.add_trace(go.Scatter(x=time, y=trace, mode="lines", name=label, line={"color": "#2563EB"})) + if bool(np.any(alarm)): + fig.add_trace( + go.Scatter( + x=time[alarm], y=np.asarray(trace)[alarm], mode="markers", name="alarm", + marker={"color": LIMIT_LINE_COLOR, "size": 8, "symbol": "x"}, + ) + ) + fig.update_layout( + template=DEFAULT_THEME, + title=f"Online {label} monitoring", + xaxis_title="Time [sequence order]", + yaxis_title=label, + ) + return fig diff --git a/tests/batch/test_batch_monitor.py b/tests/batch/test_batch_monitor.py new file mode 100644 index 00000000..a8b773e6 --- /dev/null +++ b/tests/batch/test_batch_monitor.py @@ -0,0 +1,111 @@ +"""Tests for online (Nomikos-MacGregor) batch monitoring.""" + +import numpy as np +import pandas as pd +import pytest + +from process_improve.batch._batch_monitor import BatchMonitor +from process_improve.batch._batch_pca import BatchPCA +from process_improve.batch.datasets import load_nylon +from process_improve.batch.preprocessing import resample_to_reference + + +@pytest.fixture +def aligned_nylon() -> dict: + """Nylon batches resampled to a common length.""" + batches = load_nylon() + tags = list(next(iter(batches.values())).columns) + return resample_to_reference(batches, columns_to_align=tags, reference_batch=1) + + +@pytest.fixture +def good_model_and_monitor(aligned_nylon: dict) -> tuple[BatchPCA, BatchMonitor, dict]: + """Build a BatchPCA + fitted BatchMonitor from a good subset of nylon.""" + good = {k: v for k, v in aligned_nylon.items() if 1 <= k <= 36} + model = BatchPCA(n_components=3).fit(good) + monitor = BatchMonitor(model, conf_level=0.99).fit(good) + return model, monitor, good + + +def test_predict_online_matches_diagnose_when_complete(aligned_nylon: dict) -> None: + """At full observation the PMP projection equals the offline diagnosis.""" + model = BatchPCA(n_components=3).fit(aligned_nylon) + batch = aligned_nylon[48] + online = model.predict_online(batch, upto_k=model.n_timesteps_) + offline = model.diagnose({48: batch}) + np.testing.assert_allclose(online.scores.to_numpy(), offline.scores.iloc[0].to_numpy(), atol=1e-6) + assert np.isclose(online.hotellings_t2, offline.hotellings_t2.iloc[0, -1], atol=1e-6) + assert np.isclose(online.spe, offline.spe.iloc[0], atol=1e-6) + + +def test_predict_online_bad_upto_k(aligned_nylon: dict) -> None: + """upto_k outside the valid range is rejected.""" + model = BatchPCA(n_components=2).fit(aligned_nylon) + with pytest.raises(ValueError, match="upto_k"): + model.predict_online(aligned_nylon[1], upto_k=0) + with pytest.raises(ValueError, match="upto_k"): + model.predict_online(aligned_nylon[1], upto_k=model.n_timesteps_ + 1) + + +def test_monitor_limits_shapes(good_model_and_monitor: tuple) -> None: + """The fitted limits span every time sample.""" + _, monitor, _ = good_model_and_monitor + assert monitor.spe_limit_over_time_.shape == (monitor.n_timesteps_,) + assert monitor.t2_limit_over_time_.shape == (monitor.n_timesteps_,) + assert np.all(monitor.spe_limit_over_time_ > 0) + + +def test_good_batch_stays_mostly_in_limit(good_model_and_monitor: tuple) -> None: + """A good training batch rarely breaches its own control limits.""" + _, monitor, good = good_model_and_monitor + result = monitor.monitor(good[20]) + assert result.spe.shape == (monitor.n_timesteps_,) + # good batches define the 99% limit, so alarms should be rare + assert result.spe_alarm.mean() < 0.1 + + +def test_injected_fault_breaches_spe(good_model_and_monitor: tuple, aligned_nylon: dict) -> None: + """A large sustained fault in a batch trips the SPE limit in that window.""" + _, monitor, _ = good_model_and_monitor + faulty = aligned_nylon[10].copy() + tag = faulty.columns[7] + window = slice(60, 80) + faulty.iloc[window, faulty.columns.get_loc(tag)] += 5.0 * faulty[tag].std() + result = monitor.monitor(faulty) + # the fault window (1-based samples 61..80) should raise SPE alarms + fault_mask = (result.time >= 61) & (result.time <= 80) + assert result.spe_alarm[fault_mask].any() + + +def test_monitor_upto_k_truncates(good_model_and_monitor: tuple) -> None: + """upto_k truncates the returned traces to that many samples.""" + _, monitor, good = good_model_and_monitor + result = monitor.monitor(good[20], upto_k=50) + assert len(result.time) == 50 + assert len(result.spe) == 50 + assert result.time[-1] == 50 + + +def test_monitor_bad_upto_k(good_model_and_monitor: tuple) -> None: + """An out-of-range upto_k for monitor is rejected.""" + _, monitor, good = good_model_and_monitor + with pytest.raises(ValueError, match="upto_k"): + monitor.monitor(good[20], upto_k=monitor.n_timesteps_ + 5) + + +def test_monitor_with_initial_conditions(aligned_nylon: dict) -> None: + """Monitoring threads an initial-conditions block through end to end.""" + good = {k: v for k, v in aligned_nylon.items() if 1 <= k <= 20} + z = pd.DataFrame({"charge": [float(k) for k in good]}, index=list(good.keys())) + model = BatchPCA(n_components=2).fit(good, initial_conditions=z) + monitor = BatchMonitor(model, conf_level=0.95).fit(good, initial_conditions=z) + z_one = pd.Series({"charge": 5.0}) + result = monitor.monitor(good[5], initial_conditions=z_one) + assert len(result.spe) == monitor.n_timesteps_ + + +def test_predict_online_rejects_unexpected_initial_conditions(aligned_nylon: dict) -> None: + """Passing a Z block to a model fitted without one is rejected.""" + model = BatchPCA(n_components=2).fit({k: v for k, v in aligned_nylon.items() if k <= 10}) + with pytest.raises(ValueError, match="without initial conditions"): + model.predict_online(aligned_nylon[1], upto_k=10, initial_conditions=pd.Series({"charge": 1.0})) diff --git a/tests/batch/test_batch_plots.py b/tests/batch/test_batch_plots.py index cace198d..f83d9b6e 100644 --- a/tests/batch/test_batch_plots.py +++ b/tests/batch/test_batch_plots.py @@ -4,8 +4,13 @@ import pandas as pd import pytest +from process_improve.batch._batch_monitor import BatchMonitor from process_improve.batch._batch_pca import BatchPCA -from process_improve.batch._batch_plots import contribution_at_time_plot, time_varying_loading_plot +from process_improve.batch._batch_plots import ( + contribution_at_time_plot, + online_monitoring_plot, + time_varying_loading_plot, +) from process_improve.batch.datasets import load_nylon from process_improve.batch.preprocessing import resample_to_reference @@ -94,3 +99,30 @@ def test_contribution_at_time_plot_out_of_range_time(fitted_model: BatchPCA) -> contributions = fitted_model.spe_contributions(fitted_model._scaled_wide(aligned, None)) with pytest.raises(ValueError, match="No contributions at time"): contribution_at_time_plot(contributions, k=fitted_model.n_timesteps_ + 100) + + +def test_online_monitoring_plot() -> None: + """The online monitoring plot draws the trace, limit, and good-batch mean.""" + batches = load_nylon() + tags = list(next(iter(batches.values())).columns) + aligned = resample_to_reference(batches, columns_to_align=tags, reference_batch=1) + good = {k: v for k, v in aligned.items() if 1 <= k <= 36} + model = BatchPCA(n_components=3).fit(good) + monitor = BatchMonitor(model, conf_level=0.99).fit(good) + fig = online_monitoring_plot(monitor, aligned[49], "spe") + assert isinstance(fig, go.Figure) + assert len(fig.data) >= 3 # good-batch mean, limit, and the batch trace + fig_t2 = online_monitoring_plot(monitor, aligned[49], "t2") + assert isinstance(fig_t2, go.Figure) + + +def test_online_monitoring_plot_bad_statistic() -> None: + """An unknown statistic name is rejected.""" + batches = load_nylon() + tags = list(next(iter(batches.values())).columns) + aligned = resample_to_reference(batches, columns_to_align=tags, reference_batch=1) + good = {k: v for k, v in aligned.items() if 1 <= k <= 20} + model = BatchPCA(n_components=2).fit(good) + monitor = BatchMonitor(model, conf_level=0.95).fit(good) + with pytest.raises(ValueError, match="statistic must be"): + online_monitoring_plot(monitor, aligned[1], "nonsense")