FIX load.fiff.sensor_dim(): connectivity for CTF data

eelbrain/_io/fiff.py

@@ -25,6 +25,7 @@
 from .._text import n_of
 from .._utils import ui, as_list
 from ..mne_fixes import MNE_EVOKED, MNE_RAW, MNE_VOLUME_STC
+from ..mne_fixes._channels import _adjacency_id
 
 
 try:  # mne >= 0.19
@@ -783,30 +784,37 @@ def sensor_dim(
 
     ch_type = None
     if connectivity == 'auto':
-        ch_types = find_mne_channel_types({'chs': chs})
-        if len(ch_types) == 1:
-            ch_type = ch_types[0]
-            if ch_type == 'eog':
-                connectivity = 'none'
+        if sysname and sysname.startswith('KIT-'):
+            connectivity = sysname
         else:
-            # TODO: connectivity for vector data
-            connectivity = 'none'
+            ch_types = find_mne_channel_types({'chs': chs})
+            if len(ch_types) == 1:
+                ch_type = ch_types[0]
+                if ch_type == 'eog':
+                    connectivity = 'none'
+            else:
+                # TODO: connectivity for vector data
+                connectivity = 'none'
 
     if connectivity in ('grid', 'none'):
         pass
     elif isinstance(connectivity, str):
         if connectivity == 'auto':
+            connectivity = _adjacency_id(info, ch_type)
+        if connectivity is None:
             c_matrix, names = mne.channels.find_ch_adjacency(info, ch_type)
-            fix_ch_names = any(ch not in names for ch in ch_names)
+            if any(ch not in names for ch in ch_names):
+                raise NotImplementedError("Connectivity fot this data type")
         else:
             c_matrix, names = mne.channels.read_ch_adjacency(connectivity)
-            fix_ch_names = connectivity.startswith('neuromag')
-
-        if fix_ch_names:
-            vec_ids = {name[-1] for name in ch_names}
-            if len(vec_ids) > 1:
-                raise NotImplementedError("Connectivity for Neuromag vector data")
-            names = [f'{n[:3]} {n[3:]}' for n in names]
+            # fix channel names
+            if connectivity.startswith('neuromag'):
+                vec_ids = {name[-1] for name in ch_names}
+                if len(vec_ids) > 1:
+                    raise NotImplementedError("Connectivity for Neuromag vector data")
+                names = [f'{n[:3]} {n[3:]}' for n in names]
+            elif connectivity == 'ctf275':
+                ch_names = [name[:5] for name in ch_names]
 
         # fix channel order
         if names != ch_names:

eelbrain/_io/tests/test_fiff.py

@@ -1,5 +1,6 @@
 # Author: Christian Brodbeck <christianbrodbeck@nyu.edu>
 import os
+from pathlib import Path
 from warnings import catch_warnings, filterwarnings
 
 import mne
@@ -8,12 +9,21 @@
 from numpy.testing import assert_array_equal, assert_array_almost_equal
 
 from eelbrain import load
-from eelbrain.testing import assert_dataobj_equal, requires_mne_sample_data, file_path
+from eelbrain.testing import assert_dataobj_equal, requires_mne_sample_data, requires_mne_testing_data, file_path
 
 
 FILTER_WARNING = 'The measurement information indicates a low-pass frequency of 40 Hz.'
 
 
+@requires_mne_testing_data
+def test_load_fiff_ctf():
+    path = Path(mne.datasets.testing.data_path())
+    raw_path = path / 'CTF' / 'testdata_ctf.ds'
+    raw = mne.io.read_raw_ctf(raw_path)
+    y = load.fiff.raw_ndvar(raw)
+    assert_array_equal(y.sensor.connectivity()[:3], [[0, 1], [0, 16], [0, 44]])
+
+
 @requires_mne_sample_data
 def test_load_fiff_mne():
     data_path = mne.datasets.sample.data_path()

eelbrain/mne_fixes/_channels.py

@@ -0,0 +1,83 @@
+# Author: Christian Brodbeck <christianbrodbeck@nyu.edu>
+"Private functionality from mne.channels.channels"
+import mne
+from mne.io.constants import FIFF
+
+
+def _adjacency_id(info: mne.Info, ch_type: str):
+    "Guess system ID for channel-adjacency; based on :func:`mne.channels.find_ch_adjacency`"
+    # Excerpt from mne.channels.find_ch_adjacency
+    (has_vv_mag, has_vv_grad, is_old_vv, has_4D_mag, ctf_other_types, has_CTF_grad, n_kit_grads, has_any_meg, has_eeg_coils, has_eeg_coils_and_meg, has_eeg_coils_only, has_neuromag_122_grad, has_csd_coils) = _get_ch_info(info)
+    conn_name = None
+    if has_vv_mag and ch_type == 'mag':
+        conn_name = 'neuromag306mag'
+    elif has_vv_grad and ch_type == 'grad':
+        conn_name = 'neuromag306planar'
+    elif has_4D_mag:
+        if 'MEG 248' in info['ch_names']:
+            idx = info['ch_names'].index('MEG 248')
+            grad = info['chs'][idx]['coil_type'] == FIFF.FIFFV_COIL_MAGNES_GRAD
+            mag = info['chs'][idx]['coil_type'] == FIFF.FIFFV_COIL_MAGNES_MAG
+            if ch_type == 'grad' and grad:
+                conn_name = 'bti248grad'
+            elif ch_type == 'mag' and mag:
+                conn_name = 'bti248'
+        elif 'MEG 148' in info['ch_names'] and ch_type == 'mag':
+            idx = info['ch_names'].index('MEG 148')
+            if info['chs'][idx]['coil_type'] == FIFF.FIFFV_COIL_MAGNES_MAG:
+                conn_name = 'bti148'
+    elif has_CTF_grad and ch_type == 'mag':
+        if info['nchan'] < 100:
+            conn_name = 'ctf64'
+        elif info['nchan'] > 200:
+            conn_name = 'ctf275'
+        else:
+            conn_name = 'ctf151'
+    # End excerpt
+    return conn_name
+
+
+# copied from mne.channels.channels
+def _get_ch_info(info):
+    """Get channel info for inferring acquisition device."""
+    chs = info['chs']
+    # Only take first 16 bits, as higher bits store CTF comp order
+    coil_types = {ch['coil_type'] & 0xFFFF for ch in chs}
+    channel_types = {ch['kind'] for ch in chs}
+
+    has_vv_mag = any(k in coil_types for k in
+                     [FIFF.FIFFV_COIL_VV_MAG_T1, FIFF.FIFFV_COIL_VV_MAG_T2,
+                      FIFF.FIFFV_COIL_VV_MAG_T3])
+    has_vv_grad = any(k in coil_types for k in [FIFF.FIFFV_COIL_VV_PLANAR_T1,
+                                                FIFF.FIFFV_COIL_VV_PLANAR_T2,
+                                                FIFF.FIFFV_COIL_VV_PLANAR_T3])
+    has_neuromag_122_grad = any(k in coil_types
+                                for k in [FIFF.FIFFV_COIL_NM_122])
+
+    is_old_vv = ' ' in chs[0]['ch_name']
+
+    has_4D_mag = FIFF.FIFFV_COIL_MAGNES_MAG in coil_types
+    ctf_other_types = (FIFF.FIFFV_COIL_CTF_REF_MAG,
+                       FIFF.FIFFV_COIL_CTF_REF_GRAD,
+                       FIFF.FIFFV_COIL_CTF_OFFDIAG_REF_GRAD)
+    has_CTF_grad = (FIFF.FIFFV_COIL_CTF_GRAD in coil_types or
+                    (FIFF.FIFFV_MEG_CH in channel_types and
+                     any(k in ctf_other_types for k in coil_types)))
+    # hack due to MNE-C bug in IO of CTF
+    # only take first 16 bits, as higher bits store CTF comp order
+    n_kit_grads = sum(ch['coil_type'] & 0xFFFF == FIFF.FIFFV_COIL_KIT_GRAD
+                      for ch in chs)
+
+    has_any_meg = any([has_vv_mag, has_vv_grad, has_4D_mag, has_CTF_grad,
+                       n_kit_grads])
+    has_eeg_coils = (FIFF.FIFFV_COIL_EEG in coil_types and
+                     FIFF.FIFFV_EEG_CH in channel_types)
+    has_eeg_coils_and_meg = has_eeg_coils and has_any_meg
+    has_eeg_coils_only = has_eeg_coils and not has_any_meg
+    has_csd_coils = (FIFF.FIFFV_COIL_EEG_CSD in coil_types and
+                     FIFF.FIFFV_EEG_CH in channel_types)
+
+    return (has_vv_mag, has_vv_grad, is_old_vv, has_4D_mag, ctf_other_types,
+            has_CTF_grad, n_kit_grads, has_any_meg, has_eeg_coils,
+            has_eeg_coils_and_meg, has_eeg_coils_only, has_neuromag_122_grad,
+            has_csd_coils)