def cli(matfiles, savename, rec_type, infosrc):
    """
    Convert brainstorm epochs to mne.Epochs object
    """
    if infosrc:
        if rec_type is 'ds':
            from mne.io import read_raw_ctf as read_raw
        elif rec_type is 'fif':
            from mne.io import Raw as read_raw
        with nostdout():
            raw_with_info = read_raw(infosrc)

    isFirst = True
    for fname in matfiles:
        with nostdout():
            mat_epoch = sio.loadmat(fname)
            # click.echo(mat_epoch)
        if isFirst:
            data = mat_epoch['F']
            times = mat_epoch['Time']
            # print times[0,-1]
            isFirst = False
        else:
            data = np.dstack((data, mat_epoch['F']))
        # click.echo(data.shape)
    data = data.transpose((2,0,1))


    n_channels = data.shape[1]
    sfreq = times.shape[1] / (times[0,-1] + times[0,1])
    
    
    if infosrc:
        if rec_type is 'ds':
            from mne.io import read_raw_ctf as read_raw
        elif rec_type is 'fif':
            from mne.io import Raw as read_raw

        with nostdout():
            raw_with_info = read_raw(infosrc)
        good_info = raw_with_info.info
        # click.echo(len(good_info['ch_names']))

        ch_types = [channel_type(good_info, idx) for idx in range(n_channels)]

        # click.echo(len(ch_types))

        info = create_info(ch_names=good_info['ch_names'], sfreq=sfreq, ch_types=ch_types)
    else:
        ch_types='mag'
        info = create_info(n_channels, sfreq, ch_types)

    with nostdout():
        epochs = EpochsArray(data, info)
    epochs.save(savename)

def test_picks_by_channels():
    """Test creating pick_lists."""
    rng = np.random.RandomState(909)

    test_data = rng.random_sample((4, 2000))
    ch_names = ['MEG %03d' % i for i in [1, 2, 3, 4]]
    ch_types = ['grad', 'mag', 'mag', 'eeg']
    sfreq = 250.0
    info = create_info(ch_names=ch_names, sfreq=sfreq, ch_types=ch_types)
    _assert_channel_types(info)
    raw = RawArray(test_data, info)

    pick_list = _picks_by_type(raw.info)
    assert_equal(len(pick_list), 3)
    assert_equal(pick_list[0][0], 'mag')
    pick_list2 = _picks_by_type(raw.info, meg_combined=False)
    assert_equal(len(pick_list), len(pick_list2))
    assert_equal(pick_list2[0][0], 'mag')

    pick_list2 = _picks_by_type(raw.info, meg_combined=True)
    assert_equal(len(pick_list), len(pick_list2) + 1)
    assert_equal(pick_list2[0][0], 'meg')

    test_data = rng.random_sample((4, 2000))
    ch_names = ['MEG %03d' % i for i in [1, 2, 3, 4]]
    ch_types = ['mag', 'mag', 'mag', 'mag']
    sfreq = 250.0
    info = create_info(ch_names=ch_names, sfreq=sfreq, ch_types=ch_types)
    raw = RawArray(test_data, info)
    # This acts as a set, not an order
    assert_array_equal(pick_channels(info['ch_names'], ['MEG 002', 'MEG 001']),
                       [0, 1])

    # Make sure checks for list input work.
    pytest.raises(ValueError, pick_channels, ch_names, 'MEG 001')
    pytest.raises(ValueError, pick_channels, ch_names, ['MEG 001'], 'hi')

    pick_list = _picks_by_type(raw.info)
    assert_equal(len(pick_list), 1)
    assert_equal(pick_list[0][0], 'mag')
    pick_list2 = _picks_by_type(raw.info, meg_combined=True)
    assert_equal(len(pick_list), len(pick_list2))
    assert_equal(pick_list2[0][0], 'mag')

    # pick_types type check
    pytest.raises(ValueError, raw.pick_types, eeg='string')

    # duplicate check
    names = ['MEG 002', 'MEG 002']
    assert len(pick_channels(raw.info['ch_names'], names)) == 1
    assert len(raw.copy().pick_channels(names)[0][0]) == 1

