<commit_before>"""
The :mod:`tslearn.neighbors` module gathers nearest neighbor algorithms using
time series metrics.
"""

import numpy
from sklearn import neighbors
from sklearn.neighbors import (KNeighborsClassifier, NearestNeighbors,
                               KNeighborsRegressor)
from sklearn.neighbors.base import KNeighborsMixin
from sklearn.utils import check_array
from sklearn.utils.validation import check_is_fitted
from scipy.spatial.distance import cdist as scipy_cdist

from tslearn.metrics import cdist_dtw, cdist_soft_dtw, VARIABLE_LENGTH_METRICS
from tslearn.utils import (to_time_series_dataset, to_sklearn_dataset,
                           check_dims)
from tslearn.bases import BaseModelPackage

neighbors.VALID_METRICS['brute'].extend(['dtw', 'softdtw'])


class KNeighborsTimeSeriesMixin(KNeighborsMixin):
    """Mixin for k-neighbors searches on Time Series."""

    def kneighbors(self, X=None, n_neighbors=None, return_distance=True):
        """Finds the K-neighbors of a point.

        Returns indices of and distances to the neighbors of each point.

        Parameters
        ----------
        X : array-like, shape (n_ts, sz, d)
            The query time series.
            If not provided, neighbors of each indexed point are returned.
            In this case, the query point is not considered its own neighbor.
        n_neighbors : int
            Number of neighbors to get (default is the value passed to the
            constructor).
        return_distance : boolean, optional. Defaults to True.
            If False, distances will not be returned

        Returns
        -------
        dist : array
            Array representing the distance to points, only present if
            return_distance=True
        ind : array
            Indices of the nearest points in the population matrix.
        """
        if self.metric_params is not None:
            metric_params = self.metric_params.copy()
            if "n_jobs" in metric_params.keys():
                del metric_params["n_jobs"]
            if "verbose" in metric_params.keys():
                del metric_params["verbose"]
        else:
            metric_params = {}
        self_neighbors = False
        if n_neighbors is None:
            n_neighbors = self.n_neighbors
        if X is None:
            X = self._X_fit
            self_neighbors = True
        if self.metric == "precomputed":
            full_dist_matrix = X
        else:
            parallelize = False
            if self.metric == "dtw" or self.metric == cdist_dtw:
                cdist_fun = cdist_dtw
                parallelize = True
            elif self.metric == "softdtw" or self.metric == cdist_soft_dtw:
                cdist_fun = cdist_soft_dtw
            elif self.metric in ["euclidean", "sqeuclidean", "cityblock"]:
                def cdist_fun(X, Xp):
                    return scipy_cdist(X.reshape((X.shape[0], -1)),
                                       Xp.reshape((Xp.shape[0], -1)),
                                       metric=self.metric)
            else:
                raise ValueError("Unrecognized time series metric string: %s "
                                 "(should be one of 'dtw', 'softdtw', "
                                 "'euclidean', 'sqeuclidean' "
                                 "or 'cityblock')" % self.metric)

            if X.ndim == 2:  # sklearn-format case
                X = X.reshape((X.shape[0], -1, self._d))
                fit_X = self._X_fit.reshape((self._X_fit.shape[0],
                                             -1,
                                             self._d))
            else:
                fit_X = self._X_fit
            if parallelize:
                full_dist_matrix = cdist_fun(X, fit_X, n_jobs=self.n_jobs,
                                             verbose=self.verbose, 
                                             **metric_params)
            else:
                full_dist_matrix = cdist_fun(X, fit_X, **metric_params)
        ind = numpy.argsort(full_dist_matrix, axis=1)

        if self_neighbors:
            ind = ind[:, 1:]
        if n_neighbors > full_dist_matrix.shape[1]:
            n_neighbors = full_dist_matrix.shape[1]
        ind = ind[:, :n_neighbors]

        n_ts = X.shape[0]
        sample_range = numpy.arange(n_ts)[:, None]
        dist = full_dist_matrix[sample_range, ind]

        if return_distance:
            return dist, ind
        else:
            return ind


