# -*- coding: utf-8 -*- # Author: Alexandre Gramfort # Denis Engemann # # License: BSD (3-clause) from copy import deepcopy from functools import partial import itertools as itt import os.path as op import sys import numpy as np from numpy.testing import (assert_array_almost_equal, assert_array_equal, assert_allclose, assert_equal) import pytest from mne.datasets import testing from mne.filter import filter_data from mne.io.constants import FIFF from mne.io import RawArray, concatenate_raws, read_raw_fif from mne.io.tests.test_raw import _test_concat, _test_raw_reader from mne import (concatenate_events, find_events, equalize_channels, compute_proj_raw, pick_types, pick_channels, create_info, pick_info) from mne.utils import (_TempDir, requires_pandas, object_diff, requires_mne, run_subprocess, run_tests_if_main) from mne.externals.six.moves import zip, cPickle as pickle from mne.io.proc_history import _get_rank_sss from mne.io.pick import _picks_by_type from mne.annotations import Annotations testing_path = testing.data_path(download=False) data_dir = op.join(testing_path, 'MEG', 'sample') fif_fname = op.join(data_dir, 'sample_audvis_trunc_raw.fif') ms_fname = op.join(testing_path, 'SSS', 'test_move_anon_raw.fif') skip_fname = op.join(testing_path, 'misc', 'intervalrecording_raw.fif') base_dir = op.join(op.dirname(__file__), '..', '..', 'tests', 'data') test_fif_fname = op.join(base_dir, 'test_raw.fif') test_fif_gz_fname = op.join(base_dir, 'test_raw.fif.gz') ctf_fname = op.join(base_dir, 'test_ctf_raw.fif') ctf_comp_fname = op.join(base_dir, 'test_ctf_comp_raw.fif') fif_bad_marked_fname = op.join(base_dir, 'test_withbads_raw.fif') bad_file_works = op.join(base_dir, 'test_bads.txt') bad_file_wrong = op.join(base_dir, 'test_wrong_bads.txt') hp_fname = op.join(base_dir, 'test_chpi_raw_hp.txt') hp_fif_fname = op.join(base_dir, 'test_chpi_raw_sss.fif') @testing.requires_testing_data def test_acq_skip(): """Test treatment of acquisition skips.""" raw = read_raw_fif(skip_fname, preload=True) picks = [1, 2, 10] assert_equal(len(raw.times), 17000) annotations = raw.annotations assert_equal(len(annotations), 3) # there are 3 skips assert_allclose(annotations.onset, [14, 19, 23]) assert_allclose(annotations.duration, [2., 2., 3.]) # inclusive! data, times = raw.get_data( picks, reject_by_annotation='omit', return_times=True) expected_data, expected_times = zip(raw[picks, :2000], raw[picks, 4000:7000], raw[picks, 9000:11000], raw[picks, 14000:17000]) expected_times = np.concatenate(list(expected_times), axis=-1) assert_allclose(times, expected_times) expected_data = list(expected_data) assert_allclose(data, np.concatenate(expected_data, axis=-1), atol=1e-22) # Check that acquisition skips are handled properly in filtering kwargs = dict(l_freq=None, h_freq=50., fir_design='firwin') raw_filt = raw.copy().filter(picks=picks, **kwargs) for data in expected_data: filter_data(data, raw.info['sfreq'], copy=False, **kwargs) data = raw_filt.get_data(picks, reject_by_annotation='omit') assert_allclose(data, np.concatenate(expected_data, axis=-1), atol=1e-22) # Check that acquisition skips are handled properly during I/O tempdir = _TempDir() fname = op.join(tempdir, 'test_raw.fif') raw.save(fname, fmt=raw.orig_format) # first: file size should not increase much (orig data is missing # 7 of 17 buffers, so if we write them out it should increase the file # size quite a bit. orig_size = op.getsize(skip_fname) new_size = op.getsize(fname) max_size = int(1.05 * orig_size) # almost the same + annotations assert new_size < max_size, (new_size, max_size) raw_read = read_raw_fif(fname) assert raw_read.annotations is not None assert_allclose(raw.times, raw_read.times) assert_allclose(raw_read[:][0], raw[:][0], atol=1e-17) # Saving with a bad buffer length emits warning raw.pick_channels(raw.ch_names[:2]) with pytest.warns(None) as w: raw.save(fname, buffer_size_sec=0.5, overwrite=True) assert len(w) == 0 with pytest.warns(RuntimeWarning, match='did not fit evenly'): raw.save(fname, buffer_size_sec=2., overwrite=True) def test_fix_types(): """Test fixing of channel types.""" for fname, change in ((hp_fif_fname, True), (test_fif_fname, False), (ctf_fname, False)): raw = read_raw_fif(fname) mag_picks = pick_types(raw.info, meg='mag') other_picks = np.setdiff1d(np.arange(len(raw.ch_names)), mag_picks) # we don't actually have any files suffering from this problem, so # fake it if change: for ii in mag_picks: raw.info['chs'][ii]['coil_type'] = FIFF.FIFFV_COIL_VV_MAG_T2 orig_types = np.array([ch['coil_type'] for ch in raw.info['chs']]) raw.fix_mag_coil_types() new_types = np.array([ch['coil_type'] for ch in raw.info['chs']]) if not change: assert_array_equal(orig_types, new_types) else: assert_array_equal(orig_types[other_picks], new_types[other_picks]) assert ((orig_types[mag_picks] != new_types[mag_picks]).all()) assert ((new_types[mag_picks] == FIFF.FIFFV_COIL_VV_MAG_T3).all()) def test_concat(): """Test RawFIF concatenation.""" # we trim the file to save lots of memory and some time tempdir = _TempDir() raw = read_raw_fif(test_fif_fname) raw.crop(0, 2.) test_name = op.join(tempdir, 'test_raw.fif') raw.save(test_name) # now run the standard test _test_concat(partial(read_raw_fif), test_name) @testing.requires_testing_data def test_hash_raw(): """Test hashing raw objects.""" raw = read_raw_fif(fif_fname) pytest.raises(RuntimeError, raw.__hash__) raw = read_raw_fif(fif_fname).crop(0, 0.5) raw_size = raw._size raw.load_data() raw_load_size = raw._size assert (raw_size < raw_load_size) raw_2 = read_raw_fif(fif_fname).crop(0, 0.5) raw_2.load_data() assert_equal(hash(raw), hash(raw_2)) # do NOT use assert_equal here, failing output is terrible assert_equal(pickle.dumps(raw), pickle.dumps(raw_2)) raw_2._data[0, 0] -= 1 assert hash(raw) != hash(raw_2) @testing.requires_testing_data def test_maxshield(): """Test maxshield warning.""" with pytest.warns(RuntimeWarning, match='Internal Active Shielding') as w: read_raw_fif(ms_fname, allow_maxshield=True) assert ('test_raw_fiff.py' in w[0].filename) @testing.requires_testing_data def test_subject_info(): """Test reading subject information.""" tempdir = _TempDir() raw = read_raw_fif(fif_fname).crop(0, 1) assert (raw.info['subject_info'] is None) # fake some subject data keys = ['id', 'his_id', 'last_name', 'first_name', 'birthday', 'sex', 'hand'] vals = [1, 'foobar', 'bar', 'foo', (1901, 2, 3), 0, 1] subject_info = dict() for key, val in zip(keys, vals): subject_info[key] = val raw.info['subject_info'] = subject_info out_fname = op.join(tempdir, 'test_subj_info_raw.fif') raw.save(out_fname, overwrite=True) raw_read = read_raw_fif(out_fname) for key in keys: assert_equal(subject_info[key], raw_read.info['subject_info'][key]) assert_equal(raw.info['meas_date'], raw_read.info['meas_date']) for key in ['secs', 'usecs', 'version']: assert_equal(raw.info['meas_id'][key], raw_read.info['meas_id'][key]) assert_array_equal(raw.info['meas_id']['machid'], raw_read.info['meas_id']['machid']) @testing.requires_testing_data def test_copy_append(): """Test raw copying and appending combinations.""" raw = read_raw_fif(fif_fname, preload=True).copy() raw_full = read_raw_fif(fif_fname) raw_full.append(raw) data = raw_full[:, :][0] assert_equal(data.shape[1], 2 * raw._data.shape[1]) @pytest.mark.slowtest @testing.requires_testing_data def test_rank_estimation(): """Test raw rank estimation.""" iter_tests = itt.product( [fif_fname, hp_fif_fname], # sss ['norm', dict(mag=1e11, grad=1e9, eeg=1e5)] ) for fname, scalings in iter_tests: raw = read_raw_fif(fname).crop(0, 4.).load_data() (_, picks_meg), (_, picks_eeg) = _picks_by_type(raw.info, meg_combined=True) n_meg = len(picks_meg) n_eeg = len(picks_eeg) if len(raw.info['proc_history']) == 0: expected_rank = n_meg + n_eeg else: expected_rank = _get_rank_sss(raw.info) + n_eeg assert_array_equal(raw.estimate_rank(scalings=scalings), expected_rank) assert_array_equal(raw.estimate_rank(picks=picks_eeg, scalings=scalings), n_eeg) if 'sss' in fname: raw.add_proj(compute_proj_raw(raw)) raw.apply_proj() n_proj = len(raw.info['projs']) assert_array_equal(raw.estimate_rank(tstart=0, tstop=3., scalings=scalings), expected_rank - (0 if 'sss' in fname else n_proj)) @testing.requires_testing_data def test_output_formats(): """Test saving and loading raw data using multiple formats.""" tempdir = _TempDir() formats = ['short', 'int', 'single', 'double'] tols = [1e-4, 1e-7, 1e-7, 1e-15] # let's fake a raw file with different formats raw = read_raw_fif(test_fif_fname).crop(0, 1) temp_file = op.join(tempdir, 'raw.fif') for ii, (fmt, tol) in enumerate(zip(formats, tols)): # Let's test the overwriting error throwing while we're at it if ii > 0: pytest.raises(IOError, raw.save, temp_file, fmt=fmt) raw.save(temp_file, fmt=fmt, overwrite=True) raw2 = read_raw_fif(temp_file) raw2_data = raw2[:, :][0] assert_allclose(raw2_data, raw[:, :][0], rtol=tol, atol=1e-25) assert_equal(raw2.orig_format, fmt) def _compare_combo(raw, new, times, n_times): """Compare data.""" for ti in times: # let's do a subset of points for speed orig = raw[:, ti % n_times][0] # these are almost_equals because of possible dtype differences assert_allclose(orig, new[:, ti][0]) @pytest.mark.slowtest @testing.requires_testing_data def test_multiple_files(): """Test loading multiple files simultaneously.""" # split file tempdir = _TempDir() raw = read_raw_fif(fif_fname).crop(0, 10) raw.load_data() raw.load_data() # test no operation split_size = 3. # in seconds sfreq = raw.info['sfreq'] nsamp = (raw.last_samp - raw.first_samp) tmins = np.round(np.arange(0., nsamp, split_size * sfreq)) tmaxs = np.concatenate((tmins[1:] - 1, [nsamp])) tmaxs /= sfreq tmins /= sfreq assert_equal(raw.n_times, len(raw.times)) # going in reverse order so the last fname is the first file (need later) raws = [None] * len(tmins) for ri in range(len(tmins) - 1, -1, -1): fname = op.join(tempdir, 'test_raw_split-%d_raw.fif' % ri) raw.save(fname, tmin=tmins[ri], tmax=tmaxs[ri]) raws[ri] = read_raw_fif(fname) assert_equal(len(raws[ri].times), int(round((tmaxs[ri] - tmins[ri]) * raw.info['sfreq'])) + 1) # + 1 b/c inclusive events = [find_events(r, stim_channel='STI 014') for r in raws] last_samps = [r.last_samp for r in raws] first_samps = [r.first_samp for r in raws] # test concatenation of split file pytest.raises(ValueError, concatenate_raws, raws, True, events[1:]) all_raw_1, events1 = concatenate_raws(raws, preload=False, events_list=events) assert_allclose(all_raw_1.times, raw.times) assert_equal(raw.first_samp, all_raw_1.first_samp) assert_equal(raw.last_samp, all_raw_1.last_samp) assert_allclose(raw[:, :][0], all_raw_1[:, :][0]) raws[0] = read_raw_fif(fname) all_raw_2 = concatenate_raws(raws, preload=True) assert_allclose(raw[:, :][0], all_raw_2[:, :][0]) # test proper event treatment for split files events2 = concatenate_events(events, first_samps, last_samps) events3 = find_events(all_raw_2, stim_channel='STI 014') assert_array_equal(events1, events2) assert_array_equal(events1, events3) # test various methods of combining files raw = read_raw_fif(fif_fname, preload=True) n_times = raw.n_times # make sure that all our data match times = list(range(0, 2 * n_times, 999)) # add potentially problematic points times.extend([n_times - 1, n_times, 2 * n_times - 1]) raw_combo0 = concatenate_raws([read_raw_fif(f) for f in [fif_fname, fif_fname]], preload=True) _compare_combo(raw, raw_combo0, times, n_times) raw_combo = concatenate_raws([read_raw_fif(f) for f in [fif_fname, fif_fname]], preload=False) _compare_combo(raw, raw_combo, times, n_times) raw_combo = concatenate_raws([read_raw_fif(f) for f in [fif_fname, fif_fname]], preload='memmap8.dat') _compare_combo(raw, raw_combo, times, n_times) assert_equal(raw[:, :][0].shape[1] * 2, raw_combo0[:, :][0].shape[1]) assert_equal(raw_combo0[:, :][0].shape[1], raw_combo0.n_times) # with all data preloaded, result should be preloaded raw_combo = read_raw_fif(fif_fname, preload=True) raw_combo.append(read_raw_fif(fif_fname, preload=True)) assert (raw_combo.preload is True) assert_equal(raw_combo.n_times, raw_combo._data.shape[1]) _compare_combo(raw, raw_combo, times, n_times) # with any data not preloaded, don't set result as preloaded raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=True), read_raw_fif(fif_fname, preload=False)]) assert (raw_combo.preload is False) assert_array_equal(find_events(raw_combo, stim_channel='STI 014'), find_events(raw_combo0, stim_channel='STI 014')) _compare_combo(raw, raw_combo, times, n_times) # user should be able to force data to be preloaded upon concat raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=False), read_raw_fif(fif_fname, preload=True)], preload=True) assert (raw_combo.preload is True) _compare_combo(raw, raw_combo, times, n_times) raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=False), read_raw_fif(fif_fname, preload=True)], preload='memmap3.dat') _compare_combo(raw, raw_combo, times, n_times) raw_combo = concatenate_raws([ read_raw_fif(fif_fname, preload=True), read_raw_fif(fif_fname, preload=True)], preload='memmap4.dat') _compare_combo(raw, raw_combo, times, n_times) raw_combo = concatenate_raws([ read_raw_fif(fif_fname, preload=False), read_raw_fif(fif_fname, preload=False)], preload='memmap5.dat') _compare_combo(raw, raw_combo, times, n_times) # verify that combining raws with different projectors throws an exception raw.add_proj([], remove_existing=True) pytest.raises(ValueError, raw.append, read_raw_fif(fif_fname, preload=True)) # now test event treatment for concatenated raw files events = [find_events(raw, stim_channel='STI 014'), find_events(raw, stim_channel='STI 014')] last_samps = [raw.last_samp, raw.last_samp] first_samps = [raw.first_samp, raw.first_samp] events = concatenate_events(events, first_samps, last_samps) events2 = find_events(raw_combo0, stim_channel='STI 014') assert_array_equal(events, events2) # check out the len method assert_equal(len(raw), raw.n_times) assert_equal(len(raw), raw.last_samp - raw.first_samp + 1) @testing.requires_testing_data def test_split_files(): """Test writing and reading of split raw files.""" tempdir = _TempDir() raw_1 = read_raw_fif(fif_fname, preload=True) # Test a very close corner case raw_crop = raw_1.copy().crop(0, 1.) assert_allclose(raw_1.buffer_size_sec, 10., atol=1e-2) # samp rate split_fname = op.join(tempdir, 'split_raw_meg.fif') # intended filenames split_fname_elekta_part2 = op.join(tempdir, 'split_raw_meg-1.fif') split_fname_bids_part1 = op.join(tempdir, 'split_raw_part-01_meg.fif') split_fname_bids_part2 = op.join(tempdir, 'split_raw_part-02_meg.fif') raw_1.set_annotations(Annotations([2.], [5.5], 'test')) with pytest.warns(RuntimeWarning, match='does not conform to MNE'): raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB') # check that the filenames match the intended pattern assert op.exists(split_fname_elekta_part2) # check that filenames are being formatted correctly for BIDS with pytest.warns(RuntimeWarning, match='does not conform to MNE'): raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB', split_naming='bids', overwrite=True) assert op.exists(split_fname_bids_part1) assert op.exists(split_fname_bids_part2) split_fname = op.join(tempdir, 'split_raw.fif') raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB') raw_2 = read_raw_fif(split_fname) assert_allclose(raw_2.buffer_size_sec, 1., atol=1e-2) # samp rate assert_array_almost_equal(raw_1.annotations.onset, raw_2.annotations.onset) assert_array_equal(raw_1.annotations.duration, raw_2.annotations.duration) assert_array_equal(raw_1.annotations.description, raw_2.annotations.description) data_1, times_1 = raw_1[:, :] data_2, times_2 = raw_2[:, :] assert_array_equal(data_1, data_2) assert_array_equal(times_1, times_2) with pytest.warns(RuntimeWarning, match='does not conform to MNE'): raw_bids = read_raw_fif(split_fname_bids_part1) data_bids, times_bids = raw_bids[:, :] assert_array_equal(data_1, data_bids) assert_array_equal(times_1, times_bids) # test the case where we only end up with one buffer to write # (GH#3210). These tests rely on writing meas info and annotations # taking up a certain number of bytes, so if we change those functions # somehow, the numbers below for e.g. split_size might need to be # adjusted. raw_crop = raw_1.copy().crop(0, 5) with pytest.raises(ValueError, match='after writing measurement information'): raw_crop.save(split_fname, split_size='1MB', # too small a size buffer_size_sec=1., overwrite=True) with pytest.raises(ValueError, match='too large for the given split size'): raw_crop.save(split_fname, split_size=3002276, # still too small, now after Info buffer_size_sec=1., overwrite=True) # just barely big enough here; the right size to write exactly one buffer # at a time so we hit GH#3210 if we aren't careful raw_crop.save(split_fname, split_size='4.5MB', buffer_size_sec=1., overwrite=True) raw_read = read_raw_fif(split_fname) assert_allclose(raw_crop[:][0], raw_read[:][0], atol=1e-20) # Check our buffer arithmetic # 1 buffer required raw_crop = raw_1.copy().crop(0, 1) raw_crop.save(split_fname, buffer_size_sec=1., overwrite=True) raw_read = read_raw_fif(split_fname) assert_equal(len(raw_read._raw_extras[0]), 1) assert_equal(raw_read._raw_extras[0][0]['nsamp'], 301) assert_allclose(raw_crop[:][0], raw_read[:][0]) # 2 buffers required raw_crop.save(split_fname, buffer_size_sec=0.5, overwrite=True) raw_read = read_raw_fif(split_fname) assert_equal(len(raw_read._raw_extras[0]), 2) assert_equal(raw_read._raw_extras[0][0]['nsamp'], 151) assert_equal(raw_read._raw_extras[0][1]['nsamp'], 150) assert_allclose(raw_crop[:][0], raw_read[:][0]) # 2 buffers required raw_crop.save(split_fname, buffer_size_sec=1. - 1.01 / raw_crop.info['sfreq'], overwrite=True) raw_read = read_raw_fif(split_fname) assert_equal(len(raw_read._raw_extras[0]), 2) assert_equal(raw_read._raw_extras[0][0]['nsamp'], 300) assert_equal(raw_read._raw_extras[0][1]['nsamp'], 1) assert_allclose(raw_crop[:][0], raw_read[:][0]) raw_crop.save(split_fname, buffer_size_sec=1. - 2.01 / raw_crop.info['sfreq'], overwrite=True) raw_read = read_raw_fif(split_fname) assert_equal(len(raw_read._raw_extras[0]), 2) assert_equal(raw_read._raw_extras[0][0]['nsamp'], 299) assert_equal(raw_read._raw_extras[0][1]['nsamp'], 2) assert_allclose(raw_crop[:][0], raw_read[:][0]) def test_load_bad_channels(): """Test reading/writing of bad channels.""" tempdir = _TempDir() # Load correctly marked file (manually done in mne_process_raw) raw_marked = read_raw_fif(fif_bad_marked_fname) correct_bads = raw_marked.info['bads'] raw = read_raw_fif(test_fif_fname) # Make sure it starts clean assert_array_equal(raw.info['bads'], []) # Test normal case raw.load_bad_channels(bad_file_works) # Write it out, read it in, and check raw.save(op.join(tempdir, 'foo_raw.fif')) raw_new = read_raw_fif(op.join(tempdir, 'foo_raw.fif')) assert_equal(correct_bads, raw_new.info['bads']) # Reset it raw.info['bads'] = [] # Test bad case pytest.raises(ValueError, raw.load_bad_channels, bad_file_wrong) # Test forcing the bad case with pytest.warns(RuntimeWarning, match='1 bad channel'): raw.load_bad_channels(bad_file_wrong, force=True) # write it out, read it in, and check raw.save(op.join(tempdir, 'foo_raw.fif'), overwrite=True) raw_new = read_raw_fif(op.join(tempdir, 'foo_raw.fif')) assert correct_bads == raw_new.info['bads'] # Check that bad channels are cleared raw.load_bad_channels(None) raw.save(op.join(tempdir, 'foo_raw.fif'), overwrite=True) raw_new = read_raw_fif(op.join(tempdir, 'foo_raw.fif')) assert_equal([], raw_new.info['bads']) @pytest.mark.slowtest @testing.requires_testing_data def test_io_raw(): """Test IO for raw data (Neuromag + CTF + gz).""" rng = np.random.RandomState(0) tempdir = _TempDir() # test unicode io for chars in [u'äöé', 'a']: with read_raw_fif(fif_fname) as r: assert ('Raw' in repr(r)) assert (op.basename(fif_fname) in repr(r)) r.info['description'] = chars temp_file = op.join(tempdir, 'raw.fif') r.save(temp_file, overwrite=True) with read_raw_fif(temp_file) as r2: desc2 = r2.info['description'] assert desc2 == chars # Let's construct a simple test for IO first raw = read_raw_fif(fif_fname).crop(0, 3.5) raw.load_data() # put in some data that we know the values of data = rng.randn(raw._data.shape[0], raw._data.shape[1]) raw._data[:, :] = data # save it somewhere fname = op.join(tempdir, 'test_copy_raw.fif') raw.save(fname, buffer_size_sec=1.0) # read it in, make sure the whole thing matches raw = read_raw_fif(fname) assert_allclose(data, raw[:, :][0], rtol=1e-6, atol=1e-20) # let's read portions across the 1-sec tag boundary, too inds = raw.time_as_index([1.75, 2.25]) sl = slice(inds[0], inds[1]) assert_allclose(data[:, sl], raw[:, sl][0], rtol=1e-6, atol=1e-20) # now let's do some real I/O fnames_in = [fif_fname, test_fif_gz_fname, ctf_fname] fnames_out = ['raw.fif', 'raw.fif.gz', 'raw.fif'] for fname_in, fname_out in zip(fnames_in, fnames_out): fname_out = op.join(tempdir, fname_out) raw = read_raw_fif(fname_in) nchan = raw.info['nchan'] ch_names = raw.info['ch_names'] meg_channels_idx = [k for k in range(nchan) if ch_names[k][0] == 'M'] n_channels = 100 meg_channels_idx = meg_channels_idx[:n_channels] start, stop = raw.time_as_index([0, 5], use_rounding=True) data, times = raw[meg_channels_idx, start:(stop + 1)] meg_ch_names = [ch_names[k] for k in meg_channels_idx] # Set up pick list: MEG + STI 014 - bad channels include = ['STI 014'] include += meg_ch_names picks = pick_types(raw.info, meg=True, eeg=False, stim=True, misc=True, ref_meg=True, include=include, exclude='bads') # Writing with drop_small_buffer True raw.save(fname_out, picks, tmin=0, tmax=4, buffer_size_sec=3, drop_small_buffer=True, overwrite=True) raw2 = read_raw_fif(fname_out) sel = pick_channels(raw2.ch_names, meg_ch_names) data2, times2 = raw2[sel, :] assert (times2.max() <= 3) # Writing raw.save(fname_out, picks, tmin=0, tmax=5, overwrite=True) if fname_in == fif_fname or fname_in == fif_fname + '.gz': assert_equal(len(raw.info['dig']), 146) raw2 = read_raw_fif(fname_out) sel = pick_channels(raw2.ch_names, meg_ch_names) data2, times2 = raw2[sel, :] assert_allclose(data, data2, rtol=1e-6, atol=1e-20) assert_allclose(times, times2) assert_allclose(raw.info['sfreq'], raw2.info['sfreq'], rtol=1e-5) # check transformations for trans in ['dev_head_t', 'dev_ctf_t', 'ctf_head_t']: if raw.info[trans] is None: assert (raw2.info[trans] is None) else: assert_array_equal(raw.info[trans]['trans'], raw2.info[trans]['trans']) # check transformation 'from' and 'to' if trans.startswith('dev'): from_id = FIFF.FIFFV_COORD_DEVICE else: from_id = FIFF.FIFFV_MNE_COORD_CTF_HEAD if trans[4:8] == 'head': to_id = FIFF.FIFFV_COORD_HEAD else: to_id = FIFF.FIFFV_MNE_COORD_CTF_HEAD for raw_ in [raw, raw2]: assert_equal(raw_.info[trans]['from'], from_id) assert_equal(raw_.info[trans]['to'], to_id) if fname_in == fif_fname or fname_in == fif_fname + '.gz': assert_allclose(raw.info['dig'][0]['r'], raw2.info['dig'][0]['r']) # test warnings on bad filenames raw_badname = op.join(tempdir, 'test-bad-name.fif.gz') with pytest.warns(RuntimeWarning, match='raw.fif'): raw.save(raw_badname) with pytest.warns(RuntimeWarning, match='raw.fif'): read_raw_fif(raw_badname) @testing.requires_testing_data def test_io_complex(): """Test IO with complex data types.""" rng = np.random.RandomState(0) tempdir = _TempDir() dtypes = [np.complex64, np.complex128] raw = _test_raw_reader(partial(read_raw_fif), fname=fif_fname) picks = np.arange(5) start, stop = raw.time_as_index([0, 5]) data_orig, _ = raw[picks, start:stop] for di, dtype in enumerate(dtypes): imag_rand = np.array(1j * rng.randn(data_orig.shape[0], data_orig.shape[1]), dtype) raw_cp = raw.copy() raw_cp._data = np.array(raw_cp._data, dtype) raw_cp._data[picks, start:stop] += imag_rand with pytest.warns(RuntimeWarning, match='Saving .* complex data.'): raw_cp.save(op.join(tempdir, 'raw.fif'), picks, tmin=0, tmax=5, overwrite=True) raw2 = read_raw_fif(op.join(tempdir, 'raw.fif')) raw2_data, _ = raw2[picks, :] n_samp = raw2_data.shape[1] assert_allclose(raw2_data[:, :n_samp], raw_cp._data[picks, :n_samp]) # with preloading raw2 = read_raw_fif(op.join(tempdir, 'raw.fif'), preload=True) raw2_data, _ = raw2[picks, :] n_samp = raw2_data.shape[1] assert_allclose(raw2_data[:, :n_samp], raw_cp._data[picks, :n_samp]) @testing.requires_testing_data def test_getitem(): """Test getitem/indexing of Raw.""" for preload in [False, True, 'memmap.dat']: raw = read_raw_fif(fif_fname, preload=preload) data, times = raw[0, :] data1, times1 = raw[0] assert_array_equal(data, data1) assert_array_equal(times, times1) data, times = raw[0:2, :] data1, times1 = raw[0:2] assert_array_equal(data, data1) assert_array_equal(times, times1) data1, times1 = raw[[0, 1]] assert_array_equal(data, data1) assert_array_equal(times, times1) assert_array_equal( raw[-10:-1, :][0], raw[len(raw.ch_names) - 10:len(raw.ch_names) - 1, :][0]) pytest.raises(ValueError, raw.__getitem__, (slice(-len(raw.ch_names) - 1), slice(None))) with pytest.raises(ValueError, match='start must be'): raw[-1000:] with pytest.raises(ValueError, match='stop must be'): raw[:-1000] @testing.requires_testing_data def test_proj(): """Test SSP proj operations.""" tempdir = _TempDir() for proj in [True, False]: raw = read_raw_fif(fif_fname, preload=False) if proj: raw.apply_proj() assert (all(p['active'] == proj for p in raw.info['projs'])) data, times = raw[0:2, :] data1, times1 = raw[0:2] assert_array_equal(data, data1) assert_array_equal(times, times1) # test adding / deleting proj if proj: pytest.raises(ValueError, raw.add_proj, [], {'remove_existing': True}) pytest.raises(ValueError, raw.del_proj, 0) else: projs = deepcopy(raw.info['projs']) n_proj = len(raw.info['projs']) raw.del_proj(0) assert_equal(len(raw.info['projs']), n_proj - 1) raw.add_proj(projs, remove_existing=False) # Test that already existing projections are not added. assert_equal(len(raw.info['projs']), n_proj) raw.add_proj(projs[:-1], remove_existing=True) assert_equal(len(raw.info['projs']), n_proj - 1) # test apply_proj() with and without preload for preload in [True, False]: raw = read_raw_fif(fif_fname, preload=preload) data, times = raw[:, 0:2] raw.apply_proj() data_proj_1 = np.dot(raw._projector, data) # load the file again without proj raw = read_raw_fif(fif_fname, preload=preload) # write the file with proj. activated, make sure proj has been applied raw.save(op.join(tempdir, 'raw.fif'), proj=True, overwrite=True) raw2 = read_raw_fif(op.join(tempdir, 'raw.fif')) data_proj_2, _ = raw2[:, 0:2] assert_allclose(data_proj_1, data_proj_2) assert (all(p['active'] for p in raw2.info['projs'])) # read orig file with proj. active raw2 = read_raw_fif(fif_fname, preload=preload) raw2.apply_proj() data_proj_2, _ = raw2[:, 0:2] assert_allclose(data_proj_1, data_proj_2) assert (all(p['active'] for p in raw2.info['projs'])) # test that apply_proj works raw.apply_proj() data_proj_2, _ = raw[:, 0:2] assert_allclose(data_proj_1, data_proj_2) assert_allclose(data_proj_2, np.dot(raw._projector, data_proj_2)) tempdir = _TempDir() out_fname = op.join(tempdir, 'test_raw.fif') raw = read_raw_fif(test_fif_fname, preload=True).crop(0, 0.002) raw.pick_types(meg=False, eeg=True) raw.info['projs'] = [raw.info['projs'][-1]] raw._data.fill(0) raw._data[-1] = 1. raw.save(out_fname) raw = read_raw_fif(out_fname, preload=False) raw.apply_proj() assert_allclose(raw[:, :][0][:1], raw[0, :][0]) @testing.requires_testing_data def test_preload_modify(): """Test preloading and modifying data.""" tempdir = _TempDir() rng = np.random.RandomState(0) for preload in [False, True, 'memmap.dat']: raw = read_raw_fif(fif_fname, preload=preload) nsamp = raw.last_samp - raw.first_samp + 1 picks = pick_types(raw.info, meg='grad', exclude='bads') data = rng.randn(len(picks), nsamp // 2) try: raw[picks, :nsamp // 2] = data except RuntimeError: if not preload: continue else: raise tmp_fname = op.join(tempdir, 'raw.fif') raw.save(tmp_fname, overwrite=True) raw_new = read_raw_fif(tmp_fname) data_new, _ = raw_new[picks, :nsamp // 2] assert_allclose(data, data_new) @pytest.mark.slowtest @testing.requires_testing_data def test_filter(): """Test filtering (FIR and IIR) and Raw.apply_function interface.""" raw = read_raw_fif(fif_fname).crop(0, 7) raw.load_data() sig_dec_notch = 12 sig_dec_notch_fit = 12 picks_meg = pick_types(raw.info, meg=True, exclude='bads') picks = picks_meg[:4] trans = 2.0 filter_params = dict(picks=picks, filter_length='auto', h_trans_bandwidth=trans, l_trans_bandwidth=trans, fir_design='firwin') raw_lp = raw.copy().filter(None, 8.0, **filter_params) raw_hp = raw.copy().filter(16.0, None, **filter_params) raw_bp = raw.copy().filter(8.0 + trans, 16.0 - trans, **filter_params) raw_bs = raw.copy().filter(16.0, 8.0, **filter_params) data, _ = raw[picks, :] lp_data, _ = raw_lp[picks, :] hp_data, _ = raw_hp[picks, :] bp_data, _ = raw_bp[picks, :] bs_data, _ = raw_bs[picks, :] tols = dict(atol=1e-20, rtol=1e-5) assert_allclose(bs_data, lp_data + hp_data, **tols) assert_allclose(data, lp_data + bp_data + hp_data, **tols) assert_allclose(data, bp_data + bs_data, **tols) filter_params_iir = dict(picks=picks, n_jobs=2, method='iir', iir_params=dict(output='ba')) raw_lp_iir = raw.copy().filter(None, 4.0, **filter_params_iir) raw_hp_iir = raw.copy().filter(8.0, None, **filter_params_iir) raw_bp_iir = raw.copy().filter(4.0, 8.0, **filter_params_iir) del filter_params_iir lp_data_iir, _ = raw_lp_iir[picks, :] hp_data_iir, _ = raw_hp_iir[picks, :] bp_data_iir, _ = raw_bp_iir[picks, :] summation = lp_data_iir + hp_data_iir + bp_data_iir assert_array_almost_equal(data[:, 100:-100], summation[:, 100:-100], 11) # make sure we didn't touch other channels data, _ = raw[picks_meg[4:], :] bp_data, _ = raw_bp[picks_meg[4:], :] assert_array_equal(data, bp_data) bp_data_iir, _ = raw_bp_iir[picks_meg[4:], :] assert_array_equal(data, bp_data_iir) # ... and that inplace changes are inplace raw_copy = raw.copy() assert np.may_share_memory(raw._data, raw._data) assert not np.may_share_memory(raw_copy._data, raw._data) # this could be assert_array_equal but we do this to mirror the call below assert (raw._data[0] == raw_copy._data[0]).all() raw_copy.filter(None, 20., n_jobs=2, **filter_params) assert not (raw._data[0] == raw_copy._data[0]).all() assert_equal(raw.copy().filter(None, 20., **filter_params)._data, raw_copy._data) # do a very simple check on line filtering raw_bs = raw.copy().filter(60.0 + trans, 60.0 - trans, **filter_params) data_bs, _ = raw_bs[picks, :] raw_notch = raw.copy().notch_filter( 60.0, picks=picks, n_jobs=2, method='fir', trans_bandwidth=2 * trans) data_notch, _ = raw_notch[picks, :] assert_array_almost_equal(data_bs, data_notch, sig_dec_notch) # now use the sinusoidal fitting raw_notch = raw.copy().notch_filter( None, picks=picks, n_jobs=2, method='spectrum_fit') data_notch, _ = raw_notch[picks, :] data, _ = raw[picks, :] assert_array_almost_equal(data, data_notch, sig_dec_notch_fit) # filter should set the "lowpass" and "highpass" parameters raw = RawArray(np.random.randn(3, 1000), create_info(3, 1000., ['eeg'] * 2 + ['stim'])) raw.info['lowpass'] = raw.info['highpass'] = None for kind in ('none', 'lowpass', 'highpass', 'bandpass', 'bandstop'): print(kind) h_freq = l_freq = None if kind in ('lowpass', 'bandpass'): h_freq = 70 if kind in ('highpass', 'bandpass'): l_freq = 30 if kind == 'bandstop': l_freq, h_freq = 70, 30 assert (raw.info['lowpass'] is None) assert (raw.info['highpass'] is None) kwargs = dict(l_trans_bandwidth=20, h_trans_bandwidth=20, filter_length='auto', phase='zero', fir_design='firwin') raw_filt = raw.copy().filter(l_freq, h_freq, picks=np.arange(1), **kwargs) assert (raw.info['lowpass'] is None) assert (raw.info['highpass'] is None) raw_filt = raw.copy().filter(l_freq, h_freq, **kwargs) wanted_h = h_freq if kind != 'bandstop' else None wanted_l = l_freq if kind != 'bandstop' else None assert_equal(raw_filt.info['lowpass'], wanted_h) assert_equal(raw_filt.info['highpass'], wanted_l) # Using all data channels should still set the params (GH#3259) raw_filt = raw.copy().filter(l_freq, h_freq, picks=np.arange(2), **kwargs) assert_equal(raw_filt.info['lowpass'], wanted_h) assert_equal(raw_filt.info['highpass'], wanted_l) def test_filter_picks(): """Test filtering default channel picks.""" ch_types = ['mag', 'grad', 'eeg', 'seeg', 'misc', 'stim', 'ecog', 'hbo', 'hbr'] 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 and fnirs exceptions ch_types = ('misc', 'stim', 'meg', 'eeg', 'seeg', 'ecog') # -- Filter data channels for ch_type in ('mag', 'grad', 'eeg', 'seeg', 'ecog', 'hbo', 'hbr'): picks = dict((ch, ch == ch_type) for ch in ch_types) picks['meg'] = ch_type if ch_type in ('mag', 'grad') else False picks['fnirs'] = ch_type if ch_type in ('hbo', 'hbr') else False raw_ = raw.copy().pick_types(**picks) raw_.filter(10, 30, fir_design='firwin') # -- Error if no data channel for ch_type in ('misc', 'stim'): picks = dict((ch, ch == ch_type) for ch in ch_types) raw_ = raw.copy().pick_types(**picks) pytest.raises(RuntimeError, raw_.filter, 10, 30) @testing.requires_testing_data def test_crop(): """Test cropping raw files.""" # split a concatenated file to test a difficult case raw = concatenate_raws([read_raw_fif(f) for f in [fif_fname, fif_fname]]) split_size = 10. # in seconds sfreq = raw.info['sfreq'] nsamp = (raw.last_samp - raw.first_samp + 1) # do an annoying case (off-by-one splitting) tmins = np.r_[1., np.round(np.arange(0., nsamp - 1, split_size * sfreq))] tmins = np.sort(tmins) tmaxs = np.concatenate((tmins[1:] - 1, [nsamp - 1])) tmaxs /= sfreq tmins /= sfreq raws = [None] * len(tmins) for ri, (tmin, tmax) in enumerate(zip(tmins, tmaxs)): raws[ri] = raw.copy().crop(tmin, tmax) all_raw_2 = concatenate_raws(raws, preload=False) assert_equal(raw.first_samp, all_raw_2.first_samp) assert_equal(raw.last_samp, all_raw_2.last_samp) assert_array_equal(raw[:, :][0], all_raw_2[:, :][0]) tmins = np.round(np.arange(0., nsamp - 1, split_size * sfreq)) tmaxs = np.concatenate((tmins[1:] - 1, [nsamp - 1])) tmaxs /= sfreq tmins /= sfreq # going in revere order so the last fname is the first file (need it later) raws = [None] * len(tmins) for ri, (tmin, tmax) in enumerate(zip(tmins, tmaxs)): raws[ri] = raw.copy().crop(tmin, tmax) # test concatenation of split file all_raw_1 = concatenate_raws(raws, preload=False) all_raw_2 = raw.copy().crop(0, None) for ar in [all_raw_1, all_raw_2]: assert_equal(raw.first_samp, ar.first_samp) assert_equal(raw.last_samp, ar.last_samp) assert_array_equal(raw[:, :][0], ar[:, :][0]) # test shape consistency of cropped raw data = np.zeros((1, 1002001)) info = create_info(1, 1000) raw = RawArray(data, info) for tmin in range(0, 1001, 100): raw1 = raw.copy().crop(tmin=tmin, tmax=tmin + 2) assert_equal(raw1[:][0].shape, (1, 2001)) @testing.requires_testing_data def test_resample(): """Test resample (with I/O and multiple files).""" tempdir = _TempDir() raw = read_raw_fif(fif_fname).crop(0, 3) raw.load_data() raw_resamp = raw.copy() sfreq = raw.info['sfreq'] # test parallel on upsample raw_resamp.resample(sfreq * 2, n_jobs=2, npad='auto') assert_equal(raw_resamp.n_times, len(raw_resamp.times)) raw_resamp.save(op.join(tempdir, 'raw_resamp-raw.fif')) raw_resamp = read_raw_fif(op.join(tempdir, 'raw_resamp-raw.fif'), preload=True) assert_equal(sfreq, raw_resamp.info['sfreq'] / 2) assert_equal(raw.n_times, raw_resamp.n_times / 2) assert_equal(raw_resamp._data.shape[1], raw_resamp.n_times) assert_equal(raw._data.shape[0], raw_resamp._data.shape[0]) # test non-parallel on downsample raw_resamp.resample(sfreq, n_jobs=1, npad='auto') assert_equal(raw_resamp.info['sfreq'], sfreq) assert_equal(raw._data.shape, raw_resamp._data.shape) assert_equal(raw.first_samp, raw_resamp.first_samp) assert_equal(raw.last_samp, raw.last_samp) # upsampling then downsampling doubles resampling error, but this still # works (hooray). Note that the stim channels had to be sub-sampled # without filtering to be accurately preserved # note we have to treat MEG and EEG+STIM channels differently (tols) assert_allclose(raw._data[:306, 200:-200], raw_resamp._data[:306, 200:-200], rtol=1e-2, atol=1e-12) assert_allclose(raw._data[306:, 200:-200], raw_resamp._data[306:, 200:-200], rtol=1e-2, atol=1e-7) # now check multiple file support w/resampling, as order of operations # (concat, resample) should not affect our data raw1 = raw.copy() raw2 = raw.copy() raw3 = raw.copy() raw4 = raw.copy() raw1 = concatenate_raws([raw1, raw2]) raw1.resample(10., npad='auto') raw3.resample(10., npad='auto') raw4.resample(10., npad='auto') raw3 = concatenate_raws([raw3, raw4]) assert_array_equal(raw1._data, raw3._data) assert_array_equal(raw1._first_samps, raw3._first_samps) assert_array_equal(raw1._last_samps, raw3._last_samps) assert_array_equal(raw1._raw_lengths, raw3._raw_lengths) assert_equal(raw1.first_samp, raw3.first_samp) assert_equal(raw1.last_samp, raw3.last_samp) assert_equal(raw1.info['sfreq'], raw3.info['sfreq']) # test resampling of stim channel # basic decimation stim = [1, 1, 1, 1, 0, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0] raw = RawArray([stim], create_info(1, len(stim), ['stim'])) assert_allclose(raw.resample(8., npad='auto')._data, [[1, 1, 0, 0, 1, 1, 0, 0]]) # decimation of multiple stim channels raw = RawArray(2 * [stim], create_info(2, len(stim), 2 * ['stim'])) assert_allclose(raw.resample(8., npad='auto', verbose='error')._data, [[1, 1, 0, 0, 1, 1, 0, 0], [1, 1, 0, 0, 1, 1, 0, 0]]) # decimation that could potentially drop events if the decimation is # done naively stim = [0, 0, 0, 1, 1, 0, 0, 0] raw = RawArray([stim], create_info(1, len(stim), ['stim'])) assert_allclose(raw.resample(4., npad='auto')._data, [[0, 1, 1, 0]]) # two events are merged in this case (warning) stim = [0, 0, 1, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0, 0] raw = RawArray([stim], create_info(1, len(stim), ['stim'])) with pytest.warns(RuntimeWarning, match='become unreliable'): raw.resample(8., npad='auto') # events are dropped in this case (warning) stim = [0, 1, 1, 0, 0, 1, 1, 0] raw = RawArray([stim], create_info(1, len(stim), ['stim'])) with pytest.warns(RuntimeWarning, match='become unreliable'): raw.resample(4., npad='auto') # test resampling events: this should no longer give a warning # we often have first_samp != 0, include it here too stim = [0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0] # test is on half the sfreq, but should work with trickier ones too o_sfreq, sfreq_ratio = len(stim), 0.5 n_sfreq = o_sfreq * sfreq_ratio first_samp = len(stim) // 2 raw = RawArray([stim], create_info(1, o_sfreq, ['stim']), first_samp=first_samp) events = find_events(raw) raw, events = raw.resample(n_sfreq, events=events, npad='auto') n_fsamp = int(first_samp * sfreq_ratio) # how it's calc'd in base.py # NB np.round used for rounding event times, which has 0.5 as corner case: # https://docs.scipy.org/doc/numpy/reference/generated/numpy.around.html assert_equal(events, np.array([[np.round(1 * sfreq_ratio) + n_fsamp, 0, 1], [np.round(10 * sfreq_ratio) + n_fsamp, 0, 1]])) # test copy flag stim = [1, 1, 1, 1, 0, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0] raw = RawArray([stim], create_info(1, len(stim), ['stim'])) raw_resampled = raw.copy().resample(4., npad='auto') assert (raw_resampled is not raw) raw_resampled = raw.resample(4., npad='auto') assert (raw_resampled is raw) # resample should still work even when no stim channel is present raw = RawArray(np.random.randn(1, 100), create_info(1, 100, ['eeg'])) raw.info['lowpass'] = 50. raw.resample(10, npad='auto') assert_equal(raw.info['lowpass'], 5.) assert_equal(len(raw), 10) @testing.requires_testing_data def test_hilbert(): """Test computation of analytic signal using hilbert.""" raw = read_raw_fif(fif_fname, preload=True) picks_meg = pick_types(raw.info, meg=True, exclude='bads') picks = picks_meg[:4] raw_filt = raw.copy() raw_filt.filter(10, 20, picks=picks, l_trans_bandwidth='auto', h_trans_bandwidth='auto', filter_length='auto', phase='zero', fir_window='blackman', fir_design='firwin') raw_filt_2 = raw_filt.copy() raw2 = raw.copy() raw3 = raw.copy() raw.apply_hilbert(picks, n_fft='auto') raw2.apply_hilbert(picks, n_fft='auto', envelope=True) # Test custom n_fft raw_filt.apply_hilbert(picks, n_fft='auto') n_fft = 2 ** int(np.ceil(np.log2(raw_filt_2.n_times + 1000))) raw_filt_2.apply_hilbert(picks, n_fft=n_fft) assert_equal(raw_filt._data.shape, raw_filt_2._data.shape) assert_allclose(raw_filt._data[:, 50:-50], raw_filt_2._data[:, 50:-50], atol=1e-13, rtol=1e-2) pytest.raises(ValueError, raw3.apply_hilbert, picks, n_fft=raw3.n_times - 100) env = np.abs(raw._data[picks, :]) assert_allclose(env, raw2._data[picks, :], rtol=1e-2, atol=1e-13) @testing.requires_testing_data def test_raw_copy(): """Test Raw copy.""" raw = read_raw_fif(fif_fname, preload=True) data, _ = raw[:, :] copied = raw.copy() copied_data, _ = copied[:, :] assert_array_equal(data, copied_data) assert_equal(sorted(raw.__dict__.keys()), sorted(copied.__dict__.keys())) raw = read_raw_fif(fif_fname, preload=False) data, _ = raw[:, :] copied = raw.copy() copied_data, _ = copied[:, :] assert_array_equal(data, copied_data) assert_equal(sorted(raw.__dict__.keys()), sorted(copied.__dict__.keys())) @requires_pandas def test_to_data_frame(): """Test raw Pandas exporter.""" raw = read_raw_fif(test_fif_fname, preload=True) _, times = raw[0, :10] df = raw.to_data_frame() assert ((df.columns == raw.ch_names).all()) assert_array_equal(np.round(times * 1e3), df.index.values[:10]) df = raw.to_data_frame(index=None) assert ('time' in df.index.names) assert_array_equal(df.values[:, 0], raw._data[0] * 1e13) assert_array_equal(df.values[:, 2], raw._data[2] * 1e15) def test_add_channels(): """Test raw splitting / re-appending channel types.""" rng = np.random.RandomState(0) raw = read_raw_fif(test_fif_fname).crop(0, 1).load_data() raw_nopre = read_raw_fif(test_fif_fname, preload=False) raw_eeg_meg = raw.copy().pick_types(meg=True, eeg=True) raw_eeg = raw.copy().pick_types(meg=False, eeg=True) raw_meg = raw.copy().pick_types(meg=True, eeg=False) raw_stim = raw.copy().pick_types(meg=False, eeg=False, stim=True) raw_new = raw_meg.copy().add_channels([raw_eeg, raw_stim]) assert ( all(ch in raw_new.ch_names for ch in list(raw_stim.ch_names) + list(raw_meg.ch_names)) ) raw_new = raw_meg.copy().add_channels([raw_eeg]) assert (ch in raw_new.ch_names for ch in raw.ch_names) assert_array_equal(raw_new[:, :][0], raw_eeg_meg[:, :][0]) assert_array_equal(raw_new[:, :][1], raw[:, :][1]) assert (all(ch not in raw_new.ch_names for ch in raw_stim.ch_names)) # Testing force updates raw_arr_info = create_info(['1', '2'], raw_meg.info['sfreq'], 'eeg') orig_head_t = raw_arr_info['dev_head_t'] raw_arr = rng.randn(2, raw_eeg.n_times) raw_arr = RawArray(raw_arr, raw_arr_info) # This should error because of conflicts in Info pytest.raises(ValueError, raw_meg.copy().add_channels, [raw_arr]) raw_meg.copy().add_channels([raw_arr], force_update_info=True) # Make sure that values didn't get overwritten assert_equal(object_diff(raw_arr.info['dev_head_t'], orig_head_t), '') # Now test errors raw_badsf = raw_eeg.copy() raw_badsf.info['sfreq'] = 3.1415927 raw_eeg.crop(.5) pytest.raises(RuntimeError, raw_meg.add_channels, [raw_nopre]) pytest.raises(RuntimeError, raw_meg.add_channels, [raw_badsf]) pytest.raises(AssertionError, raw_meg.add_channels, [raw_eeg]) pytest.raises(ValueError, raw_meg.add_channels, [raw_meg]) pytest.raises(TypeError, raw_meg.add_channels, raw_badsf) @testing.requires_testing_data def test_save(): """Test saving raw.""" tempdir = _TempDir() raw = read_raw_fif(fif_fname, preload=False) # can't write over file being read pytest.raises(ValueError, raw.save, fif_fname) raw = read_raw_fif(fif_fname, preload=True) # can't overwrite file without overwrite=True pytest.raises(IOError, raw.save, fif_fname) # test abspath support and annotations annot = Annotations([10], [5], ['test'], orig_time=raw.info['meas_date'][0] + raw._first_time) raw.set_annotations(annot) annot = raw.annotations new_fname = op.join(op.abspath(op.curdir), 'break_raw.fif') raw.save(op.join(tempdir, new_fname), overwrite=True) new_raw = read_raw_fif(op.join(tempdir, new_fname), preload=False) pytest.raises(ValueError, new_raw.save, new_fname) assert_array_almost_equal(annot.onset, new_raw.annotations.onset) assert_array_equal(annot.duration, new_raw.annotations.duration) assert_array_equal(annot.description, new_raw.annotations.description) assert_equal(annot.orig_time, new_raw.annotations.orig_time) @testing.requires_testing_data def test_annotation_crop(): """Test annotation sync after cropping and concatenating.""" tempdir = _TempDir() new_fname = op.join(op.abspath(op.curdir), 'break_raw.fif') annot = Annotations([5., 11., 15.], [2., 1., 3.], ['test', 'test', 'test']) raw = read_raw_fif(fif_fname, preload=False) raw.set_annotations(annot) r1 = raw.copy().crop(2.5, 7.5) r2 = raw.copy().crop(12.5, 17.5) r3 = raw.copy().crop(10., 12.) raw = concatenate_raws([r1, r2, r3]) # segments reordered onsets = raw.annotations.onset durations = raw.annotations.duration # 2*5s clips combined with annotations at 2.5s + 2s clip, annotation at 1s assert_array_almost_equal(onsets[:3], [47.95, 52.95, 56.46], decimal=2) assert_array_almost_equal([2., 2.5, 1.], durations[:3], decimal=2) # test annotation clipping annot = Annotations([0., raw.times[-1]], [2., 2.], 'test', orig_time=raw.info['meas_date'] + raw._first_time - 1.) with pytest.warns(RuntimeWarning, match='Limited .* expanding outside'): raw.set_annotations(annot) assert_allclose(raw.annotations.duration, [1., 1. + 1. / raw.info['sfreq']], atol=1e-3) # make sure we can overwrite the file we loaded when preload=True new_raw = read_raw_fif(op.join(tempdir, new_fname), preload=True) new_raw.save(op.join(tempdir, new_fname), overwrite=True) @testing.requires_testing_data def test_with_statement(): """Test with statement.""" for preload in [True, False]: with read_raw_fif(fif_fname, preload=preload) as raw_: print(raw_) def test_compensation_raw(): """Test Raw compensation.""" tempdir = _TempDir() raw_3 = read_raw_fif(ctf_comp_fname) assert_equal(raw_3.compensation_grade, 3) data_3, times = raw_3[:, :] # data come with grade 3 for ii in range(2): raw_3_new = raw_3.copy() if ii == 0: raw_3_new.load_data() raw_3_new.apply_gradient_compensation(3) assert_equal(raw_3_new.compensation_grade, 3) data_new, times_new = raw_3_new[:, :] assert_array_equal(times, times_new) assert_array_equal(data_3, data_new) # change to grade 0 raw_0 = raw_3.copy().apply_gradient_compensation(0) assert_equal(raw_0.compensation_grade, 0) data_0, times_new = raw_0[:, :] assert_array_equal(times, times_new) assert (np.mean(np.abs(data_0 - data_3)) > 1e-12) # change to grade 1 raw_1 = raw_0.copy().apply_gradient_compensation(1) assert_equal(raw_1.compensation_grade, 1) data_1, times_new = raw_1[:, :] assert_array_equal(times, times_new) assert (np.mean(np.abs(data_1 - data_3)) > 1e-12) pytest.raises(ValueError, raw_1.apply_gradient_compensation, 33) raw_bad = raw_0.copy() raw_bad.add_proj(compute_proj_raw(raw_0, duration=0.5, verbose='error')) raw_bad.apply_proj() pytest.raises(RuntimeError, raw_bad.apply_gradient_compensation, 1) # with preload tols = dict(rtol=1e-12, atol=1e-25) raw_1_new = raw_3.copy().load_data().apply_gradient_compensation(1) assert_equal(raw_1_new.compensation_grade, 1) data_1_new, times_new = raw_1_new[:, :] assert_array_equal(times, times_new) assert (np.mean(np.abs(data_1_new - data_3)) > 1e-12) assert_allclose(data_1, data_1_new, **tols) # change back raw_3_new = raw_1.copy().apply_gradient_compensation(3) data_3_new, times_new = raw_3_new[:, :] assert_allclose(data_3, data_3_new, **tols) raw_3_new = raw_1.copy().load_data().apply_gradient_compensation(3) data_3_new, times_new = raw_3_new[:, :] assert_allclose(data_3, data_3_new, **tols) for load in (False, True): for raw in (raw_0, raw_1): raw_3_new = raw.copy() if load: raw_3_new.load_data() raw_3_new.apply_gradient_compensation(3) assert_equal(raw_3_new.compensation_grade, 3) data_3_new, times_new = raw_3_new[:, :] assert_array_equal(times, times_new) assert (np.mean(np.abs(data_3_new - data_1)) > 1e-12) assert_allclose(data_3, data_3_new, **tols) # Try IO with compensation temp_file = op.join(tempdir, 'raw.fif') raw_3.save(temp_file, overwrite=True) for preload in (True, False): raw_read = read_raw_fif(temp_file, preload=preload) assert_equal(raw_read.compensation_grade, 3) data_read, times_new = raw_read[:, :] assert_array_equal(times, times_new) assert_allclose(data_3, data_read, **tols) raw_read.apply_gradient_compensation(1) data_read, times_new = raw_read[:, :] assert_array_equal(times, times_new) assert_allclose(data_1, data_read, **tols) # Now save the file that has modified compensation # and make sure the compensation is the same as it was, # but that we can undo it # These channels have norm 1e-11/1e-12, so atol=1e-18 isn't awesome, # but it's due to the single precision of the info['comps'] leading # to inexact inversions with saving/loading (casting back to single) # in between (e.g., 1->3->1 will degrade like this) looser_tols = dict(rtol=1e-6, atol=1e-18) raw_1.save(temp_file, overwrite=True) for preload in (True, False): raw_read = read_raw_fif(temp_file, preload=preload, verbose=True) assert_equal(raw_read.compensation_grade, 1) data_read, times_new = raw_read[:, :] assert_array_equal(times, times_new) assert_allclose(data_1, data_read, **looser_tols) raw_read.apply_gradient_compensation(3, verbose=True) data_read, times_new = raw_read[:, :] assert_array_equal(times, times_new) assert_allclose(data_3, data_read, **looser_tols) @requires_mne def test_compensation_raw_mne(): """Test Raw compensation by comparing with MNE-C.""" tempdir = _TempDir() def compensate_mne(fname, grad): tmp_fname = op.join(tempdir, 'mne_ctf_test_raw.fif') cmd = ['mne_process_raw', '--raw', fname, '--save', tmp_fname, '--grad', str(grad), '--projoff', '--filteroff'] run_subprocess(cmd) return read_raw_fif(tmp_fname, preload=True) for grad in [0, 2, 3]: raw_py = read_raw_fif(ctf_comp_fname, preload=True) raw_py.apply_gradient_compensation(grad) raw_c = compensate_mne(ctf_comp_fname, grad) assert_allclose(raw_py._data, raw_c._data, rtol=1e-6, atol=1e-17) assert_equal(raw_py.info['nchan'], raw_c.info['nchan']) for ch_py, ch_c in zip(raw_py.info['chs'], raw_c.info['chs']): for key in ('ch_name', 'coil_type', 'scanno', 'logno', 'unit', 'coord_frame', 'kind'): assert_equal(ch_py[key], ch_c[key]) for key in ('loc', 'unit_mul', 'range', 'cal'): assert_allclose(ch_py[key], ch_c[key]) @testing.requires_testing_data def test_drop_channels_mixin(): """Test channels-dropping functionality.""" raw = read_raw_fif(fif_fname, preload=True) drop_ch = raw.ch_names[:3] ch_names = raw.ch_names[3:] ch_names_orig = raw.ch_names dummy = raw.copy().drop_channels(drop_ch) assert_equal(ch_names, dummy.ch_names) assert_equal(ch_names_orig, raw.ch_names) assert_equal(len(ch_names_orig), raw._data.shape[0]) raw.drop_channels(drop_ch) assert_equal(ch_names, raw.ch_names) assert_equal(len(ch_names), len(raw._cals)) assert_equal(len(ch_names), raw._data.shape[0]) @testing.requires_testing_data def test_pick_channels_mixin(): """Test channel-picking functionality.""" # preload is True raw = read_raw_fif(fif_fname, preload=True) ch_names = raw.ch_names[:3] ch_names_orig = raw.ch_names dummy = raw.copy().pick_channels(ch_names) assert_equal(ch_names, dummy.ch_names) assert_equal(ch_names_orig, raw.ch_names) assert_equal(len(ch_names_orig), raw._data.shape[0]) raw.pick_channels(ch_names) # copy is False assert_equal(ch_names, raw.ch_names) assert_equal(len(ch_names), len(raw._cals)) assert_equal(len(ch_names), raw._data.shape[0]) pytest.raises(ValueError, raw.pick_channels, ch_names[0]) raw = read_raw_fif(fif_fname, preload=False) pytest.raises(RuntimeError, raw.pick_channels, ch_names) pytest.raises(RuntimeError, raw.drop_channels, ch_names) @testing.requires_testing_data def test_equalize_channels(): """Test equalization of channels.""" raw1 = read_raw_fif(fif_fname, preload=True) raw2 = raw1.copy() ch_names = raw1.ch_names[2:] raw1.drop_channels(raw1.ch_names[:1]) raw2.drop_channels(raw2.ch_names[1:2]) my_comparison = [raw1, raw2] equalize_channels(my_comparison) for e in my_comparison: assert_equal(ch_names, e.ch_names) def test_memmap(tmpdir): """Test some interesting memmapping cases.""" # concatenate_raw memmaps = [op.join(str(tmpdir), str(ii)) for ii in range(3)] raw_0 = read_raw_fif(test_fif_fname, preload=memmaps[0]) assert raw_0._data.filename == memmaps[0] raw_1 = read_raw_fif(test_fif_fname, preload=memmaps[1]) assert raw_1._data.filename == memmaps[1] raw_0.append(raw_1, preload=memmaps[2]) assert raw_0._data.filename == memmaps[2] # add_channels orig_data = raw_0[:][0] new_ch_info = pick_info(raw_0.info, [0]) new_ch_info['chs'][0]['ch_name'] = 'foo' new_ch_info._update_redundant() new_data = np.linspace(0, 1, len(raw_0.times))[np.newaxis] ch = RawArray(new_data, new_ch_info) raw_0.add_channels([ch]) if sys.platform == 'darwin': assert not hasattr(raw_0._data, 'filename') else: assert raw_0._data.filename == memmaps[2] assert_allclose(orig_data, raw_0[:-1][0], atol=1e-7) assert_allclose(new_data, raw_0[-1][0], atol=1e-7) # now let's see if .copy() actually works; it does, but eventually # we should make it optionally memmap to a new filename rather than # create an in-memory version (filename=None) raw_0 = read_raw_fif(test_fif_fname, preload=memmaps[0]) assert raw_0._data.filename == memmaps[0] assert raw_0._data[:1, 3:5].all() raw_1 = raw_0.copy() assert isinstance(raw_1._data, np.memmap) assert raw_1._data.filename is None raw_0._data[:] = 0. assert not raw_0._data.any() assert raw_1._data[:1, 3:5].all() # other things like drop_channels and crop work but do not use memmapping, # eventually we might want to add support for some of these as users # require them. run_tests_if_main()