def create_random_epochs(nep, nchan, ntime, sfreq,
                         nclasses=2, ch_types='eeg'):
    data = np.random.randn(nep*nclasses, nchan, ntime*sfreq)
    ev = create_random_events(nep, ntime*sfreq, nclasses)
    info = mne.create_info([str(i) for i in range(nchan)], sfreq, ch_types)
    ep = mne.epochs.EpochsArray(data, info, ev)
    return ep

def file_to_nparray(fname, sfreq=100.0, verbose=False):
    """
	Create a mne raw instance from csv file.
	"""
    # get channel names
    # in MNE, this means you must config ith two arrays:
    # 1) an array of channel names as strings
    # 2) corresponding array of channel types. in our case, all channels are type 'eeg'
    ch_names = getChannelNames()
    ch_type = ["eeg"] * len(ch_names)

    # add one more channel called 'class_label' as type 'stim'
    ch_names.extend(["class_label"])
    ch_type.extend(["stim"])

    # Read EEG file
    data = pd.read_table(fname, header=None, names=ch_names)
    raw_data = np.array(data[ch_names]).T
    # print raw_data.shape

    # create and populate MNE info structure
    info = create_info(ch_names, sfreq=sfreq, ch_types=ch_type)
    info["filename"] = fname

    # create raw object
    return [raw_data, info]

def test_orig_units():
    """Test the error handling for original units."""
    # Should work fine
    info = create_info(ch_names=['Cz'], sfreq=100, ch_types='eeg')
    BaseRaw(info, last_samps=[1], orig_units={'Cz': 'nV'})

    # Should complain that channel Cz does not have a corresponding original
    # unit.
    with pytest.raises(ValueError, match='has no associated original unit.'):
        info = create_info(ch_names=['Cz'], sfreq=100, ch_types='eeg')
        BaseRaw(info, last_samps=[1], orig_units={'not_Cz': 'nV'})

    # Test that a non-dict orig_units argument raises a ValueError
    with pytest.raises(ValueError, match='orig_units must be of type dict'):
        info = create_info(ch_names=['Cz'], sfreq=100, ch_types='eeg')
        BaseRaw(info, last_samps=[1], orig_units=True)

def test_filter_picks():
    """Test filtering default channel picks"""
    ch_types = ['mag', 'grad', 'eeg', 'seeg', 'misc', 'stim']
    info = create_info(ch_names=ch_types, ch_types=ch_types, sfreq=256)
    raw = RawArray(data=np.zeros((len(ch_types), 1000)), info=info)

    # -- Deal with meg mag grad exception
    ch_types = ('misc', 'stim', 'meg', 'eeg', 'seeg')

    # -- Filter data channels
    for ch_type in ('mag', 'grad', 'eeg', 'seeg'):
        picks = dict((ch, ch == ch_type) for ch in ch_types)
        picks['meg'] = ch_type if ch_type in ('mag', 'grad') else False
        raw_ = raw.pick_types(copy=True, **picks)
        # Avoid RuntimeWarning due to Attenuation
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter('always')
            raw_.filter(10, 30)
            assert_true(len(w) == 1)

    # -- Error if no data channel
    for ch_type in ('misc', 'stim'):
        picks = dict((ch, ch == ch_type) for ch in ch_types)
        raw_ = raw.pick_types(copy=True, **picks)
        assert_raises(RuntimeError, raw_.filter, 10, 30)

