[REF] Refactor smoothing transformers into a sub-package

.readthedocs.yml

@@ -13,7 +13,7 @@ python:
 build:
   os: ubuntu-24.04
   tools:
-    python: "3.10"
+    python: "3.11"
 
 sphinx:
   configuration: docs/conf.py

aeon/transformations/series/_dft.py

@@ -4,12 +4,20 @@
 __all__ = ["DFTSeriesTransformer"]
 
 
-import numpy as np
+from deprecated.sphinx import deprecated
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from aeon.transformations.series.smoothing import DiscreteFourierApproximation
 
 
-class DFTSeriesTransformer(BaseSeriesTransformer):
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="DFTSeriesTransformer is deprecated and will be removed in v1.3.0. "
+    "Please use DiscreteFourierApproximation from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class DFTSeriesTransformer(DiscreteFourierApproximation):
     """Filter a times series using Discrete Fourier Approximation (DFT).
 
     Parameters
@@ -42,47 +50,4 @@ class DFTSeriesTransformer(BaseSeriesTransformer):
     (2, 100)
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(self, r=0.5, sort=False):
-        self.r = r
-        self.sort = sort
-        super().__init__(axis=1)
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        Parameters
-        ----------
-        X : np.ndarray
-            time series in shape (n_channels, n_timepoints)
-        y : ignored argument for interface compatibility
-
-        Returns
-        -------
-        transformed version of X
-        """
-        # Compute DFT
-        dft = np.fft.fft(X)
-
-        # Mask array of terms to keep and number of terms to keep
-        mask = np.zeros_like(dft, dtype=bool)
-        keep = max(int(self.r * dft.shape[1]), 1)
-
-        # If sort is set, sort the indices by the decreasing dft amplitude
-        if self.sort:
-            sorted_indices = np.argsort(np.abs(dft))[:, ::-1]
-            for i in range(dft.shape[0]):
-                mask[i, sorted_indices[i, 0:keep]] = True
-        # Else, keep the first terms
-        else:
-            mask[:, 0:keep] = True
-
-        # Invert DFT with masked terms
-        X_ = np.fft.ifft(dft * mask).real
-
-        return X_
+    pass

aeon/transformations/series/_exp_smoothing.py

@@ -3,14 +3,21 @@
 __maintainer__ = ["Datadote"]
 __all__ = ["ExpSmoothingSeriesTransformer"]
 
-from typing import Union
 
-import numpy as np
+from deprecated.sphinx import deprecated
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from aeon.transformations.series.smoothing import ExponentialSmoothing
 
 
-class ExpSmoothingSeriesTransformer(BaseSeriesTransformer):
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="ExpSmoothingSeriesTransformer is deprecated and will be removed in v1.3.0. "
+    "Please use ExponentialSmoothing from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class ExpSmoothingSeriesTransformer(ExponentialSmoothing):
     """Filter a time series using exponential smoothing.
 
     - Exponential smoothing (EXP) is a generalisaton of moving average smoothing that
@@ -54,42 +61,4 @@ class ExpSmoothingSeriesTransformer(BaseSeriesTransformer):
      [10.     9.5    8.75   7.875]]
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(
-        self, alpha: float = 0.2, window_size: Union[int, float, None] = None
-    ) -> None:
-        if not 0 <= alpha <= 1:
-            raise ValueError(f"alpha must be in range [0, 1], got {alpha}")
-        if window_size is not None and window_size <= 0:
-            raise ValueError(f"window_size must be > 0, got {window_size}")
-        super().__init__(axis=1)
-        self.alpha = alpha if window_size is None else 2.0 / (window_size + 1)
-        self.window_size = window_size
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        private _transform containing core logic, called from transform
-
-        Parameters
-        ----------
-        X : np.ndarray
-            Data to be transformed
-        y : ignored argument for interface compatibility
-            Additional data, e.g., labels for transformation
-
-        Returns
-        -------
-        Xt: 2D np.ndarray
-            transformed version of X
-        """
-        Xt = np.zeros_like(X, dtype="float")
-        Xt[:, 0] = X[:, 0]
-        for i in range(1, Xt.shape[1]):
-            Xt[:, i] = self.alpha * X[:, i] + (1 - self.alpha) * Xt[:, i - 1]
-        return Xt
+    pass

aeon/transformations/series/_gauss.py

@@ -4,12 +4,20 @@
 __all__ = ["GaussSeriesTransformer"]
 
 
-from scipy.ndimage import gaussian_filter1d
+from deprecated.sphinx import deprecated
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from aeon.transformations.series.smoothing import GaussianFilter
 
 
-class GaussSeriesTransformer(BaseSeriesTransformer):
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="GaussSeriesTransformer is deprecated and will be removed in v1.3.0. "
+    "Please use GaussianFilter from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class GaussSeriesTransformer(GaussianFilter):
     """Filter a times series using Gaussian filter.
 
     Parameters
@@ -45,31 +53,4 @@ class GaussSeriesTransformer(BaseSeriesTransformer):
     (2, 100)
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(self, sigma=1, order=0):
-        self.sigma = sigma
-        self.order = order
-        super().__init__(axis=1)
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        Parameters
-        ----------
-        X : np.ndarray
-            time series in shape (n_channels, n_timepoints)
-        y : ignored argument for interface compatibility
-
-        Returns
-        -------
-        transformed version of X
-        """
-        # Compute Gaussian filter
-        X_ = gaussian_filter1d(X, self.sigma, self.axis, self.order)
-
-        return X_
+    pass

