ENH: Parse EyeLink BUTTON Events (e.g. from a game controller) (#13499)

Co-authored-by: Wouter Kroot <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 4 people

doc/changes/dev/13499.newfeature.rst

@@ -0,0 +1 @@
+Added support for parsing Eyelink ``BUTTON`` events (i.e. external controller button presses) to :func:`~mne.io.read_raw_eyelink` by :newcontrib:`Wouter Kroot`

doc/changes/names.inc

@@ -340,6 +340,7 @@
 .. _Victoria Peterson: https://github.com/vpeterson
 .. _Wei Xu: https://github.com/psyxw
 .. _Will Turner: https://bootstrapbill.github.io
+.. _Wouter Kroot: https://github.com/WouterKroot
 .. _Xabier de Zuazo: https://github.com/zuazo
 .. _Xiaokai Xia: https://github.com/dddd1007
 .. _Yaroslav Halchenko: http://haxbylab.dartmouth.edu/ppl/yarik.html

mne/io/eyelink/_utils.py

@@ -400,7 +400,24 @@ def _create_dataframes_for_block(block, apply_offsets):
             msgs.append([ts, offset, msg])
         df_dict["messages"] = pd.DataFrame(msgs)
 
-        # TODO: Make dataframes for other eyelink events (Buttons)
+    # make dataframes for other button events
+    if block["events"]["BUTTON"]:
+        button_events = block["events"]["BUTTON"]
+        parsed = []
+        for entry in button_events:
+            parsed.append(
+                {
+                    "time": float(entry[0]),  # onset
+                    "button_id": int(entry[1]),
+                    "button_pressed": int(entry[2]),  # 1 = press, 0 = release
+                }
+            )
+        df_dict["buttons"] = pd.DataFrame(parsed)
+        n_button = len(df_dict.get("buttons", []))
+        logger.info(f"Found {n_button} button event(s) in this file.")
+    else:
+        logger.info("No button events found in this file.")
+
     return df_dict
 
 
@@ -499,7 +516,6 @@ def _combine_block_dataframes(processed_blocks: list[dict]):
 
     for df_type in all_df_types:
         block_dfs = []
-
         for block in processed_blocks:
             if df_type in block["dfs"]:
                 # We will update the dfs in-place to conserve memory
@@ -849,7 +865,7 @@ def _make_eyelink_annots(df_dict, create_annots, apply_offsets):
             "pupil_right",
         ),
     }
-    valid_descs = ["blinks", "saccades", "fixations", "messages"]
+    valid_descs = ["blinks", "saccades", "fixations", "buttons", "messages"]
     msg = (
         "create_annotations must be True or a list containing one or"
         f" more of {valid_descs}."
@@ -875,6 +891,7 @@ def _make_eyelink_annots(df_dict, create_annots, apply_offsets):
             descriptions = key[:-1]  # i.e "blink", "fixation", "saccade"
             if key == "blinks":
                 descriptions = "BAD_" + descriptions
+
             ch_names = df["eye"].map(eye_ch_map).tolist()
             this_annot = Annotations(
                 onset=onsets,
@@ -890,21 +907,44 @@ def _make_eyelink_annots(df_dict, create_annots, apply_offsets):
                 onsets = df["time"]
             durations = [0] * onsets
             descriptions = df["event_msg"]
+            this_annot = Annotations(
+                onset=onsets, duration=durations, description=descriptions
+            )
+        elif (key == "buttons") and (key in descs):
+            required_cols = {"time", "button_id", "button_pressed"}
+            if not required_cols.issubset(df.columns):
+                raise ValueError(f"Missing column: {required_cols - set(df.columns)}")
+            # Give user a hint
+            n_presses = df["button_pressed"].sum()
+            logger.info("Found %d button press events.", n_presses)
+
+            df = df.sort_values("time")
+            onsets = df["time"]
+            durations = np.zeros_like(onsets)
+            descriptions = df.apply(_get_button_description, axis=1)
+
             this_annot = Annotations(
                 onset=onsets, duration=durations, description=descriptions
             )
         else:
-            continue  # TODO make df and annotations for Buttons
+            continue
         if not annots:
             annots = this_annot
         elif annots:
             annots += this_annot
     if not annots:
         warn(f"Annotations for {descs} were requested but none could be made.")
         return
+
     return annots
 
 
+def _get_button_description(row):
+    button_id = int(row["button_id"])
+    action = "press" if row["button_pressed"] == 1 else "release"
+    return f"button_{button_id}_{action}"
+
+
 def _make_gap_annots(raw_extras, key="recording_blocks"):
     """Create Annotations for gap periods between recording blocks."""
     df = raw_extras["dfs"][key]

mne/io/eyelink/tests/test_eyelink.py

@@ -200,7 +200,6 @@ def test_bino_to_mono(tmp_path, fname):
     """Test a file that switched from binocular to monocular mid-recording."""
     out_file = tmp_path / "tmp_eyelink.asc"
     in_file = Path(fname)
-
     lines = in_file.read_text("utf-8").splitlines()
     # We'll also add some binocular velocity data to increase our testing coverage.
     start_idx = [li for li, line in enumerate(lines) if line.startswith("START")][0]
@@ -309,6 +308,18 @@ def _simulate_eye_tracking_data(in_file, out_file):
         "SAMPLES\tPUPIL\tLEFT\tVEL\tRES\tHTARGET\tRATE\t1000.00"
         "\tTRACKING\tCR\tFILTER\t2\tINPUT"
     )
+
+    # Define your known BUTTON events
+    button_events = [
+        (7453390, 1, 1),
+        (7453410, 1, 0),
+        (7453420, 1, 1),
+        (7453430, 1, 0),
+        (7453440, 1, 1),
+        (7453450, 1, 0),
+    ]
+    button_idx = 0
+
     with out_file.open("w") as fp:
         in_recording_block = False
         events = []
@@ -332,6 +343,7 @@ def _simulate_eye_tracking_data(in_file, out_file):
                     tokens.append("INPUT")
                 elif event_type == "EBLINK":
                     continue  # simulate no blink events
+
                 elif event_type == "END":
                     pass
                 else:
@@ -354,6 +366,22 @@ def _simulate_eye_tracking_data(in_file, out_file):
                 "...\t1497\t5189\t512.5\t.............\n"
             )
 