def test_continuous_regression_with_overlap():
    """Test regression with overlap correction."""
    signal = np.zeros(100000)
    times = [1000, 2500, 3000, 5000, 5250, 7000, 7250, 8000]
    events = np.zeros((len(times), 3), int)
    events[:, 2] = 1
    events[:, 0] = times
    signal[events[:, 0]] = 1.
    effect = hann(101)
    signal = np.convolve(signal, effect)[:len(signal)]
    raw = RawArray(signal[np.newaxis, :], mne.create_info(1, 100, 'eeg'))

    assert_allclose(effect, linear_regression_raw(
        raw, events, {1: 1}, tmin=0)[1].data.flatten())

    # test that sklearn solvers can be used
    from sklearn.linear_model.ridge import ridge_regression

    def solver(X, y):
        return ridge_regression(X, y, alpha=0.)
    assert_allclose(effect, linear_regression_raw(
        raw, events, tmin=0, solver=solver)['1'].data.flatten())

    # test bad solvers
    def solT(X, y):
        return ridge_regression(X, y, alpha=0.).T
    assert_raises(ValueError, linear_regression_raw, raw, events, solver=solT)
    assert_raises(ValueError, linear_regression_raw, raw, events, solver='err')
    assert_raises(TypeError, linear_regression_raw, raw, events, solver=0)

def test_snr():
    """Test trial to trial coherence"""
    raw = mne.io.Raw(raw_fname)
    sfreq = int(raw.info['sfreq'])
    data, times = raw[0, :5 * sfreq]

    # Create fake epochs from copies of the raw + noise
    n_epochs = 40
    noise_amp = .01 * data.max()
    data = np.tile(data, [n_epochs, 1, 1])
    data += noise_amp * rng.randn(*data.shape)
    info = mne.create_info(['ch1'], raw.info['sfreq'], 'eeg')
    ev = np.vstack([np.arange(n_epochs),
                    np.zeros(n_epochs),
                    np.ones(n_epochs)]).T.astype(int)
    epochs = mne.epochs.EpochsArray(data, info, ev)

    # Test CC
    cc = snr_epochs(epochs, kind='corr')
    assert_true((cc > .99).all())

    # Test coherence
    coh, freqs = snr_epochs(epochs, fmin=2, kind='coh')
    assert_true((coh.mean(-1) > .99).all())

    # Test random signal
    data_rand = 10*rng.randn(*data.shape)
    epochs_rand = mne.epochs.EpochsArray(data_rand, info, ev)
    cc = snr_epochs(epochs_rand, kind='corr')
    assert_true(cc.mean() < .02)

    # Test incorrect inputs
    assert_raises(ValueError, snr_epochs, epochs, kind='foo')

def creat_mne_raw_object(fname,read_events):
    # Read EEG file
    data = pd.read_csv(fname)
    
    # get chanel names
    ch_names = list(data.columns[1:])
    
    # read EEG standard montage from mne
    montage = read_montage('standard_1005',ch_names)

    ch_type = ['eeg']*len(ch_names)
    data = 1e-6*np.array(data[ch_names]).T
    
    if read_events:
        # events file
        ev_fname = fname.replace('_data','_events')
        # read event file
        events = pd.read_csv(ev_fname)
        events_names = events.columns[1:]
        events_data = np.array(events[events_names]).T
        
        # define channel type, the first is EEG, the last 6 are stimulations
        ch_type.extend(['stim']*6)
        ch_names.extend(events_names)
        # concatenate event file and data
        data = np.concatenate((data,events_data))
        
    # create and populate MNE info structure
    info = create_info(ch_names,sfreq=500.0, ch_types=ch_type, montage=montage)
    info['filename'] = fname
    
    # create raw object 
    raw = RawArray(data,info,verbose=False)
    
    return raw

    def get_data(self, series, change=False):
        # Get the lift EEG data
        ws = self._series_data(series)
        lifts = self.lift_info()

        montage = mne.channels.read_montage('standard_1005', ws.names.eeg)

        ev_names = self.all_event_names()

        ch_names = list(ws.names.eeg) + ev_names

        X = []
        y = []
        # loop through the windows, creating a raw object for each one
        for win_num, window in enumerate(ws.win):
            events = np.zeros((window.eeg.shape[0], len(ev_names)))
            row = lifts[(lifts['Run'] == series) & (lifts['Lift'] == win_num + 1)]
            if change and np.isnan(row.iloc[0]['PrevW']):
                continue

            for ev_num, ev_name in enumerate(ev_names):
                t = row.iloc[0]['t' + ev_name]
                event_rows = np.abs(window.eeg_t - t) <= 0.15 + 1e-10
                events[event_rows, ev_num] = 1
            data = np.concatenate((1e-6 * window.eeg.T, events.T))
            ch_type = ['eeg'] * len(ws.names.eeg) + ['stim'] * len(ev_names)

            info = mne.create_info(ch_names, sfreq=500., ch_types=ch_type, montage=montage)
            # Return a dictionary, in case we want to incorporate non-EEG data, e.g. previous weight
            X.append({'eeg': mne.io.RawArray(data, info, verbose=False)})
            #Cast weight to integer, so that we don't have floating point problems
            yval = row.iloc[0]['CurW']==row.iloc[0]['PrevW'] if change else row.iloc[0]['CurW']
            y.append(yval)

        return X, y

