Skip to content

Commit

Permalink
Support arbitrary markers
Browse files Browse the repository at this point in the history
  • Loading branch information
@cbrnr
cbrnr committed Feb 19, 2024
1 parent 79dceaa commit 3f82816
Show file tree
Hide file tree
Showing 3 changed files with 72 additions and 111 deletions.
12 changes: 0 additions & 12 deletions pybv/_export.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -117,18 +117,6 @@ def _mne_annots2pybv_events(raw):
# defaults to type="Comment" and the full description
etype = "Comment"
description = annot["description"]
for start in ["Stimulus/S", "Response/R", "Comment/"]:
if description.startswith(start):
etype = start.split("/")[0]
description = description.replace(start, "")
break

if etype in ["Stimulus", "Response"] and description.strip().isdigit():
description = int(description.strip())
else:
# if cannot convert to int, we must use this as "Comment"
etype = "Comment"

event_dict = dict(
onset=onset, # in samples
duration=duration, # in samples
Expand Down
108 changes: 39 additions & 69 deletions pybv/io.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,15 +89,10 @@ def write_brainvision(
dimension of the `data` array.
- ``"duration"`` : int
The duration of the event in samples (defaults to ``1``).
- ``"description"`` : str | int
The description of the event. Must be a non-negative int when `type`
(see below) is either ``"Stimulus"`` or ``"Response"``, and may be a str
when `type` is ``"Comment"``.
- ``"description"`` : str
The description of the event.
- ``"type"`` : str
The type of the event, must be one of ``{"Stimulus", "Response",
"Comment"}`` (defaults to ``"Stimulus"``). Additional types like the
known BrainVision types ``"New Segment"``, ``"SyncStatus"``, etc. are
currently not supported.
The type of the event (defaults to ``"Stimulus"``).
- ``"channels"`` : str | list of {str | int}
The channels that are impacted by the event. Can be ``"all"``
(reflecting all channels), or a channel name, or a list of channel
Expand Down Expand Up @@ -151,11 +146,6 @@ def write_brainvision(
channels with non-voltage units such as °C as is (without scaling). For maximum
compatibility, all signals should be written as µV.
When passing a list of dict to `events`, the event ``type`` that can be passed is
currently limited to one of ``{"Stimulus", "Response", "Comment"}``. The BrainVision
specification itself does not limit event types, and future extensions of ``pybv``
may permit additional or even arbitrary event types.
References
----------
.. [1] https://www.brainproducts.com/support-resources/brainvision-core-data-format-1-0/
Expand Down Expand Up @@ -354,12 +344,12 @@ def _chk_events(events, ch_names, n_times):
``None``, it will be an empty list. If `events` is a list of dict, it will add
missing keys to each dict with default values, and it will, for each ith event, turn
``events[i]["channels"]`` into a list of 1-based channel name indices, where ``0``
equals ``"all"``. Event descriptions for ``"Stimulus"`` and ``"Response"`` will be
reformatted to a str of the format ``"S{:>n}"`` (or with a leading ``"R"`` for
``"Response"``), where ``n`` is determined by the description with the most digits
(minimum 3). For each ith event, the onset (``events[i]["onset"]``) will be
incremented by 1 to comply with the 1-based indexing used in BrainVision marker
files (*.vmrk*).
equals ``"all"``. Only if `events` is passed as an np.ndarray, event descriptions
will be reformatted to a str of the format ``"S{:>n}"``, where ``n`` is determined
by the description with the most digits (minimum 3). In addition, event types will
be set to ``"Stimulus"``. For each ith event, the onset (``events[i]["onset"]``)
will be incremented by 1 to comply with the 1-based indexing used in BrainVision
marker files (*.vmrk*).
Parameters
----------
Expand All @@ -376,12 +366,13 @@ def _chk_events(events, ch_names, n_times):
The preprocessed events, always provided as list of dict.
"""
if not isinstance(events, (type(None), np.ndarray, list)):
raise ValueError("events must be an array, a list of dict, or None")

# validate input: None
if isinstance(events, type(None)):
events_out = []
if events is None:
return []

# validate input: ndarray, list of dict
if not isinstance(events, (np.ndarray, list)):
raise ValueError("events must be an array, a list of dict, or None")

# default events
# NOTE: using "ch_names" as default for channels translates directly into "all" but
Expand All @@ -406,6 +397,7 @@ def _chk_events(events, ch_names, n_times):
durations = np.ones(events.shape[0], dtype=int) * event_defaults["duration"]
if events.shape[1] == 3:
durations = events[:, -1]

events_out = []
for irow, row in enumerate(events[:, 0:2]):
events_out.append(
Expand All @@ -418,6 +410,24 @@ def _chk_events(events, ch_names, n_times):
)
)

# NOTE: We format 1 -> "S 1", 10 -> "S 10", 100 -> "S100", etc.,
# https://github.com/bids-standard/pybv/issues/24#issuecomment-512746677
max_event_descr = max(
[1]
+ [
ev.get("description", "n/a")
for ev in events_out
if isinstance(ev.get("description", "n/a"), int)
]
)
twidth = max(3, int(np.ceil(np.log10(max_event_descr))))

for event in events_out:
if event["description"] < 0:
raise ValueError(f"events: descriptions must be non-negative ints.")
tformat = event["type"][0] + "{:>" + str(twidth) + "}"
event["description"] = tformat.format(event["description"])

# validate input: list of dict
if isinstance(events, list):
# we must not edit the original parameter
Expand All @@ -432,18 +442,6 @@ def _chk_events(events, ch_names, n_times):
"in list"
)

# NOTE: We format 1 -> "S 1", 10 -> "S 10", 100 -> "S100", etc.,
# https://github.com/bids-standard/pybv/issues/24#issuecomment-512746677
max_event_descr = max(
[1]
+ [
ev.get("description", "n/a")
for ev in events_out
if isinstance(ev.get("description", "n/a"), int)
]
)
twidth = max(3, int(np.ceil(np.log10(max_event_descr))))

# do full validation
for event in events_out:
# required keys
Expand All @@ -456,7 +454,7 @@ def _chk_events(events, ch_names, n_times):

# populate keys with default if missing (in-place)
for optional_key, default in event_defaults.items():
event[optional_key] = event.get(optional_key, default)
event.setdefault(optional_key, default)

# validate key types
# `onset`, `duration`
Expand Down Expand Up @@ -484,36 +482,8 @@ def _chk_events(events, ch_names, n_times):

event["onset"] = event["onset"] + 1 # VMRK uses 1-based indexing

# `type`
event_types = ["Stimulus", "Response", "Comment"]
if event["type"] not in event_types:
raise ValueError(f"events: `type` must be one of {event_types}")

# `description`
if event["type"] in ["Stimulus", "Response"]:
if not isinstance(event["description"], int):
raise ValueError(
f"events: when `type` is {event['type']}, `description` must be "
"non-negative int"
)

if event["description"] < 0:
raise ValueError(
f"events: when `type` is {event['type']}, descriptions must be "
"non-negative ints."
)

tformat = event["type"][0] + "{:>" + str(twidth) + "}"
event["description"] = tformat.format(event["description"])

else:
assert event["type"] == "Comment"
if not isinstance(event["description"], (int, str)):
raise ValueError(
f"events: when `type` is {event['type']}, `description` must be str"
" or int"
)
event["description"] = str(event["description"])
if not isinstance(event["description"], str):
raise ValueError("events: `description` must be str")

# `channels`
# "all" becomes ch_names (list of all channel names), single str 'ch_name'
Expand Down Expand Up @@ -632,8 +602,8 @@ def _write_vmrk_file(vmrk_fname, eeg_fname, events, meas_date):
# https://github.com/bids-standard/pybv/pull/77
for ch in ev["channels"]:
print(
f"Mk{iev}={ev['type']},{ev['description']},"
f"{ev['onset']},{ev['duration']},{ch}",
f"Mk{iev}={ev['type']},{ev['description']},{ev['onset']},"
f"{ev['duration']},{ch}",
file=fout,
)
iev += 1
Expand Down
63 changes: 33 additions & 30 deletions pybv/tests/test_bv_writer.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
{
"onset": 1,
"duration": 10,
"description": 1,
"description": "1",
"type": "Stimulus",
"channels": "all",
},
Expand All @@ -48,13 +48,13 @@
},
{
"onset": 1000,
"description": 2,
"description": "2",
"type": "Response",
"channels": ["ch_1", "ch_2"],
},
{
"onset": 200,
"description": 1234,
"description": "1234",
"channels": [],
},
]
Expand Down Expand Up @@ -134,40 +134,26 @@ def test_bv_writer_events_array(tmpdir, events_errormsg):
[{"onset": 100, "description": 1, "duration": 4901}],
"events: at least one event has a duration that exceeds",
),
([{"onset": 1, "description": {}}], "`description` must be str"),
([{"onset": 1, "description": 1}], "`description` must be str"),
(
[{"onset": 1, "description": 2, "type": "bogus"}],
"`type` must be one of",
),
(
[{"onset": 1, "description": "bogus"}],
"when `type` is Stimulus, `description` must be non-negative int",
),
(
[{"onset": 1, "description": {}, "type": "Comment"}],
"when `type` is Comment, `description` must be str or int",
),
(
[{"onset": 1, "description": -1}],
"when `type` is Stimulus, descriptions must be non-negative ints.",
),
(
[{"onset": 1, "description": 1, "channels": "bogus"}],
[{"onset": 1, "description": "1", "channels": "bogus"}],
"found channel .* bogus",
),
(
[{"onset": 1, "description": 1, "channels": ["ch_1", "ch_1"]}],
[{"onset": 1, "description": "1", "channels": ["ch_1", "ch_1"]}],
"events: found duplicate channel names",
),
(
[{"onset": 1, "description": 1, "channels": ["ch_1", "ch_2"]}],
[{"onset": 1, "description": "1", "channels": ["ch_1", "ch_2"]}],
"warn___feature may not be supported",
),
(
[{"onset": 1, "description": 1, "channels": 1}],
[{"onset": 1, "description": "1", "channels": 1}],
"events: `channels` must be str or list of str",
),
(
[{"onset": 1, "description": 1, "channels": [{}]}],
[{"onset": 1, "description": "1", "channels": [{}]}],
"be list of str or list of int corresponding to ch_names",
),
([], ""),
Expand Down Expand Up @@ -757,8 +743,7 @@ def test_event_writing(tmpdir):
data=data, sfreq=sfreq, ch_names=ch_names, fname_base=fname, folder_out=tmpdir
)

with pytest.warns(UserWarning, match="Such events will be written to .vmrk"):
write_brainvision(**kwargs, events=events)
write_brainvision(**kwargs, events=events)

vhdr_fname = tmpdir / fname + ".vhdr"
raw = mne.io.read_raw_brainvision(vhdr_fname=vhdr_fname, preload=True)
Expand All @@ -782,14 +767,32 @@ def test_event_writing(tmpdir):

descr = [
"Comment/Some string :-)",
"Stimulus/S 1",
"Stimulus/S1234",
"Response/R 2",
"Response/R 2",
"Stimulus/1",
"Stimulus/1234",
"Response/2",
"Response/2",
]
np.testing.assert_array_equal(raw.annotations.description, descr)

# smoke test forming events from annotations
_events, _event_id = mne.events_from_annotations(raw)
for _d in descr:
assert _d in _event_id


def test_event_array_writing(tmpdir):
"""Test writing events as an array."""
kwargs = dict(
data=data, sfreq=sfreq, ch_names=ch_names, fname_base=fname, folder_out=tmpdir
)

write_brainvision(**kwargs, events=events_array)

vhdr_fname = tmpdir / fname + ".vhdr"
raw = mne.io.read_raw_brainvision(vhdr_fname=vhdr_fname, preload=True)

descr = ["Stimulus/S 1", "Stimulus/S 1", "Stimulus/S 2", "Stimulus/S 2"]

np.testing.assert_array_equal(raw.annotations.onset, events_array[:, 0] / sfreq)
np.testing.assert_array_equal(raw.annotations.duration, np.full(4, 1 / sfreq))
np.testing.assert_array_equal(raw.annotations.description, descr)

0 comments on commit 3f82816

Please sign in to comment.