Use Annotations for read_raw_egi events (#12300)

Co-authored-by: Eric Larson <[email protected]>
Co-authored-by: nafraw <[email protected]>

doc/changes/devel/12300.apichange.rst

@@ -0,0 +1,3 @@
+A new argument ``events_as_annotations`` has been added to :func:`mne.io.read_raw_egi`
+with a default value of ``False`` that will change to ``True`` in version 1.9, by
+`Scott Huberty`_ and `Eric Larson`_.

doc/changes/devel/12300.bugfix.rst

@@ -0,0 +1 @@
+Fix bug where an event that occurred only once was excluded in :func:`mne.io.read_raw_egi`, by :newcontrib:`Ping-Keng Jao`.

doc/changes/names.inc

@@ -454,6 +454,8 @@
 
 .. _Pierre-Antoine Bannier: https://github.com/PABannier
 
+.. _Ping-Keng Jao: https://github.com/nafraw
+
 .. _Proloy Das: https://github.com/proloyd
 
 .. _Qian Chu: https://github.com/qian-chu

mne/channels/tests/test_montage.py

@@ -1074,7 +1074,11 @@ def test_egi_dig_montage(tmp_path):
     )
 
     # Test accuracy and embedding within raw object
-    raw_egi = read_raw_egi(egi_raw_fname, channel_naming="EEG %03d")
+    raw_egi = read_raw_egi(
+        egi_raw_fname,
+        channel_naming="EEG %03d",
+        events_as_annotations=True,
+    )
 
     raw_egi.set_montage(dig_montage)
     test_raw_egi = read_raw_fif(egi_fif_fname)

mne/io/egi/egi.py

@@ -13,10 +13,11 @@
 from ..._fiff.constants import FIFF
 from ..._fiff.meas_info import _empty_info
 from ..._fiff.utils import _create_chs, _read_segments_file
+from ...annotations import Annotations
 from ...utils import _check_fname, _validate_type, logger, verbose, warn
 from ..base import BaseRaw
 from .egimff import _read_raw_egi_mff
-from .events import _combine_triggers
+from .events import _combine_triggers, _triage_include_exclude
 
 
 def _read_header(fid):
@@ -101,13 +102,12 @@ def read_raw_egi(
     exclude=None,
     preload=False,
     channel_naming="E%d",
+    *,
+    events_as_annotations=None,
     verbose=None,
 ) -> "RawEGI":
     """Read EGI simple binary as raw object.
 
-    .. note:: This function attempts to create a synthetic trigger channel.
-              See the Notes section below.
-
     Parameters
     ----------
     input_fname : path-like
@@ -120,23 +120,29 @@ def read_raw_egi(
         Names of channels or list of indices that should be designated
         MISC channels. Default is None.
     include : None | list
-       The event channels to be ignored when creating the synthetic
-       trigger. Defaults to None.
+       The event channels to be included when creating the synthetic
+       trigger or annotations. Defaults to None.
        Note. Overrides ``exclude`` parameter.
     exclude : None | list
        The event channels to be ignored when creating the synthetic
-       trigger. Defaults to None. If None, channels that have more than
-       one event and the ``sync`` and ``TREV`` channels will be
-       ignored.
+       trigger or annotations. Defaults to None. If None, the ``sync`` and ``TREV``
+       channels will be ignored. This is ignored when ``include`` is not None.
     %(preload)s
 
         .. versionadded:: 0.11
     channel_naming : str
         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'``.
+        .. versionadded:: 0.14.0
 
-         .. versionadded:: 0.14.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.
+        See the Notes section for details.
+        The default will change from False to True in version 1.9.
+
+        .. versionadded:: 1.8.0
     %(verbose)s
 
     Returns
@@ -151,27 +157,51 @@ def read_raw_egi(
 
     Notes
     -----
-    The trigger channel names are based on the arbitrary user dependent event
-    codes used. However this function will attempt to generate a **synthetic
-    trigger channel** named ``STI 014`` in accordance with the general
-    Neuromag / MNE naming pattern.
-
-    The event_id assignment equals ``np.arange(n_events) + 1``. The resulting
-    ``event_id`` mapping is stored as attribute to the resulting raw object but
-    will be ignored when saving to a fiff. Note. The trigger channel is
-    artificially constructed based on timestamps received by the Netstation.
-    As a consequence, triggers have only short durations.
-
-    This step will fail if events are not mutually exclusive.
+    When ``events_from_annotations=True``, event codes on stimulus channels like
+    ``DIN1`` are stored as annotations with the ``description`` set to the stimulus
+    channel name.
+
+    When ``events_from_annotations=False`` and events are present on the included
+    stimulus channels, a new stim channel ``STI014`` will be synthesized from the
+    events. It will contain 1-sample pulses where the Netstation file had event
+    timestamps. A ``raw.event_id`` dictionary is added to the raw object that will have
+    arbitrary sequential integer IDs for the events. This will fail if any timestamps
+    are duplicated. The ``event_id`` will also not survive a save/load roundtrip.
+
+    For these reasons, it is recommended to use ``events_as_annotations=True``.
     """
     _validate_type(input_fname, "path-like", "input_fname")
     input_fname = str(input_fname)
