Repository: aeon-toolkit/aeon
Pull request #2001 (2024-08-21 14:50:25)

[ENH] Add a warping path series transformer 
After finding a warping path from elastic measures, this transformer re-aligns a series to adjust for warping corrections.
For example, on two samples of the ECG200 dataset:
![image](https://github.com/user-attachments/assets/161d3b5e-4c2a-4d5e-86ea-964a6f9dcbbe)

With a dtw path as such:
![image](https://github.com/user-attachments/assets/d2d66f96-c704-4c81-8b81-7cbfe775bc6c)

After transforming both series (see example in docstring), we get these two new series:
![image](https://github.com/user-attachments/assets/00d444b6-e4fe-4287-b628-726e1dfd493e)

And this is the new dtw path between them:
![image](https://github.com/user-attachments/assets/81973b62-5003-400e-85ca-0e5a7bfeceda)


aeon-actions-bot[bot]: 
## Thank you for contributing to `aeon`

I have added the following labels to this PR based on the title: **[ $\\color{#FEF1BE}{\\textsf{enhancement}}$ ]**.
I would have added the following labels to this PR based on the changes made: **[ $\\color{#41A8F6}{\\textsf{transformations}}$ ]**, however some package labels are already present.

The [Checks](https://github.com/aeon-toolkit/aeon/pull/2001/checks) tab will show the status of our automated tests. You can click on individual test runs in the tab or "Details" in the panel below to see more information if there is a failure.

If our `pre-commit` code quality check fails, any trivial fixes will automatically be pushed to your PR unless it is a draft.

Don't hesitate to ask questions on the `aeon` [Slack](https://join.slack.com/t/aeon-toolkit/shared_invite/zt-22vwvut29-HDpCu~7VBUozyfL_8j3dLA) channel if you have any.

### PR CI actions

These checkboxes will add labels to enable/disable CI functionality for this PR. This may not take effect immediately, and a new commit may be required to run the new configuration.

- [ ] Run `pre-commit` checks for all files
- [ ] Run all `pytest` tests and configurations
- [ ] Run all notebook example tests
- [ ] Run numba-disabled `codecov` tests
- [ ] Stop automatic `pre-commit` fixes (always disabled for drafts)
- [ ] Push an empty commit to re-run CI checks
    

Patch:
diff --git a/.all-contributorsrc b/.all-contributorsrc
index d6fff1a3..d6ea0586 100644
--- a/.all-contributorsrc
+++ b/.all-contributorsrc
@@ -2120,7 +2120,22 @@
         "bug",
         "doc",
         "test",
-        "maintenance"
+        "maintenance",
+        "review",
+        "talk",
+        "tutorial",
+        "mentoring",
+        "example"
+      ]
+    },
+    {
+      "login": "maxwell1503",
+      "name": " Maxime Devanne",
+      "avatar_url": "https://avatars.githubusercontent.com/u/28450447",
+      "profile": "https://github.com/maxwell1503",
+      "contributions": [
+        "code",
+        "bug"
       ]
     },
     {
diff --git a/aeon/transformations/series/__init__.py b/aeon/transformations/series/__init__.py
index cd83a705..bdd324bf 100644
--- a/aeon/transformations/series/__init__.py
+++ b/aeon/transformations/series/__init__.py
@@ -18,6 +18,7 @@ __all__ = [
     "ScaledLogitSeriesTransformer",
     "SIVSeriesTransformer",
     "PCASeriesTransformer",
+    "WarpingSeriesTransformer",
 ]
 
 from aeon.transformations.series._acf import (
@@ -36,4 +37,5 @@ from aeon.transformations.series._pca import PCASeriesTransformer
 from aeon.transformations.series._pla import PLASeriesTransformer
 from aeon.transformations.series._scaled_logit import ScaledLogitSeriesTransformer
 from aeon.transformations.series._sg import SGSeriesTransformer
+from aeon.transformations.series._warping import WarpingSeriesTransformer
 from aeon.transformations.series.base import BaseSeriesTransformer
diff --git a/aeon/transformations/series/_warping.py b/aeon/transformations/series/_warping.py
new file mode 100644
index 00000000..d1289598
--- /dev/null
+++ b/aeon/transformations/series/_warping.py
@@ -0,0 +1,87 @@
+"""Warping Transformer using elastic measures for warping path."""
+
+__maintainer__ = ["hadifawaz1999"]
+__all__ = ["WarpingSeriesTransformer"]
+
+import numpy as np
+
+from aeon.transformations.series.base import BaseSeriesTransformer
+
+
+class WarpingSeriesTransformer(BaseSeriesTransformer):
+    """Warping Path Transformer.
+
+    This transformer produces a longer version of the input series
+    following the warping path produced by an elastic measure.
+    The transformer assumes the path is pre-computed between the input
+    series and another one.
+
+    Parameters
+    ----------
+    series_index : int, default = 0
+        The index of the series, either 0 or 1 to choose from the warping
+        path. Given the path is generated using two series, the user
+        should choose which one is being transformed.
+    warping_path : List[Tuple[int,int]], default = None
+        The warping path used to transform the series.
+        If None, the output series is returned as is.
+
+    Examples
+    --------
+    >>> from aeon.transformations.series import WarpingSeriesTransformer
+    >>> from aeon.distances import dtw_alignment_path
+    >>> import numpy as np
+    >>> x = np.random.normal((2, 100))
+    >>> y = np.random.normal((2, 100))
+    >>> dtw_path, _ = dtw_alignment_path(x, y)
+    >>> x_transformed = WarpingSeriesTransformer(
+    ... series_index=0, warping_path=dtw_path).fit_transform(x)
+    >>> y_transformed = WarpingSeriesTransformer(
+    ... series_index=1, warping_path=dtw_path).fit_transform(y)
+    """
+
+    _tags = {
+        "X_inner_type": "np.ndarray",
+        "capability:multivariate": True,
+        "fit_is_empty": True,
+    }
+
+    def __init__(
+        self,
+        series_index: int = 0,
+        warping_path: list[tuple[int, int]] = None,
+    ) -> None:
+
+        self.series_index = series_index
+        self.warping_path = warping_path
+
+        super().__init__(axis=1)
+
+    def _transform(self, X: np.ndarray, y=None) -> np.ndarray:
+        """Transform X and return a transformed version.
+
+        Parameters
+        ----------
+        X : np.ndarray
+            time series of shape (n_channels, n_timepoints)
+        y : ignored argument for interface compatibility
+
+        Returns
+        -------
+        aligned_series : np.ndarray of shape n_channels, len(warping_path)
+        """
+        if self.warping_path is None:
+            return X
+
+        indices_0, indices_1 = zip(*self.warping_path)
+
+        if self.series_index == 0:
+            indices_ = np.array(indices_0)
+        elif self.series_index == 1:
+            indices_ = np.array(indices_1)
+        else:
+            raise ValueError("The parameter series_index can only be 0 or 1.")
+
+        aligned_series = X[:, indices_]
+
+        return aligned_series
diff --git a/docs/api_reference/transformations.rst b/docs/api_reference/transformations.rst
index 87a73de4..2caeaf4f 100644
--- a/docs/api_reference/transformations.rst
+++ b/docs/api_reference/transformations.rst
@@ -210,6 +210,17 @@ Filtering and denoising
 
     SIVSeriesTransformer
 
+Distance Based
+~~~~~~~~~~~~~~~~~~~~~~
+
+.. currentmodule:: aeon.transformations.series._warping
+
+.. autosummary::
+    :toctree: auto_generated/
+    :template: class.rst
+
+    WarpingSeriesTransformer
+
 Slope
 ~~~~~~~~~~~~~~~~~~~~~~
 


Test patch:
diff --git a/aeon/transformations/series/tests/test_warping.py b/aeon/transformations/series/tests/test_warping.py
new file mode 100644
index 00000000..ca14ab65
--- /dev/null
+++ b/aeon/transformations/series/tests/test_warping.py
@@ -0,0 +1,31 @@
+"""Test Warping series transformer."""
+
+__maintainer__ = ["hadifawaz1999"]
+
+import pytest
+
+from aeon.clustering.averaging import VALID_BA_METRICS
+from aeon.distances import get_alignment_path_function
+from aeon.testing.data_generation import make_example_2d_numpy_series
+from aeon.transformations.series import WarpingSeriesTransformer
+
+
+@pytest.mark.parametrize("distance", VALID_BA_METRICS)
+def test_warping_path_transformer(distance):
+    """Test the functionality of Warping transformation."""
+    x = make_example_2d_numpy_series(n_timepoints=20, n_channels=2)
+    y = make_example_2d_numpy_series(n_timepoints=20, n_channels=2)
+
+    alignment_path_function = get_alignment_path_function(metric=distance)
+
+    warping_path = alignment_path_function(x, y)[0]
+
+    new_x = WarpingSeriesTransformer(
+        series_index=0, warping_path=warping_path
+    ).fit_transform(x)
+    new_y = WarpingSeriesTransformer(
+        series_index=1, warping_path=warping_path
+    ).fit_transform(y)
+
+    assert int(new_x.shape[1]) == len(warping_path)
+    assert int(new_y.shape[1]) == len(warping_path)