[FIX] running numbers for duplicated channel names. (#5821)

Apply running numbers for duplicated channel names.

Fix #5820

Author: lneisenman

doc/whats_new.rst

@@ -40,6 +40,8 @@ Changelog
 Bug
 ~~~
 
+- Fix :func:`mne.io.read_raw_edf` reading duplicate channel names by `lneisenman`_
+
 - Fix :func:`set_bipolar_reference` in the case of generating all bipolar combinations and also in the case of repeated channels in both lists (anode and cathode) by `Cristóbal Moënne-Loccoz`_
 
 - Fix missing code for computing the median when ``method='median'`` in :meth:`mne.Epochs.average` by `Cristóbal Moënne-Loccoz`_
@@ -3122,3 +3124,5 @@ of commits):
 .. _Cristóbal Moënne-Loccoz: https://github.com/cmmoenne
 
 .. _David Haslacher: https://github.com/davidhaslacher
+
+.. _lneisenman:  https://github.com/lneisenman

mne/io/edf/edf.py

@@ -19,7 +19,7 @@
 from ...utils import verbose, logger, warn
 from ..utils import _blk_read_lims
 from ..base import BaseRaw, _check_update_montage
-from ..meas_info import _empty_info, DATE_NONE
+from ..meas_info import _empty_info, _unique_channel_names, DATE_NONE
 from ..constants import FIFF
 from ...filter import resample
 from ...utils import copy_function_doc_to_method_doc
@@ -563,7 +563,6 @@ def _read_edf_header(fname, exclude):
         for ch in channels:
             fid.read(80)  # transducer
         units = [fid.read(8).strip().decode() for ch in channels]
-        orig_units = dict(zip(ch_names, units))
         edf_info['units'] = list()
         for i, unit in enumerate(units):
             if i in exclude:
@@ -572,7 +571,13 @@ def _read_edf_header(fname, exclude):
                 edf_info['units'].append(1e-6)
             else:
                 edf_info['units'].append(1)
+
         ch_names = [ch_names[idx] for idx in sel]
+        units = [units[idx] for idx in sel]
+
+        # make sure channel names are unique
+        ch_names = _unique_channel_names(ch_names)
+        orig_units = dict(zip(ch_names, units))
 
         physical_min = np.array([float(fid.read(8).decode())
                                  for ch in channels])[sel]

mne/io/edf/tests/data/duplicate_channel_labels.edf

@@ -34,6 +34,8 @@
 montage_path = op.join(data_dir, 'biosemi.hpts')
 bdf_path = op.join(data_dir, 'test.bdf')
 edf_path = op.join(data_dir, 'test.edf')
+duplicate_channel_labels_path = op.join(data_dir,
+                                        'duplicate_channel_labels.edf')
 edf_uneven_path = op.join(data_dir, 'test_uneven_samp.edf')
 bdf_eeglab_path = op.join(data_dir, 'test_bdf_eeglab.mat')
 edf_eeglab_path = op.join(data_dir, 'test_edf_eeglab.mat')
@@ -61,7 +63,7 @@ def test_orig_units():
 
     # Test original units
     orig_units = raw._orig_units
-    assert len(orig_units) == 140
+    assert len(orig_units) == len(raw.ch_names)
     assert orig_units['A1'] == u'µV'  # formerly 'uV' edit by _check_orig_units
 
 
@@ -124,6 +126,15 @@ def test_edf_data():
         read_raw_edf(broken_fname, exclude=raw.ch_names[:132], preload=True)
 
 
+def test_duplicate_channel_labels_edf():
+    """Test reading edf file with duplicate channel names."""
+    EXPECTED_CHANNEL_NAMES = ['EEG F1-Ref-0', 'EEG F2-Ref', 'EEG F1-Ref-1']
+    with pytest.warns(RuntimeWarning, match='Channel names are not unique'):
+        raw = read_raw_edf(duplicate_channel_labels_path, preload=False)
+
+    assert raw.ch_names == EXPECTED_CHANNEL_NAMES
+
+
 def test_parse_annotation(tmpdir):
     """Test parsing the tal channel."""
     # test the parser

