Merge pull request #1926 from e-kenneally/fix/fs_ingress

🐛 Fix bug in abcd-options workflow

Author: sgiavasis

CPAC/anat_preproc/anat_preproc.py

@@ -914,33 +914,43 @@ def freesurfer_abcd_brain_connector(wf, cfg, strat_pool, pipe_num, opt):
     ### ABCD harmonization - anatomical brain mask generation ###
     # Ref: https://github.com/DCAN-Labs/DCAN-HCP/blob/master/PostFreeSurfer/PostFreeSurferPipeline.sh#L151-L156
 
-    wmparc_to_native = pe.Node(
-        interface=freesurfer.ApplyVolTransform(),
-        name='wmparc_to_native')
 
-    wmparc_to_native.inputs.reg_header = True
-
-    node, out = strat_pool.get_data('pipeline-fs_wmparc') 
-    wf.connect(node, out, wmparc_to_native, 'source_file')
-
-    node, out = strat_pool.get_data('pipeline-fs_raw-average')
-    wf.connect(node, out, wmparc_to_native, 'target_file')
-
-    node, out = strat_pool.get_data('pipeline-fs_xfm')
-    wf.connect(node, out, wmparc_to_native, 'xfm_reg_file')
-
-    node, out = strat_pool.get_data('freesurfer-subject-dir')
-
-    wf.connect(node, out, wmparc_to_native, 'subjects_dir')
     wmparc_to_nifti = pe.Node(util.Function(input_names=['in_file',
                                                          'reslice_like',
                                                          'args'],
                                             output_names=['out_file'],
                                             function=mri_convert),
                               name=f'wmparc_to_nifti_{pipe_num}')
-    wmparc_to_nifti.inputs.args = '-rt nearest'
+    
+    # Register wmparc file if ingressing FreeSurfer data
+    if strat_pool.check_rpool('pipeline-fs_xfm'):
+
+        wmparc_to_native = pe.Node(
+            interface=freesurfer.ApplyVolTransform(),
+            name='wmparc_to_native')
+        
+        wmparc_to_native.inputs.reg_header = True
+
+        node, out = strat_pool.get_data('pipeline-fs_wmparc') 
+        wf.connect(node, out, wmparc_to_native, 'source_file')
+
+        node, out = strat_pool.get_data('pipeline-fs_raw-average')
+        wf.connect(node, out, wmparc_to_native, 'target_file')
+
+        node, out = strat_pool.get_data('pipeline-fs_xfm')
+        wf.connect(node, out, wmparc_to_native, 'xfm_reg_file')
+
+        node, out = strat_pool.get_data('freesurfer-subject-dir')
+        wf.connect(node, out, wmparc_to_native, 'subjects_dir')
+
+        wf.connect(wmparc_to_native, 'transformed_file', wmparc_to_nifti, 'in_file')
+    
+    else:
+        
+        node, out = strat_pool.get_data('pipeline-fs_wmparc')
+        wf.connect(node, out, wmparc_to_nifti, 'in_file')
 
-    wf.connect(wmparc_to_native, 'transformed_file', wmparc_to_nifti, 'in_file')
+    wmparc_to_nifti.inputs.args = '-rt nearest'
 
     node, out = strat_pool.get_data('desc-preproc_T1w')
     wf.connect(node, out, wmparc_to_nifti, 'reslice_like')
@@ -2677,53 +2687,13 @@ def freesurfer_abcd_preproc(wf, cfg, strat_pool, pipe_num, opt=None):
     wf.connect(average_brain, 'out_stat',
                normalize_head, 'number')
 
