Fix pop_select nopoint removing two samples too few

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

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):