aeon/transformations/series/_moving_average.py

@@ -3,12 +3,21 @@
 __maintainer__ = ["Datadote"]
 __all__ = ["MovingAverageSeriesTransformer"]
 
-import numpy as np
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from deprecated.sphinx import deprecated
 
+from aeon.transformations.series.smoothing import MovingAverage
 
-class MovingAverageSeriesTransformer(BaseSeriesTransformer):
+
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="MovingAverageSeriesTransformer is deprecated and will be removed in "
+    "v1.3.0. Please use MovingAverage from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class MovingAverageSeriesTransformer(MovingAverage):
     """Calculate the moving average of an array of numbers.
 
     Slides a window across the input array, and returns the averages for each window.
@@ -41,38 +50,4 @@ class MovingAverageSeriesTransformer(BaseSeriesTransformer):
     [[-2.5 -1.5 -0.5  0.5  1.5  2.5]]
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(self, window_size: int = 5) -> None:
-        super().__init__(axis=0)
-        if window_size <= 0:
-            raise ValueError(f"window_size must be > 0, got {window_size}")
-        self.window_size = window_size
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        private _transform containing core logic, called from transform
-
-        Parameters
-        ----------
-        X : np.ndarray
-            Data to be transformed
-        y : ignored argument for interface compatibility
-            Additional data, e.g., labels for transformation
-
-        Returns
-        -------
-        Xt: 2D np.ndarray
-            transformed version of X
-        """
-        csum = np.cumsum(X, axis=0)
-        csum[self.window_size :, :] = (
-            csum[self.window_size :, :] - csum[: -self.window_size, :]
-        )
-        Xt = csum[self.window_size - 1 :, :] / self.window_size
-        return Xt
+    pass

aeon/transformations/series/_sg.py

@@ -4,12 +4,20 @@
 __all__ = ["SGSeriesTransformer"]
 
 
-from scipy.signal import savgol_filter
+from deprecated.sphinx import deprecated
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from aeon.transformations.series.smoothing import SavitzkyGolayFilter
 
 
-class SGSeriesTransformer(BaseSeriesTransformer):
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="SGSeriesTransformer is deprecated and will be removed in v1.3.0. "
+    "Please use SavitzkyGolayFilter from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class SGSeriesTransformer(SavitzkyGolayFilter):
     """Filter a times series using Savitzky-Golay (SG).
 
     Parameters
@@ -45,31 +53,4 @@ class SGSeriesTransformer(BaseSeriesTransformer):
     (2, 100)
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(self, window_length=5, polyorder=2):
-        self.window_length = window_length
-        self.polyorder = polyorder
-        super().__init__(axis=1)
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        Parameters
-        ----------
-        X : np.ndarray
-            time series in shape (n_channels, n_timepoints)
-        y : ignored argument for interface compatibility
-
-        Returns
-        -------
-        transformed version of X
-        """
-        # Compute SG
-        X_ = savgol_filter(X, self.window_length, self.polyorder)
-
-        return X_
+    pass

aeon/transformations/series/_siv.py

@@ -4,13 +4,20 @@
 __all__ = ["SIVSeriesTransformer"]
 
 
-import numpy as np
-from scipy.ndimage import median_filter
+from deprecated.sphinx import deprecated
 
-from aeon.transformations.series.base import BaseSeriesTransformer
+from aeon.transformations.series.smoothing import RecursiveMedianSieve
 
 
-class SIVSeriesTransformer(BaseSeriesTransformer):
+# TODO: Remove in v1.3.0
+@deprecated(
+    version="1.2.0",
+    reason="SIVSeriesTransformer is deprecated and will be removed in v1.3.0. "
+    "Please use RecursiveMedianSieve from "
+    "transformations.series.smoothing instead.",
+    category=FutureWarning,
+)
+class SIVSeriesTransformer(RecursiveMedianSieve):
     """Filter a times series using Recursive Median Sieve (SIV).
 
     Parameters
@@ -48,40 +55,4 @@ class SIVSeriesTransformer(BaseSeriesTransformer):
     (2, 100)
     """
 
-    _tags = {
-        "capability:multivariate": True,
-        "X_inner_type": "np.ndarray",
-        "fit_is_empty": True,
-    }
-
-    def __init__(self, window_length=None):
-        self.window_length = window_length
-        super().__init__(axis=1)
-
-    def _transform(self, X, y=None):
-        """Transform X and return a transformed version.
-
-        Parameters
-        ----------
-        X : np.ndarray
-            time series in shape (n_channels, n_timepoints)
-        y : ignored argument for interface compatibility
-
-        Returns
-        -------
-        transformed version of X
-        """
-        window_length = self.window_length
-        if window_length is None:
-            window_length = [3, 5, 7]
-        if not isinstance(window_length, list):
-            window_length = [window_length]
-
-        # Compute SIV
-        X_ = X
-
-        for w in window_length:
-            footprint = np.ones((1, w))
-            X_ = median_filter(X_, footprint=footprint)
-
-        return X_
+    pass