-    if strat_pool.check_rpool('freesurfer-subject-dir'):
-        outputs = {
+    outputs = {
             'desc-restore_T1w': (fast_correction, 'outputspec.anat_restore'),
             'desc-restore-brain_T1w': (fast_correction,
                                        'outputspec.anat_brain_restore'),
             'pipeline-fs_desc-fast_biasfield': (fast_correction, 'outputspec.bias_field')}
-        return (wf, outputs)
-
-    else:
-        
-        ### recon-all -all step ###
-        reconall = pe.Node(interface=freesurfer.ReconAll(),
-                           name=f'anat_freesurfer_{pipe_num}',
-                           mem_gb=2.7)
-
-        sub_dir = cfg.pipeline_setup['working_directory']['path']
-        freesurfer_subject_dir = os.path.join(sub_dir,
-                                              'cpac_'+cfg['subject_id'],
-                                              f'anat_preproc_freesurfer_{pipe_num}',
-                                              'anat_freesurfer')
-
-        # create the directory for FreeSurfer node
-        if not os.path.exists(freesurfer_subject_dir):
-            os.makedirs(freesurfer_subject_dir)
-
-        reconall.inputs.directive = 'all'
-        reconall.inputs.subjects_dir = freesurfer_subject_dir
-        reconall.inputs.openmp = cfg.pipeline_setup['system_config']['num_OMP_threads']
-
-        wf.connect(normalize_head, 'out_file',
-                   reconall, 'T1_files')
-
-        wf, hemisphere_outputs = freesurfer_hemispheres(wf, reconall, pipe_num)
-
-        outputs = {
-        'desc-restore_T1w': (fast_correction, 'outputspec.anat_restore'),
-        'desc-restore-brain_T1w': (fast_correction,
-                                   'outputspec.anat_brain_restore'),
-        'pipeline-fs_desc-fast_biasfield': (fast_correction, 'outputspec.bias_field'),
-        'pipeline-fs_wmparc': (reconall, 'wmparc'),
-        'freesurfer-subject-dir': (reconall, 'subjects_dir'),
-        **hemisphere_outputs
-    }
-
     return (wf, outputs)
 
-
 # we're grabbing the postproc outputs and appending them to
 # the reconall outputs
 @docstring_parameter(postproc_outputs=str(flatten_list(freesurfer_abcd_preproc, 'outputs')

CPAC/pipeline/cpac_pipeline.py

@@ -23,7 +23,7 @@
 import copy
 import faulthandler
 
-from logging import getLogger
+from CPAC.utils.monitoring.custom_logging import getLogger
 from time import strftime
 
 import nipype
@@ -887,8 +887,7 @@ def build_anat_preproc_stack(rpool, cfg, pipeline_blocks=None):
                 ]
         elif cfg.anatomical_preproc['acpc_alignment']['acpc_target'] == 'whole-head':
             if (rpool.check_rpool('space-T1w_desc-brain_mask') and \
-                cfg.anatomical_preproc['acpc_alignment']['align_brain_mask']) or \
-                    cfg.surface_analysis['freesurfer']['run_reconall']:
+                cfg.anatomical_preproc['acpc_alignment']['align_brain_mask']):
                 acpc_blocks = [
                     acpc_align_head_with_mask
                     # outputs space-T1w_desc-brain_mask for later - keep the mask (the user provided)

CPAC/pipeline/engine.py

@@ -55,7 +55,7 @@
 
 from CPAC.resources.templates.lookup_table import lookup_identifier
 
-logger = logging.getLogger('nipype.workflow')
+logger = getLogger('nipype.workflow')
 verbose_logger = logging.getLogger('engine')
 
 
@@ -448,6 +448,7 @@ def get_strats(self, resources, debug=False):
         linked_resources = []
         resource_list = []
         if debug:
+            verbose_logger = getLogger('engine')
             verbose_logger.debug('\nresources: %s', resources)
         for resource in resources:
             # grab the linked-input tuples
@@ -471,6 +472,7 @@ def get_strats(self, resources, debug=False):
         variant_pool = {}
         len_inputs = len(resource_list)
         if debug:
+            verbose_logger = getLogger('engine')
             verbose_logger.debug('linked_resources: %s',
                                  linked_resources)
             verbose_logger.debug('resource_list: %s', resource_list)
@@ -498,6 +500,7 @@ def get_strats(self, resources, debug=False):
                                 f'NO-{val[0]}')
 
             if debug:
+                verbose_logger = getLogger('engine')
                 verbose_logger.debug('%s sub_pool: %s\n', resource, sub_pool)
             total_pool.append(sub_pool)
 
@@ -533,6 +536,7 @@ def get_strats(self, resources, debug=False):
                     strat_list_list.append(strat_list)
 
             if debug:
+                verbose_logger = getLogger('engine')
                 verbose_logger.debug('len(strat_list_list): %s\n',
                                      len(strat_list_list))
             for strat_list in strat_list_list:
@@ -1320,11 +1324,11 @@ def connect_block(self, wf, cfg, rpool):
                     option_val = option_config[-1]
                     if option_val in self.grab_tiered_dct(cfg, key_list[:-1]):
                         opts.append(option_val)                
-            else:                                                           #         AND, if there are multiple option-val's (in a list) in the docstring, it gets iterated below in 'for opt in option' etc. AND THAT'S WHEN YOU HAVE TO DELINEATE WITHIN THE NODE BLOCK CODE!!!
+            else:                                           #         AND, if there are multiple option-val's (in a list) in the docstring, it gets iterated below in 'for opt in option' etc. AND THAT'S WHEN YOU HAVE TO DELINEATE WITHIN THE NODE BLOCK CODE!!!
                 opts = [None]
             all_opts += opts
-
         for name, block_dct in self.node_blocks.items():    # <--- iterates over either the single node block in the sequence, or a list of node blocks within the list of node blocks, i.e. for option forking.
+            
             switch = self.check_null(block_dct['switch'])
             config = self.check_null(block_dct['config'])
             option_key = self.check_null(block_dct['option_key'])
@@ -1395,7 +1399,6 @@ def connect_block(self, wf, cfg, rpool):
                         switch = self.grab_tiered_dct(cfg, key_list)
                 if not isinstance(switch, list):
                     switch = [switch]
-
             if True in switch:
                 for pipe_idx, strat_pool in rpool.get_strats(
                         inputs, debug).items():         # strat_pool is a ResourcePool like {'desc-preproc_T1w': { 'json': info, 'data': (node, out) }, 'desc-brain_mask': etc.}
@@ -1418,7 +1421,6 @@ def connect_block(self, wf, cfg, rpool):
                                 input_name = interface[1]
                             strat_pool.copy_resource(input_name, interface[0])
                             replaced_inputs.append(interface[0])
-                        
                         try:
                             wf, outs = block_function(wf, cfg, strat_pool,
                                                       pipe_x, opt)
@@ -1441,6 +1443,7 @@ def connect_block(self, wf, cfg, rpool):
                             node_name = f'{node_name}_{opt["Name"]}'
 
                         if debug:
+                            verbose_logger = getLogger('engine')
                             verbose_logger.debug('\n=======================')
                             verbose_logger.debug('Node name: %s', node_name)
                             prov_dct = \
@@ -1539,7 +1542,6 @@ def connect_block(self, wf, cfg, rpool):
                                                               new_json_info,
                                                               pipe_idx,
                                                               pipe_x)