class KNeighborsTimeSeries(KNeighborsTimeSeriesMixin, NearestNeighbors,
                           BaseModelPackage):
    """Unsupervised learner for implementing neighbor searches for Time Series.

    Parameters
    ----------
    n_neighbors : int (default: 5)
        Number of nearest neighbors to be considered for the decision.
    metric : {'dtw', 'softdtw', 'euclidean', 'sqeuclidean', 'cityblock'}
    (default: 'dtw')
        Metric to be used at the core of the nearest neighbor procedure.
        DTW is described in more details in :mod:`tslearn.metrics`.
        Other metrics are described in `scipy.spatial.distance doc
        <https://docs.scipy.org/doc/scipy/reference/spatial.distance.html>`_.
    metric_params : dict or None (default: None)
        Dictionnary of metric parameters.
        For metrics that accept parallelization of the cross-distance matrix
        computations, `n_jobs` and `verbose` keys passed in `metric_params` 
        are overridden by the `n_jobs` and `verbose` arguments.

    n_jobs : int or None, optional (default=None)
        The number of jobs to run in parallel for cross-distance matrix
        computations.
        Ignored if the cross-distance matrix cannot be computed using
        parallelization.
        ``None`` means 1 unless in a :obj:`joblib.parallel_backend` context.
        ``-1`` means using all processors. See scikit-learns'
        `Glossary <https://scikit-learn.org/stable/glossary.html#term-n-jobs>`__
        for more details.
        
    Notes
    -----
        The training data are saved to disk if this model is
        serialized and may result in a large model file if the training
        dataset is large.

    Examples
    --------
    >>> time_series = to_time_series_dataset([[1, 2, 3, 4],
    ...                                       [3, 3, 2, 0],
    ...                                       [1, 2, 2, 4]])
    >>> knn = KNeighborsTimeSeries(n_neighbors=1).fit(time_series)
    >>> dataset = to_time_series_dataset([[1, 1, 2, 2, 2, 3, 4]])
    >>> dist, ind = knn.kneighbors(dataset, return_distance=True)
    >>> dist
    array([[0.]])
    >>> ind
    array([[0]])
    >>> knn2 = KNeighborsTimeSeries(n_neighbors=10,
    ...                             metric="euclidean").fit(time_series)
    >>> knn2.kneighbors(return_distance=False)
    array([[2, 1],
           [2, 0],
           [0, 1]])    
    """  # noqa: E501
    def __init__(self, n_neighbors=5, metric="dtw", metric_params=None,
                 n_jobs=None, verbose=0):
        NearestNeighbors.__init__(self,
                                  n_neighbors=n_neighbors,
                                  algorithm='brute')
        self.metric = metric
        self.metric_params = metric_params
        self.n_jobs = n_jobs
        self.verbose = verbose

    def _is_fitted(self):
        if self.metric in VARIABLE_LENGTH_METRICS:
            check_is_fitted(self, '_ts_fit')
        else:<commit_msg>remote whitespace<commit_after>"""
The :mod:`tslearn.neighbors` module gathers nearest neighbor algorithms using
time series metrics.
"""

import numpy
from sklearn import neighbors
from sklearn.neighbors import (KNeighborsClassifier, NearestNeighbors,
                               KNeighborsRegressor)
from sklearn.neighbors.base import KNeighborsMixin
from sklearn.utils import check_array
from sklearn.utils.validation import check_is_fitted
from scipy.spatial.distance import cdist as scipy_cdist

from tslearn.metrics import cdist_dtw, cdist_soft_dtw, VARIABLE_LENGTH_METRICS
from tslearn.utils import (to_time_series_dataset, to_sklearn_dataset,
                           check_dims)
from tslearn.bases import BaseModelPackage

neighbors.VALID_METRICS['brute'].extend(['dtw', 'softdtw'])


class KNeighborsTimeSeriesMixin(KNeighborsMixin):
    """Mixin for k-neighbors searches on Time Series."""

    def kneighbors(self, X=None, n_neighbors=None, return_distance=True):
        """Finds the K-neighbors of a point.

        Returns indices of and distances to the neighbors of each point.

        Parameters
        ----------
        X : array-like, shape (n_ts, sz, d)
            The query time series.
            If not provided, neighbors of each indexed point are returned.
            In this case, the query point is not considered its own neighbor.
        n_neighbors : int
            Number of neighbors to get (default is the value passed to the
            constructor).
        return_distance : boolean, optional. Defaults to True.
            If False, distances will not be returned

        Returns
        -------
        dist : array
            Array representing the distance to points, only present if
            return_distance=True
        ind : array
            Indices of the nearest points in the population matrix.
        """
        if self.metric_params is not None:
            metric_params = self.metric_params.copy()
            if "n_jobs" in metric_params.keys():
                del metric_params["n_jobs"]
            if "verbose" in metric_params.keys():
                del metric_params["verbose"]
        else:
            metric_params = {}
        self_neighbors = False
        if n_neighbors is None:
            n_neighbors = self.n_neighbors
        if X is None:
            X = self._X_fit
            self_neighbors = True
        if self.metric == "precomputed":
            full_dist_matrix = X
        else:
            parallelize = False
            if self.metric == "dtw" or self.metric == cdist_dtw:
                cdist_fun = cdist_dtw
                parallelize = True
            elif self.metric == "softdtw" or self.metric == cdist_soft_dtw:
                cdist_fun = cdist_soft_dtw
            elif self.metric in ["euclidean", "sqeuclidean", "cityblock"]:
                def cdist_fun(X, Xp):
                    return scipy_cdist(X.reshape((X.shape[0], -1)),
                                       Xp.reshape((Xp.shape[0], -1)),
                                       metric=self.metric)
            else:
                raise ValueError("Unrecognized time series metric string: %s "
                                 "(should be one of 'dtw', 'softdtw', "
                                 "'euclidean', 'sqeuclidean' "
                                 "or 'cityblock')" % self.metric)

            if X.ndim == 2:  # sklearn-format case
                X = X.reshape((X.shape[0], -1, self._d))
                fit_X = self._X_fit.reshape((self._X_fit.shape[0],
                                             -1,
                                             self._d))
            else:
                fit_X = self._X_fit
            if parallelize:
                full_dist_matrix = cdist_fun(X, fit_X, n_jobs=self.n_jobs,
                                             verbose=self.verbose, 
                                             **metric_params)
            else:
                full_dist_matrix = cdist_fun(X, fit_X, **metric_params)
        ind = numpy.argsort(full_dist_matrix, axis=1)

        if self_neighbors:
            ind = ind[:, 1:]
        if n_neighbors > full_dist_matrix.shape[1]:
            n_neighbors = full_dist_matrix.shape[1]
        ind = ind[:, :n_neighbors]

        n_ts = X.shape[0]
        sample_range = numpy.arange(n_ts)[:, None]
        dist = full_dist_matrix[sample_range, ind]

        if return_distance:
            return dist, ind
        else:
            return ind


