ENH: support safe operations on variable-duration Epochs

With trials of differing length the methods divide into three kinds, and
guessing which one you are calling is how a wrong answer gets returned
quietly.

Selecting epochs, selecting channels, dropping and shifting the time
origin do not care how long each trial is, so they work as they always
did; the per-epoch bounds travel with the epochs they describe. That
needs one branch each in GetEpochsMixin._getitem, shift_time and
_pick_drop_channels, since those hold the data as one array.
_pick_drop_channels replaces the list contents rather than the attribute,
which keeps `_data` an ndarray for Raw, Evoked and the rest.

Reductions across a shared time axis decline and say what they would
need: padding makes the number of contributing epochs a function of time,
which no single nave describes. Measuring it is what settled this - on 43
epochs spanning 2.0-3.6 s, average() returns an Evoked that is 44% NaN
while nave reports 43 where 3 epochs remain.

Per-trial operations with no ragged implementation decline too, rather
than running on a padded copy and returning a wrong answer instead of a
slow one. plot() is among them for now; the next commit implements it.
to_data_frame keeps a warning fallback, since its result is only read.

Author: snesmaeili

mne/channels/channels.py

@@ -636,7 +636,14 @@ def _pick_drop_channels(self, idx, *, verbose=None):
         else:  # All others (Evoked, Epochs, Raw) have chs axis=-2
             axis = -2
         if hasattr(self, "_data"):  # skip non-preloaded Raw
-            self._data = self._data.take(idx, axis=axis)
+            if isinstance(self._data, list):
+                # variable-duration epochs: one array per epoch, channels are
+                # regular within each, so the pick applies the same way to all.
+                # Replacing the contents rather than the attribute keeps `_data`
+                # an ndarray everywhere else this mixin is used.
+                self._data[:] = [epoch.take(idx, axis=axis) for epoch in self._data]
+            else:
+                self._data = self._data.take(idx, axis=axis)
         else:
             assert isinstance(self, BaseRaw) and not self.preload
 

mne/epochs.py

@@ -10,7 +10,7 @@
 from collections import Counter
 from collections.abc import Callable, Iterable, Iterator
 from copy import deepcopy
-from functools import partial
+from functools import partial, wraps
 from inspect import getfullargspec
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal
@@ -4042,6 +4042,165 @@ def _events_from_annotations(raw, events, event_id, annotations, on_missing):
     return events, event_id, annotations
 
 
