# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.
import sys
from collections import OrderedDict
from datetime import datetime, timedelta, timezone
from itertools import repeat
from pathlib import Path
import numpy as np
import pytest
from numpy.testing import (
assert_allclose,
assert_array_almost_equal,
assert_array_equal,
assert_equal,
)
from pytest import approx
import mne
from mne import (
Annotations,
Epochs,
annotations_from_events,
count_annotations,
create_info,
events_from_annotations,
read_annotations,
)
from mne.annotations import (
_handle_meas_date,
_read_annotations_txt_parse_header,
_sync_onset,
)
from mne.datasets import testing
from mne.io import RawArray, concatenate_raws, read_raw_fif
from mne.utils import (
_dt_to_stamp,
_raw_annot,
_record_warnings,
_stamp_to_dt,
assert_and_remove_boundary_annot,
catch_logging,
check_version,
)
data_path = testing.data_path(download=False)
data_dir = data_path / "MEG" / "sample"
fif_fname = Path(__file__).parents[1] / "io" / "tests" / "data" / "test_raw.fif"
first_samps = pytest.mark.parametrize("first_samp", (0, 10000))
edf_reduced = data_path / "EDF" / "test_reduced.edf"
edf_annot_only = data_path / "EDF" / "SC4001EC-Hypnogram.edf"
needs_pandas = pytest.mark.skipif(not check_version("pandas"), reason="Needs pandas")
# On Windows, datetime.fromtimestamp throws an error for negative times.
# We mimic this behavior on non-Windows platforms for ease of testing.
class _windows_datetime(datetime):
@classmethod
def fromtimestamp(cls, timestamp, tzinfo=None):
if timestamp < 0:
raise OSError("[Errno 22] Invalid argument")
return datetime.fromtimestamp(timestamp, tzinfo)
@pytest.fixture(scope="function")
def windows_like_datetime(monkeypatch):
"""Ensure datetime.fromtimestamp is Windows-like."""
if not sys.platform.startswith("win"):
monkeypatch.setattr("mne.annotations.datetime", _windows_datetime)
yield
def test_basics():
"""Test annotation class."""
raw = read_raw_fif(fif_fname)
assert raw.annotations is not None
assert len(raw.annotations.onset) == 0
pytest.raises(OSError, read_annotations, fif_fname)
onset = np.array(range(10))
duration = np.ones(10)
description = np.repeat("test", 10)
dt = raw.info["meas_date"]
assert isinstance(dt, datetime)
stamp = _dt_to_stamp(dt)
# Test time shifts.
for orig_time in [None, dt, stamp[0], stamp]:
annot = Annotations(onset, duration, description, orig_time)
if orig_time is None:
assert annot.orig_time is None
else:
assert isinstance(annot.orig_time, datetime)
assert annot.orig_time.tzinfo is timezone.utc
pytest.raises(ValueError, Annotations, onset, duration, description[:9])
pytest.raises(ValueError, Annotations, [onset, 1], duration, description)
pytest.raises(ValueError, Annotations, onset, [duration, 1], description)
# Test combining annotations with concatenate_raws
raw2 = raw.copy()
delta = raw.times[-1] + 1.0 / raw.info["sfreq"]
orig_time = stamp[0] + stamp[1] * 1e-6 + raw2._first_time
offset = _dt_to_stamp(_handle_meas_date(raw2.info["meas_date"]))
offset = offset[0] + offset[1] * 1e-6
offset = orig_time - offset
assert_allclose(offset, raw._first_time)
annot = Annotations(onset, duration, description, orig_time)
assert annot.orig_time is not None
assert " segments" in repr(annot)
raw2.set_annotations(annot)
assert_allclose(raw2.annotations.onset, onset + offset)
assert raw2.annotations is not annot
assert raw2.annotations.orig_time is not None
concatenate_raws([raw, raw2])
assert_and_remove_boundary_annot(raw)
assert_allclose(onset + offset + delta, raw.annotations.onset, rtol=1e-5)
assert_array_equal(annot.duration, raw.annotations.duration)
assert_array_equal(raw.annotations.description, np.repeat("test", 10))
def test_annot_sanitizing(tmp_path):
"""Test description sanitizing."""
annot = Annotations([0], [1], ["a;:b"])
fname = tmp_path / "custom-annot.fif"
annot.save(fname)
annot_read = read_annotations(fname)
_assert_annotations_equal(annot, annot_read)
# make sure pytest raises error on char-sequence that is not allowed
with pytest.raises(ValueError, match="in description not supported"):
Annotations([0], [1], ["a{COLON}b"])
def test_raw_array_orig_times():
"""Test combining with RawArray and orig_times."""
data = np.random.randn(2, 1000) * 10e-12
sfreq = 100.0
info = create_info(ch_names=["MEG1", "MEG2"], ch_types=["grad"] * 2, sfreq=sfreq)
meas_date = _handle_meas_date(np.pi)
with info._unlock():
info["meas_date"] = meas_date
raws = []
for first_samp in [12300, 100, 12]:
raw = RawArray(data.copy(), info, first_samp=first_samp)
ants = Annotations([1.0, 2.0], [0.5, 0.5], "x", np.pi + first_samp / sfreq)
raw.set_annotations(ants)
raws.append(raw)
assert_allclose(raws[0].annotations.onset, [124, 125])
raw = RawArray(data.copy(), info)
assert not len(raw.annotations)
raw.set_annotations(Annotations([1.0], [0.5], "x", None))
assert_allclose(raw.annotations.onset, [1.0])
raws.append(raw)
raw = concatenate_raws(raws, verbose="debug")
assert raw.info["meas_date"] == raw.annotations.orig_time == meas_date
assert_and_remove_boundary_annot(raw, 3)
assert_array_equal(
raw.annotations.onset, [124.0, 125.0, 134.0, 135.0, 144.0, 145.0, 154.0]
)
raw.annotations.delete(2)
assert_array_equal(
raw.annotations.onset, [124.0, 125.0, 135.0, 144.0, 145.0, 154.0]
)
raw.annotations.append(5, 1.5, "y")
assert_array_equal(
raw.annotations.onset, [5.0, 124.0, 125.0, 135.0, 144.0, 145.0, 154.0]
)
assert_array_equal(raw.annotations.duration, [1.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5])
assert_array_equal(raw.annotations.description, ["y", "x", "x", "x", "x", "x", "x"])
# These three things should be equivalent
stamp = _dt_to_stamp(raw.info["meas_date"])
orig_time = _handle_meas_date(stamp)
for empty_annot in (
Annotations([], [], [], stamp),
Annotations([], [], [], orig_time),
Annotations([], [], [], None),
None,
):
raw.set_annotations(empty_annot)
assert isinstance(raw.annotations, Annotations)
assert len(raw.annotations) == 0
assert raw.annotations.orig_time == orig_time
def test_crop(tmp_path):
"""Test cropping with annotations."""
raw = read_raw_fif(fif_fname)
events = mne.find_events(raw)
onset = events[events[:, 2] == 1, 0] / raw.info["sfreq"]
duration = np.full_like(onset, 0.5)
description = [f"bad {k}" for k in range(len(onset))]
annot = mne.Annotations(
onset, duration, description, orig_time=raw.info["meas_date"]
)
raw.set_annotations(annot)
split_time = raw.times[-1] / 2.0 + 2.0
split_idx = len(onset) // 2 + 1
raw_cropped_left = raw.copy().crop(0.0, split_time - 1.0 / raw.info["sfreq"])
assert_array_equal(
raw_cropped_left.annotations.description,
raw.annotations.description[:split_idx],
)
assert_allclose(
raw_cropped_left.annotations.duration, raw.annotations.duration[:split_idx]
)
assert_allclose(
raw_cropped_left.annotations.onset, raw.annotations.onset[:split_idx]
)
raw_cropped_right = raw.copy().crop(split_time, None)
assert_array_equal(
raw_cropped_right.annotations.description,
raw.annotations.description[split_idx:],
)
assert_allclose(
raw_cropped_right.annotations.duration, raw.annotations.duration[split_idx:]
)
assert_allclose(
raw_cropped_right.annotations.onset, raw.annotations.onset[split_idx:]
)
raw_concat = mne.concatenate_raws(
[raw_cropped_left, raw_cropped_right], verbose="debug"
)
assert_allclose(raw_concat.times, raw.times)
assert_allclose(raw_concat[:][0], raw[:][0], atol=1e-20)
assert_and_remove_boundary_annot(raw_concat)
# Ensure we annotations survive round-trip crop->concat
assert_array_equal(raw_concat.annotations.description, raw.annotations.description)
for attr in ("onset", "duration"):
assert_allclose(
getattr(raw_concat.annotations, attr),
getattr(raw.annotations, attr),
err_msg=f"Failed for {attr}:",
)
raw.set_annotations(None) # undo
# Test concatenating annotations with and without orig_time.
raw2 = raw.copy()
raw.set_annotations(Annotations([45.0], [3], "test", raw.info["meas_date"]))
raw2.set_annotations(Annotations([2.0], [3], "BAD", None))
expected_onset = [45.0, 2.0 + raw._last_time]
raw = concatenate_raws([raw, raw2])
assert_and_remove_boundary_annot(raw)
assert_array_almost_equal(raw.annotations.onset, expected_onset, decimal=2)
# Test IO
fname = tmp_path / "test-annot.fif"
raw.annotations.save(fname)
annot_read = read_annotations(fname)
for attr in ("onset", "duration"):
assert_allclose(getattr(annot_read, attr), getattr(raw.annotations, attr))
assert annot_read.orig_time == raw.annotations.orig_time
assert_array_equal(annot_read.description, raw.annotations.description)
annot = Annotations((), (), ())
annot.save(fname, overwrite=True)
pytest.raises(OSError, read_annotations, fif_fname) # none in old raw
annot = read_annotations(fname)
assert isinstance(annot, Annotations)
assert len(annot) == 0
annot.crop() # test if cropping empty annotations doesn't raise an error
# Test that empty annotations can be saved with an object
fname = tmp_path / "test_raw.fif"
raw.set_annotations(annot)
raw.save(fname)
raw_read = read_raw_fif(fname)
assert isinstance(raw_read.annotations, Annotations)
assert len(raw_read.annotations) == 0
raw.set_annotations(None)
raw.save(fname, overwrite=True)
raw_read = read_raw_fif(fname)
assert raw_read.annotations is not None
assert len(raw_read.annotations.onset) == 0
# test saving and reloading cropped annotations in raw instance
info = create_info(
[f"EEG{i + 1}" for i in range(3)], ch_types=["eeg"] * 3, sfreq=50
)
raw = RawArray(np.zeros((3, 50 * 20)), info)
annotation = mne.Annotations([8, 12, 15], [2] * 3, [1, 2, 3])
raw = raw.set_annotations(annotation)
raw_copied = raw.copy().crop(5, 18)
fname = tmp_path / "test_raw.fif"
raw_copied.save(fname, overwrite=True)
raw_loaded = mne.io.read_raw(str(fname))
for attr in ("onset", "duration"):
assert_allclose(
getattr(raw.annotations, attr), getattr(raw_copied.annotations, attr)
)
assert_allclose(
getattr(raw_copied.annotations, attr), getattr(raw_loaded.annotations, attr)
)
@first_samps
def test_chunk_duration(first_samp):
"""Test chunk_duration."""
# create dummy raw
raw = RawArray(
data=np.empty([10, 10], dtype=np.float64),
info=create_info(ch_names=10, sfreq=1.0),
first_samp=first_samp,
)
with raw.info._unlock():
raw.info["meas_date"] = _handle_meas_date(0)
raw.set_annotations(
Annotations(description="foo", onset=[0], duration=[10], orig_time=None)
)
assert raw.annotations.orig_time == raw.info["meas_date"]
assert_allclose(raw.annotations.onset, [first_samp])
# expected_events = [[0, 0, 1], [0, 0, 1], [1, 0, 1], [1, 0, 1], ..
# [9, 0, 1], [9, 0, 1]]
expected_events = np.atleast_2d(np.repeat(range(10), repeats=2)).T
expected_events = np.insert(expected_events, 1, 0, axis=1)
expected_events = np.insert(expected_events, 2, 1, axis=1)
expected_events[:, 0] += first_samp
events, events_id = events_from_annotations(
raw, chunk_duration=0.5, use_rounding=False
)
assert_array_equal(events, expected_events)
# test chunk durations that do not fit equally in annotation duration
expected_events = np.zeros((3, 3))
expected_events[:, -1] = 1
expected_events[:, 0] = np.arange(0, 9, step=3) + first_samp
events, events_id = events_from_annotations(raw, chunk_duration=3.0)
assert_array_equal(events, expected_events)
def test_events_from_annotation_orig_time_none():
"""Tests events_from_annotation with orig_time None and first_sampe > 0."""
# Create fake data
sfreq, duration_s = 100, 10
data = np.random.RandomState(42).randn(1, sfreq * duration_s)
info = mne.create_info(ch_names=["EEG1"], ch_types=["eeg"], sfreq=sfreq)
raw = mne.io.RawArray(data, info)
# Add annotation toward the end
onset = [8]
duration = [1]
description = ["0"]
annots = mne.Annotations(onset, duration, description)
raw = raw.set_annotations(annots)
# Crop start of raw
raw.crop(tmin=7)
# Extract epochs
events, event_ids = mne.events_from_annotations(raw)
epochs = mne.Epochs(
raw, events, tmin=0, tmax=1, baseline=None, on_missing="warning"
)
# epochs is empty
assert_array_equal(epochs.get_data()[0], data[:, 800:901])
def test_crop_more():
"""Test more cropping."""
raw = mne.io.read_raw_fif(fif_fname).crop(0, 11).load_data()
raw._data[:] = np.random.RandomState(0).randn(*raw._data.shape)
onset = np.array([0.47058824, 2.49773765, 6.67873287, 9.15837097])
duration = np.array([0.89592767, 1.13574672, 1.09954739, 0.48868752])
annotations = mne.Annotations(onset, duration, "BAD")
raw.set_annotations(annotations)
assert len(raw.annotations) == 4
delta = 1.0 / raw.info["sfreq"]
offset = raw.first_samp * delta
raw_concat = mne.concatenate_raws(
[
raw.copy().crop(0, 4 - delta),
raw.copy().crop(4, 8 - delta),
raw.copy().crop(8, None),
]
)
assert_allclose(raw_concat.times, raw.times)
assert_allclose(raw_concat[:][0], raw[:][0])
assert raw_concat.first_samp == raw.first_samp
assert_and_remove_boundary_annot(raw_concat, 2)
assert len(raw_concat.annotations) == 4
assert_array_equal(raw_concat.annotations.description, raw.annotations.description)
assert_allclose(raw.annotations.duration, duration)
assert_allclose(raw_concat.annotations.duration, duration)
assert_allclose(raw.annotations.onset, onset + offset)
assert_allclose(
raw_concat.annotations.onset, onset + offset, atol=1.0 / raw.info["sfreq"]
)
@testing.requires_testing_data
def test_read_brainstorm_annotations():
"""Test reading for Brainstorm events file."""
fname = data_dir / "events_sample_audvis_raw_bst.mat"
annot = read_annotations(fname)
assert len(annot) == 238
assert annot.onset.min() > 40 # takes into account first_samp
assert np.unique(annot.description).size == 5
@testing.requires_testing_data
@pytest.mark.parametrize(
"fname, n_annot",
[
(edf_annot_only, 154),
(edf_reduced, 5),
],
)
def test_read_edf_annotations(fname, n_annot):
"""Test reading EDF annotations."""
annot = read_annotations(fname)
assert len(annot) == n_annot
@first_samps
def test_raw_reject(first_samp):
"""Test raw data getter with annotation reject."""
sfreq = 100.0
info = create_info(["a", "b", "c", "d", "e"], sfreq, ch_types="eeg")
raw = RawArray(np.ones((5, 15000)), info, first_samp=first_samp)
with pytest.warns(RuntimeWarning, match="outside the data range"):
raw.set_annotations(Annotations([2, 100, 105, 148], [2, 8, 5, 8], "BAD"))
data, times = raw.get_data(
[0, 1, 3, 4],
100,
11200,
"omit",
return_times=True, # 1-112 s
)
bad_times = np.concatenate(
[np.arange(200, 400), np.arange(10000, 10800), np.arange(10500, 11000)]
)
expected_times = np.setdiff1d(np.arange(100, 11200), bad_times) / sfreq
assert_allclose(times, expected_times)
# with orig_time and complete overlap
raw = read_raw_fif(fif_fname)
raw.set_annotations(
Annotations(
onset=np.array([1, 4, 5], float) + raw._first_time,
duration=[1, 3, 1],
description="BAD",
orig_time=raw.info["meas_date"],
)
)
t_stop = 18.0
assert raw.times[-1] > t_stop
n_stop = int(round(t_stop * raw.info["sfreq"]))
n_drop = int(round(4 * raw.info["sfreq"]))
assert len(raw.times) >= n_stop
data, times = raw.get_data(range(10), 0, n_stop, "omit", True)
assert data.shape == (10, n_stop - n_drop)
assert times[-1] == raw.times[n_stop - 1]
assert_array_equal(data[:, -100:], raw[:10, n_stop - 100 : n_stop][0])
data, times = raw.get_data(range(10), 0, n_stop, "NaN", True)
assert_array_equal(data.shape, (10, n_stop))
assert times[-1] == raw.times[n_stop - 1]
t_1, t_2 = raw.time_as_index([1, 2], use_rounding=True)
assert np.isnan(data[:, t_1:t_2]).all() # 1s -2s
assert not np.isnan(data[:, :t_1].any())
assert not np.isnan(data[:, t_2:].any())
assert_array_equal(data[:, -100:], raw[:10, n_stop - 100 : n_stop][0])
assert_array_equal(raw.get_data(), raw[:][0])
# Test _sync_onset
times = np.array([10, -88, 190], float)
onsets = _sync_onset(raw, times)
assert_array_almost_equal(onsets, times - raw.first_samp / raw.info["sfreq"])
assert_array_almost_equal(times, _sync_onset(raw, onsets, True))
@first_samps
def test_annotation_filtering(first_samp):
"""Test that annotations work properly with filtering."""
# Create data with just a DC component
data = np.ones((1, 1000))
info = create_info(1, 1000.0, "eeg")
raws = [RawArray(data * (ii + 1), info, first_samp=first_samp) for ii in range(4)]
kwargs_pass = dict(l_freq=None, h_freq=50.0, fir_design="firwin")
kwargs_stop = dict(l_freq=50.0, h_freq=None, fir_design="firwin")
# lowpass filter, which should not modify the data
raws_pass = [raw.copy().filter(**kwargs_pass) for raw in raws]
# highpass filter, which should zero it out
raws_stop = [raw.copy().filter(**kwargs_stop) for raw in raws]
# concat the original and the filtered segments
raws_concat = concatenate_raws([raw.copy() for raw in raws])
raws_zero = raws_concat.copy().apply_function(lambda x: x * 0)
raws_pass_concat = concatenate_raws(raws_pass)
raws_stop_concat = concatenate_raws(raws_stop)
# make sure we did something reasonable with our individual-file filtering
assert_allclose(raws_concat[0][0], raws_pass_concat[0][0], atol=1e-14)
assert_allclose(raws_zero[0][0], raws_stop_concat[0][0], atol=1e-14)
# ensure that our Annotations cut up the filtering properly
raws_concat_pass = raws_concat.copy().filter(
skip_by_annotation="edge", **kwargs_pass
)
assert_allclose(raws_concat[0][0], raws_concat_pass[0][0], atol=1e-14)
raws_concat_stop = raws_concat.copy().filter(
skip_by_annotation="edge", **kwargs_stop
)
assert_allclose(raws_zero[0][0], raws_concat_stop[0][0], atol=1e-14)
# test notch_filtering
raw_notch = concatenate_raws([raws_concat.copy(), raws_concat.copy()])
raw_notch.annotations.append(3.0 + raw_notch._first_time, 0.2, "foo_notch")
raw_notch.annotations.append(5.0 + raw_notch._first_time, 0.2, "foo_notch")
n_times = raw_notch._data.shape[1]
with catch_logging() as log:
raw_notch.notch_filter(
60.0,
fir_design="firwin",
trans_bandwidth=5.0,
skip_by_annotation="foo_notch",
verbose="info",
)
log = log.getvalue()
assert "3 contiguous segment" in log
# check that data has same shape before/after filtering
assert n_times == raw_notch._data.shape[1]
# one last test: let's cut out a section entirely:
# here the 1-3 second window should be skipped
raw = raws_concat.copy()
raw.annotations.append(1.0 + raw._first_time, 2.0, "foo")
with catch_logging() as log:
raw.filter(
l_freq=50.0,
h_freq=None,
fir_design="firwin",
skip_by_annotation="foo",
verbose="info",
)
log = log.getvalue()
assert "2 contiguous segments" in log
raw.annotations.append(2.0 + raw._first_time, 1.0, "foo") # shouldn't change
with catch_logging() as log:
raw.filter(
l_freq=50.0,
h_freq=None,
fir_design="firwin",
skip_by_annotation="foo",
verbose="info",
)
log = log.getvalue()
assert "2 contiguous segments" in log
# our filter will zero out anything not skipped:
mask = np.concatenate((np.zeros(1000), np.ones(2000), np.zeros(1000)))
expected_data = raws_concat[0][0][0] * mask
assert_allclose(raw[0][0][0], expected_data, atol=1e-14)
# Let's try another one
raw = raws[0].copy()
raw.set_annotations(Annotations([0.0], [0.5], ["BAD_ACQ_SKIP"]))
my_data, times = raw.get_data(reject_by_annotation="omit", return_times=True)
assert_allclose(times, raw.times[500:])
assert my_data.shape == (1, 500)
raw_filt = raw.copy().filter(skip_by_annotation="bad_acq_skip", **kwargs_stop)
expected = data.copy()
expected[:, 500:] = 0
assert_allclose(raw_filt[:][0], expected, atol=1e-14)
raw = raws[0].copy()
raw.set_annotations(Annotations([0.5], [0.5], ["BAD_ACQ_SKIP"]))
my_data, times = raw.get_data(reject_by_annotation="omit", return_times=True)
assert_allclose(times, raw.times[:500])
assert my_data.shape == (1, 500)
raw_filt = raw.copy().filter(skip_by_annotation="bad_acq_skip", **kwargs_stop)
expected = data.copy()
expected[:, :500] = 0
assert_allclose(raw_filt[:][0], expected, atol=1e-14)
@first_samps
def test_annotation_omit(first_samp):
"""Test raw.get_data with annotations."""
data = np.concatenate([np.ones((1, 1000)), 2 * np.ones((1, 1000))], -1)
info = create_info(1, 1000.0, "eeg")
raw = RawArray(data, info, first_samp=first_samp)
raw.set_annotations(Annotations([0.5], [1], ["bad"]))
expected = raw[0][0]
assert_allclose(raw.get_data(reject_by_annotation=None), expected)
# nan
expected[0, 500:1500] = np.nan
assert_allclose(raw.get_data(reject_by_annotation="nan"), expected)
got = np.concatenate(
[
raw.get_data(start=start, stop=stop, reject_by_annotation="nan")
for start, stop in ((0, 1000), (1000, 2000))
],
-1,
)
assert_allclose(got, expected)
# omit
expected = expected[:, np.isfinite(expected[0])]
assert_allclose(raw.get_data(reject_by_annotation="omit"), expected)
got = np.concatenate(
[
raw.get_data(start=start, stop=stop, reject_by_annotation="omit")
for start, stop in ((0, 1000), (1000, 2000))
],
-1,
)
assert_allclose(got, expected)
pytest.raises(ValueError, raw.get_data, reject_by_annotation="foo")
def test_annotation_epoching():
"""Test that annotations work properly with concatenated edges."""
# Create data with just a DC component
data = np.ones((1, 1000))
info = create_info(1, 1000.0, "eeg")
raw = concatenate_raws([RawArray(data, info) for ii in range(3)])
assert raw.annotations is not None
assert len(raw.annotations) == 4
assert np.isin(raw.annotations.description, ["BAD boundary"]).sum() == 2
assert np.isin(raw.annotations.description, ["EDGE boundary"]).sum() == 2
assert_array_equal(raw.annotations.duration, 0.0)
events = np.array([[a, 0, 1] for a in [0, 500, 1000, 1500, 2000]])
epochs = Epochs(
raw, events, tmin=0, tmax=0.999, baseline=None, preload=True
) # 1000 samples long
assert_equal(len(epochs.drop_log), len(events))
assert_equal(len(epochs), 3)
assert_equal([0, 2, 4], epochs.selection)
def test_annotation_concat():
"""Test if two Annotations objects can be concatenated."""
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"], ch_names=[["1"], ["2"], []])
b = Annotations([11, 12, 13], [1, 2, 2], ["x", "y", "z"], ch_names=[[], ["3"], []])
# test + operator (does not modify a or b)
c = a + b
assert_array_equal(c.onset, [1, 2, 3, 11, 12, 13])
assert_array_equal(c.duration, [5, 5, 8, 1, 2, 2])
assert_array_equal(c.description, ["a", "b", "c", "x", "y", "z"])
assert_equal(len(a), 3)
assert_equal(len(b), 3)
assert_equal(len(c), 6)
# c should have updated channel names
want_names = np.array([("1",), ("2",), (), (), ("3",), ()], dtype="O")
assert_array_equal(c.ch_names, want_names)
# test += operator (modifies a in place)
a += b
assert_array_equal(a.onset, [1, 2, 3, 11, 12, 13])
assert_array_equal(a.duration, [5, 5, 8, 1, 2, 2])
assert_array_equal(a.description, ["a", "b", "c", "x", "y", "z"])
assert_equal(len(a), 6)
assert_equal(len(b), 3)
# test += operator (modifies a in place)
b._orig_time = _handle_meas_date(1038942070.7201)
with pytest.raises(ValueError, match="orig_time should be the same"):
a += b
def test_annotations_crop():
"""Test basic functionality of annotation crop."""
onset = np.arange(1, 10)
duration = np.full_like(onset, 10)
description = ["yy"] * onset.shape[0]
a = Annotations(
onset=onset, duration=duration, description=description, orig_time=0
)
# cropping window larger than annotations --> do not modify
a_ = a.copy().crop(tmin=-10, tmax=42)
assert_array_equal(a_.onset, a.onset)
assert_array_equal(a_.duration, a.duration)
# cropping with left shifted window
with _record_warnings() as w:
a_ = a.copy().crop(tmin=0, tmax=4.2)
assert_array_equal(a_.onset, [1.0, 2.0, 3.0, 4.0])
assert_allclose(a_.duration, [3.2, 2.2, 1.2, 0.2])
assert len(w) == 0
# cropping with right shifted window
with _record_warnings() as w:
a_ = a.copy().crop(tmin=17.8, tmax=22)
assert_array_equal(a_.onset, [17.8, 17.8])
assert_allclose(a_.duration, [0.2, 1.2])
assert len(w) == 0
# cropping with centered small window
a_ = a.copy().crop(tmin=11, tmax=12)
assert_array_equal(a_.onset, [11, 11, 11, 11, 11, 11, 11, 11, 11])
assert_array_equal(a_.duration, [0, 1, 1, 1, 1, 1, 1, 1, 1])
# cropping with out-of-bounds window
with _record_warnings() as w:
a_ = a.copy().crop(tmin=42, tmax=100)
assert_array_equal(a_.onset, [])
assert_array_equal(a_.duration, [])
assert len(w) == 0
# test error raising
with pytest.raises(ValueError, match="tmax should be greater than.*tmin"):
a.copy().crop(tmin=42, tmax=0)
# test warnings
with pytest.warns(RuntimeWarning, match="Omitted .* were outside"):
a.copy().crop(tmin=42, tmax=100, emit_warning=True)
with pytest.warns(RuntimeWarning, match="Limited .* expanding outside"):
a.copy().crop(tmin=0, tmax=12, emit_warning=True)
@testing.requires_testing_data
def test_events_from_annot_in_raw_objects():
"""Test basic functionality of events_fron_annot for raw objects."""
raw = read_raw_fif(fif_fname)
events = mne.find_events(raw)
event_id = {
"Auditory/Left": 1,
"Auditory/Right": 2,
"Visual/Left": 3,
"Visual/Right": 4,
"Visual/Smiley": 32,
"Motor/Button": 5,
}
event_map = {v: k for k, v in event_id.items()}
annot = Annotations(
onset=raw.times[events[:, 0] - raw.first_samp],
duration=np.zeros(len(events)),
description=[event_map[vv] for vv in events[:, 2]],
orig_time=None,
)
raw.set_annotations(annot)
events2, event_id2 = events_from_annotations(raw, event_id=event_id, regexp=None)
assert_array_equal(events, events2)
assert_equal(event_id, event_id2)
events3, event_id3 = events_from_annotations(raw, event_id=None, regexp=None)
assert_array_equal(events[:, 0], events3[:, 0])
assert set(event_id.keys()) == set(event_id3.keys())
# ensure that these actually got sorted properly
expected_event_id = {
desc: idx + 1 for idx, desc in enumerate(sorted(event_id.keys()))
}
assert event_id3 == expected_event_id
first = np.unique(events3[:, 2])
second = np.arange(1, len(event_id) + 1, 1).astype(first.dtype)
assert_array_equal(first, second)
first = np.unique(list(event_id3.values()))
second = np.arange(1, len(event_id) + 1, 1).astype(first.dtype)
assert_array_equal(first, second)
events4, event_id4 = events_from_annotations(raw, event_id=None, regexp=".*Left")
expected_event_id4 = {k: v for k, v in event_id.items() if "Left" in k}
assert_equal(event_id4.keys(), expected_event_id4.keys())
expected_events4 = events[(events[:, 2] == 1) | (events[:, 2] == 3)]
assert_array_equal(expected_events4[:, 0], events4[:, 0])
events5, event_id5 = events_from_annotations(
raw, event_id=event_id, regexp=".*Left"
)
expected_event_id5 = {k: v for k, v in event_id.items() if "Left" in k}
assert_equal(event_id5, expected_event_id5)
expected_events5 = events[(events[:, 2] == 1) | (events[:, 2] == 3)]
assert_array_equal(expected_events5, events5)
with pytest.raises(ValueError, match="not find any of the events"):
events_from_annotations(raw, regexp="not_there")
with pytest.raises(ValueError, match="Invalid type for event_id"):
events_from_annotations(raw, event_id="wrong")
# concat does not introduce BAD or EDGE
raw_concat = concatenate_raws([raw.copy(), raw.copy()])
_, event_id = events_from_annotations(raw_concat)
assert isinstance(event_id, dict)
assert len(event_id) > 0
for kind in ("BAD", "EDGE"):
assert f"{kind} boundary" in raw_concat.annotations.description
for key in event_id.keys():
assert kind not in key
# remove all events
raw.set_annotations(None)
events7, _ = events_from_annotations(raw)
assert_array_equal(events7, np.empty((0, 3), dtype=int))
def test_events_from_annot_onset_alingment():
"""Test events and annotations onset are the same."""
raw = _raw_annot(meas_date=1, orig_time=1.5)
# s 0 1 2 3
# raw . |--------xxxxxxxxx
# annot . |---xx
# raw.annot . |--------xx
# latency . 0 1 2
# . 0 0
assert raw.annotations.orig_time == _handle_meas_date(1)
assert raw.annotations.onset[0] == 1
assert raw.first_samp == 10
event_latencies, event_id = events_from_annotations(raw)
assert event_latencies[0, 0] == 10
assert raw.first_samp == event_latencies[0, 0]
@pytest.mark.parametrize(
"use_rounding,tol,shape,onsets,descriptions",
[
pytest.param(True, 0, (2, 3), [202, 402], [0, 2], id="rounding-notol"),
pytest.param(True, 1e-8, (3, 3), [202, 302, 402], [0, 1, 2], id="rounding-tol"),
pytest.param(False, 0, (2, 3), [202, 401], [0, 2], id="norounding-notol"),
pytest.param(
False, 1e-8, (3, 3), [202, 302, 401], [0, 1, 2], id="norounding-tol"
),
pytest.param(None, None, (3, 3), [202, 302, 402], [0, 1, 2], id="default"),
],
)
def test_events_from_annot_with_tolerance(
use_rounding, tol, shape, onsets, descriptions
):
"""Test events_from_annotations w/ and w/o tolerance."""
info = create_info(ch_names=1, sfreq=100)
raw = RawArray(data=np.empty((1, 1000)), info=info, first_samp=0)
meas_date = _handle_meas_date(0)
with raw.info._unlock(check_after=True):
raw.info["meas_date"] = meas_date
chunk_duration = 1
annot = Annotations([2.02, 3.02, 4.02], chunk_duration, ["0", "1", "2"], 0)
raw.set_annotations(annot)
event_id = {"0": 0, "1": 1, "2": 2}
if use_rounding is None:
events, _ = events_from_annotations(
raw, event_id=event_id, chunk_duration=chunk_duration
)
else:
events, _ = events_from_annotations(
raw,
event_id=event_id,
chunk_duration=chunk_duration,
use_rounding=use_rounding,
tol=tol,
)
assert events.shape == shape
assert (events[:, 0] == onsets).all()
assert (events[:, 2] == descriptions).all()
def _create_annotation_based_on_descr(
description, annotation_start_sampl=0, duration=0, orig_time=0
):
"""Create a raw object with annotations from descriptions.
The returning raw object contains as many annotations as description given.
All starting at `annotation_start_sampl`.
"""
# create dummy raw
raw = RawArray(
data=np.empty([10, 10], dtype=np.float64),
info=create_info(ch_names=10, sfreq=1000.0),
first_samp=0,
)
raw.set_meas_date(0)
# create dummy annotations based on the descriptions
onset = raw.times[annotation_start_sampl]
onset_matching_desc = np.full_like(description, onset, dtype=type(onset))
duration_matching_desc = np.full_like(description, duration, dtype=type(duration))
annot = Annotations(
description=description,
onset=onset_matching_desc,
duration=duration_matching_desc,
orig_time=orig_time,
)
if duration != 0:
with pytest.warns(RuntimeWarning, match="Limited.*expanding outside"):
# duration 0.1s is larger than the raw data expand
raw.set_annotations(annot)
else:
raw.set_annotations(annot)
# Make sure that set_annotations(annot) works
assert all(raw.annotations.onset == onset)
if duration != 0:
expected_duration = (len(raw.times) / raw.info["sfreq"]) - onset
else:
expected_duration = 0
_duration = raw.annotations.duration[0]
assert _duration == approx(expected_duration)
assert all(raw.annotations.duration == _duration)
assert all(raw.annotations.description == description)
return raw
def test_event_id_function_default():
"""Test[unit_test] for event_id_function default in event_from_annotations.
The expected behavior is give numeric label for all those annotations not
present in event_id, starting at 1.
"""
# No event_id given
description = ["a", "b", "c", "d", "e", "f", "g"]
expected_event_id = dict(zip(description, range(1, 100)))
expected_events = np.array(
[[3, 3, 3, 3, 3, 3, 3], [0, 0, 0, 0, 0, 0, 0], [1, 2, 3, 4, 5, 6, 7]]
).T
raw = _create_annotation_based_on_descr(
description, annotation_start_sampl=3, duration=100
)
events, event_id = events_from_annotations(raw, event_id=None)
assert_array_equal(events, expected_events)
assert event_id == expected_event_id
def test_event_id_function_using_custom_function():
"""Test [unit_test] arbitrary function to create the ids."""
def _constant_id(*args, **kwargs):
return 42
description = ["a", "b", "c", "d", "e", "f", "g"]
expected_event_id = dict(zip(description, repeat(42)))
expected_events = np.repeat([[0, 0, 42]], len(description), axis=0)
raw = _create_annotation_based_on_descr(description)
events, event_id = events_from_annotations(raw, event_id=_constant_id)
assert_array_equal(events, expected_events)
assert event_id == expected_event_id
# Test for IO with .csv files
def _assert_annotations_equal(a, b, tol=0):
__tracebackhide__ = True
assert_allclose(a.onset, b.onset, rtol=0, atol=tol, err_msg="onset")
assert_allclose(a.duration, b.duration, rtol=0, atol=tol, err_msg="duration")
assert_array_equal(a.description, b.description, err_msg="description")
assert_array_equal(a.ch_names, b.ch_names, err_msg="ch_names")
a_orig_time = a.orig_time
b_orig_time = b.orig_time
assert a_orig_time == b_orig_time, "orig_time"
_ORIG_TIME = datetime.fromtimestamp(1038942071.7201, timezone.utc)
@pytest.fixture(scope="function", params=("ch_names", "fmt"))
def dummy_annotation_file(tmp_path_factory, ch_names, fmt):
"""Create csv file for testing."""
if fmt == "csv":
content = (
"onset,duration,description\n"
"2002-12-03 19:01:11.720100,1.0,AA\n"
"2002-12-03 19:01:20.720100,2.425,BB"
)
elif fmt == "txt":
content = (
"# MNE-Annotations\n"
"# orig_time : 2002-12-03 19:01:11.720100\n"
"# onset, duration, description\n"
"0, 1, AA \n"
"9, 2.425, BB"
)
else:
assert fmt == "fif"
content = Annotations([0, 9], [1, 2.425], ["AA", "BB"], orig_time=_ORIG_TIME)
if ch_names:
if isinstance(content, Annotations):
# this is a bit of a hack but it works
content.ch_names[:] = ((), ("MEG0111", "MEG2563"))
else:
content = content.splitlines()
content[-3] += ",ch_names"
content[-2] += ","
content[-1] += ",MEG0111:MEG2563"
content = "\n".join(content)
fname = tmp_path_factory.mktemp("data") / f"annotations-annot.{fmt}"
if isinstance(content, str):
with open(fname, "w") as f:
f.write(content)
else:
content.save(fname)
return fname
@pytest.mark.parametrize("ch_names", (False, True))
@pytest.mark.parametrize("fmt", [pytest.param("csv", marks=needs_pandas), "txt", "fif"])
def test_io_annotation(dummy_annotation_file, tmp_path, fmt, ch_names):
"""Test CSV, TXT, and FIF input/output (which support ch_names)."""
annot = read_annotations(dummy_annotation_file)
assert annot.orig_time == _ORIG_TIME
kwargs = dict(orig_time=_ORIG_TIME)
if ch_names:
kwargs["ch_names"] = ((), ("MEG0111", "MEG2563"))
_assert_annotations_equal(
annot, Annotations([0.0, 9.0], [1.0, 2.425], ["AA", "BB"], **kwargs), tol=1e-6
)
# Now test writing
fname = tmp_path / f"annotations-annot.{fmt}"
annot.save(fname)
annot2 = read_annotations(fname)
_assert_annotations_equal(annot, annot2)
# Now without an orig_time
annot._orig_time = None
annot.save(fname, overwrite=True)
annot2 = read_annotations(fname)
_assert_annotations_equal(annot, annot2)
def test_broken_csv(tmp_path):
"""Test broken .csv that does not use timestamps."""
pytest.importorskip("pandas")
content = "onset,duration,description\n1.,1.0,AA\n3.,2.425,BB"
fname = tmp_path / "annotations_broken.csv"
with open(fname, "w") as f:
f.write(content)
with pytest.warns(RuntimeWarning, match="save your CSV as a TXT"):
read_annotations(fname)
# Test for IO with .txt files
@pytest.fixture(scope="function", params=("ch_names",))
def dummy_annotation_txt_file(tmp_path_factory, ch_names):
"""Create txt file for testing."""
content = "3.14, 42, AA \n6.28, 48, BB"
if ch_names:
content = content.splitlines()
content[0] = content[0].strip() + ","
content[1] = content[1].strip() + ", MEG0111:MEG2563"
content = "\n".join(content)
fname = tmp_path_factory.mktemp("data") / "annotations.txt"
with open(fname, "w") as f:
f.write(content)
return fname
@pytest.mark.parametrize("ch_names", (False, True))
def test_io_annotation_txt(dummy_annotation_txt_file, tmp_path_factory, ch_names):
"""Test TXT input/output without meas_date."""
annot = read_annotations(str(dummy_annotation_txt_file))
assert annot.orig_time is None
kwargs = dict()
if ch_names:
kwargs["ch_names"] = [(), ("MEG0111", "MEG2563")]
_assert_annotations_equal(
annot, Annotations([3.14, 6.28], [42.0, 48], ["AA", "BB"], **kwargs)
)
# Now test writing
fname = tmp_path_factory.mktemp("data") / "annotations.txt"
annot.save(fname)
annot2 = read_annotations(fname)
_assert_annotations_equal(annot, annot2)
# Now with an orig_time
assert annot.orig_time is None
annot._orig_time = _handle_meas_date(1038942071.7201)
assert annot.orig_time is not None
annot.save(fname, overwrite=True)
annot2 = read_annotations(fname)
assert annot2.orig_time is not None
_assert_annotations_equal(annot, annot2)
@pytest.mark.parametrize(
"meas_date, out",
[
pytest.param("toto", None, id="invalid string"),
pytest.param(None, None, id="None"),
pytest.param(42, 42.0, id="Scalar"),
pytest.param(3.14, 3.14, id="Float"),
pytest.param((3, 140000), 3.14, id="Scalar tuple"),
pytest.param(
"2002-12-03 19:01:11.720100", 1038942071.7201, id="valid iso8601 string"
),
pytest.param("2002-12-03T19:01:11.720100", None, id="invalid iso8601 string"),
],
)
def test_handle_meas_date(meas_date, out):
"""Test meas date formats."""
if out is not None:
assert out >= 0 # otherwise it'll break on Windows
out = datetime.fromtimestamp(out, timezone.utc)
assert _handle_meas_date(meas_date) == out
def test_read_annotation_txt_header(tmp_path):
"""Test TXT orig_time recovery."""
content = (
"# A something \n"
"# orig_time : 42\n"
"# orig_time : 2002-12-03 19:01:11.720100\n"
"# orig_time : 42\n"
"# C\n"
"Done"
)
fname = tmp_path / "header.txt"
with open(fname, "w") as f:
f.write(content)
orig_time = _read_annotations_txt_parse_header(fname)
want = datetime.fromtimestamp(1038942071.7201, timezone.utc)
assert orig_time == want
def test_read_annotation_txt_one_segment(tmp_path):
"""Test empty TXT input/output."""
content = "# MNE-Annotations\n# onset, duration, description\n3.14, 42, AA"
fname = tmp_path / "one-annotations.txt"
with open(fname, "w") as f:
f.write(content)
annot = read_annotations(fname)
_assert_annotations_equal(annot, Annotations(3.14, 42, ["AA"]))
def test_read_annotation_txt_empty(tmp_path):
"""Test empty TXT input/output."""
content = "# MNE-Annotations\n# onset, duration, description\n"
fname = tmp_path / "empty-annotations.txt"
with open(fname, "w") as f:
f.write(content)
annot = read_annotations(fname)
_assert_annotations_equal(annot, Annotations([], [], []))
def test_annotations_simple_iteration():
"""Test indexing Annotations."""
NUM_ANNOT = 5
EXPECTED_ELEMENTS_TYPE = (np.float64, np.float64, np.str_)
EXPECTED_ONSETS = EXPECTED_DURATIONS = [x for x in range(NUM_ANNOT)]
EXPECTED_DESCS = [x.__repr__() for x in range(NUM_ANNOT)]
annot = Annotations(
onset=EXPECTED_ONSETS,
duration=EXPECTED_DURATIONS,
description=EXPECTED_DESCS,
orig_time=None,
)
for ii, elements in enumerate(annot[:2]):
assert isinstance(elements, OrderedDict)
expected_values = (ii, ii, str(ii))
for elem, expected_type, expected_value in zip(
elements.values(), EXPECTED_ELEMENTS_TYPE, expected_values
):
assert np.isscalar(elem)
assert isinstance(elem, expected_type)
assert elem == expected_value
def test_annotations_slices():
"""Test indexing Annotations."""
NUM_ANNOT = 5
EXPECTED_ONSETS = EXPECTED_DURATIONS = [x for x in range(NUM_ANNOT)]
EXPECTED_DESCS = [x.__repr__() for x in range(NUM_ANNOT)]
annot = Annotations(
onset=EXPECTED_ONSETS,
duration=EXPECTED_DURATIONS,
description=EXPECTED_DESCS,
orig_time=None,
)
# Indexing returns a copy. So this has no effect in annot
annot[0]["onset"] = 42
annot[0]["duration"] = 3.14
annot[0]["description"] = "foobar"
annot[:1].onset[0] = 42
annot[:1].duration[0] = 3.14
annot[:1].description[0] = "foobar"
# Slicing with single element returns a dictionary
for ii in EXPECTED_ONSETS:
assert annot[ii] == dict(
zip(
["onset", "duration", "description", "orig_time"],
[ii, ii, str(ii), None],
)
)
# Slices should give back Annotations
for current in (
annot[slice(0, None, 2)],
annot[[bool(ii % 2) for ii in range(len(annot))]],
annot[:1],
annot[[0, 2, 2]],
annot[(0, 2, 2)],
annot[np.array([0, 2, 2])],
annot[1::2],
):
assert isinstance(current, Annotations)
assert len(current) != len(annot)
for bad_ii in [len(EXPECTED_ONSETS), 42, "foo"]:
with pytest.raises(IndexError):
annot[bad_ii]
def test_sorting():
"""Test annotation sorting."""
annot = Annotations([10, 20, 30], [1, 2, 3], "BAD")
# assert_array_equal(annot.onset, [0, 5, 10])
annot.append([5, 15, 25, 35], 0.5, "BAD")
onset = list(range(5, 36, 5))
duration = list(annot.duration)
assert_array_equal(annot.onset, onset)
assert_array_equal(annot.duration, duration)
annot.append([10, 10], [0.1, 9], "BAD") # 0.1 should be before, 9 after
want_before = onset.index(10)
duration.insert(want_before, 0.1)
duration.insert(want_before + 2, 9)
onset.insert(want_before, 10)
onset.insert(want_before, 10)
assert_array_equal(annot.onset, onset)
assert_array_equal(annot.duration, duration)
def test_date_none(tmp_path):
"""Test that DATE_NONE is used properly."""
# Regression test for gh-5908
n_chans = 139
n_samps = 20
data = np.random.random_sample((n_chans, n_samps))
ch_names = [f"E{x}" for x in range(n_chans)]
ch_types = ["eeg"] * n_chans
info = create_info(ch_names=ch_names, ch_types=ch_types, sfreq=2048)
assert info["meas_date"] is None
raw = RawArray(data=data, info=info)
fname = tmp_path / "test-raw.fif"
raw.save(fname)
raw_read = read_raw_fif(fname, preload=True)
assert raw_read.info["meas_date"] is None
def test_negative_meas_dates(windows_like_datetime):
"""Test meas_date previous to 1970."""
# Regression test for gh-6621
raw = RawArray(
data=np.empty((1, 1), dtype=np.float64), info=create_info(ch_names=1, sfreq=1.0)
)
raw.set_meas_date((-908196946, 988669))
raw.set_annotations(
Annotations(description="foo", onset=[0], duration=[0], orig_time=None)
)
events, _ = events_from_annotations(raw)
assert events[:, 0] == 0
def test_crop_when_negative_orig_time(windows_like_datetime):
"""Test cropping with orig_time, tmin and tmax previous to 1970."""
# Regression test for gh-6621
orig_time_stamp = -908196945.011331 # 1941-03-22 11:04:14.988669
annot = Annotations(
description="foo",
onset=np.arange(0, 0.999, 0.1),
duration=[0],
orig_time=orig_time_stamp,
)
stamp = _dt_to_stamp(annot.orig_time)
assert_allclose(stamp[0] + stamp[1] * 1e-6, orig_time_stamp)
t = stamp[0] + stamp[1] * 1e-6
assert t == orig_time_stamp
assert len(annot) == 10
# do not raise
annot.crop(verbose="debug")
assert len(annot) == 10
# Crop with negative tmin, tmax
tmin, tmax = (orig_time_stamp + t for t in (0.25, 0.75))
assert tmin < 0 and tmax < 0
crop_annot = annot.crop(tmin=tmin, tmax=tmax)
assert_allclose(crop_annot.onset, [0.3, 0.4, 0.5, 0.6, 0.7])
orig_dt = _stamp_to_dt(stamp)
assert crop_annot.orig_time == orig_dt # orig_time does not change
def test_crop_with_none(windows_like_datetime):
"""Test cropping with None in arguments."""
orig_time_stamp = 100
annot = Annotations(
description="foo",
onset=np.arange(5, 10, 1),
duration=[1],
orig_time=orig_time_stamp,
)
annot.crop(tmin=None, tmax=None)
assert len(annot) == 5
annot.crop(tmin=(7.5 + orig_time_stamp), tmax=None)
assert len(annot) == 3
def test_crop_wo_orig_time(windows_like_datetime):
"""Test cropping without orig_time."""
orig_time_stamp = 100
annot = Annotations(
description="foo",
onset=np.arange(5, 10, 1),
duration=[1],
orig_time=orig_time_stamp,
)
annot.crop(tmin=(7.5), tmax=None, use_orig_time=False)
assert len(annot) == 3
def test_allow_nan_durations():
"""Deal with "n/a" strings in BIDS events with nan durations."""
raw = RawArray(
data=np.empty([2, 10], dtype=np.float64),
info=create_info(ch_names=2, sfreq=1.0),
first_samp=0,
)
raw.set_meas_date(0)
ons = [1, 2.0, 15.0, 17.0]
dus = [np.nan, 1.0, 0.5, np.nan]
descriptions = ["A"] * 4
onsets = np.asarray(ons, dtype=float)
durations = np.asarray(dus, dtype=float)
annot = mne.Annotations(onset=onsets, duration=durations, description=descriptions)
with pytest.warns(RuntimeWarning, match="Omitted 2 annotation"):
raw.set_annotations(annot)
@testing.requires_testing_data
def test_annotations_from_events():
"""Test events to annotations conversion."""
raw = read_raw_fif(fif_fname)
events = mne.find_events(raw)
# 1. Automatic event description
# -------------------------------------------------------------------------
annots = annotations_from_events(
events, raw.info["sfreq"], first_samp=raw.first_samp, orig_time=None
)
assert len(annots) == events.shape[0]
# Convert back to events
raw.set_annotations(annots)
events_out, _ = events_from_annotations(raw, event_id=int)
assert_array_equal(events, events_out)
# 2. Explicit event mapping
# -------------------------------------------------------------------------
event_desc = {1: "one", 2: "two", 3: "three", 32: None}
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=event_desc,
first_samp=raw.first_samp,
orig_time=None,
)
assert np.all([a in ["one", "two", "three"] for a in annots.description])
assert len(annots) == events[events[:, 2] <= 3].shape[0]
# 3. Pass list
# -------------------------------------------------------------------------
event_desc = [1, 2, 3]
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=event_desc,
first_samp=raw.first_samp,
orig_time=None,
)
assert np.all([a in ["1", "2", "3"] for a in annots.description])
assert len(annots) == events[events[:, 2] <= 3].shape[0]
# 4. Try passing callable
# -------------------------------------------------------------------------
event_desc = lambda d: f"event{d}" # noqa:E731
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=event_desc,
first_samp=raw.first_samp,
orig_time=None,
)
assert np.all(["event" in a for a in annots.description])
assert len(annots) == events.shape[0]
# 5. Pass numpy array
# -------------------------------------------------------------------------
event_desc = np.array([[1, 2, 3], [1, 2, 3]])
with pytest.raises(ValueError, match="event_desc must be 1D"):
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=event_desc,
first_samp=raw.first_samp,
orig_time=None,
)
with pytest.raises(ValueError, match="Invalid type for event_desc"):
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=1,
first_samp=raw.first_samp,
orig_time=None,
)
event_desc = np.array([1, 2, 3])
annots = annotations_from_events(
events,
sfreq=raw.info["sfreq"],
event_desc=event_desc,
first_samp=raw.first_samp,
orig_time=None,
)
assert np.all([a in ["1", "2", "3"] for a in annots.description])
assert len(annots) == events[events[:, 2] <= 3].shape[0]
def test_repr():
"""Test repr of Annotations."""
# short annotation repr (< 79 characters)
r = repr(Annotations(range(3), [0] * 3, list("abc")))
assert r == "<Annotations | 3 segments: a (1), b (1), c (1)>"
# long annotation repr (> 79 characters, will be shortened)
r = repr(Annotations(range(14), [0] * 14, list("abcdefghijklmn")))
assert r == (
"<Annotations | 14 segments: a (1), b (1), c (1), d (1), e (1), f (1), g ...>"
)
# empty Annotations
r = repr(Annotations([], [], []))
assert r == "<Annotations | 0 segments>"
@pytest.mark.parametrize("time_format", (None, "ms", "datetime", "timedelta"))
def test_annotation_to_data_frame(time_format):
"""Test annotation class to data frame conversion."""
pytest.importorskip("pandas")
onset = np.arange(1, 10)
durations = np.full_like(onset, [4, 5, 6, 4, 5, 6, 4, 5, 6])
description = ["yy"] * onset.shape[0]
a = Annotations(
onset=onset, duration=durations, description=description, orig_time=0
)
df = a.to_data_frame(time_format=time_format)
for col in ["onset", "duration", "description"]:
assert col in df.columns
assert df.description[0] == "yy"
want = 1000 if time_format == "ms" else 1
got = df.onset[1] - df.onset[0]
if time_format in ("datetime", "timedelta"):
got = got.seconds
assert want == got
assert df.groupby("description").count().onset["yy"] == 9
def test_annotation_ch_names():
"""Test annotation ch_names updating and pruning."""
info = create_info(10, 1000.0, "eeg")
raw = RawArray(np.zeros((10, 1000)), info)
onset = [0.1, 0.3, 0.6]
duration = [0.05, 0.1, 0.2]
description = ["first", "second", "third"]
ch_names = [[], raw.ch_names[4:6], raw.ch_names[5:7]]
annot = Annotations(onset, duration, description, ch_names=ch_names)
raw.set_annotations(annot)
# renaming
rename = {name: name + "new" for name in raw.ch_names}
raw_2 = raw.copy().rename_channels(rename)
for ch_rename, ch in zip(raw_2.annotations.ch_names, annot.ch_names):
assert all(name in raw_2.ch_names for name in ch_rename)
assert all(name in raw.ch_names for name in ch)
assert not any(name in raw.ch_names for name in ch_rename)
assert not any(name in raw_2.ch_names for name in ch)
raw_2.rename_channels({val: key for key, val in rename.items()})
_assert_annotations_equal(raw.annotations, raw_2.annotations)
# dropping
raw_2.drop_channels(raw.ch_names[5:])
annot_pruned = raw_2.annotations
assert len(raw_2.annotations) == 2 # dropped the last one
assert raw_2.annotations.ch_names[1] == tuple(raw.ch_names[4:5])
for ch_drop in raw_2.annotations.ch_names:
assert all(name in raw_2.ch_names for name in ch_drop)
with pytest.raises(ValueError, match="channel name in annotations missing"):
raw_2.set_annotations(annot)
with pytest.warns(RuntimeWarning, match="channel name in annotations missing"):
raw_2.set_annotations(annot, on_missing="warn")
assert raw_2.annotations is not annot_pruned
_assert_annotations_equal(raw_2.annotations, annot_pruned)
def test_annotation_rename():
"""Test annotation renaming works."""
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
assert isinstance(a.description, np.ndarray)
assert len(a) == 3
assert "a" in a.description
assert "b" in a.description
assert "c" in a.description
assert "new_name" not in a.description
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
a.rename({"a": "new_name"})
assert isinstance(a.description, np.ndarray)
assert len(a) == 3
assert "a" not in a.description
assert "new_name" in a.description
assert np.where([d == "new_name" for d in a.description])[0] == 0
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
a.rename({"a": "new_name", "b": "new name b"})
assert len(a) == 3
assert "a" not in a.description
assert "new_name" in a.description
assert "b" not in a.description
assert "new name b" in a.description
assert np.where([d == "new_name" for d in a.description])[0] == 0
assert np.where([d == "new name b" for d in a.description])[0] == 1
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
a.rename({"b": "new_name", "c": "new name c"})
assert isinstance(a.description, np.ndarray)
assert len(a) == 3
assert "b" not in a.description
assert "new_name" in a.description
assert "c" not in a.description
assert "new name c" in a.description
assert "a" in a.description
assert np.where([d == "new_name" for d in a.description])[0] == 1
assert np.where([d == "new name c" for d in a.description])[0] == 2
assert len(np.where([d == "new name b" for d in a.description])[0]) == 0
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
with pytest.raises(ValueError, match="not present in data"):
a.rename({"aaa": "does not exist"})
with pytest.raises(ValueError, match="[' a']"):
a.rename({" a": "does not exist"})
with pytest.raises(TypeError, match="dict, got <class 'str'> instead"):
a.rename("wrong")
with pytest.raises(TypeError, match="dict, got <class 'list'> instead"):
a.rename(["wrong"])
with pytest.raises(TypeError, match="dict, got <class 'set'> instead"):
a.rename({"wrong"})
a = np.array([[0, 0, 11], [1000, 0, 1], [1230, 0, 111]])
a = mne.annotations_from_events(a, 256)
a.rename({"1": "A", "11": "B", "111": "C"})
assert_array_equal(a.description, ["B", "A", "C"])
def test_annotation_duration_setting():
"""Test annotation duration setting works."""
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
assert isinstance(a.duration, np.ndarray)
assert len(a) == 3
assert a.duration[0] == 5
assert a.duration[2] == 8
a.set_durations({"a": 3})
assert a.duration[0] == 3
assert a.duration[2] == 8
a.set_durations({"a": 313, "c": 18})
assert a.duration[0] == 313
assert a.duration[2] == 18
a.set_durations({"a": 1, "b": 13})
assert a.duration[0] == 1
assert a.duration[1] == 13
a = Annotations([1, 2, 3], [5, 5, 8], ["a", "b", "c"])
assert len(a) == 3
assert a.duration[0] == 5
assert a.duration[2] == 8
a.set_durations(7.2)
assert isinstance(a.duration, np.ndarray)
assert a.duration[0] == 7.2
assert a.duration[2] == 7.2
a.set_durations(2)
assert a.duration[0] == 2
with pytest.raises(ValueError, match="not present in data"):
a.set_durations({"aaa": 2.2})
with pytest.raises(TypeError, match=" got <class 'set'> instead"):
a.set_durations({"aaa", 2.2})
@pytest.mark.parametrize("meas_date", (None, 1))
@pytest.mark.parametrize("set_meas_date", ("before", "after"))
@pytest.mark.parametrize("first_samp", (0, 100, 3000))
def test_annot_noop(meas_date, first_samp, set_meas_date):
"""Show some unintuitive behavior of annotations."""
sfreq = 1000.0
info = create_info(1, sfreq, "eeg")
onset = 0.5
annot_kwargs = dict()
if set_meas_date == "before":
with info._unlock():
info["meas_date"] = _handle_meas_date(meas_date)
if meas_date is not None:
onset += first_samp / sfreq
annot_kwargs["orig_time"] = meas_date
raw = RawArray(np.zeros((1, 2000)), info, first_samp=first_samp)
annot = Annotations(onset, 0.1, "bad", **annot_kwargs)
raw.set_annotations(annot, verbose="debug")
if set_meas_date == "after":
raw.set_meas_date(meas_date)
first_annot = raw.annotations
if meas_date is None:
first_annot.onset -= raw.first_time
raw.set_annotations(first_annot, verbose="debug") # should be a no-op...
second_annot = raw.annotations
want = first_annot.onset[0]
# it has been shifted when meas_date is None!
if meas_date is None:
want = want + raw.first_time
assert_allclose(second_annot.onset[0], want)
@pytest.mark.parametrize("setting", ("before", "after"))
@pytest.mark.parametrize("meas_date", ("first", "second", "both", None))
@pytest.mark.parametrize("first_samp_2", (0, 320))
@pytest.mark.parametrize("first_samp_1", (160, 0))
def test_annot_concat_crop(meas_date, first_samp_1, first_samp_2, setting):
"""Test that annotation and cropping works properly."""
n_ch = 2
sfreq = 160
duration = 0.1
meas_date_1 = meas_date_2 = None
assert meas_date in (None, "first", "second", "both")
if meas_date in ("first", "both"):
meas_date_1 = datetime(2022, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
if meas_date in ("second", "both"):
meas_date_2 = datetime(2022, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
del meas_date
def _create_raw(eeg, sfreq, onset, description, meas_date, first_samp, setting):
info = mne.create_info(eeg.shape[0], ch_types="eeg", sfreq=sfreq)
raw = mne.io.RawArray(eeg, info, first_samp=first_samp)
if setting == "before":
annot = mne.Annotations(onset, duration, description)
raw = raw.set_annotations(annot)
raw.set_meas_date(meas_date)
else:
assert setting == "after"
raw.set_meas_date(meas_date)
delta = first_samp / sfreq if meas_date is not None else 0
annot = mne.Annotations(
onset + delta, duration, description, orig_time=meas_date
)
raw = raw.set_annotations(annot)
return raw
data_1 = np.array([list(range(40)) * 4 * 10] * n_ch) * 5 * 1e-7
onset_1 = np.array([2.5, 5, 6, 7, 8])
description_1 = [12, "on", 1, 2, "off"]
raw_1 = _create_raw(
data_1, sfreq, onset_1, description_1, meas_date_1, first_samp_1, setting
)
assert_allclose(raw_1.annotations.onset, onset_1 + first_samp_1 / sfreq)
data_2 = np.array([([1e-5] * int(sfreq / 2) + [0] * int(sfreq / 2)) * 10] * n_ch)
onset_2 = np.array([1.5, 2, 2.7, 5])
description_2 = ["on", 3, 4, "off"]
raw_2 = _create_raw(
data_2, sfreq, onset_2, description_2, meas_date_2, first_samp_2, setting
)
assert_allclose(raw_2.annotations.onset, onset_2 + first_samp_2 / sfreq)
onset = np.concatenate([onset_1, np.round(onset_2 + len(raw_1.times) / sfreq, 6)])
assert onset[0] == 2.5
assert_allclose(raw_1.annotations.onset[0], 2.5 + first_samp_1 / sfreq)
onset = np.round(onset + first_samp_1 / sfreq, 6)
want_annot = mne.Annotations(
onset=onset,
duration=duration,
description=description_1 + description_2,
orig_time=meas_date_1,
)
raw_copy = concatenate_raws([raw_1.copy()])
assert_allclose(raw_copy.annotations.onset[0], 2.5 + first_samp_1 / sfreq)
raw = concatenate_raws([raw_1, raw_2])
assert raw.first_samp == raw_1.first_samp == first_samp_1
del raw_1, raw_2
assert_allclose(raw.annotations.onset[0], 2.5 + first_samp_1 / sfreq)
assert raw.info["meas_date"] == meas_date_1
gap_idx = len(description_1)
assert list(raw.annotations.description[gap_idx : gap_idx + 2]) == [
"BAD boundary",
"EDGE boundary",
]
raw.annotations.delete([gap_idx, gap_idx + 1])
start_idx = np.where(raw.annotations.description == "on")[0]
end_idx = np.where(raw.annotations.description == "off")[0]
tmins = raw.annotations.onset[start_idx]
tmaxs = raw.annotations.onset[end_idx]
tmins -= raw.first_time
tmaxs -= raw.first_time
assert len(tmins) == len(tmaxs) == 2
assert raw.info["meas_date"] == meas_date_1
_assert_annotations_equal(raw.annotations, want_annot)
# test a round-trip set -- see test_annot_noop for why we need conditional
if meas_date_1 is None:
want_annot.onset -= first_samp_1 / sfreq
raw.set_annotations(want_annot)
if meas_date_1 is None: # put it back to what it was before
want_annot.onset += first_samp_1 / sfreq
_assert_annotations_equal(raw.annotations, want_annot)
want_descs = list()
for start, stop in zip(start_idx, end_idx):
want_descs.append(list(raw.annotations.description[start : stop + 1]))
for tmin, tmax, descs in zip(tmins, tmaxs, want_descs):
sess = raw.copy()
_assert_annotations_equal(sess.annotations, raw.annotations)
_assert_annotations_equal(sess.annotations, want_annot)
# let's manually print what the logger.debug should say if it's
# doing something correctly
if meas_date_1 is not None:
md = raw.info["meas_date"]
print(f"\nmeas_info set to {md}")
md = md + timedelta(seconds=raw._first_time)
print(f"Data starts at {md}")
md = md + timedelta(seconds=tmin)
print(f"Cropping data to {md}")
md = raw.info["meas_date"] + timedelta(seconds=raw.annotations.onset[1])
print(f"Second annot at {md}")
assert sess.first_samp == first_samp_1
sess.crop(tmin, tmax, verbose="debug")
want_first_samp = first_samp_1 + int(round(tmin * sfreq))
assert sess.first_samp == want_first_samp
assert sess.annotations.orig_time == meas_date_1
assert list(sess.annotations.description) == descs
@pytest.mark.parametrize("first_samp", (0, 10000))
@pytest.mark.parametrize("meas_date", (None, 24 * 60 * 60))
def test_annot_meas_date_first_samp_crop(meas_date, first_samp):
"""Test yet another meas_date / first_samp issue."""
sfreq = 1000.0
info = mne.create_info(1, sfreq, "eeg")
raw = mne.io.RawArray(
np.random.RandomState(0).randn(1, 3000), info, first_samp=first_samp
)
raw.set_meas_date(meas_date)
onset = np.array([0, 1, 2], float)
if meas_date is not None:
onset += first_samp / sfreq
annot = mne.Annotations(
onset=onset,
duration=[0.1, 0.2, 0.3],
description=["a", "b", "c"],
orig_time=raw.info["meas_date"],
)
assert len(annot) == 3
raw.set_annotations(annot)
assert len(raw.annotations) == 3
raw_crop = raw.copy().crop(0, 1.5, verbose="debug")
assert len(raw_crop.annotations) == 2
assert_array_equal(raw_crop.annotations.description, annot.description[:2])
assert_array_equal(raw_crop.annotations.duration, annot.duration[:2])
# these two should be the equivalent
raw_crop = raw.copy().crop(2, 2.5, verbose="debug")
raw_crop_2 = raw.copy().crop(1, None).crop(1, 1.5)
assert_allclose(raw_crop.get_data(), raw_crop_2.get_data())
assert raw_crop.first_samp == raw_crop_2.first_samp
want_onset = onset[2:]
if meas_date is None:
want_onset = want_onset + raw.first_time
for this_raw in (raw_crop, raw_crop_2):
assert len(this_raw.annotations) == 1
assert_allclose(this_raw.annotations.onset, want_onset)
assert_allclose(this_raw.annotations.duration, annot.duration[2:])
def test_count_annotations():
"""Test counting unique annotations."""
annotations = Annotations([0, 1, 2], [1, 2, 1], ["T0", "T1", "T0"])
assert annotations.count() == {"T0": 2, "T1": 1}
assert count_annotations(annotations) == {"T0": 2, "T1": 1}
@pytest.mark.parametrize("split_size", ["1GB", "5MB"])
def test_append_splits_boundary(tmp_path, split_size):
"""Test that split files don't break bad boundary annotations."""
fname = tmp_path / "test_append_raw.fif"
raw_orig = RawArray(
np.random.default_rng(0).normal(size=(100, 100000)),
create_info(100, 1000.0, "eeg"),
)
onset = len(raw_orig.times) / raw_orig.info["sfreq"] + raw_orig.first_time
raw_orig.save(fname, split_size=split_size)
raw = read_raw_fif(fname).load_data()
if split_size == "1GB":
assert len(raw.filenames) == 1
else:
assert len(raw.filenames) == 10
assert_allclose(raw.get_data(), raw_orig.get_data(), rtol=1e-5)
raw.append(raw.copy())
assert len(raw.annotations) == 2
assert raw.annotations.description[0] == "BAD boundary"
assert_allclose(raw.annotations.onset, [onset] * 2)
Datasets
Models
