python_embed/Lib/site-packages/mne/io/fieldtrip/utils.py

# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.

import numpy as np

from ..._fiff._digitization import DigPoint, _ensure_fiducials_head
from ..._fiff.constants import FIFF
from ..._fiff.meas_info import create_info
from ..._fiff.pick import pick_info
from ...transforms import rotation3d_align_z_axis
from ...utils import _check_pandas_installed, warn

_supported_megs = ["neuromag306"]

_unit_dict = {
    "m": 1,
    "cm": 1e-2,
    "mm": 1e-3,
    "V": 1,
    "mV": 1e-3,
    "uV": 1e-6,
    "T": 1,
    "T/m": 1,
    "T/cm": 1e2,
}

NOINFO_WARNING = (
    "Importing FieldTrip data without an info dict from the "
    "original file. Channel locations, orientations and types "
    "will be incorrect. The imported data cannot be used for "
    "source analysis, channel interpolation etc."
)


def _validate_ft_struct(ft_struct):
    """Run validation checks on the ft_structure."""
    if isinstance(ft_struct, list):
        raise RuntimeError("Loading of data in cell arrays is not supported")


def _create_info(ft_struct, raw_info):
    """Create MNE info structure from a FieldTrip structure."""
    if raw_info is None:
        warn(NOINFO_WARNING)

    sfreq = _set_sfreq(ft_struct)
    ch_names = ft_struct["label"]
    if raw_info:
        info = raw_info.copy()
        missing_channels = set(ch_names) - set(info["ch_names"])
        if missing_channels:
            warn(
                "The following channels are present in the FieldTrip data "
                f"but cannot be found in the provided info: {missing_channels}.\n"
                "These channels will be removed from the resulting data!"
            )

            missing_chan_idx = [ch_names.index(ch) for ch in missing_channels]
            new_chs = [ch for ch in ch_names if ch not in missing_channels]
            ch_names = new_chs
            ft_struct["label"] = ch_names

            if "trial" in ft_struct:
                ft_struct["trial"] = _remove_missing_channels_from_trial(
                    ft_struct["trial"], missing_chan_idx
                )

            if "avg" in ft_struct:
                if ft_struct["avg"].ndim == 2:
                    ft_struct["avg"] = np.delete(
                        ft_struct["avg"], missing_chan_idx, axis=0
                    )

        with info._unlock():
            info["sfreq"] = sfreq
        ch_idx = [info["ch_names"].index(ch) for ch in ch_names]
        pick_info(info, ch_idx, copy=False)
    else:
        info = create_info(ch_names, sfreq)
        chs, dig = _create_info_chs_dig(ft_struct)
        with info._unlock(update_redundant=True):
            info.update(chs=chs, dig=dig)

    return info


def _remove_missing_channels_from_trial(trial, missing_chan_idx):
    if isinstance(trial, list):
        for idx_trial in range(len(trial)):
            trial[idx_trial] = _remove_missing_channels_from_trial(
                trial[idx_trial], missing_chan_idx
            )
    elif isinstance(trial, np.ndarray):
        if trial.ndim == 2:
            trial = np.delete(trial, missing_chan_idx, axis=0)
    else:
        raise ValueError(
            '"trial" field of the FieldTrip structure has an unknown format.'
        )

    return trial


