ADD: load.cnd()

AMD

doc/reference.rst

@@ -86,6 +86,7 @@ Functions and modules for loading specific file formats as Eelbrain object:
    load.eyelink
    load.fiff
    load.txt
+   load.cnd
    load.besa
 
 

eelbrain/_io/cnd.py

@@ -0,0 +1,123 @@
+# Author: Christian Brodbeck <christianbrodbeck@nyu.edu>
+"""Continuous neural data (CND) format used for mTRF-Toolbox"""
+from pathlib import Path
+from typing import Optional, Sequence, Union
+
+import mne
+from mne.externals.pymatreader import read_mat
+import numpy
+
+from .._data_obj import Dataset, Factor, NDVar, Scalar, Sensor, UTS, Var, combine, _matrix_graph
+from .._io.fiff import mne_neighbor_files
+from .._types import PathArg
+from .._utils import ui
+
+
+FILETYPES = [("CND files", "*.mat")]
+
+
+def read_cnd(
+        filename: PathArg = None,
+        connectivity: Union[str, Sequence, float] = None,
+) -> Optional[Dataset]:
+    """Load continuous neural data (CND) file used in the mTRF-Toolbox
+
+    Parameters
+    ----------
+    filename
+        Path to the data file (``*.mat``). If unspecified, open a file dialog.
+    connectivity
+        Sensor adjacency graph for EEG sensors. Can be a string to use a `FieldTrip neighbor file
+        <https://www.fieldtriptoolbox.org/template/neighbours/>`_ (e.g., ``'biosemi64'``;
+        Use a :class:`float` for disance-based connectivity (see :meth:`Sensor.set_connectivity`).
+        For more options see :class:`Sensor`.
+
+    Notes
+    -----
+    This format is experimental and the returned data format might change in the future.
+    """
+    if filename is None:
+        path = ui.ask_file("Load CND File", "Select CND file to load as NDVar", FILETYPES)
+        if not path:
+            return
+    else:
+        path = Path(filename)
+        if not path.suffix and not path.exists():
+            path = path.with_suffix('.mat')
+    data = read_mat(path)
+    if 'eeg' in data:
+        data_type = data['eeg']['dataType']
+        # EEG sensor properties
+        dist_connectivity = None
+        sysname = data['eeg']['deviceName']
+        ch_names = data['eeg']['chanlocs']['labels']
+        # connectivity default
+        if connectivity is None:
+            available = mne_neighbor_files()
+            if sysname in available:
+                connectivity = sysname
+            else:
+                connectivity = 1.6
+        # find connectivity
+        if isinstance(connectivity, float):
+            dist_connectivity = connectivity
+            connectivity = 'none'
+        elif connectivity is False:
+            connectivity = 'none'
+        elif isinstance(connectivity, str) and connectivity not in ('grid', 'none'):
+            adj_matrix, adj_names = mne.channels.read_ch_adjacency(connectivity)
+            # fix channel order
+            if adj_names != ch_names:
+                index = numpy.array([adj_names.index(name) for name in ch_names])
+                adj_matrix = adj_matrix[index][:, index]
+            connectivity = _matrix_graph(adj_matrix)
+        locs = numpy.vstack([
+            -numpy.array(data['eeg']['chanlocs']['Y']),
+            data['eeg']['chanlocs']['X'],
+            data['eeg']['chanlocs']['Z'],
+        ]).T
+        sensor = Sensor(locs, ch_names, sysname, connectivity=connectivity)
+        if dist_connectivity:
+            sensor.set_connectivity(connect_dist=dist_connectivity)
+        # EEG data
+        tstep = 1 / data['eeg']['fs']
+        eeg = []
+        for trial_data in data['eeg']['data']:
+            uts = UTS(0, tstep, trial_data.shape[0])
+            eeg.append(NDVar(trial_data, (uts, sensor), 'eeg'))
+        # Trial position
+        ds = Dataset({
+            data_type.lower(): combine(eeg, to_list=True),
+            'origTrialPosition': Var(data['eeg']['origTrialPosition'] - 1),
+        })
+        # Extra channels
+        if 'extChan' in data['eeg']:
+            desc = ds.info['extChan'] = data['eeg']['extChan']['description']
+            extra_data = []
+            for trial_data in data['eeg']['data']:
+                uts = UTS(0, tstep, trial_data.shape[0])
+                channel = Scalar('channel', range(trial_data.shape[1]))
+                extra_data.append(NDVar(trial_data, (uts, channel), desc))
+            ds['extChan'] = extra_data
+        if 'reRef' in data['eeg']:
+            ds.info['reRef'] = data['eeg']['reRef']
+        return ds
+    if 'stim' in data:
+        ds = Dataset({'stimIdxs': Var(data['stim']['stimIdxs'] - 1)})
+        # stim.data has size nStim x nRuns.
+        tstep = 1 / data['stim']['fs']
+        stim_names = data['stim']['names']
+        for name, stim_data in zip(stim_names, data['stim']['data']):
+            stim_ndvars = []
+            for run_data in stim_data:
+                uts = UTS(0, tstep, len(run_data))
+                stim_ndvars.append(NDVar(run_data, uts, name))
+            key = Dataset.as_key(name)
+            ds[key] = combine(stim_ndvars, to_list=True)
+        # Condition labels
+        if 'condIdxs' in data['stim']:
+            ds['condIdxs'] = Var(data['stim']['condIdxs'] - 1)
+            labels = dict(enumerate(data['stim']['condNames'], 1))
+            ds['condition'] = Factor(data['stim']['condIdxs'], labels=labels)
+        return ds
+    raise IOError("File contains neither 'eeg' or 'stim' entry")

eelbrain/_io/fiff.py

@@ -47,6 +47,11 @@
 PicksArg = Any
 
 
+def mne_neighbor_files():
+    neighbor_dir = Path(mne.channels.__file__).parent / 'data' / 'neighbors'
+    return [re.match(r'([\w-]+)_neighb', path.stem).group(1) for path in neighbor_dir.glob('*_neighb.mat')]
+
+
 def mne_raw(path=None, proj=False, **kwargs):
     """Load a :class:`mne.io.Raw` object
 
@@ -696,7 +701,7 @@ def sensor_dim(
         inferred automatically for KIT data converted with a recent version of
         MNE-Python).
     connectivity : str | list of (str, str) | array of int, (n_edges, 2)
-        Connectivity between elements. Can be specified as:
+        Sensor connectivity (adjacency graph). Can be specified as:
 
         - ``"none"`` for no connections
         - list of connections (e.g., ``[('OZ', 'O1'), ('OZ', 'O2'), ...]``)
@@ -716,7 +721,7 @@ def sensor_dim(
     if not isinstance(info, mne.Info):
         info_ = getattr(info, 'info', info)
         if not isinstance(info_, mne.Info):
-            raise TypeError("No mne.Info object: %r" % (info,))
+            raise TypeError(f"{info=}: no mne.Info object")
         info = info_
 
     if picks is None:

eelbrain/load/__init__.py

@@ -21,6 +21,7 @@
 from . import txt
 
 from .txt import tsv
+from .._io.cnd import read_cnd as cnd
 from .._io.feather import load_feather as feather
 from .._io.pickle import unpickle, update_subjects_dir, convert_pickle_protocol
 from .._io.pyarrow_context import load_arrow as arrow