ENH: Add button events to Eyelink

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,7 +891,9 @@ 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()
+            # breakpoint()  # debug
             this_annot = Annotations(
                 onset=onsets,
                 duration=durations,
@@ -890,18 +908,37 @@ 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)}")
+
+            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}"
+
+            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
 
 

mne/io/eyelink/download_test_data.py

@@ -0,0 +1,7 @@
+from mne.datasets import testing
+
+# testing.data_path(force_update=True, download=True)
+testing.data_path(force_update=False)
+
+path = testing.data_path()
+print(path)

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,35 @@ 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)
+
+    print("\n=== Annotations ===")
+    for onset, duration, desc in zip(
+        raw.annotations.onset, raw.annotations.duration, raw.annotations.description
+    ):
+        print(f"{onset:.3f}s  dur={duration:.3f}s  {desc}")
+
+    print("\n=== Recording block markers ===")
+    for onset, desc in zip(raw.annotations.onset, raw.annotations.description):
+        if desc == "BAD_ACQ_SKIP":
+            print(f"BAD_ACQ_SKIP at {onset:.3f}s")
+
+    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,20 +514,24 @@ 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
         tokens = line.split()
         if line.startswith("ESACC"):
             href_sacc_vals = ["9999", "9999", "9999", "9999", "99.99", "999"]
             tokens[5:5] = href_sacc_vals  # add href saccade values
+            # print(f"\nLine {li}: {line}")
+            # print(f"Tokens ({len(tokens)}): {tokens}")
         elif line.startswith("EFIX"):
             tokens = line.split()
             href_fix_vals = ["9999.9", "9999.9", "999"]
             tokens[5:3] = href_fix_vals
         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