Aluode/PerceptionLabPortable
0
1# Authors: The MNE-Python contributors.2# License: BSD-3-Clause3# Copyright the MNE-Python contributors.4 5import json6import re7import warnings8from collections import Counter, OrderedDict, UserDict, UserList9from collections.abc import Iterable10from copy import deepcopy11from datetime import datetime, timedelta, timezone12from itertools import takewhile13from textwrap import shorten14 15import numpy as np16from scipy.io import loadmat17 18from ._fiff.constants import FIFF19from ._fiff.open import fiff_open20from ._fiff.tag import read_tag21from ._fiff.tree import dir_tree_find22from ._fiff.write import (23 _safe_name_list,24 end_block,25 start_and_end_file,26 start_block,27 write_double,28 write_float,29 write_name_list_sanitized,30 write_string,31)32from .utils import (33 _check_dict_keys,34 _check_dt,35 _check_fname,36 _check_option,37 _check_pandas_installed,38 _check_time_format,39 _convert_times,40 _DefaultEventParser,41 _dt_to_stamp,42 _is_numeric,43 _mask_to_onsets_offsets,44 _on_missing,45 _pl,46 _stamp_to_dt,47 _validate_type,48 check_fname,49 fill_doc,50 int_like,51 logger,52 verbose,53 warn,54)55 56# For testing windows_like_datetime, we monkeypatch "datetime" in this module.57# Keep the true datetime object around for _validate_type use.58_datetime = datetime59 60 61class _AnnotationsExtrasDict(UserDict):62 """A dictionary for storing extra fields of annotations.63 64 The keys of the dictionary are strings, and the values can be65 strings, integers, floats, or None.66 """67 68 def __setitem__(self, key: str, value: str | int | float | None) -> None:69 _validate_type(key, str, "key")70 if key in ("onset", "duration", "description", "ch_names"):71 raise ValueError(f"Key '{key}' is reserved and cannot be used in extras.")72 _validate_type(73 value,74 (str, int, float, None),75 "value",76 )77 super().__setitem__(key, value)78 79 80class _AnnotationsExtrasList(UserList):81 """A list of dictionaries for storing extra fields of annotations.82 83 Each dictionary in the list corresponds to an annotation and contains84 extra fields.85 The keys of the dictionaries are strings, and the values can be86 strings, integers, floats, or None.87 """88 89 def __repr__(self):90 return repr(self.data)91 92 @staticmethod93 def _validate_value(94 value: dict | _AnnotationsExtrasDict | None,95 ) -> _AnnotationsExtrasDict:96 _validate_type(97 value,98 (dict, _AnnotationsExtrasDict, None),99 "extras dict value",100 "dict or None",101 )102 return (103 value104 if isinstance(value, _AnnotationsExtrasDict)105 else _AnnotationsExtrasDict(value or {})106 )107 108 def __init__(self, initlist=None):109 if not (isinstance(initlist, _AnnotationsExtrasList) or initlist is None):110 initlist = [self._validate_value(v) for v in initlist]111 super().__init__(initlist)112 113 def __setitem__( # type: ignore[override]114 self,115 key: int | slice,116 value,117 ) -> None:118 _validate_type(key, (int, slice), "key", "int or slice")119 if isinstance(key, int):120 iterable = False121 value = [value]122 else:123 _validate_type(value, Iterable, "value", "Iterable when key is a slice")124 iterable = True125 126 new_values = [self._validate_value(v) for v in value]127 if not iterable:128 new_values = new_values[0]129 super().__setitem__(key, new_values)130 131 def __iadd__(self, other):132 if not isinstance(other, _AnnotationsExtrasList):133 other = _AnnotationsExtrasList(other)134 super().__iadd__(other)135 136 def append(self, item):137 super().append(self._validate_value(item))138 139 def insert(self, i, item):140 super().insert(i, self._validate_value(item))141 142 def extend(self, other):143 if not isinstance(other, _AnnotationsExtrasList):144 other = _AnnotationsExtrasList(other)145 super().extend(other)146 147 148def _validate_extras(extras, length: int):149 _validate_type(extras, (None, list, _AnnotationsExtrasList), "extras")150 if extras is not None and len(extras) != length:151 raise ValueError(152 f"extras must be None or a list of length {length}, got {len(extras)}."153 )154 if isinstance(extras, _AnnotationsExtrasList):155 return extras156 return _AnnotationsExtrasList(extras or [None] * length)157 158 159def _check_o_d_s_c_e(onset, duration, description, ch_names, extras):160 onset = np.atleast_1d(np.array(onset, dtype=float))161 if onset.ndim != 1:162 raise ValueError(163 f"Onset must be a one dimensional array, got {onset.ndim} (shape "164 f"{onset.shape})."165 )166 duration = np.array(duration, dtype=float)167 if duration.ndim == 0 or duration.shape == (1,):168 duration = np.repeat(duration, len(onset))169 if duration.ndim != 1:170 raise ValueError(171 f"Duration must be a one dimensional array, got {duration.ndim}."172 )173 174 description = np.array(description, dtype=str)175 if description.ndim == 0 or description.shape == (1,):176 description = np.repeat(description, len(onset))177 if description.ndim != 1:178 raise ValueError(179 f"Description must be a one dimensional array, got {description.ndim}."180 )181 _safe_name_list(description, "write", "description")182 183 # ch_names: convert to ndarray of tuples184 _validate_type(ch_names, (None, tuple, list, np.ndarray), "ch_names")185 if ch_names is None:186 ch_names = [()] * len(onset)187 ch_names = list(ch_names)188 for ai, ch in enumerate(ch_names):189 _validate_type(ch, (list, tuple, np.ndarray), f"ch_names[{ai}]")190 ch_names[ai] = tuple(ch)191 for ci, name in enumerate(ch_names[ai]):192 _validate_type(name, str, f"ch_names[{ai}][{ci}]")193 ch_names = _ndarray_ch_names(ch_names)194 195 if not (len(onset) == len(duration) == len(description) == len(ch_names)):196 raise ValueError(197 "Onset, duration, description, and ch_names must be "198 f"equal in sizes, got {len(onset)}, {len(duration)}, "199 f"{len(description)}, and {len(ch_names)}."200 )201 202 extras = _validate_extras(extras, len(onset))203 return onset, duration, description, ch_names, extras204 205 206def _ndarray_ch_names(ch_names):207 # np.array(..., dtype=object) if all entries are empty will give208 # an empty array of shape (n_entries, 0) which is not helpful. So let's209 # force it to give us an array of shape (n_entries,) full of empty210 # tuples211 out = np.empty(len(ch_names), dtype=object)212 out[:] = ch_names213 return out214 215 216@fill_doc217class Annotations:218 """Annotation object for annotating segments of raw data.219 220 .. note::221 To convert events to `~mne.Annotations`, use222 `~mne.annotations_from_events`. To convert existing `~mne.Annotations`223 to events, use `~mne.events_from_annotations`.224 225 Parameters226 ----------227 onset : array of float, shape (n_annotations,)228 The starting time of annotations in seconds after ``orig_time``.229 duration : array of float, shape (n_annotations,) | float230 Durations of the annotations in seconds. If a float, all the231 annotations are given the same duration.232 description : array of str, shape (n_annotations,) | str233 Array of strings containing description for each annotation. If a234 string, all the annotations are given the same description. To reject235 epochs, use description starting with keyword 'bad'. See example above.236 orig_time : float | str | datetime | tuple of int | None237 A POSIX Timestamp, datetime or a tuple containing the timestamp as the238 first element and microseconds as the second element. Determines the239 starting time of annotation acquisition. If None (default),240 starting time is determined from beginning of raw data acquisition.241 In general, ``raw.info['meas_date']`` (or None) can be used for syncing242 the annotations with raw data if their acquisition is started at the243 same time. If it is a string, it should conform to the ISO8601 format.244 More precisely to this '%%Y-%%m-%%d %%H:%%M:%%S.%%f' particular case of245 the ISO8601 format where the delimiter between date and time is ' ' and at most246 microsecond precision (nanoseconds are not supported).247 %(ch_names_annot)s248 249 .. versionadded:: 0.23250 extras : list[dict[str, int | float | str | None] | None] | None251 Optional list of dicts containing extra fields for each annotation.252 The number of items must match the number of annotations.253 254 .. versionadded:: 1.10255 256 See Also257 --------258 mne.annotations_from_events259 mne.events_from_annotations260 261 Notes262 -----263 Annotations are added to instance of :class:`mne.io.Raw` as the attribute264 :attr:`raw.annotations <mne.io.Raw.annotations>`.265 266 To reject bad epochs using annotations, use267 annotation description starting with 'bad' keyword. The epochs with268 overlapping bad segments are then rejected automatically by default.269 270 To remove epochs with blinks you can do:271 272 >>> eog_events = mne.preprocessing.find_eog_events(raw) # doctest: +SKIP273 >>> n_blinks = len(eog_events) # doctest: +SKIP274 >>> onset = eog_events[:, 0] / raw.info['sfreq'] - 0.25 # doctest: +SKIP275 >>> duration = np.repeat(0.5, n_blinks) # doctest: +SKIP276 >>> description = ['bad blink'] * n_blinks # doctest: +SKIP277 >>> annotations = mne.Annotations(onset, duration, description) # doctest: +SKIP278 >>> raw.set_annotations(annotations) # doctest: +SKIP279 >>> epochs = mne.Epochs(raw, events, event_id, tmin, tmax) # doctest: +SKIP280 281 **ch_names**282 283 Specifying channel names allows the creation of channel-specific284 annotations. Once the annotations are assigned to a raw instance with285 :meth:`mne.io.Raw.set_annotations`, if channels are renamed by the raw286 instance, the annotation channels also get renamed. If channels are dropped287 from the raw instance, any channel-specific annotation that has no channels288 left in the raw instance will also be removed.289 290 **orig_time**291 292 If ``orig_time`` is None, the annotations are synced to the start of the293 data (0 seconds). Otherwise the annotations are synced to sample 0 and294 ``raw.first_samp`` is taken into account the same way as with events.295 296 When setting annotations, the following alignments297 between ``raw.info['meas_date']`` and ``annotation.orig_time`` take place:298 299 ::300 301 ----------- meas_date=XX, orig_time=YY -----------------------------302 303 | +------------------+304 |______________| RAW |305 | | |306 | +------------------+307 meas_date first_samp308 .309 . | +------+310 . |_________| ANOT |311 . | | |312 . | +------+313 . orig_time onset[0]314 .315 | +------+316 |___________________| |317 | | |318 | +------+319 orig_time onset[0]'320 321 ----------- meas_date=XX, orig_time=None ---------------------------322 323 | +------------------+324 |______________| RAW |325 | | |326 | +------------------+327 . N +------+328 . o_________| ANOT |329 . n | |330 . e +------+331 .332 | +------+333 |________________________| |334 | | |335 | +------+336 orig_time onset[0]'337 338 ----------- meas_date=None, orig_time=YY ---------------------------339 340 N +------------------+341 o______________| RAW |342 n | |343 e +------------------+344 | +------+345 |_________| ANOT |346 | | |347 | +------+348 349 [[[ CRASH ]]]350 351 ----------- meas_date=None, orig_time=None -------------------------352 353 N +------------------+354 o______________| RAW |355 n | |356 e +------------------+357 . N +------+358 . o_________| ANOT |359 . n | |360 . e +------+361 .362 N +------+363 o________________________| |364 n | |365 e +------+366 orig_time onset[0]'367 368 .. warning::369 This means that when ``raw.info['meas_date'] is None``, doing370 ``raw.set_annotations(raw.annotations)`` will not alter ``raw`` if and371 only if ``raw.first_samp == 0``. When it's non-zero,372 ``raw.set_annotations`` will assume that the "new" annotations refer to373 the original data (with ``first_samp==0``), and will be re-referenced to374 the new time offset!375 376 **Specific annotation**377 378 ``BAD_ACQ_SKIP`` annotation leads to specific reading/writing file379 behaviours. See :meth:`mne.io.read_raw_fif` and380 :meth:`Raw.save() <mne.io.Raw.save>` notes for details.381 """ # noqa: E501382 383 def __init__(384 self,385 onset,386 duration,387 description,388 orig_time=None,389 ch_names=None,390 *,391 extras=None,392 ):393 self._orig_time = _handle_meas_date(orig_time)394 if isinstance(orig_time, str) and self._orig_time is None:395 try: # only warn if `orig_time` is not the default '1970-01-01 00:00:00'396 if _handle_meas_date(0) == datetime.strptime(397 orig_time, "%Y-%m-%d %H:%M:%S"398 ).replace(tzinfo=timezone.utc):399 pass400 except ValueError: # error if incorrect datetime format AND not the default401 warn(402 "The format of the `orig_time` string is not recognised. It "403 "must conform to the ISO8601 format with at most microsecond "404 "precision and where the delimiter between date and time is "405 f"' '. Got: {orig_time}. Defaulting `orig_time` to None.",406 RuntimeWarning,407 )408 self.onset, self.duration, self.description, self.ch_names, self._extras = (409 _check_o_d_s_c_e(onset, duration, description, ch_names, extras)410 )411 self._sort() # ensure we're sorted412 413 @property414 def orig_time(self):415 """The time base of the Annotations."""416 return self._orig_time417 418 @property419 def extras(self):420 """The extras of the Annotations.421 422 The ``extras`` attribute is a list of dictionaries.423 It can easily be converted to a pandas DataFrame using:424 ``pd.DataFrame(extras)``.425 """426 return self._extras427 428 @extras.setter429 def extras(self, extras):430 self._extras = _validate_extras(extras, len(self.onset))431 432 @property433 def _extras_columns(self) -> set[str]:434 """The set containing all the keys in all extras dicts."""435 return {k for d in self.extras for k in d}436 437 def __eq__(self, other):438 """Compare to another Annotations instance."""439 if not isinstance(other, Annotations):440 return False441 return (442 np.array_equal(self.onset, other.onset)443 and np.array_equal(self.duration, other.duration)444 and np.array_equal(self.description, other.description)445 and np.array_equal(self.ch_names, other.ch_names)446 and self.orig_time == other.orig_time447 )448 449 def __repr__(self):450 """Show the representation."""451 counter = Counter(self.description)452 kinds = ", ".join(["{} ({})".format(*k) for k in sorted(counter.items())])453 kinds = (": " if len(kinds) > 0 else "") + kinds454 ch_specific = ", channel-specific" if self._any_ch_names() else ""455 s = (456 f"Annotations | {len(self.onset)} segment"457 f"{_pl(len(self.onset))}{ch_specific}{kinds}"458 )459 return "<" + shorten(s, width=77, placeholder=" ...") + ">"460 461 def __len__(self):462 """Return the number of annotations.463 464 Returns465 -------466 n_annot : int467 The number of annotations.468 """469 return len(self.duration)470 471 def __add__(self, other):472 """Add (concatencate) two Annotation objects."""473 out = self.copy()474 out += other475 return out476 477 def __iadd__(self, other):478 """Add (concatencate) two Annotation objects in-place.479 480 Both annotations must have the same orig_time481 """482 if len(self) == 0:483 self._orig_time = other.orig_time484 if self.orig_time != other.orig_time:485 raise ValueError(486 "orig_time should be the same to add/concatenate 2 annotations (got "487 f"{self.orig_time} != {other.orig_time})"488 )489 return self.append(490 other.onset,491 other.duration,492 other.description,493 other.ch_names,494 extras=other.extras,495 )496 497 def __iter__(self):498 """Iterate over the annotations."""499 # Figure this out once ahead of time for consistency and speed (for500 # thousands of annotations)501 with_ch_names = self._any_ch_names()502 for idx in range(len(self.onset)):503 yield self.__getitem__(idx, with_ch_names=with_ch_names)504 505 def __getitem__(self, key, *, with_ch_names=None, with_extras=True):506 """Propagate indexing and slicing to the underlying numpy structure."""507 if isinstance(key, int_like):508 out_keys = ("onset", "duration", "description", "orig_time")509 out_vals = (510 self.onset[key],511 self.duration[key],512 self.description[key],513 self.orig_time,514 )515 if with_ch_names or (with_ch_names is None and self._any_ch_names()):516 out_keys += ("ch_names",)517 out_vals += (self.ch_names[key],)518 if with_extras:519 out_keys += ("extras",)520 out_vals += (self.extras[key],)521 return OrderedDict(zip(out_keys, out_vals))522 else:523 key = list(key) if isinstance(key, tuple) else key524 return Annotations(525 onset=self.onset[key],526 duration=self.duration[key],527 description=self.description[key],528 orig_time=self.orig_time,529 ch_names=self.ch_names[key],530 extras=[self.extras[i] for i in np.arange(len(self.extras))[key]],531 )532 533 @fill_doc534 def append(self, onset, duration, description, ch_names=None, *, extras=None):535 """Add an annotated segment. Operates inplace.536 537 Parameters538 ----------539 onset : float | array-like540 Annotation time onset from the beginning of the recording in541 seconds.542 duration : float | array-like543 Duration of the annotation in seconds.544 description : str | array-like545 Description for the annotation. To reject epochs, use description546 starting with keyword 'bad'.547 %(ch_names_annot)s548 549 .. versionadded:: 0.23550 extras : list[dict[str, int | float | str | None] | None] | None551 Optional list of dicts containing extras fields for each annotation.552 The number of items must match the number of annotations.553 554 .. versionadded:: 1.10555 556 Returns557 -------558 self : mne.Annotations559 The modified Annotations object.560 561 Notes562 -----563 The array-like support for arguments allows this to be used similarly564 to not only ``list.append``, but also565 `list.extend <https://docs.python.org/3/library/stdtypes.html#mutable-sequence-types>`__.566 """ # noqa: E501567 onset, duration, description, ch_names, extras = _check_o_d_s_c_e(568 onset, duration, description, ch_names, extras569 )570 self.onset = np.append(self.onset, onset)571 self.duration = np.append(self.duration, duration)572 self.description = np.append(self.description, description)573 self.ch_names = np.append(self.ch_names, ch_names)574 self.extras.extend(extras)575 self._sort()576 return self577 578 def copy(self):579 """Return a copy of the Annotations.580 581 Returns582 -------583 inst : instance of Annotations584 A copy of the object.585 """586 return deepcopy(self)587 588 def delete(self, idx):589 """Remove an annotation. Operates inplace.590 591 Parameters592 ----------593 idx : int | array-like of int594 Index of the annotation to remove. Can be array-like to595 remove multiple indices.596 """597 self.onset = np.delete(self.onset, idx)598 self.duration = np.delete(self.duration, idx)599 self.description = np.delete(self.description, idx)600 self.ch_names = np.delete(self.ch_names, idx)601 if isinstance(idx, int_like):602 del self.extras[idx]603 elif len(idx) > 0:604 # convert slice-like idx to ints, and delete list items in reverse order605 for i in np.sort(np.arange(len(self.extras))[idx])[::-1]:606 del self.extras[i]607 608 @fill_doc609 def to_data_frame(self, time_format="datetime"):610 """Export annotations in tabular structure as a pandas DataFrame.611 612 Parameters613 ----------614 %(time_format_df_raw)s615 Default is ``datetime``.616 617 .. versionadded:: 1.7618 619 Returns620 -------621 result : pandas.DataFrame622 Returns a pandas DataFrame with onset, duration, and623 description columns. A column named ch_names is added if any624 annotations are channel-specific.625 """626 pd = _check_pandas_installed(strict=True)627 valid_time_formats = ["ms", "timedelta", "datetime"]628 dt = _handle_meas_date(self.orig_time)629 if dt is None:630 dt = _handle_meas_date(0)631 time_format = _check_time_format(time_format, valid_time_formats, dt)632 dt = dt.replace(tzinfo=None)633 times = _convert_times(self.onset, time_format, meas_date=dt, drop_nano=True)634 df = dict(onset=times, duration=self.duration, description=self.description)635 if self._any_ch_names():636 df.update(ch_names=self.ch_names)637 df = pd.DataFrame(df)638 extras_df = pd.DataFrame(self.extras)639 df = pd.concat([df, extras_df], axis=1)640 return df641 642 def count(self):643 """Count annotations.644 645 Returns646 -------647 counts : dict648 A dictionary containing unique annotation descriptions as keys with their649 counts as values.650 """651 return count_annotations(self)652 653 def _any_ch_names(self):654 return any(len(ch) for ch in self.ch_names)655 656 def _prune_ch_names(self, info, on_missing):657 # this prunes channel names and if a given channel-specific annotation658 # no longer has any channels left, it gets dropped659 keep = set(info["ch_names"])660 ch_names = self.ch_names661 warned = False662 drop_idx = list()663 for ci, ch in enumerate(ch_names):664 if len(ch):665 names = list()666 for name in ch:667 if name not in keep:668 if not warned:669 _on_missing(670 on_missing,671 "At least one channel name in "672 f"annotations missing from info: {name}",673 )674 warned = True675 else:676 names.append(name)677 ch_names[ci] = tuple(names)678 if not len(ch_names[ci]):679 drop_idx.append(ci)680 if len(drop_idx):681 self.delete(drop_idx)682 return self683 684 @verbose685 def save(self, fname, *, overwrite=False, verbose=None):686 """Save annotations to FIF, CSV or TXT.687 688 Typically annotations get saved in the FIF file for raw data689 (e.g., as ``raw.annotations``), but this offers the possibility690 to also save them to disk separately in different file formats691 which are easier to share between packages.692 693 Parameters694 ----------695 fname : path-like696 The filename to use.697 %(overwrite)s698 699 .. versionadded:: 0.23700 %(verbose)s701 702 Notes703 -----704 The format of the information stored in the saved annotation objects705 depends on the chosen file format. :file:`.csv` files store the onset706 as timestamps (e.g., ``2002-12-03 19:01:56.676071``),707 whereas :file:`.txt` files store onset as seconds since start of the708 recording (e.g., ``45.95597082905339``).709 """710 check_fname(711 fname,712 "annotations",713 (714 "-annot.fif",715 "-annot.fif.gz",716 "_annot.fif",717 "_annot.fif.gz",718 ".txt",719 ".csv",720 ),721 )722 fname = _check_fname(fname, overwrite=overwrite)723 if fname.suffix == ".txt":724 _write_annotations_txt(fname, self)725 elif fname.suffix == ".csv":726 _write_annotations_csv(fname, self)727 else:728 with start_and_end_file(fname) as fid:729 _write_annotations(fid, self)730 731 def _sort(self):732 """Sort in place."""733 # instead of argsort here we use sorted so that it gives us734 # the onset-then-duration hierarchy735 vals = sorted(zip(self.onset, self.duration, range(len(self))))736 order = list(list(zip(*vals))[-1]) if len(vals) else []737 self.onset = self.onset[order]738 self.duration = self.duration[order]739 self.description = self.description[order]740 self.ch_names = self.ch_names[order]741 self.extras = [self.extras[i] for i in order]742 743 @verbose744 def crop(745 self, tmin=None, tmax=None, emit_warning=False, use_orig_time=True, verbose=None746 ):747 """Remove all annotation that are outside of [tmin, tmax].748 749 The method operates inplace.750 751 Parameters752 ----------753 tmin : float | datetime | None754 Start time of selection in seconds.755 tmax : float | datetime | None756 End time of selection in seconds.757 emit_warning : bool758 Whether to emit warnings when limiting or omitting annotations.759 Defaults to False.760 use_orig_time : bool761 Whether to use orig_time as an offset.762 Defaults to True.763 %(verbose)s764 765 Returns766 -------767 self : instance of Annotations768 The cropped Annotations object.769 """770 if len(self) == 0:771 return self # no annotations, nothing to do772 if not use_orig_time or self.orig_time is None:773 offset = _handle_meas_date(0)774 else:775 offset = self.orig_time776 if tmin is None:777 tmin = timedelta(seconds=self.onset.min()) + offset778 if tmax is None:779 tmax = timedelta(seconds=(self.onset + self.duration).max()) + offset780 for key, val in [("tmin", tmin), ("tmax", tmax)]:781 _validate_type(782 val, ("numeric", _datetime), key, "numeric, datetime, or None"783 )784 absolute_tmin = _handle_meas_date(tmin)785 absolute_tmax = _handle_meas_date(tmax)786 del tmin, tmax787 if absolute_tmin > absolute_tmax:788 raise ValueError(789 f"tmax should be greater than or equal to tmin ({absolute_tmin} < "790 f"{absolute_tmax})."791 )792 logger.debug(f"Cropping annotations {absolute_tmin} - {absolute_tmax}")793 794 onsets, durations, descriptions, ch_names, extras = [], [], [], [], []795 out_of_bounds, clip_left_elem, clip_right_elem = [], [], []796 for idx, (onset, duration, description, ch, extra) in enumerate(797 zip(self.onset, self.duration, self.description, self.ch_names, self.extras)798 ):799 # if duration is NaN behave like a zero800 if np.isnan(duration):801 duration = 0.0802 # convert to absolute times803 absolute_onset = timedelta(seconds=onset) + offset804 absolute_offset = absolute_onset + timedelta(seconds=duration)805 out_of_bounds.append(806 absolute_onset > absolute_tmax or absolute_offset < absolute_tmin807 )808 if out_of_bounds[-1]:809 clip_left_elem.append(False)810 clip_right_elem.append(False)811 logger.debug(812 f" [{idx}] Dropping "813 f"({absolute_onset} - {absolute_offset}: {description})"814 )815 else:816 # clip the left side817 clip_left_elem.append(absolute_onset < absolute_tmin)818 if clip_left_elem[-1]:819 absolute_onset = absolute_tmin820 clip_right_elem.append(absolute_offset > absolute_tmax)821 if clip_right_elem[-1]:822 absolute_offset = absolute_tmax823 if clip_left_elem[-1] or clip_right_elem[-1]:824 durations.append((absolute_offset - absolute_onset).total_seconds())825 else:826 durations.append(duration)827 onsets.append((absolute_onset - offset).total_seconds())828 logger.debug(829 f" [{idx}] Keeping "830 f"({absolute_onset} - {absolute_offset} -> "831 f"{onset} - {onset + duration})"832 )833 descriptions.append(description)834 ch_names.append(ch)835 extras.append(extra)836 logger.debug(f"Cropping complete (kept {len(onsets)})")837 self.onset = np.array(onsets, float)838 self.duration = np.array(durations, float)839 assert (self.duration >= 0).all()840 self.description = np.array(descriptions, dtype=str)841 self.ch_names = _ndarray_ch_names(ch_names)842 self.extras = extras843 844 if emit_warning:845 omitted = np.array(out_of_bounds).sum()846 if omitted > 0:847 warn(f"Omitted {omitted} annotation(s) that were outside data range.")848 limited = (np.array(clip_left_elem) | np.array(clip_right_elem)).sum()849 if limited > 0:850 warn(851 f"Limited {limited} annotation(s) that were expanding outside the"852 " data range."853 )854 855 return self856 857 @verbose858 def set_durations(self, mapping, verbose=None):859 """Set annotation duration(s). Operates inplace.860 861 Parameters862 ----------863 mapping : dict | float864 A dictionary mapping the annotation description to a duration in865 seconds e.g. ``{'ShortStimulus' : 3, 'LongStimulus' : 12}``.866 Alternatively, if a number is provided, then all annotations867 durations are set to the single provided value.868 %(verbose)s869 870 Returns871 -------872 self : mne.Annotations873 The modified Annotations object.874 875 Notes876 -----877 .. versionadded:: 0.24.0878 """879 _validate_type(mapping, (int, float, dict))880 881 if isinstance(mapping, dict):882 _check_dict_keys(883 mapping,884 self.description,885 valid_key_source="data",886 key_description="Annotation description(s)",887 )888 for stim in mapping:889 map_idx = [desc == stim for desc in self.description]890 self.duration[map_idx] = mapping[stim]891 892 elif _is_numeric(mapping):893 self.duration = np.ones(self.description.shape) * mapping894 895 else:896 raise ValueError(897 "Setting durations requires the mapping of "898 "descriptions to times to be provided as a dict. "899 f"Instead {type(mapping)} was provided."900 )901 902 return self903 904 @verbose905 def rename(self, mapping, verbose=None):906 """Rename annotation description(s). Operates inplace.907 908 Parameters909 ----------910 mapping : dict911 A dictionary mapping the old description to a new description,912 e.g. {'1.0' : 'Control', '2.0' : 'Stimulus'}.913 %(verbose)s914 915 Returns916 -------917 self : mne.Annotations918 The modified Annotations object.919 920 Notes921 -----922 .. versionadded:: 0.24.0923 """924 _validate_type(mapping, dict)925 _check_dict_keys(926 mapping,927 self.description,928 valid_key_source="data",929 key_description="Annotation description(s)",930 )931 self.description = np.array([str(mapping.get(d, d)) for d in self.description])932 return self933 934 935class EpochAnnotationsMixin:936 """Mixin class for Annotations in Epochs."""937 938 @property939 def annotations(self): # noqa: D102940 return self._annotations941 942 @verbose943 def set_annotations(self, annotations, on_missing="raise", *, verbose=None):944 """Setter for Epoch annotations from Raw.945 946 This method does not handle offsetting the times based947 on first_samp or measurement dates, since that is expected948 to occur in Raw.set_annotations().949 950 Parameters951 ----------952 annotations : instance of mne.Annotations | None953 Annotations to set.954 %(on_missing_ch_names)s955 %(verbose)s956 957 Returns958 -------959 self : instance of Epochs960 The epochs object with annotations.961 962 Notes963 -----964 Annotation onsets and offsets are stored as time in seconds (not as965 sample numbers).966 967 If you have an ``-epo.fif`` file saved to disk created before 1.0,968 annotations can be added correctly only if no decimation or969 resampling was performed. We thus suggest to regenerate your970 :class:`mne.Epochs` from raw and re-save to disk with 1.0+ if you971 want to safely work with :class:`~mne.Annotations` in epochs.972 973 Since this method does not handle offsetting the times based974 on first_samp or measurement dates, the recommended way to add975 Annotations is::976 977 raw.set_annotations(annotations)978 annotations = raw.annotations979 epochs.set_annotations(annotations)980 981 .. versionadded:: 1.0982 """983 _validate_type(annotations, (Annotations, None), "annotations")984 if annotations is None:985 self._annotations = None986 else:987 if getattr(self, "_unsafe_annot_add", False):988 warn(989 "Adding annotations to Epochs created (and saved to disk) before "990 "1.0 will yield incorrect results if decimation or resampling was "991 "performed on the instance, we recommend regenerating the Epochs "992 "and re-saving them to disk."993 )994 new_annotations = annotations.copy()995 new_annotations._prune_ch_names(self.info, on_missing)996 self._annotations = new_annotations997 return self998 999 def get_annotations_per_epoch(self, *, with_extras=False):1000 """Get a list of annotations that occur during each epoch.1001 1002 Parameters1003 ----------1004 with_extras : bool1005 Whether to include the annotations extra fields in the output,1006 as an additional last element of the tuple. Default is False.1007 1008 .. versionadded:: 1.101009 1010 Returns1011 -------1012 epoch_annots : list1013 A list of lists (with length equal to number of epochs) where each1014 inner list contains any annotations that overlap the corresponding1015 epoch. Annotations are stored as a :class:`tuple` of onset,1016 duration, description (not as a :class:`~mne.Annotations` object),1017 where the onset is now relative to time=0 of the epoch, rather than1018 time=0 of the original continuous (raw) data.1019 """1020 # create a list of annotations for each epoch1021 epoch_annot_list = [[] for _ in range(len(self.events))]1022 1023 # check if annotations exist1024 if self.annotations is None:1025 return epoch_annot_list1026 1027 # when each epoch and annotation starts/stops1028 # no need to account for first_samp here...1029 epoch_tzeros = self.events[:, 0] / self._raw_sfreq1030 epoch_starts, epoch_stops = (1031 np.atleast_2d(epoch_tzeros) + np.atleast_2d(self.times[[0, -1]]).T1032 )1033 # ... because first_samp isn't accounted for here either1034 annot_starts = self._annotations.onset1035 annot_stops = annot_starts + self._annotations.duration1036 1037 # the first two cases (annot_straddles_epoch_{start|end}) will both1038 # (redundantly) capture cases where an annotation fully encompasses1039 # an epoch (e.g., annot from 1-4s, epoch from 2-3s). The redundancy1040 # doesn't matter because results are summed and then cast to bool (all1041 # we care about is presence/absence of overlap).1042 annot_straddles_epoch_start = np.logical_and(1043 np.atleast_2d(epoch_starts) >= np.atleast_2d(annot_starts).T,1044 np.atleast_2d(epoch_starts) < np.atleast_2d(annot_stops).T,1045 )1046 1047 annot_straddles_epoch_end = np.logical_and(1048 np.atleast_2d(epoch_stops) > np.atleast_2d(annot_starts).T,1049 np.atleast_2d(epoch_stops) <= np.atleast_2d(annot_stops).T,1050 )1051 1052 # this captures the only remaining case we care about: annotations1053 # fully contained within an epoch (or exactly coextensive with it).1054 annot_fully_within_epoch = np.logical_and(1055 np.atleast_2d(epoch_starts) <= np.atleast_2d(annot_starts).T,1056 np.atleast_2d(epoch_stops) >= np.atleast_2d(annot_stops).T,1057 )1058 1059 # combine all cases to get array of shape (n_annotations, n_epochs).1060 # Nonzero entries indicate overlap between the corresponding1061 # annotation (row index) and epoch (column index).1062 all_cases = (1063 annot_straddles_epoch_start1064 + annot_straddles_epoch_end1065 + annot_fully_within_epoch1066 )1067 1068 # for each Epoch-Annotation overlap occurrence:1069 for annot_ix, epo_ix in zip(*np.nonzero(all_cases)):1070 this_annot = self._annotations[annot_ix]1071 this_tzero = epoch_tzeros[epo_ix]1072 # adjust annotation onset to be relative to epoch tzero...1073 annot = (1074 this_annot["onset"] - this_tzero,1075 this_annot["duration"],1076 this_annot["description"],1077 )1078 if with_extras:1079 annot += (this_annot["extras"],)1080 # ...then add it to the correct sublist of `epoch_annot_list`1081 epoch_annot_list[epo_ix].append(annot)1082 return epoch_annot_list1083 1084 def add_annotations_to_metadata(self, overwrite=False, *, with_extras=True):1085 """Add raw annotations into the Epochs metadata data frame.1086 1087 Adds three columns to the ``metadata`` consisting of a list1088 in each row:1089 - ``annot_onset``: the onset of each Annotation within1090 the Epoch relative to the start time of the Epoch (in seconds).1091 - ``annot_duration``: the duration of each Annotation1092 within the Epoch in seconds.1093 - ``annot_description``: the free-form text description of each1094 Annotation.1095 1096 Parameters1097 ----------1098 overwrite : bool1099 Whether to overwrite existing columns in metadata or not.1100 Default is False.1101 with_extras : bool1102 Whether to include the annotations extra fields in the output,1103 as an additional last element of the tuple. Default is True.1104 1105 .. versionadded:: 1.101106 1107 Returns1108 -------1109 self : instance of Epochs1110 The modified instance (instance is also modified inplace).1111 1112 Notes1113 -----1114 .. versionadded:: 1.01115 """1116 pd = _check_pandas_installed()1117 1118 # check if annotations exist1119 if self.annotations is None:1120 warn(1121 f"There were no Annotations stored in {self}, so "1122 "metadata was not modified."1123 )1124 return self1125 1126 # get existing metadata DataFrame or instantiate an empty one1127 if self._metadata is not None:1128 metadata = self._metadata1129 else:1130 data = np.empty((len(self.events), 0))1131 metadata = pd.DataFrame(data=data)1132 1133 if (1134 any(1135 name in metadata.columns1136 for name in ["annot_onset", "annot_duration", "annot_description"]1137 )1138 and not overwrite1139 ):1140 raise RuntimeError(1141 "Metadata for Epochs already contains columns "1142 '"annot_onset", "annot_duration", or "annot_description".'1143 )1144 1145 # get the Epoch annotations, then convert to separate lists for1146 # onsets, durations, and descriptions1147 epoch_annot_list = self.get_annotations_per_epoch(with_extras=with_extras)1148 onset, duration, description = [], [], []1149 extras = {k: [] for k in self.annotations._extras_columns}1150 for epoch_annot in epoch_annot_list:1151 for ix, annot_prop in enumerate((onset, duration, description)):1152 entry = [annot[ix] for annot in epoch_annot]1153 1154 # round onset and duration to avoid IO round trip mismatch1155 if ix < 2:1156 entry = np.round(entry, decimals=12).tolist()1157 1158 annot_prop.append(entry)1159 for k in extras.keys():1160 entry = [annot[3].get(k, None) for annot in epoch_annot]1161 extras[k].append(entry)1162 1163 # Create a new Annotations column that is instantiated as an empty1164 # list per Epoch.1165 metadata["annot_onset"] = pd.Series(onset)1166 metadata["annot_duration"] = pd.Series(duration)1167 metadata["annot_description"] = pd.Series(description)1168 for k, v in extras.items():1169 metadata[f"annot_{k}"] = pd.Series(v)1170 1171 # reset the metadata1172 self.metadata = metadata1173 return self1174 1175 1176def _combine_annotations(1177 one, two, one_n_samples, one_first_samp, two_first_samp, sfreq1178):1179 """Combine a tuple of annotations."""1180 assert one is not None1181 assert two is not None1182 shift = one_n_samples / sfreq # to the right by the number of samples1183 shift += one_first_samp / sfreq # to the right by the offset1184 shift -= two_first_samp / sfreq # undo its offset1185 onset = np.concatenate([one.onset, two.onset + shift])1186 duration = np.concatenate([one.duration, two.duration])1187 description = np.concatenate([one.description, two.description])1188 ch_names = np.concatenate([one.ch_names, two.ch_names])1189 return Annotations(onset, duration, description, one.orig_time, ch_names)1190 1191 1192def _handle_meas_date(meas_date):1193 """Convert meas_date to datetime or None.1194 1195 If `meas_date` is a string, it should conform to the ISO8601 format.1196 More precisely to this '%Y-%m-%d %H:%M:%S.%f' particular case of the1197 ISO8601 format where the delimiter between date and time is ' '.1198 Note that ISO8601 allows for ' ' or 'T' as delimiters between date and1199 time.1200 """