FIX: add recover_epochs fallback for EGI .mff epoch metadata mismatch

doc/changes/dev/13992.bugfix.rst

@@ -0,0 +1 @@
+Add ``recover_epochs`` parameter to :func:`~mne.io.read_raw_egi` to gracefully recover epoch structure from binary signal data when ``epochs.xml`` metadata is inconsistent (e.g., improperly closed or interrupted EGI recordings), by :newcontrib:`Krishnaveni Parvataneni`.

doc/changes/names.inc

@@ -188,6 +188,7 @@
 .. _Keith Doelling: https://github.com/kdoelling1919
 .. _Konstantinos Tsilimparis: https://contsili.github.io/
 .. _Kostiantyn Maksymenko: https://github.com/makkostya
+.. _Krishnaveni Parvataneni: https://github.com/kveni12
 .. _Kristijan Armeni: https://github.com/kristijanarmeni
 .. _Kyle Mathewson: https://github.com/kylemath
 .. _Larry Eisenman:  https://github.com/lneisenman

mne/io/egi/egi.py

@@ -100,6 +100,7 @@ def read_raw_egi(
     preload=False,
     channel_naming="E%d",
     *,
+    recover_epochs=False,
     events_as_annotations=True,
     verbose=None,
 ) -> "RawEGI":
@@ -134,6 +135,15 @@ def read_raw_egi(
 
         .. versionadded:: 0.14.0
 
+    recover_epochs : bool
+        If True and an MFF file has inconsistent epoch metadata (e.g., from an
+        improperly closed recording or interrupted acquisition), attempt to recover
+        the epoch structure from the binary signal data instead of raising an error.
+        A :class:`RuntimeWarning` is emitted when recovery is performed.
+        Default is False.
+
+        .. versionadded:: 1.13.0
+
     events_as_annotations : bool
         If True, annotations are created from experiment events. If False (default),
         a synthetic trigger channel ``STI 014`` is created from experiment events.
@@ -181,6 +191,7 @@ def read_raw_egi(
             exclude,
             preload,
             channel_naming,
+            recover_epochs=recover_epochs,
             events_as_annotations=events_as_annotations,
             verbose=verbose,
         )

mne/io/egi/egimff.py

@@ -35,7 +35,7 @@
 REFERENCE_NAMES = ("VREF", "Vertex Reference")
 
 
-def _read_mff_header(filepath):
+def _read_mff_header(filepath, recover_epochs=False):
     """Read mff header."""
     _soft_import("mffpy", "reading EGI MFF data")
     from mffpy.xml_files import XML
@@ -96,10 +96,29 @@ def _read_mff_header(filepath):
         or not (epochs["first_samps"][1:] >= epochs["last_samps"][:-1]).all()
     )
     if bad:
