ENH: crop variable-duration epochs

Cropping asks for a window in seconds, and that question has an answer for
each trial on its own: keep the samples inside it. No epoch has to be
padded, stretched or compared with any other, so crop leaves the
not-implemented table.

The requested window is applied to every epoch independently and clamped
to that epoch's own bounds where it reaches past them, which is what the
fixed path does against its single interval. Selections for every epoch
are computed before anything is written, so a window that misses one
epoch fails and leaves the object as it was rather than dropping it. That
failure comes from _time_mask seeing an inverted interval once tmax has
been clamped back, which is the same route the fixed path takes.

Clamping is reported once per bound rather than once per epoch, and only
when it happened. A clamped tmax keeps that epoch's last sample even when
include_tmax is False, matching the fixed path.

Bounds are taken from the samples that survived, never from the requested
float, so len(get_times(i)) continues to describe the block.

Cropping can also remove the variation: when every epoch ends up on the
same axis the blocks are stacked and the object becomes an ordinary
Epochs again, which is checked by sample index and length rather than by
comparing floats. The reductions then return on their own, since the
wrappers ask about _variable_duration when they are called.

ExtendedTimeMixin is untouched. It is shared with Raw, Evoked and TFR,
and this behaviour belongs to Epochs.

Author: snesmaeili

mne/epochs.py

@@ -97,6 +97,7 @@
     _prepare_read_metadata,
     _prepare_write_metadata,
     _scale_dataframe_data,
+    _time_mask,
     _validate_type,
     check_fname,
     check_random_state,