+    if events_as_annotations is None:
+        warn(
+            "events_as_annotations defaults to False in 1.8 but will change to "
+            "True in 1.9, set it explicitly to avoid this warning",
+            FutureWarning,
+        )
+        events_as_annotations = False
+
     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=events_as_annotations,
+            verbose=verbose,
         )
     return RawEGI(
-        input_fname, eog, misc, include, exclude, preload, channel_naming, verbose
+        input_fname,
+        eog,
+        misc,
+        include,
+        exclude,
+        preload,
+        channel_naming,
+        events_as_annotations=events_as_annotations,
+        verbose=verbose,
     )
 
 
@@ -190,6 +220,8 @@ def __init__(
         exclude=None,
         preload=False,
         channel_naming="E%d",
+        *,
+        events_as_annotations=True,
         verbose=None,
     ):
         input_fname = str(_check_fname(input_fname, "read", True, "input_fname"))
@@ -209,58 +241,18 @@ def __init__(
 
         logger.info("    Assembling measurement info ...")
 
-        event_codes = []
-        if egi_info["n_events"] > 0:
-            event_codes = list(egi_info["event_codes"])
-            if include is None:
-                exclude_list = ["sync", "TREV"] if exclude is None else exclude
-                exclude_inds = [
-                    i for i, k in enumerate(event_codes) if k in exclude_list
-                ]
-                more_excludes = []
-                if exclude is None:
-                    for ii, event in enumerate(egi_events):
-                        if event.sum() <= 1 and event_codes[ii]:
-                            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.",
-                        RuntimeWarning,
-                    )
-                else:
-                    exclude_inds.extend(more_excludes)
-
-                exclude_inds.sort()
-                include_ = [
-                    i for i in np.arange(egi_info["n_events"]) if i not in exclude_inds
-                ]
-                include_names = [k for i, k in enumerate(event_codes) if i in include_]
-            else:
-                include_ = [i for i, k in enumerate(event_codes) if k in include]
-                include_names = include
-
-            for kk, v in [("include", include_names), ("exclude", exclude)]:
-                if isinstance(v, list):
-                    for k in v:
-                        if k not in event_codes:
-                            raise ValueError(f'Could find event named "{k}"')
-                elif v is not None:
-                    raise ValueError(f"`{kk}` must be None or of type list")
-
-            event_ids = np.arange(len(include_)) + 1
+        event_codes = egi_info["event_codes"]
+        include = _triage_include_exclude(include, exclude, egi_events, egi_info)
+        if egi_info["n_events"] > 0 and not events_as_annotations:
+            event_ids = np.arange(len(include)) + 1
             logger.info('    Synthesizing trigger channel "STI 014" ...')
-            excl_events = ", ".join(
-                k for i, k in enumerate(event_codes) if i not in include_
-            )
-            logger.info(f"    Excluding events {{{excl_events}}} ...")
             egi_info["new_trigger"] = _combine_triggers(
-                egi_events[include_], remapping=event_ids
+                egi_events[[e in include for e in event_codes]], remapping=event_ids
             )
             self.event_id = dict(
-                zip([e for e in event_codes if e in include_names], event_ids)
+                zip([e for e in event_codes if e in include], event_ids)
             )
         else:
-            # No events
             self.event_id = None
             egi_info["new_trigger"] = None
         info = _empty_info(egi_info["samp_rate"])
@@ -275,11 +267,12 @@ def __init__(
         my_timestamp = time.mktime(my_time.timetuple())
         info["meas_date"] = (my_timestamp, 0)
         ch_names = [channel_naming % (i + 1) for i in range(egi_info["n_channels"])]
-        ch_names.extend(list(egi_info["event_codes"]))
+        cals = np.repeat(cal, len(ch_names))
+        ch_names.extend(list(event_codes))
+        cals = np.concatenate([cals, np.ones(egi_info["n_events"])])
         if egi_info["new_trigger"] is not None:
             ch_names.append("STI 014")  # our new_trigger
-        nchan = len(ch_names)
-        cals = np.repeat(cal, nchan)
+            cals = np.concatenate([cals, [1.0]])
         ch_coil = FIFF.FIFFV_COIL_EEG
         ch_kind = FIFF.FIFFV_EEG_CH
         chs = _create_chs(ch_names, cals, ch_coil, ch_kind, eog, (), (), misc)
@@ -292,7 +285,6 @@ def __init__(
             chs[idx].update(
                 {
                     "unit_mul": FIFF.FIFF_UNITM_NONE,
-                    "cal": 1.0,
                     "kind": FIFF.FIFFV_STIM_CH,
                     "coil_type": FIFF.FIFFV_COIL_NONE,
                     "unit": FIFF.FIFF_UNIT_NONE,
@@ -314,6 +306,17 @@ def __init__(
             raw_extras=[egi_info],
             verbose=verbose,
         )
+        if events_as_annotations:
+            annot = dict(onset=list(), duration=list(), description=list())
+            for code, row in zip(egi_info["event_codes"], egi_events):
+                if code not in include:
+                    continue
+                onset = np.where(row)[0] / self.info["sfreq"]
+                annot["onset"].extend(onset)
+                annot["duration"].extend([0.0] * len(onset))
+                annot["description"].extend([code] * len(onset))
+            if annot:
+                self.set_annotations(Annotations(**annot))
 
     def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
         """Read a segment of data from a file."""