MRG: Test fnirs channel structures for chromophore data

mne/preprocessing/nirs/_beer_lambert_law.py

@@ -12,7 +12,7 @@
 from ...io.constants import FIFF
 from ...utils import _validate_type
 from ..nirs import source_detector_distances, _channel_frequencies,\
-    _check_channels_ordered
+    _check_channels_ordered, _channel_chromophore
 
 
 def beer_lambert_law(raw, ppf=0.1):
@@ -55,6 +55,9 @@ def beer_lambert_law(raw, ppf=0.1):
             raw.rename_channels({
                 ch['ch_name']: '%s %s' % (ch['ch_name'][:-4], kind)})
 
+    # Validate the format of data after transformation is valid
+    chroma = np.unique(_channel_chromophore(raw))
+    _check_channels_ordered(raw, chroma)
     return raw
 
 

mne/preprocessing/nirs/nirs.py

@@ -77,53 +77,95 @@ def _channel_chromophore(raw):
     return chroma
 
 
-def _check_channels_ordered(raw, freqs):
+def _check_channels_ordered(raw, pair_vals):
     """Check channels follow expected fNIRS format."""
     # Every second channel should be same SD pair
     # and have the specified light frequencies.
-    picks = _picks_to_idx(raw.info, ['fnirs_cw_amplitude', 'fnirs_od'],
-                          exclude=[], allow_empty=True)
-    if len(picks) % 2 != 0:
+
+    # All wavelength based fNIRS data.
+    picks_wave = _picks_to_idx(raw.info, ['fnirs_cw_amplitude', 'fnirs_od'],
+                               exclude=[], allow_empty=True)
+    # All chromophore fNIRS data
+    picks_chroma = _picks_to_idx(raw.info, ['hbo', 'hbr'],
+                                 exclude=[], allow_empty=True)
+    # All continuous wave fNIRS data
+    picks_cw = np.hstack([picks_chroma, picks_wave])
+
+    if len(picks_cw) % 2 != 0:
         raise ValueError(
             'NIRS channels not ordered correctly. An even number of NIRS '
-            'channels is required. %d channels were provided: %r'
-            % (len(raw.ch_names), raw.ch_names))
+            f'channels is required. {len(raw.ch_names)} channels were'
+            f'provided: {raw.ch_names}')
 
-    all_freqs = [raw.info["chs"][ii]["loc"][9] for ii in picks]
+    # Ensure wavelength info exists for waveform data
+    all_freqs = [raw.info["chs"][ii]["loc"][9] for ii in picks_wave]
     if np.any(np.isnan(all_freqs)):
         raise ValueError(
             'NIRS channels is missing wavelength information in the'
             f'info["chs"] structure. The encoded wavelengths are {all_freqs}.')
 
-    for ii in picks[::2]:
+    for ii in picks_cw[::2]:
         ch1_name_info = re.match(r'S(\d+)_D(\d+) (\d+)',
                                  raw.info['chs'][ii]['ch_name'])
         ch2_name_info = re.match(r'S(\d+)_D(\d+) (\d+)',
                                  raw.info['chs'][ii + 1]['ch_name'])
 
-        if raw.info['chs'][ii]['loc'][9] != \
-                float(ch1_name_info.groups()[2]) or \
-                raw.info['chs'][ii + 1]['loc'][9] != \
-                float(ch2_name_info.groups()[2]):
-            raise ValueError(
-                'NIRS channels not ordered correctly. Channel name and NIRS'
-                ' frequency do not match: %s -> %s & %s -> %s'
-                % (raw.info['chs'][ii]['ch_name'],
-                   raw.info['chs'][ii]['loc'][9],
-                   raw.info['chs'][ii + 1]['ch_name'],
-                   raw.info['chs'][ii + 1]['loc'][9]))
+        if bool(ch2_name_info) & bool(ch1_name_info):
+
+            if raw.info['chs'][ii]['loc'][9] != \
+                    float(ch1_name_info.groups()[2]) or \
+                    raw.info['chs'][ii + 1]['loc'][9] != \
+                    float(ch2_name_info.groups()[2]):
+                raise ValueError(
+                    'NIRS channels not ordered correctly. '
+                    'Channel name and NIRS'
+                    ' frequency do not match: %s -> %s & %s -> %s'
+                    % (raw.info['chs'][ii]['ch_name'],
+                       raw.info['chs'][ii]['loc'][9],
+                       raw.info['chs'][ii + 1]['ch_name'],
+                       raw.info['chs'][ii + 1]['loc'][9]))
+
+            first_value = int(ch1_name_info.groups()[2])
+            second_value = int(ch2_name_info.groups()[2])
+            error_word = "frequencies"
+
+        else:
+            ch1_name_info = re.match(r'S(\d+)_D(\d+) (\w+)',
+                                     raw.info['chs'][ii]['ch_name'])
+            ch2_name_info = re.match(r'S(\d+)_D(\d+) (\w+)',
+                                     raw.info['chs'][ii + 1]['ch_name'])
+
+            if bool(ch2_name_info) & bool(ch1_name_info):
+
+                first_value = ch1_name_info.groups()[2]
+                second_value = ch2_name_info.groups()[2]
+                error_word = "chromophore"
+
+                if (first_value not in ["hbo", "hbr"] or
+                        second_value not in ["hbo", "hbr"]):
+                    raise ValueError(
+                        "NIRS channels have specified naming conventions."
+                        "Chromophore data must be labeled either hbo or hbr."
+                        "Failing channels are "
+                        f"{raw.info['chs'][ii]['ch_name']}, "
+                        f"{raw.info['chs'][ii + 1]['ch_name']}")
+
+            else:
+                raise ValueError(
+                    'NIRS channels have specified naming conventions.'
+                    'The provided channel names can not be parsed.'
+                    f'Channels are {raw.ch_names}')
 
         if (ch1_name_info.groups()[0] != ch2_name_info.groups()[0]) or \
            (ch1_name_info.groups()[1] != ch2_name_info.groups()[1]) or \