-        raise RuntimeError(
-            "EGI epoch first/last samps could not be parsed:\n"
-            f"{list(epochs['first_samps'])}\n{list(epochs['last_samps'])}"
+        if not recover_epochs:
+            raise RuntimeError(
+                "EGI epoch first/last samps could not be parsed:\n"
+                f"{list(epochs['first_samps'])}\n{list(epochs['last_samps'])}"
+            )
+        warn(
+            "EGI epoch metadata mismatch; recovering epoch structure from binary data",
+            RuntimeWarning,
+            stacklevel=2,
+        )
+        # Rebuild epoch boundaries from the actual binary block sizes
+        block_sizes = signal_blocks["samples_block"]
+        total_samps = int(block_sizes.sum())
+        n_epochs = len(epochs["first_samps"])
+        samps_per_epoch = total_samps // n_epochs
+        epochs["first_samps"] = np.array(
+            [i * samps_per_epoch for i in range(n_epochs)], dtype=np.int64
+        )
+        epochs["last_samps"] = np.array(
+            [min((i + 1) * samps_per_epoch, total_samps) for i in range(n_epochs)],
+            dtype=np.int64,
         )
+        n_samps_epochs = (epochs["last_samps"] - epochs["first_samps"]).sum()
     summaryinfo.update(epochs)
 
     disk_samps = np.full(epochs["last_samps"][-1], -1)
@@ -193,21 +212,23 @@ def _read_mff_header(filepath):
     return summaryinfo
 
 
-def _read_header(input_fname):
+def _read_header(input_fname, recover_epochs=False):
     """Obtain the headers from the file package mff.
 
     Parameters
     ----------
     input_fname : path-like
         Path for the file
+    recover_epochs : bool
+        If True, recover epoch structure from binary data when metadata is inconsistent.
 
     Returns
     -------
     info : dict
         Main headers set.
     """
     input_fname = str(input_fname)  # cast to str any Paths
-    mff_hdr = _read_mff_header(input_fname)
+    mff_hdr = _read_mff_header(input_fname, recover_epochs=recover_epochs)
     with open(input_fname + "/signal1.bin", "rb") as fid:
         version = np.fromfile(fid, np.int32, 1)[0]
     """
@@ -346,6 +367,7 @@ def _read_raw_egi_mff(
     preload=False,
     channel_naming="E%d",
     *,
+    recover_epochs=False,
     events_as_annotations=True,
     verbose=None,
 ):
@@ -358,6 +380,7 @@ def _read_raw_egi_mff(
         exclude,
         preload,
         channel_naming,
+        recover_epochs=recover_epochs,
         events_as_annotations=events_as_annotations,
         verbose=verbose,
     )
@@ -379,6 +402,7 @@ def __init__(
         preload=False,
         channel_naming="E%d",
         *,
+        recover_epochs=False,
         events_as_annotations=True,
         verbose=None,
     ):
@@ -393,7 +417,7 @@ def __init__(
             )
         )
         logger.info(f"Reading EGI MFF Header from {input_fname}...")
-        egi_info = _read_header(input_fname)
+        egi_info = _read_header(input_fname, recover_epochs=recover_epochs)
         if eog is None:
             eog = []
         if misc is None:

mne/io/egi/tests/test_egi.py

@@ -598,3 +598,45 @@ def test_egi_mff_bad_xml(tmp_path):
             raw = read_raw_egi(mff_fname)
     # little check that the bad XML doesn't affect the parsing of other xml files
     assert "DIN1" in raw.annotations.description
+
+
+@requires_testing_data
+def test_recover_epochs_raises_by_default(tmp_path):
+    """Test that mismatched epoch metadata raises RuntimeError by default."""
+    import re as _re
+
+    mff_fname = copytree_rw(egi_mff_fname, tmp_path / "test_egi_bad_epochs.mff")
+    epochs_xml = mff_fname / "epochs.xml"
+    content = epochs_xml.read_text(encoding="utf-8")
+
+    def _corrupt_end(m):
+        val = int(m.group(1))
+        return f"<endTime>{val * 100}</endTime>"
+
+    bad_content = _re.sub(r"<endTime>(\d+)</endTime>", _corrupt_end, content)
+    epochs_xml.write_text(bad_content, encoding="utf-8")
+    with pytest.raises(
+        RuntimeError, match="EGI epoch first/last samps could not be parsed"
+    ):
+        read_raw_egi(mff_fname, recover_epochs=False)
+
+
+@requires_testing_data
+def test_recover_epochs_succeeds_with_flag(tmp_path):
+    """Test that recover_epochs=True loads a file with mismatched epoch metadata."""
+    import re as _re
+
+    mff_fname = copytree_rw(egi_mff_fname, tmp_path / "test_egi_recover_epochs.mff")
+    epochs_xml = mff_fname / "epochs.xml"
+    content = epochs_xml.read_text(encoding="utf-8")
+
+    def _corrupt_end(m):
+        val = int(m.group(1))
+        return f"<endTime>{val * 100}</endTime>"
+
+    bad_content = _re.sub(r"<endTime>(\d+)</endTime>", _corrupt_end, content)
+    epochs_xml.write_text(bad_content, encoding="utf-8")
+    with pytest.warns(RuntimeWarning, match="EGI epoch metadata mismatch"):
+        raw = read_raw_egi(mff_fname, recover_epochs=True)
+    assert raw is not None
+    assert raw.n_times > 0