+        for timestamp in np.arange(7453390, 7453490):  # 100 samples 7453100
+            fp.write(
+                f"{timestamp}\t-2434.0\t-1760.0\t840.0\t100\t20\t45\t45\t127.0\t"
+                "...\t1497\t5189\t512.5\t.............\n"
+            )
+            # Check and insert button events at this timestamp
+            if (
+                button_idx < len(button_events)
+                and button_events[button_idx][0] == timestamp
+            ):
+                t, btn_id, state = button_events[button_idx]
+                fp.write(
+                    f"BUTTON\t{t}\t{btn_id}\t{state}\t100\t20\t45\t45\t127.0\t"
+                    "1497.0\t5189.0\t512.5\t.............\n"
+                )
+                button_idx += 1
         fp.write("END\t7453390\tRIGHT\tSAMPLES\tEVENTS\n")
 
 
@@ -397,14 +425,24 @@ def test_multi_block_misc_channels(fname, tmp_path):
 
     assert raw.ch_names == chs_in_file
     assert raw.annotations.description[1] == "SYNCTIME"
-    assert raw.annotations.description[-1] == "BAD_ACQ_SKIP"
-    assert np.isclose(raw.annotations.onset[-1], 1.001)
-    assert np.isclose(raw.annotations.duration[-1], 0.1)
+
+    assert raw.annotations.description[-7] == "BAD_ACQ_SKIP"
+    assert np.isclose(raw.annotations.onset[-7], 1.001)
+    assert np.isclose(raw.annotations.duration[-7], 0.1)
 
     data, times = raw.get_data(return_times=True)
     assert not np.isnan(data[0, np.where(times < 1)[0]]).any()
     assert np.isnan(data[0, np.logical_and(times > 1, times <= 1.1)]).all()
 
+    assert raw.annotations.description[-6] == "button_1_press"
+    button_idx = [
+        ii
+        for ii, desc in enumerate(raw.annotations.description)
+        if "button" in desc.lower()
+    ]
+    assert len(button_idx) == 6
+    assert_allclose(raw.annotations.onset[button_idx[0]], 2.102, atol=1e-3)
+
     # smoke test for reading events with missing samples (should not emit a warning)
     find_events(raw, verbose=True)
 
@@ -465,6 +503,7 @@ def test_href_eye_events(tmp_path):
     """Test Parsing file where Eye Event Data option was set to 'HREF'."""
     out_file = tmp_path / "tmp_eyelink.asc"
     lines = fname_href.read_text("utf-8").splitlines()
+
     for li, line in enumerate(lines):
         if not line.startswith(("ESACC", "EFIX")):
             continue
@@ -479,6 +518,7 @@ def test_href_eye_events(tmp_path):
         new_line = "\t".join(tokens) + "\n"
         lines[li] = new_line
     out_file.write_text("\n".join(lines), encoding="utf-8")
+
     raw = read_raw_eyelink(out_file)
     # Just check that we actually parsed the Saccade and Fixation events
     assert "saccade" in raw.annotations.description