@@ -2729,6 +2730,10 @@ def crop(
         # XXX this could be made to work on non-preloaded data...
         _check_preload(self, "Modifying data of epochs")
 
+        if self._variable_duration:
+            self._crop_variable(tmin, tmax, include_tmax)
+            return self
+
         super().crop(tmin=tmin, tmax=tmax, include_tmax=include_tmax)
 
         # Adjust rejection period
@@ -2746,6 +2751,115 @@ def crop(
             self.reject_tmax = self.tmax
         return self
 
+    def _crop_variable(self, tmin, tmax, include_tmax):
+        """Crop each epoch on its own time axis.
+
+        The requested window is a physical interval in seconds, so it is applied
+        to every epoch independently and clamped to the epoch's own bounds when
+        it reaches past them. Nothing is padded, interpolated or aligned; an
+        epoch that the window misses entirely makes the whole call fail, the
+        same way it would for that epoch on its own.
+
+        Parameters
+        ----------
+        tmin : float | None
+            Start of the window, or ``None`` for each epoch's own start.
+        tmax : float | None
+            End of the window, or ``None`` for each epoch's own end.
+        include_tmax : bool
+            Whether to keep the sample at ``tmax``.
+        """
+        for name in ("reject_tmin", "reject_tmax"):
+            if getattr(self, name, None) is not None:
+                raise NotImplementedError(
+                    f"{name} is not implemented for variable-duration epochs, "
+                    "because the window is not guaranteed to exist in every "
+                    "epoch."
+                )
+
+        sfreq = float(self.info["sfreq"])
+        # First pass: work out every selection while changing nothing, so that a
+        # window that misses one epoch leaves the object as it was.
+        masks = list()
+        clamped_tmin = clamped_tmax = False
+        for ii in range(len(self.events)):
+            times = self.get_times(ii)
+            this_tmin, this_tmax = tmin, tmax
+            this_include_tmax = include_tmax
+            if this_tmin is None:
+                this_tmin = times[0]
+            elif this_tmin < times[0]:
+                clamped_tmin = True
+                this_tmin = times[0]
+            if this_tmax is None:
+                this_tmax = times[-1]
+            elif this_tmax > times[-1]:
+                clamped_tmax = True
+                this_tmax = times[-1]
+                # matches the fixed path: a clamped end keeps its last sample
+                this_include_tmax = True
+            # _time_mask raises when the window is inverted, which is what an
+            # entirely out-of-range request collapses to once tmax is clamped
+            mask = _time_mask(
+                times,
+                this_tmin,
+                this_tmax,
+                sfreq=sfreq,
+                include_tmax=this_include_tmax,
+            )
+            if not mask.any():
+                raise ValueError(
+                    f"tmin ({tmin}) and tmax ({tmax}) would leave epoch {ii} "
+                    f"with no samples; it spans {times[0]:g} to {times[-1]:g} s."
+                )
+            masks.append(mask)
+
+        # One warning per bound however many epochs needed clamping
+        if clamped_tmin:
+            warn(
+                "tmin is not in time interval for every epoch. tmin is set to "
+                "each of those epochs' own first sample."
+            )
+        if clamped_tmax:
+            warn(
+                "tmax is not in time interval for every epoch. tmax is set to "
+                "each of those epochs' own last sample."
+            )
+
+        # Second pass: apply
+        ragged = self._data
+        assert ragged is not None  # variable-duration epochs are always preloaded
+        starts = np.empty(len(masks))
+        stops = np.empty(len(masks))
+        for ii, mask in enumerate(masks):
+            kept = self.get_times(ii)[mask]
+            ragged[ii] = ragged[ii][..., mask]
+            starts[ii], stops[ii] = kept[0], kept[-1]
+        self._tmin_per_epoch = starts
+        self._tmax_per_epoch = stops
+
+        # Cropping can remove the variation entirely. Compare the axes by their
+        # sample index and length rather than by float equality: every epoch is
+        # regularly sampled at one sfreq, so that pair identifies an axis
+        # exactly and does not depend on how the bound was rounded.
+        first_idx = np.round(starts * sfreq).astype(int)
+        lengths = np.array([epoch.shape[-1] for epoch in ragged])
+        if (
+            len(masks)
+            and (first_idx == first_idx[0]).all()
+            and (lengths == lengths[0]).all()
+        ):
+            self._data = np.stack(list(ragged))
+            self._variable_duration = False
+            self._tmin_per_epoch = None  # ty: ignore[invalid-assignment]
+            self._tmax_per_epoch = None  # ty: ignore[invalid-assignment]
+            start_idx, stop_idx = first_idx[0], first_idx[0] + lengths[0] - 1
+        else:
+            start_idx = int(round(starts.min() * sfreq))
+            stop_idx = int(round(stops.max() * sfreq))
+        self._raw_times = np.arange(start_idx, stop_idx + 1) / sfreq
+        self._set_times(self._raw_times)
+
     def copy(self) -> Self:
         """Return copy of Epochs instance.
 
@@ -4068,7 +4182,6 @@ def _events_from_annotations(raw, events, event_id, annotations, on_missing):
     "filter": "filtering",
     "apply_function": "applying a function",
     "apply_baseline": "baseline correction",
-    "crop": "cropping",
     "decimate": "decimation",
     "resample": "resampling",
     "save": "writing to FIF",

mne/tests/test_epochs_variable_duration.py

@@ -4,6 +4,8 @@
 # License: BSD-3-Clause
 # Copyright the MNE-Python contributors.
 
+import warnings
+
 import numpy as np
 import pytest
 from numpy.testing import assert_allclose, assert_array_equal
@@ -508,3 +510,177 @@ def test_pick_does_not_reach_back_into_the_parent(variable):
     subset.pick(["a"])
     assert [epoch.shape for epoch in variable.get_data()] == before
     assert all(epoch.shape[0] == 1 for epoch in subset.get_data())
+
+
+# -- crop -------------------------------------------------------------------
+def _crop_oracle(epochs, idx, **kwargs):
+    """Crop epoch ``idx`` as an ordinary one-epoch Epochs, for comparison."""
+    data = epochs.get_data()[idx]
+    tmin = np.atleast_1d(epochs.tmin)[idx]
+    one = EpochsArray(
+        data[None],
+        create_info(list(epochs.ch_names), SFREQ, "eeg"),
+        tmin=float(tmin),
+        baseline=None,
+        verbose=False,
+    )
+    return one.crop(**kwargs)
+
+
+def _assert_crop_matches_mne(epochs, **kwargs):
+    """Assert every cropped epoch equals ordinary MNE cropping it alone."""
+    wanted = [_crop_oracle(epochs, ii, **kwargs) for ii in range(len(epochs))]
+    got = epochs.copy().crop(**kwargs)
+    data = got.get_data()
+    for ii, one in enumerate(wanted):
+        assert_allclose(data[ii], one.get_data()[0])
+        times = got.get_times(ii) if got.variable_duration else got.times
+        assert_allclose(times, one.times)
+    return got
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        dict(tmin=0.0, tmax=0.4),  # both bounds
+        dict(tmin=0.1),  # only tmin
+        dict(tmax=0.45),  # only tmax
+        dict(tmin=-0.05, tmax=0.45, include_tmax=True),
+        dict(tmin=-0.05, tmax=0.45, include_tmax=False),
+    ],
+)
+def test_crop_matches_mne_per_epoch(variable, kwargs):
+    """Test that cropping equals ordinary MNE applied to each epoch alone."""
+    _assert_crop_matches_mne(variable, **kwargs)
+
+
+def test_crop_matches_mne_with_unequal_tmin(kwargs=None):
+    """Test parity when both bounds differ between epochs."""
+    epochs = _make([-0.2, -0.35, 0.0, -0.1], [0.5, 0.9, 0.7, 0.6])
+    _assert_crop_matches_mne(epochs, tmin=0.05, tmax=0.4)
+    _assert_crop_matches_mne(epochs, tmax=0.5)
+
+
+def test_crop_keeps_the_object_ragged(variable):
+    """Test that unequal durations survive a crop that does not equalise them."""
+    cropped = variable.copy().crop(tmin=0.0)
+    assert cropped.variable_duration
+    assert len(np.unique(cropped.durations)) > 1
+    assert isinstance(cropped._data, list)
+
+
+def test_crop_clamps_each_epoch_and_warns_once(variable):
+    """Test that a bound past some epochs clamps per epoch, warning once."""
+    before = variable.durations.copy()
+    with pytest.warns(RuntimeWarning, match="tmax is not in time interval") as rec:
+        cropped = variable.copy().crop(tmax=99.0)
+    assert len(rec) == 1  # not one per epoch
+    # each epoch kept everything it had
+    assert_allclose(cropped.durations, before)
+    assert cropped.variable_duration
+
+    with pytest.warns(RuntimeWarning, match="tmin is not in time interval") as rec:
+        cropped = variable.copy().crop(tmin=-99.0)
+    assert len(rec) == 1
+    assert_allclose(cropped.durations, before)
+
+
+def test_crop_does_not_warn_when_nothing_is_clamped(variable):
+    """Test that a window inside every epoch is silent."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        variable.copy().crop(tmin=0.0, tmax=0.4)
+
+
+def test_crop_clamped_tmax_keeps_the_last_sample(variable):
+    """Test that clamping tmax includes that epoch's final sample."""
+    lengths = [epoch.shape[-1] for epoch in variable.get_data()]
+    with pytest.warns(RuntimeWarning, match="tmax is not in time interval"):
+        # include_tmax=False must not drop the endpoint that clamping produced
+        cropped = variable.copy().crop(tmax=99.0, include_tmax=False)
+    assert [epoch.shape[-1] for epoch in cropped.get_data()] == lengths
+
+
+def test_crop_outside_every_sample_fails_cleanly(variable):
+    """Test that a window missing an epoch refuses and changes nothing."""
+    before_data = [epoch.copy() for epoch in variable.get_data()]
+    before_tmin = np.array(variable.tmin)
+    before_tmax = np.array(variable.tmax)
+    with pytest.raises(ValueError, match="must be less than or equal to"):
+        variable.crop(tmin=5.0)
+    # the failure left the object exactly as it was
+    for got, want in zip(variable.get_data(), before_data):
+        assert_array_equal(got, want)
+    assert_array_equal(np.array(variable.tmin), before_tmin)
+    assert_array_equal(np.array(variable.tmax), before_tmax)
+    assert variable.variable_duration
+
+
+def test_crop_bounds_come_from_retained_samples(variable):
+    """Test that the stored bounds are sample positions, not the request."""
+    # only tmin, so the differing ends keep the object ragged
+    cropped = variable.copy().crop(tmin=0.013)
+    assert cropped.variable_duration
+    # the request fell between samples and was snapped to one
+    assert not np.isclose(np.atleast_1d(cropped.tmin)[0], 0.013)
+    for ii in range(len(cropped)):
+        times = cropped.get_times(ii)
+        assert times[0] == pytest.approx(np.atleast_1d(cropped.tmin)[ii])
+        assert times[-1] == pytest.approx(np.atleast_1d(cropped.tmax)[ii])
+        # and the axis still describes the block exactly
+        assert len(times) == cropped.get_data()[ii].shape[-1]
+    assert_allclose(
+        cropped.durations, np.atleast_1d(cropped.tmax) - np.atleast_1d(cropped.tmin)
+    )
+
+
+def test_crop_that_equalises_axes_returns_fixed_epochs(variable):
+    """Test that removing the variation gives an ordinary Epochs back."""
+    cropped = variable.copy().crop(tmax=0.5)
+    assert not cropped.variable_duration
+    assert isinstance(cropped._data, np.ndarray)
+    assert cropped._tmin_per_epoch is None
+    assert cropped._tmax_per_epoch is None
+    assert isinstance(cropped.tmin, float)
+    assert isinstance(cropped.tmax, float)
+    # times is answerable again, and agrees with the data
+    assert len(cropped.times) == cropped.get_data().shape[-1]
+    # and the reductions come back on their own, without touching the tables
+    evoked = cropped.average()
+    assert evoked.nave == len(cropped)
+    assert_allclose(evoked.times, cropped.times)
+
+
+def test_crop_keeps_epoch_bookkeeping(variable):
+    """Test that events, metadata and drop_log travel unchanged."""
+    import pandas as pd
+
+    pytest.importorskip("pandas")
+    variable.metadata = pd.DataFrame(dict(kind=list("abcd")))
+    events = variable.events.copy()
+    drop_log = variable.drop_log
+    selection = variable.selection.copy()
+
+    cropped = variable.copy().crop(tmin=0.0, tmax=0.4)
+    assert_array_equal(cropped.events, events)
+    assert cropped.drop_log == drop_log
+    assert_array_equal(cropped.selection, selection)
+    assert list(cropped.metadata["kind"]) == list("abcd")
+
+
+def test_crop_does_not_fall_back_to_as_fixed(variable):
+    """Test that cropping is native, never a padded copy."""
+
+    def _boom(*args, **kwargs):
+        raise AssertionError("crop() fell back to as_fixed()")
+
+    variable.as_fixed = _boom
+    cropped = variable.crop(tmin=0.0)
+    assert not np.isnan(np.concatenate(cropped.get_data(), axis=-1)).any()
+
+
+def test_crop_refuses_when_rejection_windows_are_set(variable):
+    """Test that a stray rejection window is refused, not compared to an array."""
+    variable.reject_tmin = 0.0  # the constructor forbids this; be defensive
+    with pytest.raises(NotImplementedError, match="reject_tmin is not implemented"):
+        variable.crop(tmin=0.0)