def test_annotation_property_deprecation_warning():
    """Test that assigning annotations warns and nowhere else."""
    with pytest.warns(None) as w:
        raw = RawArray(np.random.rand(1, 1), create_info(1, 1))
    assert len(w) is 0
    with pytest.warns(DeprecationWarning, match='by assignment is deprecated'):
        raw.annotations = None

def _raw_annot(meas_date, orig_time):
    info = create_info(ch_names=10, sfreq=10.)
    raw = RawArray(data=np.empty((10, 10)), info=info, first_samp=10)
    raw.info['meas_date'] = meas_date
    annot = Annotations([.5], [.2], ['dummy'], orig_time)
    raw.set_annotations(annotations=annot)
    return raw

def _generate_coherence_data():
    """Create an epochs object with coherence at 22Hz between channels 1 and 3.

    A base 10 Hz sine wave is generated for all channels, but with different
    phases, which means no actual coherence. A  22Hz sine wave is laid on top
    for channels 1 and 3, with the same phase, so there is coherence between
    these channels.
    """
    ch_names = ['CH1', 'CH2', 'CH3']
    sfreq = 50.
    info = mne.create_info(ch_names, sfreq, 'eeg')
    tstep = 1. / sfreq
    n_samples = int(10 * sfreq)  # 10 seconds of data
    times = np.arange(n_samples) * tstep
    events = np.array([[0, 1, 1]])  # one event

    # Phases for the signals
    phases = np.arange(info['nchan']) * 0.3 * np.pi

    # Generate 10 Hz sine waves with different phases
    signal = np.vstack([np.sin(times * 2 * np.pi * 10 + phase)
                        for phase in phases])

    data = np.zeros((1, info['nchan'], n_samples))
    data[0, :, :] = signal

    # Generate 22Hz sine wave at the first and last electrodes with the same
    # phase.
    signal = np.sin(times * 2 * np.pi * 22)
    data[0, [0, -1], :] += signal

    return mne.EpochsArray(data, info, events, baseline=(0, times[-1]))

def test_apply_function_verbose():
    """Test apply function verbosity
    """
    n_chan = 2
    n_times = 3
    ch_names = [str(ii) for ii in range(n_chan)]
    raw = RawArray(np.zeros((n_chan, n_times)),
                   create_info(ch_names, 1., 'mag'))
    # test return types in both code paths (parallel / 1 job)
    assert_raises(TypeError, raw.apply_function, bad_1,
                  None, None, 1)
    assert_raises(ValueError, raw.apply_function, bad_2,
                  None, None, 1)
    assert_raises(TypeError, raw.apply_function, bad_1,
                  None, None, 2)
    assert_raises(ValueError, raw.apply_function, bad_2,
                  None, None, 2)

    # check our arguments
    tempdir = _TempDir()
    test_name = op.join(tempdir, 'test.log')
    set_log_file(test_name)
    try:
        raw.apply_function(printer, None, None, 1, verbose=False)
        with open(test_name) as fid:
            assert_equal(len(fid.readlines()), 0)
        raw.apply_function(printer, None, None, 1, verbose=True)
        with open(test_name) as fid:
            assert_equal(len(fid.readlines()), n_chan)
    finally:
        set_log_file(None)

