Event Keys

Author: scott-huberty

mne/io/egi/egi.py

@@ -101,6 +101,7 @@ def read_raw_egi(
     channel_naming="E%d",
     *,
     events_as_annotations=True,
+    event_key=None,
     verbose=None,
 ) -> "RawEGI":
     """Read EGI simple binary as raw object.
@@ -141,6 +142,15 @@ def read_raw_egi(
         The default will change from False to True in version 1.9.
 
         .. versionadded:: 1.8.0
+    event_key : str | None
+        The MFF event key whose value is appended to annotation descriptions and extras
+        when ``events_as_annotations=True``, and to stim channel names when
+        ``events_as_annotations=False``. For example, ``event_key='cel#'`` will convert
+        an MFF event code ``'stim'`` with ``cel#=1`` to ``'stim_1'``. These will also
+        appear as ``'event_key_cel#': 1`` in annotations.extras
+        Only supported for MFF files.
+
+        .. versionadded:: 1.11.0
     %(verbose)s
 
     Returns
@@ -155,22 +165,31 @@ def read_raw_egi(
 
     Notes
     -----
-    When ``events_from_annotations=True``, event codes on stimulus channels like
+    When ``events_as_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
+    When ``events_as_annotations=False`` and events are present on the included
+    stimulus channels, a new stim channel ``STI 014`` 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``.
+
+    MFF event track XML files can store additional metadata in a
+    ``<keys>`` child element of each ``<event>`` element. This corresponds to
+    the ECI Event Data Stream "Key List" described in the
+    `EGI Amp Server Pro SDK User Guide
+    <https://www.egi.com/images/stories/manuals/amp-server-pro-sdk-3-0-network-apis-user-guide-rev-01.pdf>`__.
+    For example, E-Prime driven experiments sometimes store the experimental condition
+    in a ``'cel#'`` event key.
     """
     _validate_type(input_fname, "path-like", "input_fname")
     input_fname = str(input_fname)
     _validate_type(events_as_annotations, bool, "events_as_annotations")
