# 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} @fill_doc 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 """ @verbose 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." ) @copy_doc(RawANT) 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, )