def test_pick_seeg_ecog():
    """Test picking with sEEG and ECoG
    """
    names = 'A1 A2 Fz O OTp1 OTp2 E1 OTp3 E2 E3'.split()
    types = 'mag mag eeg eeg seeg seeg ecog seeg ecog ecog'.split()
    info = create_info(names, 1024., types)
    idx = channel_indices_by_type(info)
    assert_array_equal(idx['mag'], [0, 1])
    assert_array_equal(idx['eeg'], [2, 3])
    assert_array_equal(idx['seeg'], [4, 5, 7])
    assert_array_equal(idx['ecog'], [6, 8, 9])
    assert_array_equal(pick_types(info, meg=False, seeg=True), [4, 5, 7])
    for i, t in enumerate(types):
        assert_equal(channel_type(info, i), types[i])
    raw = RawArray(np.zeros((len(names), 10)), info)
    events = np.array([[1, 0, 0], [2, 0, 0]])
    epochs = Epochs(raw, events, {'event': 0}, -1e-5, 1e-5, add_eeg_ref=False)
    evoked = epochs.average(pick_types(epochs.info, meg=True, seeg=True))
    e_seeg = evoked.copy().pick_types(meg=False, seeg=True)
    for l, r in zip(e_seeg.ch_names, [names[4], names[5], names[7]]):
        assert_equal(l, r)
    # Deal with constant debacle
    raw = read_raw_fif(op.join(io_dir, 'tests', 'data',
                               'test_chpi_raw_sss.fif'), add_eeg_ref=False)
    assert_equal(len(pick_types(raw.info, meg=False, seeg=True, ecog=True)), 0)

def test_multitaper_psd():
    """ Test multi-taper PSD computation """

    import nitime as ni
    n_times = 1000
    n_channels = 5
    data = np.random.RandomState(0).randn(n_channels, n_times)
    sfreq = 500
    info = create_info(n_channels, sfreq, 'eeg')
    raw = RawArray(data, info)
    assert_raises(ValueError, psd_multitaper, raw, sfreq, normalization='foo')
    ni_5 = (LooseVersion(ni.__version__) >= LooseVersion('0.5'))
    norm = 'full' if ni_5 else 'length'

    for adaptive, n_jobs in zip((False, True, True), (1, 1, 2)):
        psd, freqs = psd_multitaper(raw, adaptive=adaptive,
                                    n_jobs=n_jobs,
                                    normalization=norm)
        freqs_ni, psd_ni, _ = ni.algorithms.spectral.multi_taper_psd(
            data, sfreq, adaptive=adaptive, jackknife=False)

        # for some reason nitime returns n_times + 1 frequency points
        # causing the value at 0 to be different
        assert_array_almost_equal(psd[:, 1:], psd_ni[:, 1:-1], decimal=3)
        assert_array_almost_equal(freqs, freqs_ni[:-1])

def write_mnefiff(data, filename):
    """Export data to MNE using FIFF format.

    Parameters
    ----------
    data : instance of ChanTime
        data with only one trial
    filename : path to file
        file to export to (include '.mat')

    Notes
    -----
    It cannot store data larger than 2 GB.
    The data is assumed to have only EEG electrodes.
    It overwrites a file if it exists.
    """
    from mne import create_info, set_log_level
    from mne.io import RawArray

    set_log_level(WARNING)

    TRIAL = 0
    info = create_info(list(data.axis['chan'][TRIAL]), data.s_freq, ['eeg', ] *
                       data.number_of('chan')[TRIAL])

    UNITS = 1e-6  # mne wants data in uV
    fiff = RawArray(data.data[0] * UNITS, info)

    if data.attr['chan']:
        fiff.set_channel_positions(data.attr['chan'].return_xyz(),
                                   data.attr['chan'].return_label())

    fiff.save(filename, overwrite=True)