+#: Methods whose result is only looked at. They warn and run on ``as_fixed()``,
+#: which is enough for inspection because the padding is visible to whoever is
+#: looking. Anything numeric is not in this table: see the two below.
+_VARIABLE_FALLBACK = {
+    "to_data_frame": "",
+}
+
+#: Methods that combine epochs across a shared time axis. Padding them makes the
+#: number of contributing epochs a function of time, which no scalar ``nave`` can
+#: describe, so they ask for a policy instead of inventing one. See :gh:`14206`.
+_VARIABLE_NEEDS_POLICY = {
+    "average": "averaging",
+    "standard_error": "estimating the standard error",
+    "subtract_evoked": "subtracting an evoked response",
+    "iter_evoked": "iterating as evoked responses",
+    "compute_tfr": "computing a time-frequency representation",
+    "compute_psd": "computing a spectrum",
+}
+
+#: Methods that are mathematically per-trial and simply have no ragged
+#: implementation yet. Running them on a padded copy would return a wrong answer
+#: rather than a slow one, so they raise until implemented natively.
+_VARIABLE_NOT_IMPLEMENTED = {
+    "filter": "filtering",
+    "plot": "browsing",
+    "apply_function": "applying a function",
+    "apply_baseline": "baseline correction",
+    "crop": "cropping",
+    "decimate": "decimation",
+    "resample": "resampling",
+    "save": "writing to FIF",
+    "export": "exporting",
+    # these render an image over one axis and cannot draw the NaN padding that
+    # as_fixed() introduces, so the fallback has nothing useful to show
+    "plot_image": "plotting as an image",
+    "plot_topo_image": "plotting as a topographic image",
+}
+
+
+def _wrap_variable_fallback(func, name, note):
+    """Warn and fall back to ``as_fixed()`` for variable-duration epochs.
+
+    Parameters
+    ----------
+    func : callable
+        The original method.
+    name : str
+        Its name, used to look it up on the fixed-duration copy.
+    note : str
+        Extra sentence appended to the warning, or an empty string.
+
+    Returns
+    -------
+    wrapper : callable
+        The wrapped method.
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        if not getattr(self, "_variable_duration", False):
+            return func(self, *args, **kwargs)
+        message = (
+            f"{name}() needs one time axis, which these variable-duration "
+            f"epochs do not have, so it ran on as_fixed(): every epoch padded "
+            f"to span {self.tmin.min():g} to {self.tmax.max():g} s."
+        )
+        if note:
+            message += " " + note
+        message += " Call as_fixed() yourself to make this explicit."
+        warn(message, RuntimeWarning)
+        fixed, _ = self.as_fixed()
+        return getattr(fixed, name)(*args, **kwargs)
+
+    return wrapper
+
+
+def _raise_needs_policy(func, name, what):
+    """Raise for reductions across a time axis the epochs do not share.
+
+    Parameters
+    ----------
+    func : callable
+        The original method.
+    name : str
+        Its name.
+    what : str
+        Short description of the operation, used in the message.
+
+    Returns
+    -------
+    wrapper : callable
+        The wrapped method.
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        if not getattr(self, "_variable_duration", False):
+            return func(self, *args, **kwargs)
+        raise NotImplementedError(
+            f"{name}() combines epochs across a shared time axis, and these "
+            f"epochs do not share one. {what.capitalize()} them needs an "
+            "explicit policy, because the number of contributing epochs varies "
+            "across the window and no single nave describes it. Either call "
+            "as_fixed(), which pads to the union window and returns that count "
+            "alongside the data, or align the epochs first. See "
+            "https://github.com/mne-tools/mne-python/issues/14206."
+        )
+
+    return wrapper
+
+
+def _raise_not_implemented(func, name, what):
+    """Raise for per-trial operations with no ragged implementation yet.
+
+    Parameters
+    ----------
+    func : callable
+        The original method.
+    name : str
+        Its name.
+    what : str
+        Short description of the operation, used in the message.
+
+    Returns
+    -------
+    wrapper : callable
+        The wrapped method.
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        if not getattr(self, "_variable_duration", False):
+            return func(self, *args, **kwargs)
+        raise NotImplementedError(
+            f"{name}() is not implemented for variable-duration epochs. "
+            f"{what.capitalize()} is per-trial and could work here, but running "
+            "it on a padded copy would change the result rather than just slow "
+            "it down, so it raises until implemented. See "
+            "https://github.com/mne-tools/mne-python/issues/14206."
+        )
+
+    return wrapper
+
+
+for _name, _note in _VARIABLE_FALLBACK.items():
+    _orig = getattr(BaseEpochs, _name, None)
+    if _orig is not None:
+        setattr(BaseEpochs, _name, _wrap_variable_fallback(_orig, _name, _note))
+for _name, _what in _VARIABLE_NEEDS_POLICY.items():
+    _orig = getattr(BaseEpochs, _name, None)
+    if _orig is not None:
+        setattr(BaseEpochs, _name, _raise_needs_policy(_orig, _name, _what))
+for _name, _what in _VARIABLE_NOT_IMPLEMENTED.items():
+    _orig = getattr(BaseEpochs, _name, None)
+    if _orig is not None:
+        setattr(BaseEpochs, _name, _raise_not_implemented(_orig, _name, _what))
+del _name, _note, _what, _orig
+
+
 @fill_doc
 class Epochs(BaseEpochs):
     """Epochs extracted from a Raw instance.

mne/tests/test_epochs_variable_duration.py

@@ -9,6 +9,11 @@
 from numpy.testing import assert_allclose, assert_array_equal
 
 from mne import EpochsArray, create_info
+from mne.epochs import (
+    _VARIABLE_FALLBACK,
+    _VARIABLE_NEEDS_POLICY,
+    _VARIABLE_NOT_IMPLEMENTED,
+)
 
 SFREQ = 100.0
 CH_NAMES = ["a", "b", "c"]
@@ -204,9 +209,91 @@ def test_as_fixed_on_fixed_epochs_is_a_copy():
 
 
 # -- dispatch --------------------------------------------------------------
+@pytest.mark.parametrize("meth", sorted(_VARIABLE_NEEDS_POLICY))
+def test_reductions_ask_for_a_policy(variable, meth):
+    """Test that combining epochs across a time axis they lack is refused.
+
+    Padding first and reducing afterwards is not a slower answer, it is a
+    different one: one short epoch turns a whole time point into NaN, and the
+    scalar ``nave`` keeps reporting the full count.
+    """
+    with pytest.raises(NotImplementedError, match="explicit policy"):
+        result = getattr(variable, meth)()
+        list(result)  # iter_evoked is a generator
+
+
+def test_policy_message_names_the_varying_count(variable):
+    """Test that the refusal explains itself rather than just declining."""
+    with pytest.raises(NotImplementedError) as excinfo:
+        variable.average()
+    message = str(excinfo.value)
+    assert "varies across the window" in message
+    assert "as_fixed" in message
+
+
+def test_compute_tfr_does_not_silently_pad(variable):
+    """Test that the transform is not quietly given padded data.
+
+    Padding before a time-frequency transform is the opposite of the order this
+    work argues for, which is to transform at native duration and warp the
+    result. Doing it silently inside ``compute_tfr`` would ship the thing being
+    argued against.
+    """
+    with pytest.raises(NotImplementedError, match="explicit policy"):
+        variable.compute_tfr("morlet", freqs=np.arange(10.0, 20.0, 2.0), n_cycles=2)
+
+
+@pytest.mark.parametrize("meth", sorted(_VARIABLE_NOT_IMPLEMENTED))
+def test_per_trial_methods_raise_until_implemented(variable, meth):
+    """Test that per-trial work refuses rather than running on a padded copy."""
+    with pytest.raises(NotImplementedError, match="not implemented"):
+        getattr(variable, meth)()
+
+
+@pytest.mark.parametrize("meth", sorted(_VARIABLE_FALLBACK))
+def test_display_methods_warn_and_fall_back(variable, meth):
+    """Test that the remaining inspection method degrades rather than refuses."""
+    if meth == "to_data_frame":
+        pytest.importorskip("pandas")
+    with pytest.warns(RuntimeWarning, match="ran on as_fixed"):
+        assert getattr(variable, meth)() is not None
 
 
 # -- operations that stay native -------------------------------------------