+    _validate_type(event_key, (str, None), "event_key")
 
     if input_fname.rstrip("/\\").endswith(".mff"):  # allows .mff or .mff/
         return _read_raw_egi_mff(
@@ -182,8 +201,10 @@ def read_raw_egi(
             preload,
             channel_naming,
             events_as_annotations=events_as_annotations,
+            event_key=event_key,
             verbose=verbose,
         )
+
     return RawEGI(
         input_fname,
         eog,

mne/io/egi/egimff.py

@@ -369,6 +369,7 @@ def _read_raw_egi_mff(
     channel_naming="E%d",
     *,
     events_as_annotations=True,
+    event_key=None,
     verbose=None,
 ):
     """Read EGI mff binary as raw object."""
@@ -381,6 +382,7 @@ def _read_raw_egi_mff(
         preload,
         channel_naming,
         events_as_annotations=events_as_annotations,
+        event_key=event_key,
         verbose=verbose,
     )
 
@@ -402,6 +404,7 @@ def __init__(
         channel_naming="E%d",
         *,
         events_as_annotations=True,
+        event_key=None,
         verbose=None,
     ):
         """Init the RawMff class."""
@@ -422,7 +425,9 @@ def __init__(
             misc = np.where(np.array(egi_info["chan_type"]) != "eeg")[0].tolist()
 
         logger.info("    Reading events ...")
-        egi_events, egi_info, mff_events = _read_events(input_fname, egi_info)
+        egi_events, egi_info, mff_events = _read_events(
+            input_fname, egi_info, event_key=event_key
+        )
         cals = np.array([CAL_SCALES[t] for t in egi_info["chan_unit"]])
         logger.info("    Assembling measurement info ...")
         event_codes = egi_info["event_codes"]

mne/io/egi/events.py

@@ -10,7 +10,7 @@
 from ...utils import _soft_import, _validate_type, logger, warn
 
 
-def _read_events(input_fname, info):
+def _read_events(input_fname, info, *, event_key=None):
     """Read events for the record.
 
     Parameters
@@ -22,7 +22,7 @@ def _read_events(input_fname, info):
     """
     n_samples = info["last_samps"][-1]
     mff_events, event_codes = _read_mff_events(
-        input_fname, info["sfreq"], info["meas_dt_local"]
+        input_fname, info["sfreq"], info["meas_dt_local"], event_key=event_key
     )
     info["n_events"] = len(event_codes)
     info["event_codes"] = event_codes
@@ -36,7 +36,7 @@ def _read_events(input_fname, info):
     return events, info, mff_events
 
 
-def _read_mff_events(filename, sfreq, start_time):
+def _read_mff_events(filename, sfreq, start_time, *, event_key=None):
     """Extract the events with mffpy."""
     mffpy = _soft_import("mffpy", purpose="reading EGI MFF data", min_version="0.11")
     from mffpy.xml_files import XML
@@ -69,9 +69,10 @@ def _read_mff_events(filename, sfreq, start_time):
     for track in tracks:
         for event in track.events:
             code_str = event["code"]
-            cel = event.get("keys", {}).get("cel#")
-            if cel is not None:
-                code_str = f"{code_str}_{int(cel)}"
+            keys = event.get("keys", {})
+            key_value = keys.get(event_key) if event_key is not None else None
+            if key_value is not None:
+                code_str = f"{code_str}_{key_value}"
             if code_str not in code:
                 code.append(code_str)
 
@@ -86,6 +87,14 @@ def _read_mff_events(filename, sfreq, start_time):
                 )
             else:
                 extras = {}
+            extras.update(
+                {
+                    f"event_key_{key}": value.item()
+                    if isinstance(value, np.generic)
+                    else value
+                    for key, value in keys.items()
+                }
+            )
 
             markers.append(
                 {

mne/io/egi/tests/test_egi.py

@@ -600,39 +600,20 @@ def test_egi_mff_bad_xml(tmp_path):
     # 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_egi_mff_cel_condition_codes():
-    """Test that cel# compound codes are only generated for repeated trial events.
-
-    EGI records a ``cel#`` key on each event to identify the experimental
-    condition.  When a code appears multiple times per condition (e.g. ``stm+``
-    firing on every trial) ``read_raw_egi`` should append the cel# value and
-    produce codes like ``stm+_1`` / ``stm+_2``.
-
-    However, some files use a once-per-condition-block ``CELL`` marker where
-    each cel# value (1–5) fires exactly once.  Splitting those into separate
-    channels would create consecutive-sample trigger values that confuse
-    ``find_events`` and prevent ``STI 014`` from being synthesised.  The reader
-    must leave such codes unsplit.
-    """
-    import mffpy
-
-    for fname in (egi_pause_fname, egi_eprime_pause_fname):
-        r = mffpy.Reader(str(fname))
-        _, codes = _read_mff_events(str(fname), 250.0, r.startdatetime)
-        assert "CELL" in codes, f"CELL missing in {fname.name}"
-        assert not any(c.startswith("CELL_") for c in codes), (
-            f"CELL should not be split into per-condition channels in "
-            f"{fname.name}; got {codes}"
-        )
-
-    # Also verify end-to-end: STI 014 must be synthesised and CELL must appear
-    # in raw.event_id (not split) when reading the full file.
-    for fname in (egi_pause_fname, egi_eprime_pause_fname):
-        raw = read_raw_egi(fname, events_as_annotations=False)
-        assert "CELL" in raw.event_id, f"CELL missing from event_id in {fname.name}"
-        assert not any(k.startswith("CELL_") for k in raw.event_id), (
-            f"CELL should not be split in {fname.name}; got {list(raw.event_id)}"
+@pytest.mark.parametrize(
+    "fname, expected",
+    [pytest.param(egi_pause_fname, "AM40_3", id="paused")],
+)
+def test_read_event_keys(fname, expected):
+    """Test event metadata extraction from``<keys>`` child elements."""
+    with pytest.warns(RuntimeWarning, match="Acquisition skips detected"):
+        raw = _test_raw_reader(
+            read_raw_egi,
+            input_fname=fname,
+            test_scaling=False,
+            test_rank="less",
+            events_as_annotations=True,
+            event_key="cel#",
         )
-        assert "STI 014" in raw.ch_names, f"STI 014 not synthesised in {fname.name}"
+    assert expected in raw.annotations.description