Spaces:
Running
Running
| # Authors: The MNE-Python contributors. | |
| # License: BSD-3-Clause | |
| # Copyright the MNE-Python contributors. | |
| from __future__ import annotations | |
| import re | |
| from collections import defaultdict | |
| from typing import TYPE_CHECKING | |
| import numpy as np | |
| from ..._fiff.constants import FIFF | |
| from ..._fiff.meas_info import create_info | |
| from ...annotations import Annotations | |
| from ...utils import ( | |
| _check_fname, | |
| _soft_import, | |
| _validate_type, | |
| copy_doc, | |
| fill_doc, | |
| logger, | |
| verbose, | |
| warn, | |
| ) | |
| from ..base import BaseRaw | |
| if TYPE_CHECKING: | |
| from pathlib import Path | |
| from numpy.typing import NDArray | |
| _UNITS: dict[str, float] = {"uv": 1e-6, "µv": 1e-6} | |
| class RawANT(BaseRaw): | |
| r"""Reader for Raw ANT files in .cnt format. | |
| Parameters | |
| ---------- | |
| fname : file-like | |
| Path to the ANT raw file to load. The file should have the extension ``.cnt``. | |
| eog : str | None | |
| Regex pattern to find EOG channel labels. If None, no EOG channels are | |
| automatically detected. | |
| misc : str | None | |
| Regex pattern to find miscellaneous channels. If None, no miscellaneous channels | |
| are automatically detected. The default pattern ``"BIP\d+"`` will mark all | |
| bipolar channels as ``misc``. | |
| .. note:: | |
| A bipolar channel might actually contain ECG, EOG or other signal types | |
| which might have a dedicated channel type in MNE-Python. In this case, use | |
| :meth:`mne.io.Raw.set_channel_types` to change the channel type of the | |
| channel. | |
| bipolars : list of str | tuple of str | None | |
| The list of channels to treat as bipolar EEG channels. Each element should be | |
| a string of the form ``'anode-cathode'`` or in ANT terminology as ``'label- | |
| reference'``. If None, all channels are interpreted as ``'eeg'`` channels | |
| referenced to the same reference electrode. Bipolar channels are treated | |
| as EEG channels with a special coil type in MNE-Python, see also | |
| :func:`mne.set_bipolar_reference` | |
| .. warning:: | |
| Do not provide auxiliary channels in this argument, provide them in the | |
| ``eog`` and ``misc`` arguments. | |
| impedance_annotation : str | |
| The string to use for impedance annotations. Defaults to ``"impedance"``, | |
| however, the impedance measurement might mark the end of a segment and the | |
| beginning of a new segment, in which case a discontinuity similar to what | |
| :func:`mne.concatenate_raws` produces is present. In this case, it's better to | |
| include a ``BAD_xxx`` annotation to mark the discontinuity. | |
| .. note:: | |
| Note that the impedance annotation will likely have a duration of ``0``. | |
| If the measurement marks a discontinuity, the duration should be modified to | |
| cover the discontinuity in its entirety. | |
| encoding : str | |
| Encoding to use for :class:`str` in the CNT file. Defaults to ``'latin-1'``. | |
| %(preload)s | |
| %(verbose)s | |
| """ | |
| def __init__( | |
| self, | |
| fname: str | Path, | |
| eog: str | None, | |
| misc: str | None, | |
| bipolars: list[str] | tuple[str, ...] | None, | |
| impedance_annotation: str, | |
| *, | |
| encoding: str = "latin-1", | |
| preload: bool | NDArray, | |
| verbose=None, | |
| ) -> None: | |
| logger.info("Reading ANT file %s", fname) | |
| _soft_import("antio", "reading ANT files", min_version="0.5.0") | |
| from antio import read_cnt | |
| from antio.parser import ( | |
| read_device_info, | |
| read_info, | |
| read_meas_date, | |
| read_subject_info, | |
| read_triggers, | |
| ) | |
| fname = _check_fname(fname, overwrite="read", must_exist=True, name="fname") | |
| _validate_type(eog, (str, None), "eog") | |
| _validate_type(misc, (str, None), "misc") | |
| _validate_type(bipolars, (list, tuple, None), "bipolar") | |
| _validate_type(impedance_annotation, (str,), "impedance_annotation") | |
| if len(impedance_annotation) == 0: | |
| raise ValueError("The impedance annotation cannot be an empty string.") | |
| cnt = read_cnt(fname) | |
| # parse channels, sampling frequency, and create info | |
| ch_names, ch_units, ch_refs, _, _ = read_info(cnt, encoding=encoding) | |
| ch_types = _parse_ch_types(ch_names, eog, misc, ch_refs) | |
| if bipolars is not None: # handle bipolar channels | |
| bipolars_idx = _handle_bipolar_channels(ch_names, ch_refs, bipolars) | |
| for idx, ch in zip(bipolars_idx, bipolars): | |
| if ch_types[idx] != "eeg": | |
| warn( | |
| f"Channel {ch} was not parsed as an EEG channel, changing to " | |
| "EEG channel type since bipolar EEG was requested." | |
| ) | |
| ch_names[idx] = ch | |
| ch_types[idx] = "eeg" | |
| info = create_info( | |
| ch_names, sfreq=cnt.get_sample_frequency(), ch_types=ch_types | |
| ) | |
| info.set_meas_date(read_meas_date(cnt)) | |
| make, model, serial, site = read_device_info(cnt, encoding=encoding) | |
| info["device_info"] = dict(type=make, model=model, serial=serial, site=site) | |
| his_id, name, sex, birthday = read_subject_info(cnt, encoding=encoding) | |
| info["subject_info"] = dict( | |
| his_id=his_id, | |
| first_name=name, | |
| sex=sex, | |
| ) | |
| if birthday is not None: | |
| info["subject_info"]["birthday"] = birthday | |
| if bipolars is not None: | |
| with info._unlock(): | |
| for idx in bipolars_idx: | |
| info["chs"][idx]["coil_type"] = FIFF.FIFFV_COIL_EEG_BIPOLAR | |
| first_samps = np.array((0,)) | |
| last_samps = (cnt.get_sample_count() - 1,) | |
| raw_extras = { | |
| "orig_nchan": cnt.get_channel_count(), | |
| "orig_ch_units": ch_units, | |
| "first_samples": np.array(first_samps), | |
| "last_samples": np.array(last_samps), | |
| } | |
| super().__init__( | |
| info, | |
| preload=preload, | |
| first_samps=first_samps, | |
| last_samps=last_samps, | |
| filenames=[fname], | |
| verbose=verbose, | |
| raw_extras=[raw_extras], | |
| ) | |
| # look for annotations (called trigger by ant) | |
| onsets, durations, descriptions, _, disconnect = read_triggers(cnt) | |
| onsets, durations, descriptions = _prepare_annotations( | |
| onsets, durations, descriptions, disconnect, impedance_annotation | |
| ) | |
| onsets = np.array(onsets) / self.info["sfreq"] | |
| durations = np.array(durations) / self.info["sfreq"] | |
| annotations = Annotations(onsets, duration=durations, description=descriptions) | |
| self.set_annotations(annotations) | |
| def _read_segment_file(self, data, idx, fi, start, stop, cals, mult): | |
| from antio import read_cnt | |
| from antio.parser import read_data | |
| ch_units = self._raw_extras[0]["orig_ch_units"] | |
| first_samples = self._raw_extras[0]["first_samples"] | |
| n_times = self._raw_extras[0]["last_samples"] + 1 | |
| for first_samp, this_n_times in zip(first_samples, n_times): | |
| i_start = max(start, first_samp) | |
| i_stop = min(stop, this_n_times + first_samp) | |
| # read and scale data array | |
| cnt = read_cnt(self.filenames[fi]) | |
| one = read_data(cnt, i_start, i_stop) | |
| _scale_data(one, ch_units) | |
| data_view = data[:, i_start - start : i_stop - start] | |
| if isinstance(idx, slice): | |
| data_view[:] = one[idx] | |
| else: | |
| # faster than doing one = one[idx] | |
| np.take(one, idx, axis=0, out=data_view) | |
| def _handle_bipolar_channels( | |
| ch_names: list[str], ch_refs: list[str], bipolars: list[str] | tuple[str, ...] | |
| ) -> list[int]: | |
| """Handle bipolar channels.""" | |
| bipolars_idx = [] | |
| for ch in bipolars: | |
| _validate_type(ch, (str,), "bipolar_channel") | |
| if "-" not in ch: | |
| raise ValueError( | |
| "Bipolar channels should be provided as 'anode-cathode' or " | |
| f"'label-reference'. '{ch}' is not valid." | |
| ) | |
| anode, cathode = ch.split("-") | |
| if anode not in ch_names: | |
| raise ValueError(f"Anode channel {anode} not found in the channels.") | |
| idx = ch_names.index(anode) | |
| if cathode != ch_refs[idx]: | |
| raise ValueError( | |
| f"Reference electrode for {anode} is {ch_refs[idx]}, not {cathode}." | |
| ) | |
| # store idx for later FIFF coil type change | |
| bipolars_idx.append(idx) | |
| return bipolars_idx | |
| def _parse_ch_types( | |
| ch_names: list[str], eog: str | None, misc: str | None, ch_refs: list[str] | |
| ) -> list[str]: | |
| """Parse the channel types.""" | |
| eog = re.compile(eog) if eog is not None else None | |
| misc = re.compile(misc) if misc is not None else None | |
| ch_types = [] | |
| for ch in ch_names: | |
| if eog is not None and re.fullmatch(eog, ch): | |
| ch_types.append("eog") | |
| elif misc is not None and re.fullmatch(misc, ch): | |
| ch_types.append("misc") | |
| else: | |
| ch_types.append("eeg") | |
| eeg_refs = [ch_refs[k] for k, elt in enumerate(ch_types) if elt == "eeg"] | |
| if len(set(eeg_refs)) == 1: | |
| logger.info( | |
| "All %i EEG channels are referenced to %s.", len(eeg_refs), eeg_refs[0] | |
| ) | |
| else: | |
| warn("All EEG channels are not referenced to the same electrode.") | |
| return ch_types | |
| def _prepare_annotations( | |
| onsets: list[int], | |
| durations: list[int], | |
| descriptions: list[str], | |
| disconnect: dict[str, list[int]], | |
| impedance_annotation: str, | |
| ) -> tuple[list[int], list[int], list[str]]: | |
| """Parse the ANT triggers into better Annotations.""" | |
| # first, let's replace the description 'impedance' with impedance_annotation | |
| for k, desc in enumerate(descriptions): | |
| if desc.lower() == "impedance": | |
| descriptions[k] = impedance_annotation | |
| # next, let's look for amplifier connection/disconnection and let's try to create | |
| # BAD_disconnection annotations from them. | |
| if ( | |
| len(disconnect["start"]) == len(disconnect["stop"]) | |
| and len(disconnect["start"]) != 0 | |
| and all( | |
| 0 <= stop - start | |
| for start, stop in zip(disconnect["start"], disconnect["stop"]) | |
| ) | |
| ): | |
| for start, stop in zip(disconnect["start"], disconnect["stop"]): | |
| onsets.append(start) | |
| durations.append(stop - start) | |
| descriptions.append("BAD_disconnection") | |
| else: | |
| for elt in disconnect["start"]: | |
| onsets.append(elt) | |
| durations.append(0) | |
| descriptions.append("Amplifier disconnected") | |
| for elt in disconnect["stop"]: | |
| onsets.append(elt) | |
| durations.append(0) | |
| descriptions.append("Amplifier reconnected") | |
| return onsets, durations, descriptions | |
| def _scale_data(data: NDArray[np.float64], ch_units: list[str]) -> None: | |
| """Scale the data array based on the human-readable units reported by ANT. | |
| Operates in-place. | |
| """ | |
| units_index = defaultdict(list) | |
| for idx, unit in enumerate(ch_units): | |
| units_index[unit].append(idx) | |
| for unit, value in units_index.items(): | |
| if unit in _UNITS: | |
| data[np.array(value, dtype=np.int16), :] *= _UNITS[unit] | |
| else: | |
| warn( | |
| f"Unit {unit} not recognized, not scaling. Please report the unit on " | |
| "a github issue on https://github.com/mne-tools/mne-python." | |
| ) | |
| def read_raw_ant( | |
| fname, | |
| eog=None, | |
| misc=r"BIP\d+", | |
| bipolars=None, | |
| impedance_annotation="impedance", | |
| *, | |
| encoding: str = "latin-1", | |
| preload=False, | |
| verbose=None, | |
| ) -> RawANT: | |
| """ | |
| Returns | |
| ------- | |
| raw : instance of RawANT | |
| A Raw object containing ANT data. | |
| See :class:`mne.io.Raw` for documentation of attributes and methods. | |
| Notes | |
| ----- | |
| .. versionadded:: 1.9 | |
| """ | |
| return RawANT( | |
| fname, | |
| eog=eog, | |
| misc=misc, | |
| bipolars=bipolars, | |
| impedance_annotation=impedance_annotation, | |
| encoding=encoding, | |
| preload=preload, | |
| verbose=verbose, | |
| ) | |