# Path: jr/stats/base.py
# def align_signals(a, b):
#     """Finds optimal delay to align two 1D signals
#     maximizes hstack((zeros(shift), b)) = a
# 
#     Parameters
#     ----------
#     a : np.array, shape(n)
#     b : np.array, shape(m)
# 
#     Output
#     ------
#     shift : int
#         Integer that maximizes hstack((zeros(shift), b)) - a = 0
#     """
#     # check inputs
#     a = np.asarray(a)
#     b = np.asarray(b)
#     if np.prod(a.ndim) > 1 or np.prod(b.ndim) > 1:
#         raise ValueError('Can only vectorize vectors')
#     # longest first
#     sign = 1
#     if len(b) > len(a):
#         sign = -1
#         a, b = b, a
#     r = cross_correlation_fft(a, b)
#     shift = np.argmax(r) - len(a) + len(a) / 2
#     # deal with odd / even lengths (b doubles in size by cross_correlation_fft)
#     if len(a) % 2 and len(b) % 2:
#         shift += 1
#     if len(a) > len(b) and len(a) % 2 and not(len(b) % 2):
#         shift += 1
#     return sign * shift
# 
# Path: jr/stats/base.py
# def fast_mannwhitneyu(Y, X, use_continuity=True, n_jobs=-1):
#     from mne.parallel import parallel_func
#     X = np.array(X)
#     Y = np.array(Y)
#     nx, ny = len(X), len(Y)
#     dims = X.shape
#     X = np.reshape(X, [nx, -1])
#     Y = np.reshape(Y, [ny, -1])
#     parallel, p_time_gen, n_jobs = parallel_func(_loop_mannwhitneyu, n_jobs)
#     n_chunks = np.min([n_jobs, X.shape[1]])
#     chunks = np.array_split(range(X.shape[1]), n_chunks)
#     out = parallel(p_time_gen(X[..., chunk],
#                               Y[..., chunk], use_continuity=use_continuity)
#                    for chunk in chunks)
#     # Unpack estimators into time slices X folds list of lists.
#     U, p_value = map(list, zip(*out))
#     U = np.hstack(U).reshape(dims[1:])
#     p_value = np.hstack(p_value).reshape(dims[1:])
#     AUC = U / (nx * ny)
#     # XXX FIXME this introduces a bug
#     # # correct directionality of U stats imposed by mannwhitneyu
#     # if nx > ny:
#     #     AUC = 1 - AUC
#     return U, p_value, AUC


import numpy as np
from numpy.testing import assert_equal, assert_array_almost_equal
from .. import align_signals
from .. import fast_mannwhitneyu
    from sklearn.metrics import roc_auc_score


def test_align_signal():
    a = np.asarray(np.random.rand(1000) > .9, float)
    for pad in [10, 11, 0]:
        b = np.hstack((np.zeros(pad), a))
        for a_, b_, sign in ((a, b, 1), (b, a, -1)):
            # default
            assert_equal(align_signals(a_, b_), sign * pad)
            # even / odd lengths
            assert_equal(align_signals(a_[:-1], b_[:-1]), sign * pad)
            assert_equal(align_signals(a_[:-1], b_), sign * pad)
            assert_equal(align_signals(a_, b_[:-1]), sign * pad)


def test_auc():
    for ii in range(10):
        X = np.random.rand(20, 50)
        y = np.random.randint(0, 2, 20)

# Path: jr/stats/base.py
# def align_signals(a, b):
#     """Finds optimal delay to align two 1D signals
#     maximizes hstack((zeros(shift), b)) = a
# 
#     Parameters
#     ----------
#     a : np.array, shape(n)
#     b : np.array, shape(m)
# 
#     Output
#     ------
#     shift : int
#         Integer that maximizes hstack((zeros(shift), b)) - a = 0
#     """
#     # check inputs
#     a = np.asarray(a)
#     b = np.asarray(b)
#     if np.prod(a.ndim) > 1 or np.prod(b.ndim) > 1:
#         raise ValueError('Can only vectorize vectors')
#     # longest first
#     sign = 1
#     if len(b) > len(a):
#         sign = -1
#         a, b = b, a
#     r = cross_correlation_fft(a, b)
#     shift = np.argmax(r) - len(a) + len(a) / 2
#     # deal with odd / even lengths (b doubles in size by cross_correlation_fft)
#     if len(a) % 2 and len(b) % 2:
#         shift += 1
#     if len(a) > len(b) and len(a) % 2 and not(len(b) % 2):
#         shift += 1
#     return sign * shift
# 
# Path: jr/stats/base.py
# def fast_mannwhitneyu(Y, X, use_continuity=True, n_jobs=-1):
#     from mne.parallel import parallel_func
#     X = np.array(X)
#     Y = np.array(Y)
#     nx, ny = len(X), len(Y)
#     dims = X.shape
#     X = np.reshape(X, [nx, -1])
#     Y = np.reshape(Y, [ny, -1])
#     parallel, p_time_gen, n_jobs = parallel_func(_loop_mannwhitneyu, n_jobs)
#     n_chunks = np.min([n_jobs, X.shape[1]])
#     chunks = np.array_split(range(X.shape[1]), n_chunks)
#     out = parallel(p_time_gen(X[..., chunk],
#                               Y[..., chunk], use_continuity=use_continuity)
#                    for chunk in chunks)
#     # Unpack estimators into time slices X folds list of lists.
#     U, p_value = map(list, zip(*out))
#     U = np.hstack(U).reshape(dims[1:])
#     p_value = np.hstack(p_value).reshape(dims[1:])
#     AUC = U / (nx * ny)
#     # XXX FIXME this introduces a bug
#     # # correct directionality of U stats imposed by mannwhitneyu
#     # if nx > ny:
#     #     AUC = 1 - AUC
#     return U, p_value, AUC


# Path: jr/stats/tests/test_base.py
import numpy as np
from numpy.testing import assert_equal, assert_array_almost_equal
from .. import align_signals
from .. import fast_mannwhitneyu
    from sklearn.metrics import roc_auc_score



def test_align_signal():
    a = np.asarray(np.random.rand(1000) > .9, float)
    for pad in [10, 11, 0]:
        b = np.hstack((np.zeros(pad), a))
        for a_, b_, sign in ((a, b, 1), (b, a, -1)):
            # default
            assert_equal(align_signals(a_, b_), sign * pad)
            # even / odd lengths
            assert_equal(align_signals(a_[:-1], b_[:-1]), sign * pad)
            assert_equal(align_signals(a_[:-1], b_), sign * pad)
            assert_equal(align_signals(a_, b_[:-1]), sign * pad)


def test_auc():
    for ii in range(10):
        X = np.random.rand(20, 50)
        y = np.random.randint(0, 2, 20)

        _, _, auc = fast_mannwhitneyu(X[y == 0, ...],