+def test_pick_keeps_durations(variable):
+    """Test that channel selection leaves the time axis alone."""
+    before = variable.durations.copy()
+    picked = variable.copy().pick(["a", "c"])
+    assert picked.ch_names == ["a", "c"]
+    assert_allclose(picked.durations, before)
+    for epoch in picked.get_data():
+        assert epoch.shape[0] == 2
+
+
+def test_getitem_keeps_per_epoch_bounds(variable):
+    """Test that indexing carries the bounds with the epochs."""
+    subset = variable[[0, 2]]
+    assert len(subset) == 2
+    assert_allclose(subset.durations, variable.durations[[0, 2]])
+    for got, want in zip(subset.get_data(), [variable.get_data()[i] for i in (0, 2)]):
+        assert_array_equal(got, want)
+
+
+def test_drop_keeps_per_epoch_bounds(variable):
+    """Test that dropping an epoch drops its bounds too."""
+    kept = variable.copy().drop([1])
+    assert len(kept) == 3
+    assert_allclose(kept.durations, variable.durations[[0, 2, 3]])
+
+
+def test_shift_time_moves_bounds_not_samples(variable):
+    """Test that shifting the origin does not resample anything."""
+    before_lengths = [epoch.shape[1] for epoch in variable.get_data()]
+    before_durations = variable.durations.copy()
+    shifted = variable.copy().shift_time(0.1)
+    assert_allclose(shifted.tmin, variable.tmin + 0.1)
+    assert_allclose(shifted.durations, before_durations)
+    assert [epoch.shape[1] for epoch in shifted.get_data()] == before_lengths
 
 
 # -- the time axis ---------------------------------------------------------
@@ -245,6 +332,30 @@ def test_fixed_epochs_still_have_times():
     assert epochs.average().data.shape == (len(CH_NAMES), 71)
 
 
+def test_nothing_reaches_the_user_as_an_internal_error(variable):
+    """Test that no public method leaks a NumPy error about lists."""
+    import warnings
+
+    names = (
+        sorted(_VARIABLE_FALLBACK)
+        + sorted(_VARIABLE_NEEDS_POLICY)
+        + sorted(_VARIABLE_NOT_IMPLEMENTED)
+    )
+    for name in names:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            try:
+                getattr(variable.copy(), name)()
+            except (NotImplementedError, RuntimeError):
+                pass
+            except TypeError as exc:
+                assert "argument" in str(exc), f"{name}: {exc}"
+            except (AttributeError, IndexError) as exc:
+                raise AssertionError(f"{name} leaked an internal error: {exc}")
+            except Exception:
+                pass
+
+
 # -- construction from Raw -------------------------------------------------
 def _raw(n_seconds=30.0, sfreq=SFREQ):
     """Return a small continuous recording."""
@@ -352,3 +463,41 @@ def test_from_raw_scalar_bounds_still_scalar():
     assert not epochs.variable_duration
     assert isinstance(epochs.tmin, float)
     assert epochs.get_data().shape == (2, len(CH_NAMES), 51)
+
+
+@pytest.mark.parametrize(
+    "item", [slice(None, 2), slice(1, None), slice(None, None, 2), slice(None)]
+)
+def test_getitem_slice_selects_epochs_not_the_slice(variable, item):
+    """Test that slicing subsets the epochs rather than wrapping the slice."""
+    want = np.arange(len(variable))[item]
+    subset = variable[item]
+
+    assert len(subset) == len(want)
+    # a slice used to survive into the data list as a single nested element
+    assert all(isinstance(d, np.ndarray) and d.ndim == 2 for d in subset.get_data())
+    assert_allclose(subset.durations, variable.durations[want])
+    for got, idx in zip(subset.get_data(), want):
+        assert_array_equal(got, variable.get_data()[idx])
+
+
+def test_apply_function_refuses(variable):
+    """Test that apply_function refuses instead of indexing a list with a tuple."""
+    with pytest.raises(NotImplementedError, match="not implemented"):
+        variable.apply_function(lambda x: x * 2)
+
+
+def test_pick_does_not_reach_back_into_the_parent(variable):
+    """Test that picking replaces one object's epochs and no other's."""
+    before = [epoch.shape for epoch in variable.get_data()]
+
+    # _pick_drop_channels replaces the list contents in place, so anything
+    # sharing that list would be picked too
+    variable.copy().pick(["a", "c"])
+    assert [epoch.shape for epoch in variable.get_data()] == before
+
+    subset = variable[:2]
+    assert subset._data is not variable._data
+    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())