def creat_mne_raw_object(fname, read_events = True):

    print ("loading data from %s" %fname)
    data = pd.read_csv(fname)

    ch_names = list(data.columns[1:])

    montage = read_montage('standard_1005', ch_names)
    ch_type = ['eeg']*len(ch_names)
    data = 1e-6*np.array(data[ch_names]).T
    
    if read_events:
        ev_fname = fname.replace('_data', '_events')
        print (ev_fname)
        events = pd.read_csv(ev_fname)
        events_names = events.columns[1:]
        events_data = np.array(events[events_names]).T

        ch_type.extend(['stim']*6)
        ch_names.extend(events_names)
        data = np.concatenate((data, events_data))
        
    info = create_info(ch_names, sfreq=500.0, ch_types=ch_type, montage=montage)
    info['filename'] = fname
    raw = RawArray(data, info, verbose=True)

    return raw

def test_multitaper_psd():
    """Test multi-taper PSD computation."""
    import nitime as ni
    for n_times in (100, 101):
        n_channels = 5
        data = np.random.RandomState(0).randn(n_channels, n_times)
        sfreq = 500
        info = create_info(n_channels, sfreq, 'eeg')
        raw = RawArray(data, info)
        pytest.raises(ValueError, psd_multitaper, raw, sfreq,
                      normalization='foo')
        ni_5 = (LooseVersion(ni.__version__) >= LooseVersion('0.5'))
        norm = 'full' if ni_5 else 'length'
        for adaptive, n_jobs in zip((False, True, True), (1, 1, 2)):
            psd, freqs = psd_multitaper(raw, adaptive=adaptive,
                                        n_jobs=n_jobs,
                                        normalization=norm)
            with warnings.catch_warnings(record=True):  # nitime integers
                freqs_ni, psd_ni, _ = ni.algorithms.spectral.multi_taper_psd(
                    data, sfreq, adaptive=adaptive, jackknife=False)
            assert_array_almost_equal(psd, psd_ni, decimal=4)
            if n_times % 2 == 0:
                # nitime's frequency definitions must be incorrect,
                # they give the same values for 100 and 101 samples
                assert_array_almost_equal(freqs, freqs_ni)
        with pytest.raises(ValueError, match='use a value of at least'):
            psd_multitaper(raw, bandwidth=4.9)

def test_add_channels():
    """Test tfr splitting / re-appending channel types
    """
    data = np.zeros((6, 2, 3))
    times = np.array([0.1, 0.2, 0.3])
    freqs = np.array([0.10, 0.20])
    info = mne.create_info(
        ["MEG 001", "MEG 002", "MEG 003", "EEG 001", "EEG 002", "STIM 001"],
        1000.0,
        ["mag", "mag", "mag", "eeg", "eeg", "stim"],
    )
    tfr = AverageTFR(info, data=data, times=times, freqs=freqs, nave=20, comment="test", method="crazy-tfr")
    tfr_eeg = tfr.pick_types(meg=False, eeg=True, copy=True)
    tfr_meg = tfr.pick_types(meg=True, copy=True)
    tfr_stim = tfr.pick_types(meg=False, stim=True, copy=True)
    tfr_eeg_meg = tfr.pick_types(meg=True, eeg=True, copy=True)
    tfr_new = tfr_meg.add_channels([tfr_eeg, tfr_stim], copy=True)
    assert_true(all(ch in tfr_new.ch_names for ch in tfr_stim.ch_names + tfr_meg.ch_names))
    tfr_new = tfr_meg.add_channels([tfr_eeg], copy=True)

    assert_true(ch in tfr_new.ch_names for ch in tfr.ch_names)
    assert_array_equal(tfr_new.data, tfr_eeg_meg.data)
    assert_true(all(ch not in tfr_new.ch_names for ch in tfr_stim.ch_names))

    # Now test errors
    tfr_badsf = tfr_eeg.copy()
    tfr_badsf.info["sfreq"] = 3.1415927
    tfr_eeg = tfr_eeg.crop(-0.1, 0.1)

    assert_raises(RuntimeError, tfr_meg.add_channels, [tfr_badsf])
    assert_raises(AssertionError, tfr_meg.add_channels, [tfr_eeg])
    assert_raises(ValueError, tfr_meg.add_channels, [tfr_meg])
    assert_raises(AssertionError, tfr_meg.add_channels, tfr_badsf)