-           (int(ch1_name_info.groups()[2]) != freqs[0]) or \
-           (int(ch2_name_info.groups()[2]) != freqs[1]):
+           (first_value != pair_vals[0]) or \
+           (second_value != pair_vals[1]):
             raise ValueError(
                 'NIRS channels not ordered correctly. Channels must be ordered'
                 ' as source detector pairs with alternating'
-                ' frequencies: %d & %d'
-                % (freqs[0], freqs[1]))
+                f' {error_word}: {pair_vals[0]} & {pair_vals[1]}')
 
-    return picks
+    return picks_cw
 
 
 def _fnirs_check_bads(raw):

mne/preprocessing/nirs/tests/test_nirs.py

@@ -176,7 +176,7 @@ def test_fnirs_channel_naming_and_order_readers(fname):
     raw_names_reversed = raw.copy().ch_names
     raw_names_reversed.reverse()
     raw_reversed = raw.copy().pick_channels(raw_names_reversed, ordered=True)
-    with pytest.raises(ValueError, match='not ordered correctly'):
+    with pytest.raises(ValueError, match='not ordered .* frequencies'):
         _check_channels_ordered(raw_reversed, freqs)
     # So if we flip the second argument it should pass again
     _check_channels_ordered(raw_reversed, [850, 760])
@@ -197,6 +197,10 @@ def test_fnirs_channel_naming_and_order_readers(fname):
     assert len(_channel_chromophore(raw)) == len(raw.ch_names)
     chroma = np.unique(_channel_chromophore(raw))
     assert_array_equal(chroma, ["hbo", "hbr"])
+    picks = _check_channels_ordered(raw, chroma)
+    assert len(picks) == len(raw.ch_names)
+    with pytest.raises(ValueError, match='not ordered .* chromophore'):
+        _check_channels_ordered(raw, ["hbx", "hbr"])
 
 
 def test_fnirs_channel_naming_and_order_custom_raw():
@@ -306,3 +310,54 @@ def test_fnirs_channel_naming_and_order_custom_optical_density():
     # and this is how you would fix the ordering, then it should pass
     raw.pick(picks=[0, 3, 1, 4, 2, 5])
     _check_channels_ordered(raw, [760, 850])
+
+
+def test_fnirs_channel_naming_and_order_custom_chroma():
+    """Ensure fNIRS channel checking on manually created data."""
+    data = np.random.normal(size=(6, 10))
+
+    # Start with a correctly named raw intensity dataset
+    # These are the steps required to build an fNIRS Raw object from scratch
+    ch_names = ['S1_D1 hbo', 'S1_D1 hbr', 'S2_D1 hbo', 'S2_D1 hbr',
+                'S3_D1 hbo', 'S3_D1 hbr']
+    ch_types = np.tile(["hbo", "hbr"], 3)
+    info = create_info(ch_names=ch_names, ch_types=ch_types, sfreq=1.0)
+    raw = RawArray(data, info, verbose=True)
+
+    chroma = np.unique(_channel_chromophore(raw))
+    picks = _check_channels_ordered(raw, chroma)
+    assert len(picks) == len(raw.ch_names)
+    assert len(picks) == 6
+
+    # Test block creation fails
+    ch_names = ['S1_D1 hbo', 'S2_D1 hbo', 'S3_D1 hbo',
+                'S1_D1 hbr', 'S2_D1 hbr', 'S3_D1 hbr']
+    ch_types = np.repeat(["hbo", "hbr"], 3)
+    info = create_info(ch_names=ch_names, ch_types=ch_types, sfreq=1.0)
+    raw = RawArray(data, info, verbose=True)
+    with pytest.raises(ValueError, match='not ordered .* chromophore'):
+        _check_channels_ordered(raw, ["hbo", "hbr"])
+    # Reordering should fix
+    raw.pick(picks=[0, 3, 1, 4, 2, 5])
+    _check_channels_ordered(raw, ["hbo", "hbr"])
+    # Wrong names should fail
+    with pytest.raises(ValueError, match='not ordered .* chromophore'):
+        _check_channels_ordered(raw, ["hbb", "hbr"])
+
+    # Test weird naming
+    ch_names = ['S1_D1 hbb', 'S1_D1 hbr', 'S2_D1 hbb', 'S2_D1 hbr',
+                'S3_D1 hbb', 'S3_D1 hbr']
+    ch_types = np.tile(["hbo", "hbr"], 3)
+    info = create_info(ch_names=ch_names, ch_types=ch_types, sfreq=1.0)
+    raw = RawArray(data, info, verbose=True)
+    with pytest.raises(ValueError, match='naming conventions'):
+        _check_channels_ordered(raw, ["hbb", "hbr"])
+
+    # Check more weird naming
+    ch_names = ['S1_DX hbo', 'S1_DX hbr', 'S2_D1 hbo', 'S2_D1 hbr',
+                'S3_D1 hbo', 'S3_D1 hbr']
+    ch_types = np.tile(["hbo", "hbr"], 3)
+    info = create_info(ch_names=ch_names, ch_types=ch_types, sfreq=1.0)
+    raw = RawArray(data, info, verbose=True)
+    with pytest.raises(ValueError, match='can not be parsed'):
+        _check_channels_ordered(raw, ["hbo", "hbr"])