mne/io/edf/tests/test_edf.py

@@ -125,6 +125,35 @@ def _stamp_to_dt(stamp):
             datetime.timedelta(0, 0, stamp[1]))  # day, sec, μs
 
 
+def _unique_channel_names(ch_names):
+    """Ensure unique channel names."""
+    FIFF_CH_NAME_MAX_LENGTH = 15
+    unique_ids = np.unique(ch_names, return_index=True)[1]
+    if len(unique_ids) != len(ch_names):
+        dups = set(ch_names[x]
+                   for x in np.setdiff1d(range(len(ch_names)), unique_ids))
+        warn('Channel names are not unique, found duplicates for: '
+             '%s. Applying running numbers for duplicates.' % dups)
+        for ch_stem in dups:
+            overlaps = np.where(np.array(ch_names) == ch_stem)[0]
+            # We need an extra character since we append '-'.
+            # np.ceil(...) is the maximum number of appended digits.
+            n_keep = (FIFF_CH_NAME_MAX_LENGTH - 1 -
+                      int(np.ceil(np.log10(len(overlaps)))))
+            n_keep = min(len(ch_stem), n_keep)
+            ch_stem = ch_stem[:n_keep]
+            for idx, ch_idx in enumerate(overlaps):
+                ch_name = ch_stem + '-%s' % idx
+                if ch_name not in ch_names:
+                    ch_names[ch_idx] = ch_name
+                else:
+                    raise ValueError('Adding a running number for a '
+                                     'duplicate resulted in another '
+                                     'duplicate name %s' % ch_name)
+
+    return ch_names
+
+
 # XXX Eventually this should be de-duplicated with the MNE-MATLAB stuff...
 class Info(dict):
     """Measurement information.
@@ -530,22 +559,9 @@ def _check_consistency(self):
         self._check_ch_name_length()
 
         # make sure channel names are unique
-        unique_ids = np.unique(self['ch_names'], return_index=True)[1]
-        if len(unique_ids) != self['nchan']:
-            dups = set(self['ch_names'][x]
-                       for x in np.setdiff1d(range(self['nchan']), unique_ids))
-            warn('Channel names are not unique, found duplicates for: '
-                 '%s. Applying running numbers for duplicates.' % dups)
-            for ch_stem in dups:
-                overlaps = np.where(np.array(self['ch_names']) == ch_stem)[0]
-                n_keep = min(len(ch_stem),
-                             14 - int(np.ceil(np.log10(len(overlaps)))))
-                ch_stem = ch_stem[:n_keep]
-                for idx, ch_idx in enumerate(overlaps):
-                    ch_name = ch_stem + '-%s' % idx
-                    assert ch_name not in self['ch_names']
-                    self['ch_names'][ch_idx] = ch_name
-                    self['chs'][ch_idx]['ch_name'] = ch_name
+        self['ch_names'] = _unique_channel_names(self['ch_names'])
+        for idx, ch_name in enumerate(self['ch_names']):
+            self['chs'][idx]['ch_name'] = ch_name
 
         if 'filename' in self:
             warn('the "filename" key is misleading '

mne/io/meas_info.py

@@ -113,6 +113,17 @@ def test_make_info():
     assert info['meas_date'] is None
 
 
+def test_duplicate_name_correction():
+    """Test duplicate channel names with running number."""
+    # When running number is possible
+    info = create_info(['A', 'A', 'A'], 1000., verbose='error')
+    assert info['ch_names'] == ['A-0', 'A-1', 'A-2']
+
+    # When running number is not possible
+    with pytest.raises(ValueError, match='Adding a running number'):
+        create_info(['A', 'A', 'A-0'], 1000., verbose='error')
+
+
 def test_fiducials_io():
     """Test fiducials i/o."""
     tempdir = _TempDir()