Auto-detect the trigger channel based on retrieved channel names

phys2bids/cli/run.py

@@ -77,8 +77,9 @@ def _get_parser():
                           type=int,
                           help='The column number of the trigger channel. '
                                'Channel numbering starts with 1. '
-                               'Default is 1.',
-                          default=1)
+                               'Default is 0. If chtrig is left as zero phys2bids will '
+                               'perform an automatic trigger channel search by channel names.',
+                          default=0)
     optional.add_argument('-chsel', '--channel-selection',
                           dest='chsel',
                           nargs='*',

phys2bids/phys2bids.py

@@ -127,7 +127,7 @@ def print_json(outfile, samp_freq, time_offset, ch_name):
     description='The BIDS specification',
     cite_module=True)
 def phys2bids(filename, info=False, indir='.', outdir='.', heur_file=None,
-              sub=None, ses=None, chtrig=1, chsel=None, num_timepoints_expected=None,
+              sub=None, ses=None, chtrig=0, chsel=None, num_timepoints_expected=None,
               tr=None, thr=None, pad=9, ch_name=[], yml='', debug=False, quiet=False):
     """
     Run main workflow of phys2bids.
@@ -191,8 +191,11 @@ def phys2bids(filename, info=False, indir='.', outdir='.', heur_file=None,
     # Check options to make them internally coherent pt. II
     # #!# This can probably be done while parsing?
     indir = utils.check_input_dir(indir)
-    if chtrig < 1:
-        raise Exception('Wrong trigger channel. Channel indexing starts with 1!')
+    if chtrig <= 0:
+        raise Exception('Wrong trigger channel. Channel indexing cannot be negative!')
+    elif chtrig == 0:
+        LGR.warning('No trigger channel specified. phys2bids will run an automatic search based '
+                    'on  channel names.')
 
     filename, ftype = utils.check_input_type(filename,
                                              indir)

phys2bids/physio_obj.py

@@ -4,11 +4,14 @@
 """I/O objects for phys2bids."""
 
 import logging
+import re
 from copy import deepcopy
 from itertools import groupby
 
 import numpy as np
 
+TRIGGER_NAMES = ["trig", "trigger"]
+
 LGR = logging.getLogger(__name__)
 
 
@@ -239,6 +242,12 @@ def __init__(self, timeseries, freq, ch_name, units, trigger_idx,
         self.num_timepoints_found = deepcopy(num_timepoints_found)
         self.thr = deepcopy(thr)
         self.time_offset = deepcopy(time_offset)
+        if trigger_idx == 0:
+            self.auto_trigger_selection()
+        else:
+            if ch_name[trigger_idx] not in TRIGGER_NAMES:
+                LGR.warning('Trigger channel name is not in our trigger channel name alias list. '
+                            'Please make sure you choose the proper channel.')
 
     @property
     def ch_amount(self):
@@ -543,6 +552,39 @@ def print_info(self, filename):
 
         LGR.info(info)
 
+    def auto_trigger_selection(self):
+        """
+        Find a trigger index matching the channels with a regular expresion.
+
+        Parameters
+        ----------
+        self
+         Raises
+        ------
+        Exception
+        More than one possible trigger channel was automatically found.
+        Exception
+        No trigger channel automatically found
+        Notes
+        -----
+        Outcome:
+            trigger_idx : int
+                Automatically retrieved trigger index
+        """
+        LGR.info('Running automatic trigger detection.')
+        results = [re.search('|'.join(TRIGGER_NAMES),
+                             case, re.IGNORECASE) for case in self.ch_name]
+        indexes = [i for i, v in enumerate(results) if v]
+        if len(indexes) == 1:
+            self.trigger_idx = indexes[0]
+            LGR.info(f'{self.ch_name[self.trigger_idx]} selected as trigger channel')
+        if len(indexes) > 1:
+            raise Exception('More than one possible trigger channel was automatically found. '
+                            'Please run phys2bids specifying the -chtrig argument.')
+        if len(indexes) < 1:
+            raise Exception('No trigger channel automatically found. Please run phys2bids '
+                            'specifying the -chtrig argument.')
+
 
 class BlueprintOutput():
     """

phys2bids/tests/test_phys2bids.py

@@ -52,7 +52,7 @@ def test_raise_exception(samefreq_full_acq_file):
 
     with raises(Exception) as errorinfo:
         phys2bids.phys2bids(filename=test_filename, num_timepoints_expected=[70], tr=[1.3, 2],
-                            indir=test_path, outdir=test_path)
+                            indir=test_path, outdir=test_path, chtrig=3)
     assert "doesn't match" in str(errorinfo.value)
 
     with raises(Exception) as errorinfo:

phys2bids/tests/test_physio_obj.py

@@ -2,7 +2,6 @@
 
 import numpy as np
 from pytest import raises
-
 from phys2bids import physio_obj as po
 
 
@@ -260,3 +259,39 @@ def test_BlueprintOutput():
 
     # Test __eq__
     assert blueprint_out == blueprint_out
+
+
+def test_auto_trigger_selection(caplog):
+    """Test auto_trigger_selection."""
+    test_time = np.array([0, 1, 2, 3, 4])
+    test_trigger = np.array([0, 1, 2, 3, 4])
+    test_half = np.array([0, 1, 2])
+    test_twice = np.array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
+    test_timeseries = [test_time, test_trigger, test_half, test_twice]
+    test_freq = [1, 1, 0.5, 2]
+    test_chn_name = ['time', 'trigger', 'half', 'CO2']
+    test_units = ['s', 'V', 'V', 'V']
+
+    # test when trigger is not the name of the channel
+    test_chtrig = 2
+    phys_in = po.BlueprintInput(test_timeseries, test_freq, test_chn_name,
+                                test_units, test_chtrig)
+    assert 'Trigger channel name is not' in caplog.text
+
+    # test when trigger is 0 and that the trigger channel is recognized by name:
+    test_chtrig = 0
+    phys_in = po.BlueprintInput(test_timeseries, test_freq, test_chn_name,
+                                test_units, test_chtrig)
+    assert phys_in.trigger_idx == 1
+    # test when no trigger is found
+    test_chn_name = ['time', 'dummy', 'half', 'CO2']
+    with raises(Exception) as errorinfo:
+        phys_in = po.BlueprintInput(test_timeseries, test_freq, test_chn_name,
+                                    test_units, test_chtrig)
+        assert 'No trigger channel automatically found' in str(errorinfo.value)
+    # test when no trigger is found
+    test_chn_name = ['time', 'trigger', 'TRIGGER', 'CO2']
+    with raises(Exception) as errorinfo:
+        phys_in = po.BlueprintInput(test_timeseries, test_freq, test_chn_name,
+                                    test_units, test_chtrig)
+        assert 'More than one possible trigger channel' in str(errorinfo.value)