-
Notifications
You must be signed in to change notification settings - Fork 91
Refactor _handle_events_reading to allow extracting annotation information stand-alone #1389
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 16 commits
c2d078a
61fd6bb
c3f3317
8428c75
f17ef79
44a885e
49fa312
2942010
f98ac2b
2dd91cc
8868355
4fdf132
d7f92da
10f634c
4e6b554
02d26b2
cce5558
9c75050
b62ae70
a4ce5e5
6559e55
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -523,8 +523,70 @@ def _handle_info_reading(sidecar_fname, raw): | |
return raw | ||
|
||
|
||
def _handle_events_reading(events_fname, raw): | ||
"""Read associated events.tsv and convert valid events to annotations on Raw.""" | ||
def events_file_to_annotation_kwargs(events_fname: str | Path) -> dict: | ||
r""" | ||
Read the `events.tsv` file and extract onset, duration, and description. | ||
|
||
This function reads an events file in TSV format and extracts the onset, | ||
duration, and description of events. | ||
|
||
Parameters | ||
---------- | ||
events_fname : str | ||
The file path to the `events.tsv` file. | ||
|
||
Returns | ||
------- | ||
dict | ||
A dictionary containing the following keys: | ||
- 'onset' : np.ndarray | ||
The onset times of the events in seconds. | ||
drammock marked this conversation as resolved.
Show resolved
Hide resolved
|
||
- 'duration' : np.ndarray | ||
The durations of the events in seconds. | ||
- 'description' : np.ndarray | ||
The descriptions of the events. | ||
- 'event_id' : dict | ||
A dictionary mapping event descriptions to integer event IDs. | ||
|
||
Notes | ||
----- | ||
The function handles the following cases: | ||
- If the `trial_type` column is available, it uses it for event descriptions. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add blank line before bullet list starts |
||
- If the `stim_type` column is available, it uses it for backward compatibility. | ||
- If the `value` column is available, it uses it to create the `event_id`. | ||
- If none of the above columns are available, it defaults to using 'n/a' for | ||
descriptions and 1 for event IDs. | ||
|
||
Examples | ||
-------- | ||
>>> import pandas as pd | ||
>>> from pathlib import Path | ||
>>> import tempfile | ||
>>> | ||
>>> # Create a sample DataFrame | ||
>>> data = { | ||
... 'onset': [0.1, 0.2, 0.3], | ||
... 'duration': [0.1, 0.1, 0.1], | ||
... 'trial_type': ['event1', 'event2', 'event1'], | ||
... 'value': [1, 2, 1], | ||
... 'sample': [10, 20, 30] | ||
... } | ||
>>> df = pd.DataFrame(data) | ||
>>> | ||
>>> # Write the DataFrame to a temporary file | ||
>>> temp_dir = tempfile.gettempdir() | ||
>>> events_file = Path(temp_dir) / 'events.tsv' | ||
>>> df.to_csv(events_file, sep='\t', index=False) | ||
>>> | ||
>>> # Read the events file using the function | ||
>>> events_dict = events_file_to_annotation_kwargs(events_file) | ||
>>> events_dict | ||
{'onset': array([0.1, 0.2, 0.3]), | ||
'duration': array([0.1, 0.1, 0.1]), | ||
'description': array(['event1', 'event2', 'event1'], dtype='<U6'), | ||
'event_id': {'event1': 1, 'event2': 2}} | ||
|
||
""" | ||
logger.info(f"Reading events from {events_fname}.") | ||
events_dict = _from_tsv(events_fname) | ||
|
||
|
@@ -601,9 +663,21 @@ def _handle_events_reading(events_fname, raw): | |
[0 if du == "n/a" else du for du in events_dict["duration"]], dtype=float | ||
) | ||
|
||
return {"onset": ons, "duration": durs, "description": descrs, "event_id": event_id} | ||
|
||
|
||
def _handle_events_reading(events_fname, raw): | ||
"""Read associated events.tsv and convert valid events to annotations on Raw.""" | ||
annotations_info = events_file_to_annotation_kwargs(events_fname) | ||
event_id = annotations_info["event_id"] | ||
|
||
# Add events as Annotations, but keep essential Annotations present in raw file | ||
annot_from_raw = raw.annotations.copy() | ||
annot_from_events = mne.Annotations(onset=ons, duration=durs, description=descrs) | ||
annot_from_events = mne.Annotations( | ||
onset=annotations_info["onset"], | ||
duration=annotations_info["duration"], | ||
description=annotations_info["description"], | ||
) | ||
raw.set_annotations(annot_from_events) | ||
|
||
annot_idx_to_keep = [ | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -14,6 +14,7 @@ | |
|
||
import mne | ||
import numpy as np | ||
import pandas as pd | ||
import pytest | ||
from mne.datasets import testing | ||
from mne.io.constants import FIFF | ||
|
@@ -32,6 +33,7 @@ | |
_handle_events_reading, | ||
_handle_scans_reading, | ||
_read_raw, | ||
events_file_to_annotation_kwargs, | ||
get_head_mri_trans, | ||
read_raw_bids, | ||
) | ||
|
@@ -1466,3 +1468,76 @@ def test_gsr_and_temp_reading(): | |
raw = read_raw_bids(bids_path) | ||
assert raw.get_channel_types(["GSR"]) == ["gsr"] | ||
assert raw.get_channel_types(["Temperature"]) == ["temperature"] | ||
|
||
|
||
def test_events_file_to_annotation_kwargs(tmp_path): | ||
"""Test that events file is read correctly.""" | ||
bids_path = BIDSPath( | ||
subject="01", session="eeg", task="rest", datatype="eeg", root=tiny_bids_root | ||
) | ||
events_fname = _find_matching_sidecar(bids_path, suffix="events", extension=".tsv") | ||
|
||
# ---------------- plain read -------------------------------------------- | ||
df = pd.read_csv(events_fname, sep="\t") | ||
ev_kwargs = events_file_to_annotation_kwargs(events_fname=events_fname) | ||
assert (ev_kwargs["onset"] == df["onset"].values).all() | ||
assert (ev_kwargs["duration"] == df["duration"].values).all() | ||
assert (ev_kwargs["description"] == df["trial_type"].values).all() | ||
matthiasdold marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
# ---------------- filtering out n/a values ------------------------------ | ||
tmp_tsv_file = tmp_path / "events.tsv" | ||
dext = pd.concat( | ||
[df.copy().assign(onset=df.onset + i) for i in range(5)] | ||
).reset_index(drop=True) | ||
|
||
dext = dext.assign( | ||
ix=range(len(dext)), | ||
value=dext.trial_type.map({"start_experiment": 1, "show_stimulus": 2}), | ||
duration=1.0, | ||
) | ||
|
||
# nan values for `_drop` must be string values, `_drop` is called on | ||
# `onset`, `value` and `trial_type`. `duration` n/a should end up as float 0 | ||
for c in ["onset", "value", "trial_type", "duration"]: | ||
dext[c] = dext[c].astype(str) | ||
|
||
dext.loc[0, "onset"] = "n/a" | ||
dext.loc[1, "duration"] = "n/a" | ||
dext.loc[4, "trial_type"] = "n/a" | ||
dext.loc[4, "value"] = ( | ||
"n/a" # to check that filtering is also applied when we drop the `trial_type` | ||
) | ||
dext.to_csv(tmp_tsv_file, sep="\t", index=False) | ||
|
||
ev_kwargs_filtered = events_file_to_annotation_kwargs(events_fname=tmp_tsv_file) | ||
|
||
dext_f = dext[ | ||
(dext["onset"] != "n/a") | ||
& (dext["trial_type"] != "n/a") | ||
& (dext["value"] != "n/a") | ||
] | ||
|
||
assert (ev_kwargs_filtered["onset"] == dext_f["onset"].astype(float).values).all() | ||
assert ( | ||
ev_kwargs_filtered["duration"] | ||
== dext_f["duration"].replace("n/a", "0.0").astype(float).values | ||
).all() | ||
assert (ev_kwargs_filtered["description"] == dext_f["trial_type"].values).all() | ||
assert ( | ||
ev_kwargs_filtered["duration"][0] == 0.0 | ||
) # now idx=0, as first row is filtered out | ||
Comment on lines
+1521
to
+1529
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. idem. Could also consider There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. As we are using a dict for the return type, I would just stick to comparing the iterables here. Implemented all other suggestions though. |
||
|
||
# ---------------- default if missing trial_type ------------------------ | ||
tmp_tsv_file = tmp_path / "events.tsv" | ||
matthiasdold marked this conversation as resolved.
Show resolved
Hide resolved
|
||
dext.drop(columns="trial_type").to_csv(tmp_tsv_file, sep="\t", index=False) | ||
|
||
ev_kwargs_default = events_file_to_annotation_kwargs(events_fname=tmp_tsv_file) | ||
assert (ev_kwargs_default["onset"] == dext_f["onset"].astype(float).values).all() | ||
assert ( | ||
ev_kwargs_default["duration"] | ||
== dext_f["duration"].replace("n/a", "0.0").astype(float).values | ||
).all() | ||
assert ( | ||
np.sort(np.unique(ev_kwargs_default["description"])) | ||
== np.sort(dext_f["value"].unique()) | ||
).all() | ||
matthiasdold marked this conversation as resolved.
Show resolved
Hide resolved
|
Uh oh!
There was an error while loading. Please reload this page.