Use Annotations for read_raw_egi events

mne/io/egi/egi.py

@@ -103,6 +103,7 @@ def read_raw_egi(
     exclude=None,
     preload=False,
     channel_naming="E%d",
+    events_as_annotations=False,
     verbose=None,
 ) -> "RawEGI":
     """Read EGI simple binary as raw object.
@@ -137,6 +138,10 @@ def read_raw_egi(
         Channel naming convention for the data channels. Defaults to ``'E%%d'``
         (resulting in channel names ``'E1'``, ``'E2'``, ``'E3'``...). The
         effective default prior to 0.14.0 was ``'EEG %%03d'``.
+    events_as_annotations : bool
+        If True, annotations are created from experiment events. If False (default),
+        synthetic trigger channels are created from experiment events. See the Notes
+        section for details.
 
          .. versionadded:: 0.14.0
     %(verbose)s
@@ -170,7 +175,15 @@ def read_raw_egi(
     input_fname = str(input_fname)
     if input_fname.rstrip("/\\").endswith(".mff"):  # allows .mff or .mff/
         return _read_raw_egi_mff(
-            input_fname, eog, misc, include, exclude, preload, channel_naming, verbose
+            input_fname,
+            eog,
+            misc,
+            include,
+            exclude,
+            preload,
+            channel_naming,
+            events_as_annotations,
+            verbose,
         )
     return RawEGI(
         input_fname, eog, misc, include, exclude, preload, channel_naming, verbose

mne/io/egi/egimff.py

@@ -17,6 +17,7 @@
 from ..._fiff.utils import _create_chs, _mult_cal_one
 from ...annotations import Annotations
 from ...channels.montage import make_dig_montage
+from ...event import find_events
 from ...evoked import EvokedArray
 from ...utils import _check_fname, _check_option, _soft_import, logger, verbose, warn
 from ..base import BaseRaw
@@ -370,6 +371,7 @@ def _read_raw_egi_mff(
     exclude=None,
     preload=False,
     channel_naming="E%d",
+    events_as_annotations=False,
     verbose=None,
 ):
     """Read EGI mff binary as raw object.
@@ -401,6 +403,10 @@ def _read_raw_egi_mff(
         Channel naming convention for the data channels. Defaults to 'E%%d'
         (resulting in channel names 'E1', 'E2', 'E3'...). The effective default
         prior to 0.14.0 was 'EEG %%03d'.
+    events_as_annotations : bool
+        If True, annotations are created from experiment events. If False (default),
+        synthetic trigger channels are created from experiment events. See the Notes
+        section for details.
     %(verbose)s
 
     Returns
@@ -431,7 +437,15 @@ def _read_raw_egi_mff(
     .. versionadded:: 0.15.0
     """
     return RawMff(
-        input_fname, eog, misc, include, exclude, preload, channel_naming, verbose
+        input_fname,
+        eog,
+        misc,
+        include,
+        exclude,
+        preload,
+        channel_naming,
+        events_as_annotations,
+        verbose,
     )
 
 
@@ -450,6 +464,7 @@ def __init__(
         exclude=None,
         preload=False,
         channel_naming="E%d",
+        events_as_annotations=False,
         verbose=None,
     ):
         """Init the RawMff class."""
@@ -487,7 +502,7 @@ def __init__(
                             more_excludes.append(ii)
                 if len(exclude_inds) + len(more_excludes) == len(event_codes):
                     warn(
-                        "Did not find any event code with more than one " "event.",
+                        "Did not find any event code with more than one event.",
                         RuntimeWarning,
                     )
                 else:
@@ -615,7 +630,7 @@ def __init__(
             np.concatenate([idx[key] for key in keys]), np.arange(len(chs))
         ):
             raise ValueError(
-                "Currently interlacing EEG and PNS channels" "is not supported"
+                "Currently interlacing EEG and PNS channels is not supported"
             )
         egi_info["kind_bounds"] = [0]
         for key in keys:
@@ -672,6 +687,15 @@ def __init__(
         if len(annot["onset"]):
             self.set_annotations(Annotations(**annot, orig_time=None))
 
+        # create events from annotations
+        if events_as_annotations:
+            ev = find_events(self)
+            event_dict = {v: k for k, v in self.event_id.items()}
+            annot["onset"].extend(ev[:, 0] / self.info["sfreq"])
+            annot["duration"].extend(np.zeros(ev.shape[0]))
+            annot["description"].extend([event_dict[e] for e in ev[:, 2]])
+            self.set_annotations(Annotations(**annot))
+
     def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
         """Read a chunk of data."""
         logger.debug(f"Reading MFF {start:6d} ... {stop:6d} ...")

mne/io/egi/tests/test_egi.py

@@ -142,15 +142,16 @@ def test_egi_mff_pause_chunks(fname, tmp_path):
 
 
 @requires_testing_data
-def test_io_egi_mff():
+@pytest.mark.parametrize("events_as_annotations", (True, False))
+def test_io_egi_mff(events_as_annotations):
     """Test importing EGI MFF simple binary files."""
     pytest.importorskip("defusedxml")
     # want vars for n chans
     n_ref = 1
     n_eeg = 128
     n_card = 3
 
-    raw = read_raw_egi(egi_mff_fname, include=None)
+    raw = read_raw_egi(egi_mff_fname, events_as_annotations=events_as_annotations)
     assert "RawMff" in repr(raw)
     assert raw.orig_format == "single"
     include = ["DIN1", "DIN2", "DIN3", "DIN4", "DIN5", "DIN7"]
@@ -160,6 +161,7 @@ def test_io_egi_mff():
         include=include,
         channel_naming="EEG %03d",
         test_scaling=False,  # XXX probably some bug
+        events_as_annotations=events_as_annotations,
     )
     assert raw.info["sfreq"] == 1000.0
     assert len(raw.info["dig"]) == n_card + n_eeg + n_ref
@@ -198,6 +200,20 @@ def test_io_egi_mff():
     for ch in include:
         assert ch in raw.event_id
         assert raw.event_id[ch] == int(ch[-1])
+    # test converting stim triggers to annotations
+    if events_as_annotations:
+        # Grab the first annotation. Should be the first "DIN1" event.
+        assert len(raw.annotations)
+        onset, dur, desc, _ = raw.annotations[0].values()
+        assert np.isclose(onset, 2.438)
+        assert np.isclose(dur, 0)
+        assert desc == "DIN1"
+        # grab the DIN1 channel
+        din1 = raw.get_data(picks="DIN1")
+        # Check that the time in sec of first event is the same as the first annotation
+        pin_hi_idx = np.where(din1 == 1)[1]
+        pin_hi_sec = pin_hi_idx / raw.info["sfreq"]
+        assert np.isclose(pin_hi_sec[0], onset)
 
 
 def test_io_egi():