def _create_info_chs_dig(ft_struct):
    """Create the chs info field from the FieldTrip structure."""
    all_channels = ft_struct["label"]
    ch_defaults = dict(
        coord_frame=FIFF.FIFFV_COORD_UNKNOWN,
        cal=1.0,
        range=1.0,
        unit_mul=FIFF.FIFF_UNITM_NONE,
        loc=np.array([0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 1]),
        unit=FIFF.FIFF_UNIT_V,
    )
    try:
        elec = ft_struct["elec"]
    except KeyError:
        elec = None

    try:
        grad = ft_struct["grad"]
    except KeyError:
        grad = None

    if elec is None and grad is None:
        warn(
            "The supplied FieldTrip structure does not have an elec or grad "
            "field. No channel locations will extracted and the kind of "
            "channel might be inaccurate."
        )
    if "chanpos" not in (elec or grad or {"chanpos": None}):
        raise RuntimeError(
            "This file was created with an old version of FieldTrip. You can "
            "convert the data to the new version by loading it into FieldTrip "
            "and applying ft_selectdata with an empty cfg structure on it. "
            "Otherwise you can supply the Info field."
        )

    chs = list()
    dig = list()
    counter = 0
    for idx_chan, cur_channel_label in enumerate(all_channels):
        cur_ch = ch_defaults.copy()
        cur_ch["ch_name"] = cur_channel_label
        cur_ch["logno"] = idx_chan + 1
        cur_ch["scanno"] = idx_chan + 1
        if elec and cur_channel_label in elec["label"]:
            cur_ch = _process_channel_eeg(cur_ch, elec)
            assert cur_ch["coord_frame"] == FIFF.FIFFV_COORD_HEAD
            # Ref gets ident=0 and we don't have it, so start at 1
            counter += 1
            d = DigPoint(
                r=cur_ch["loc"][:3],
                coord_frame=FIFF.FIFFV_COORD_HEAD,
                kind=FIFF.FIFFV_POINT_EEG,
                ident=counter,
            )
            dig.append(d)
        elif grad and cur_channel_label in grad["label"]:
            cur_ch = _process_channel_meg(cur_ch, grad)
        else:
            if cur_channel_label.startswith("EOG"):
                cur_ch["kind"] = FIFF.FIFFV_EOG_CH
                cur_ch["coil_type"] = FIFF.FIFFV_COIL_EEG
            elif cur_channel_label.startswith("ECG"):
                cur_ch["kind"] = FIFF.FIFFV_ECG_CH
                cur_ch["coil_type"] = FIFF.FIFFV_COIL_EEG_BIPOLAR
            elif cur_channel_label.startswith("STI"):
                cur_ch["kind"] = FIFF.FIFFV_STIM_CH
                cur_ch["coil_type"] = FIFF.FIFFV_COIL_NONE
            else:
                warn(
                    f"Cannot guess the correct type of channel {cur_channel_label}. "
                    "Making it a MISC channel."
                )
                cur_ch["kind"] = FIFF.FIFFV_MISC_CH
                cur_ch["coil_type"] = FIFF.FIFFV_COIL_NONE

        chs.append(cur_ch)
    _ensure_fiducials_head(dig)

    return chs, dig


def _set_sfreq(ft_struct):
    """Set the sample frequency."""
    try:
        sfreq = ft_struct["fsample"]
    except KeyError:
        try:
            time = ft_struct["time"]
        except KeyError:
            raise ValueError("No Source for sfreq found")
        else:
            t1, t2 = float(time[0]), float(time[1])
            sfreq = 1 / (t2 - t1)
    try:
        sfreq = float(sfreq)
    except TypeError:
        warn(
            "FieldTrip structure contained multiple sample rates, trying the "
            f"first of:\n{sfreq} Hz"
        )
        sfreq = float(sfreq.ravel()[0])
    return sfreq


def _set_tmin(ft_struct):
    """Set the start time before the event in evoked data if possible."""
    times = ft_struct["time"]
    time_check = all(times[i][0] == times[i - 1][0] for i, x in enumerate(times))
    if time_check:
        tmin = times[0][0]
    else:
        raise RuntimeError(
            "Loading data with non-uniform times per epoch is not supported"
        )
    return tmin


def _create_events(ft_struct, trialinfo_column):
    """Create an event matrix from the FieldTrip structure."""
    if "trialinfo" not in ft_struct:
        return None

    event_type = ft_struct["trialinfo"]
    event_number = range(len(event_type))

    if trialinfo_column < 0:
        raise ValueError("trialinfo_column must be positive")

    available_ti_cols = 1
    if event_type.ndim == 2:
        available_ti_cols = event_type.shape[1]

    if trialinfo_column > (available_ti_cols - 1):
        raise ValueError(
            "trialinfo_column is higher than the amount of columns in trialinfo."
        )

    event_trans_val = np.zeros(len(event_type))

    if event_type.ndim == 2:
        event_type = event_type[:, trialinfo_column]

    events = (
        np.vstack([np.array(event_number), event_trans_val, event_type]).astype("int").T
    )

    return events


def _create_event_metadata(ft_struct):
    """Create event metadata from trialinfo."""
    pandas = _check_pandas_installed(strict=False)
    if not pandas:
        warn(
            "The Pandas library is not installed. Not returning the original "
            "trialinfo matrix as metadata."
        )
        return None

    metadata = pandas.DataFrame(ft_struct["trialinfo"])

    return metadata


