diff --git a/src/eegprep/functions/popfunc/pop_select.py b/src/eegprep/functions/popfunc/pop_select.py index b5023621..ce7650cd 100644 --- a/src/eegprep/functions/popfunc/pop_select.py +++ b/src/eegprep/functions/popfunc/pop_select.py @@ -93,9 +93,6 @@ def _has_content(x): return False return True - # Track whether notime came directly from rmtime (to match MATLAB boundary adjustment logic) - notime_from_rmtime = _has_content(g['rmtime']) - if _has_content(g['rmtrial']): g['notrial'] = g['rmtrial'] if _has_content(g['rmtime']): @@ -402,25 +399,20 @@ def _clip_time_matrix(mat): if cur < xmax: bounds.append([cur, xmax]) notime_mat = np.array(bounds, dtype=float) if bounds else np.empty((0, 2)) + # EEGLAB shifts the interior edges of the derived complement by one + # sample so the kept samples are exactly [t0, t1]. User-supplied + # notime/nopoint/rmtime/rmpoint ranges are passed through unchanged. + for i in range(notime_mat.shape[0]): + if notime_mat[i, 0] != xmin: + notime_mat[i, 0] += 1.0 / srate + if notime_mat[i, 1] != xmax: + notime_mat[i, 1] -= 1.0 / srate # now reject notime_mat intervals from continuous data if notime_mat.size: - # EEGLAB only adjusts interior edges when notime was derived from time, not when it came from rmtime - if notime_from_rmtime: - # Skip boundary adjustment when notime came directly from rmtime - adjusted = notime_mat.copy() - else: - # EEGLAB adjusts interior edges by +/- one sample; replicate - adjusted = notime_mat.copy() - for i in range(adjusted.shape[0]): - # shift interior boundaries off-sample - if adjusted[i, 0] != xmin: - adjusted[i, 0] += 1.0 / srate - if adjusted[i, 1] != xmax: - adjusted[i, 1] -= 1.0 / srate # map to 1-based sample indices - nbtimes = adjusted.size - pts, _ = eeg_lat2point(adjusted.reshape(-1), np.ones(nbtimes), srate, [xmin, xmax]) + nbtimes = notime_mat.size + pts, _ = eeg_lat2point(notime_mat.reshape(-1), np.ones(nbtimes), srate, [xmin, xmax]) pts = pts.reshape((-1, 2)) # drop empty ranges keep_rows = (pts[:, 1] - pts[:, 0]) != 0 diff --git a/tests/test_pop_select.py b/tests/test_pop_select.py index a8b6fe34..b9222c91 100644 --- a/tests/test_pop_select.py +++ b/tests/test_pop_select.py @@ -8,6 +8,7 @@ from eegprep.functions.popfunc.pop_loadset import pop_loadset from eegprep.functions.popfunc.pop_select import pop_select from eegprep.functions.popfunc.pop_epoch import pop_epoch +from tests.fixtures import SAMPLE_DATASET_PATH # where the test resources web_root = 'https://sccntestdatasets.s3.us-east-2.amazonaws.com/' @@ -132,6 +133,81 @@ def test_parity_rmtime_continuous(self): np.allclose(EEG_py_out['data'][:, :min_pnts], EEG_mat_out['data'][:, :min_pnts], atol=1e-7, equal_nan=True) ) + def test_parity_nopoint_continuous(self): + if int(self.EEG_py.get('trials', 1)) > 1: + self.skipTest("Dataset is epoched; skipping continuous nopoint parity test") + + EEG_py_out = pop_select(copy.deepcopy(self.EEG_py), nopoint=[276, 525]) + EEG_mat_out = self.eeglab.pop_select(copy.deepcopy(self.EEG_py), 'nopoint', np.array([276.0, 525.0])) + + self.assertEqual(EEG_py_out['pnts'], EEG_mat_out['pnts']) + # pre-existing boundary events carry NaN duration, so compare NaN-aware + np.testing.assert_array_equal(np.array(_boundary_events(EEG_py_out)), np.array(_boundary_events(EEG_mat_out))) + self.assertTrue(np.allclose(EEG_py_out['data'], EEG_mat_out['data'], atol=1e-7, equal_nan=True)) + + +def _boundary_events(EEG): + """Return (latency, duration) pairs of boundary events, in order.""" + return [(float(ev['latency']), float(ev['duration'])) for ev in EEG['event'] if str(ev.get('type')) == 'boundary'] + + +class TestPopSelectContinuousRemoval(unittest.TestCase): + """Sample-exact removal/keep semantics on continuous data. + + Expected values were confirmed against EEGLAB pop_select on + sample_data/eeglab_data.set (30504 samples, 128 Hz). Point ranges are + 1-based and inclusive at both ends; a boundary event sits half a sample + before the first removed sample with duration equal to the removed count. + """ + + def setUp(self): + self.EEG = pop_loadset(str(SAMPLE_DATASET_PATH)) + self.assertEqual(self.EEG['pnts'], 30504) + + def _check(self, EEG_out, pnts, boundaries): + self.assertEqual(EEG_out['pnts'], pnts) + self.assertEqual(EEG_out['data'].shape[1], pnts) + self.assertEqual(_boundary_events(EEG_out), boundaries) + + def test_nopoint_removes_inclusive_range(self): + for key in ('nopoint', 'rmpoint'): + EEG_out = pop_select(copy.deepcopy(self.EEG), **{key: [276, 525]}) + self._check(EEG_out, 30254, [(275.5, 250.0)]) + + def test_nopoint_first_and_last_samples(self): + EEG_out = pop_select(copy.deepcopy(self.EEG), nopoint=[1, 10]) + self._check(EEG_out, 30494, [(0.5, 10.0)]) + np.testing.assert_array_equal(EEG_out['data'], self.EEG['data'][:, 10:]) + + EEG_out = pop_select(copy.deepcopy(self.EEG), nopoint=[30495, 30504]) + self._check(EEG_out, 30494, [(30494.5, 10.0)]) + np.testing.assert_array_equal(EEG_out['data'], self.EEG['data'][:, :30494]) + + def test_nopoint_two_regions(self): + EEG_out = pop_select(copy.deepcopy(self.EEG), nopoint=[[100, 200], [300, 400]]) + self._check(EEG_out, 30302, [(99.5, 101.0), (198.5, 101.0)]) + + def test_notime_removes_inclusive_range(self): + for key in ('notime', 'rmtime'): + EEG_out = pop_select(copy.deepcopy(self.EEG), **{key: [2.15, 4.1]}) + self._check(EEG_out, 30253, [(275.5, 251.0)]) + + def test_point_and_time_keep_range(self): + EEG_out = pop_select(copy.deepcopy(self.EEG), point=[276, 525]) + self._check(EEG_out, 250, [(0.5, 275.0), (250.5, 29979.0)]) + + EEG_out = pop_select(copy.deepcopy(self.EEG), time=[2.15, 4.1]) + self._check(EEG_out, 251, [(0.5, 275.0), (251.5, 29978.0)]) + + def test_point_keeps_exactly_what_nopoint_removes(self): + a, b = 276, 525 + kept = pop_select(copy.deepcopy(self.EEG), point=[a, b])['data'] + rest = pop_select(copy.deepcopy(self.EEG), nopoint=[a, b])['data'] + np.testing.assert_array_equal(kept, self.EEG['data'][:, a - 1 : b]) + np.testing.assert_array_equal( + np.concatenate([rest[:, : a - 1], kept, rest[:, a - 1 :]], axis=1), self.EEG['data'] + ) + class TestPopSelectFunctional(unittest.TestCase): def setUp(self):