# 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
Datasets
Models