-
         return wf
 
 
@@ -1736,7 +1738,11 @@ def ingress_raw_anat_data(wf, rpool, cfg, data_paths, unique_id, part_id,
                     dl_dir=cfg.pipeline_setup['working_directory']['path'])
                 rpool.set_data(key, fs_ingress, 'outputspec.data',
                                {}, "", f"fs_{key}_ingress")
-
+            else:
+                warnings.warn(str(
+                        LookupError("\n[!] Path does not exist for "
+                                        f"{fullpath}.\n")))
+                
     return rpool
 
 
@@ -1776,6 +1782,7 @@ def ingress_raw_func_data(wf, rpool, cfg, data_paths, unique_id, part_id,
         # pylint: disable=protected-access
         wf._local_func_scans = local_func_scans
         if cfg.pipeline_setup['Debugging']['verbose']:
+            verbose_logger = getLogger('engine')
             verbose_logger.debug('local_func_scans: %s', local_func_scans)
     del local_func_scans
 

CPAC/resources/configs/pipeline_config_abcd-options.yml

@@ -36,8 +36,7 @@ surface_analysis:
     # Ingress freesurfer recon-all folder
     ingress_reconall: On
 
-    # FreeSurfer will run as part of configured brain extraction in this specific configuration, so FreeSurfer's independent run was automatically disabled.
-    run_reconall: Off
+    run_reconall: On
 
   # Run ABCD-HCP post FreeSurfer and fMRISurface pipeline
   post_freesurfer:

CPAC/surface/globals.py

@@ -1,4 +1,5 @@
 from logging import raiseExceptions
+from CPAC.utils.monitoring.custom_logging import log_subprocess
 import os
 import nipype.interfaces.utility as util
 from CPAC.utils.interfaces.function import Function
@@ -37,7 +38,6 @@ def run_surface(post_freesurfer_folder,
     """
 
     import os
-    import subprocess
 
     recon_all_path = os.path.join(freesurfer_folder, 'recon_all')
 
@@ -54,7 +54,7 @@ def run_surface(post_freesurfer_folder,
         '--hiresmesh', high_res_mesh, '--lowresmesh', low_res_mesh, \
         '--subcortgraylabels', subcortical_gray_labels, '--freesurferlabels', freesurfer_labels]
 
-    subprocess.check_output(cmd)
+    log_subprocess(cmd)
 
     # DCAN-HCP fMRISurface
     # https://github.com/DCAN-Labs/DCAN-HCP/blob/master/fMRISurface/GenericfMRISurfaceProcessingPipeline.sh
@@ -64,7 +64,7 @@ def run_surface(post_freesurfer_folder,
            scout_bold, '--lowresmesh', low_res_mesh, '--grayordinatesres',
            gray_ordinates_res, '--fmrires', fmri_res, '--smoothingFWHM',
            smooth_fwhm]
-    subprocess.check_output(cmd)
+    log_subprocess(cmd)
 
     dtseries = os.path.join(post_freesurfer_folder,
                             'MNINonLinear/Results/task-rest01/'

CPAC/surface/surf_preproc.py

@@ -22,7 +22,6 @@
 from warnings import warn
 import pkg_resources as p
 import yaml
-from CPAC.surface.globals import DOUBLERUN_GUARD_MESSAGE
 from CPAC.utils.utils import load_preconfig
 from .diff import dct_diff
 
@@ -118,18 +117,6 @@ def __init__(self, config_map=None):
                 # replace spaces with hyphens in Regressor 'Name's
                 regressor['Name'] = regressor['Name'].replace(' ', '-')
 
-        # Don't double-run FreeSurfer
-        try:
-            if 'FreeSurfer-ABCD' in config_map['anatomical_preproc'][
-                    'brain_extraction']['using']:
-                self.set_nested(config_map,
-                                ['surface_analysis', 'freesurfer', 
-                                 'run_reconall'],
-                                False)
-                warn(DOUBLERUN_GUARD_MESSAGE)
-        except (KeyError, TypeError):
-            pass
-
         config_map = schema(config_map)
 
         # remove 'FROM' before setting attributes now that it's imported

CPAC/utils/configuration/configuration.py

@@ -20,7 +20,7 @@
 from datetime import datetime
 from hashlib import sha1
 import yaml
-from CPAC.surface.globals import DOUBLERUN_GUARD_MESSAGE
+
 from CPAC.utils.configuration import Configuration, Preconfiguration, \
                                      preconfig_yaml
 from CPAC.utils.utils import set_nested_value, update_config_dict, \
@@ -87,18 +87,7 @@ def dump(self, new_dict, parents=None):
         -------
         str
         """
-        # Initialize variable for automatically updated value
-        if parents == ['surface_analysis', 'freesurfer'] or parents is None:
-            freesurfer_extraction = False
-            try:
-                brain_extractions = self.get_nested(
-                    new_dict, ['anatomical_preproc', 'brain_extraction',
-                               'using'])
-            except KeyError:
-                brain_extractions = []
-            if (brain_extractions is not None and
-                    'FreeSurfer-ABCD' in brain_extractions):
-                freesurfer_extraction = True
+    
         # SSOT FSLDIR
         try:  # Get from current config
             fsldir = self.get_nested(new_dict,
@@ -117,17 +106,6 @@ def dump(self, new_dict, parents=None):
             _dump = ['%YAML 1.1', '---']
             if 'pipeline_setup' not in new_dict:
                 new_dict['pipeline_setup'] = None
-            # Insert automatically-changed value in original dict
-            if freesurfer_extraction:
-                new_dict = set_nested_value(
-                    new_dict, ['surface_analysis', 'freesurfer',
-                    'run_reconall'], False)
-            ingress_keys = ['surface_analysis', 'freesurfer', 'ingress_reconall']
-            try:
-                self.get_nested(new_dict, ingress_keys)
-            except KeyError:
-                new_dict = set_nested_value(new_dict, ingress_keys,
-                                            self.get_nested(self._dict, ingress_keys))
         else:
             _dump = []
         # Prepare for indentation
@@ -150,12 +128,7 @@ def dump(self, new_dict, parents=None):
                 value = self.get_nested(new_dict, keys)
             except KeyError:  # exclude unincluded keys
                 continue
-            # Add comment for automatically changed value
-            if (keys == ['surface_analysis', 'freesurfer', 'run_reconall'] and
-                    freesurfer_extraction):
-                if comment is None:
-                    comment = []
-                comment.append(f'# {DOUBLERUN_GUARD_MESSAGE}')
+
             # Print comment if there's one above this key in the template
             if comment:
                 if key != 'pipeline_setup':