def _process_channel_eeg(cur_ch, elec):
    """Convert EEG channel from FieldTrip to MNE.

    Parameters
    ----------
    cur_ch: dict
        Channel specific dictionary to populate.

    elec: dict
        elec dict as loaded from the FieldTrip structure

    Returns
    -------
    cur_ch: dict
        The original dict (cur_ch) with the added information
    """
    all_labels = np.asanyarray(elec["label"])
    chan_idx_in_elec = np.where(all_labels == cur_ch["ch_name"])[0][0]
    position = np.squeeze(elec["chanpos"][chan_idx_in_elec, :])
    # chanunit = elec['chanunit'][chan_idx_in_elec]  # not used/needed yet
    position_unit = elec["unit"]

    position = position * _unit_dict[position_unit]
    cur_ch["loc"] = np.hstack((position, np.zeros((9,))))
    cur_ch["unit"] = FIFF.FIFF_UNIT_V
    cur_ch["kind"] = FIFF.FIFFV_EEG_CH
    cur_ch["coil_type"] = FIFF.FIFFV_COIL_EEG
    cur_ch["coord_frame"] = FIFF.FIFFV_COORD_HEAD

    return cur_ch


def _process_channel_meg(cur_ch, grad):
    """Convert MEG channel from FieldTrip to MNE.

    Parameters
    ----------
    cur_ch: dict
        Channel specific dictionary to populate.

    grad: dict
        grad dict as loaded from the FieldTrip structure

    Returns
    -------
    dict: The original dict (cur_ch) with the added information
    """
    all_labels = np.asanyarray(grad["label"])
    chan_idx_in_grad = np.where(all_labels == cur_ch["ch_name"])[0][0]
    gradtype = grad["type"]
    chantype = grad["chantype"][chan_idx_in_grad]
    position_unit = grad["unit"]
    position = np.squeeze(grad["chanpos"][chan_idx_in_grad, :])
    position = position * _unit_dict[position_unit]

    if gradtype == "neuromag306" and "tra" in grad and "coilpos" in grad:
        # Try to regenerate original channel pos.
        idx_in_coilpos = np.where(grad["tra"][chan_idx_in_grad, :] != 0)[0]
        cur_coilpos = grad["coilpos"][idx_in_coilpos, :]
        cur_coilpos = cur_coilpos * _unit_dict[position_unit]
        cur_coilori = grad["coilori"][idx_in_coilpos, :]
        if chantype == "megmag":
            position = cur_coilpos[0] - 0.0003 * cur_coilori[0]
        if chantype == "megplanar":
            tmp_pos = cur_coilpos - 0.0003 * cur_coilori
            position = np.average(tmp_pos, axis=0)

    original_orientation = np.squeeze(grad["chanori"][chan_idx_in_grad, :])
    try:
        orientation = rotation3d_align_z_axis(original_orientation).T
    except AssertionError:
        orientation = np.eye(3)
    assert orientation.shape == (3, 3)
    orientation = orientation.flatten()
    # chanunit = grad['chanunit'][chan_idx_in_grad]  # not used/needed yet

    cur_ch["loc"] = np.hstack((position, orientation))
    cur_ch["kind"] = FIFF.FIFFV_MEG_CH
    if chantype == "megmag":
        cur_ch["coil_type"] = FIFF.FIFFV_COIL_POINT_MAGNETOMETER
        cur_ch["unit"] = FIFF.FIFF_UNIT_T
    elif chantype == "megplanar":
        cur_ch["coil_type"] = FIFF.FIFFV_COIL_VV_PLANAR_T1
        cur_ch["unit"] = FIFF.FIFF_UNIT_T_M
    elif chantype == "refmag":
        cur_ch["coil_type"] = FIFF.FIFFV_COIL_MAGNES_REF_MAG
        cur_ch["unit"] = FIFF.FIFF_UNIT_T
    elif chantype == "refgrad":
        cur_ch["coil_type"] = FIFF.FIFFV_COIL_MAGNES_REF_GRAD
        cur_ch["unit"] = FIFF.FIFF_UNIT_T
    elif chantype == "meggrad":
        cur_ch["coil_type"] = FIFF.FIFFV_COIL_AXIAL_GRAD_5CM
        cur_ch["unit"] = FIFF.FIFF_UNIT_T
    else:
        raise RuntimeError(f"Unexpected coil type: {chantype}.")

    cur_ch["coord_frame"] = FIFF.FIFFV_COORD_HEAD

    return cur_ch