class KNeighborsTimeSeries(KNeighborsTimeSeriesMixin, NearestNeighbors,
                           BaseModelPackage):
    """Unsupervised learner for implementing neighbor searches for Time Series.

    Parameters
    ----------
    n_neighbors : int (default: 5)
        Number of nearest neighbors to be considered for the decision.
    metric : {'dtw', 'softdtw', 'euclidean', 'sqeuclidean', 'cityblock'}
    (default: 'dtw')
        Metric to be used at the core of the nearest neighbor procedure.
        DTW is described in more details in :mod:`tslearn.metrics`.
        Other metrics are described in `scipy.spatial.distance doc
        <https://docs.scipy.org/doc/scipy/reference/spatial.distance.html>`_.
    metric_params : dict or None (default: None)
        Dictionnary of metric parameters.
        For metrics that accept parallelization of the cross-distance matrix
        computations, `n_jobs` and `verbose` keys passed in `metric_params` 
        are overridden by the `n_jobs` and `verbose` arguments.

    n_jobs : int or None, optional (default=None)
        The number of jobs to run in parallel for cross-distance matrix
        computations.
        Ignored if the cross-distance matrix cannot be computed using
        parallelization.
        ``None`` means 1 unless in a :obj:`joblib.parallel_backend` context.
        ``-1`` means using all processors. See scikit-learns'
        `Glossary <https://scikit-learn.org/stable/glossary.html#term-n-jobs>`__
        for more details.
        
    Notes
    -----
        The training data are saved to disk if this model is
        serialized and may result in a large model file if the training
        dataset is large.

    Examples
    --------
    >>> time_series = to_time_series_dataset([[1, 2, 3, 4],
    ...                                       [3, 3, 2, 0],
    ...                                       [1, 2, 2, 4]])
    >>> knn = KNeighborsTimeSeries(n_neighbors=1).fit(time_series)
    >>> dataset = to_time_series_dataset([[1, 1, 2, 2, 2, 3, 4]])
    >>> dist, ind = knn.kneighbors(dataset, return_distance=True)
    >>> dist
    array([[0.]])
    >>> ind
    array([[0]])
    >>> knn2 = KNeighborsTimeSeries(n_neighbors=10,
    ...                             metric="euclidean").fit(time_series)
    >>> knn2.kneighbors(return_distance=False)
    array([[2, 1],
           [2, 0],
           [0, 1]])
    """  # noqa: E501
    def __init__(self, n_neighbors=5, metric="dtw", metric_params=None,
                 n_jobs=None, verbose=0):
        NearestNeighbors.__init__(self,
                                  n_neighbors=n_neighbors,
                                  algorithm='brute')
        self.metric = metric
        self.metric_params = metric_params
        self.n_jobs = n_jobs
        self.verbose = verbose

    def _is_fitted(self):
        if self.metric in VARIABLE_LENGTH_METRICS:
            check_is_fitted(self, '_ts_fit')
        else:
{'fraction_non_alphanumeric': 0.09234687039270875, 'fraction_numerical': 0.007237635705669481, 'mean_word_length': 1.8071858540255832, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 57, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 40, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '770116', 'n_tokens_mistral': 4424, 'n_tokens_neox': 4163, 'n_words': 1160}