Merge remote-tracking branch 'upstream/master' into fix/search

satra · satra · commit d4ff0edda19c · 2018-02-05T09:46:02.000-05:00
* upstream/master: (23 commits)
  fix out_file template bugs
  FIX: Do not cache grab_exts
  fix: remove deprecated output from _list_outputs
  ENH: Reorder traits for better error message
  ENH: Add info to ImageFile for better error messages
  ENH: Simplify listifying logic
  Revert "changing in_files in RealignInputSpec"
  Revert "ENH: Re-enable spm.Realign to take lists of lists of files"
  ENH: Re-enable spm.Realign to take lists of lists of files
  FIX: argstr for mask in BlurToFWHM
  updated PR
  FIX: Default value for sbatch_args (SLURMGraph)
  BF: fix ortvec argstr
  do not access _id on nodes for logging
  minor fixes
  [skip ci] Add missing changelog entry
  redo changes
  make ignore_exception a class attribute
  improve message, fix error in logging
  add longer description to warning
  ...
diff --git a/CHANGES b/CHANGES
@@ -6,6 +6,7 @@ Upcoming release
 
 ###### [Full changelog](https://github.com/nipy/nipype/milestone/16?closed=1)
 
+* FIX: Change to interface workdir within ``Interface.run()`` instead Node (https://github.com/nipy/nipype/pull/2384)
 * FIX: PBS plugin submissions (https://github.com/nipy/nipype/pull/2344)
 * FIX: Graph plugins submissions (https://github.com/nipy/nipype/pull/2359)
 * FIX: Logging error if % in interface command (https://github.com/nipy/nipype/pull/2364)
diff --git a/doc/users/saving_workflows.rst b/doc/users/saving_workflows.rst
@@ -66,34 +66,34 @@ This will create a file "outputtestsave.py" with the following content:
 	inode = Node(IdentityInterface(fields=['a'], mandatory_inputs=True), name="inode")
 	# Node: testsave.testfunc
 	testfunc = Node(Function(input_names=['a'], output_names=['output']), name="testfunc")
+	testfunc.interface.ignore_exception = False
 	def testfunc_1(in1):
 	    """dummy func
 	    """
 	    out = in1 + 'foo' + "out1"
 	    return out
 
 	testfunc.inputs.function_str = getsource(testfunc_1)
-	testfunc.inputs.ignore_exception = False
 	testfunc.inputs.in1 = '-sub'
 	testsave.connect(inode, "a", testfunc, "in1")
 	# Node: testsave.bet2
 	bet2 = MapNode(BET(), iterfield=['infile'], name="bet2")
+	bet2.interface.ignore_exception = False
 	bet2.iterables = ('frac', [0.4, 0.5])
 	bet2.inputs.environ = {'FSLOUTPUTTYPE': 'NIFTI_GZ'}
-	bet2.inputs.ignore_exception = False
 	bet2.inputs.output_type = 'NIFTI_GZ'
 	bet2.terminal_output = 'stream'
 	# Node: testsave.bet
 	bet = Node(BET(), name="bet")
+	bet.interface.ignore_exception = False
 	bet.iterables = ('frac', [0.3, 0.4])
 	bet.inputs.environ = {'FSLOUTPUTTYPE': 'NIFTI_GZ'}
-	bet.inputs.ignore_exception = False
 	bet.inputs.output_type = 'NIFTI_GZ'
 	bet.terminal_output = 'stream'
 	# Node: testsave.maths
 	maths = Node(ImageMaths(), name="maths")
+	maths.interface.ignore_exception = False
 	maths.inputs.environ = {'FSLOUTPUTTYPE': 'NIFTI_GZ'}
-	maths.inputs.ignore_exception = False
 	maths.inputs.output_type = 'NIFTI_GZ'
 	maths.terminal_output = 'stream'
 	testsave.connect(bet2, ('mask_file', func), maths, "in_file2")
diff --git a/nipype/algorithms/icc.py b/nipype/algorithms/icc.py
@@ -80,7 +80,6 @@ def _run_interface(self, runtime):
     def _list_outputs(self):
         outputs = self._outputs().get()
         outputs['icc_map'] = os.path.abspath('icc_map.nii')
-        outputs['sessions_F_map'] = os.path.abspath('sessions_F_map.nii')
         outputs['session_var_map'] = os.path.abspath('session_var_map.nii')
         outputs['subject_var_map'] = os.path.abspath('subject_var_map.nii')
         return outputs
diff --git a/nipype/caching/memory.py b/nipype/caching/memory.py
@@ -94,7 +94,7 @@ def __call__(self, **kwargs):
         return out
 
     def __repr__(self):
-        return '{}({}.{}}, base_dir={})'.format(
+        return '{}({}.{}), base_dir={})'.format(
             self.__class__.__name__, self.interface.__module__,
             self.interface.__name__, self.base_dir)
 
diff --git a/nipype/interfaces/afni/model.py b/nipype/interfaces/afni/model.py
@@ -142,7 +142,7 @@ class DeconvolveInputSpec(AFNICommandInputSpec):
         desc='this option lets you input a rectangular array of 1 or more '
         'baseline vectors from a file. This method is a fast way to '
         'include a lot of baseline regressors in one step. ',
-        argstr='ortvec %s')
+        argstr='-ortvec %s %s')
     x1D = File(desc='specify name for saved X matrix', argstr='-x1D %s')
     x1D_stop = traits.Bool(
         desc='stop running after writing .xmat.1D file', argstr='-x1D_stop')
diff --git a/nipype/interfaces/afni/preprocess.py b/nipype/interfaces/afni/preprocess.py
@@ -930,7 +930,7 @@ class BlurToFWHMInputSpec(AFNICommandInputSpec):
     mask = File(
         desc='Mask dataset, if desired. Voxels NOT in mask will be set to zero '
         'in output.',
-        argstr='-blurmaster %s',
+        argstr='-mask %s',
         exists=True)
 
 
diff --git a/nipype/interfaces/afni/tests/test_auto_BlurToFWHM.py b/nipype/interfaces/afni/tests/test_auto_BlurToFWHM.py
@@ -23,7 +23,7 @@ def test_BlurToFWHM_inputs():
             argstr='-input %s',
             mandatory=True,
         ),
-        mask=dict(argstr='-blurmaster %s', ),
+        mask=dict(argstr='-mask %s', ),
         num_threads=dict(
             nohash=True,
             usedefault=True,
diff --git a/nipype/interfaces/afni/tests/test_auto_Deconvolve.py b/nipype/interfaces/afni/tests/test_auto_Deconvolve.py
@@ -73,7 +73,7 @@ def test_Deconvolve_inputs():
             nohash=True,
             usedefault=True,
         ),
-        ortvec=dict(argstr='ortvec %s', ),
+        ortvec=dict(argstr='-ortvec %s %s', ),
         out_file=dict(argstr='-bucket %s', ),
         outputtype=dict(),
         polort=dict(argstr='-polort %d', ),
diff --git a/nipype/interfaces/afni/tests/test_auto_Merge.py b/nipype/interfaces/afni/tests/test_auto_Merge.py
@@ -32,7 +32,7 @@ def test_Merge_inputs():
         ),
         out_file=dict(
             argstr='-prefix %s',
-            name_source='in_file',
+            name_source='in_files',
             name_template='%s_merge',
         ),
         outputtype=dict(),
diff --git a/nipype/interfaces/afni/tests/test_auto_Zcat.py b/nipype/interfaces/afni/tests/test_auto_Zcat.py
@@ -36,7 +36,8 @@ def test_Zcat_inputs():
         ),
         out_file=dict(
             argstr='-prefix %s',
-            name_template='zcat',
+            name_source='in_files',
+            name_template='%s_zcat',
         ),
         outputtype=dict(),
         terminal_output=dict(
diff --git a/nipype/interfaces/afni/utils.py b/nipype/interfaces/afni/utils.py
@@ -1434,7 +1434,7 @@ class MergeInputSpec(AFNICommandInputSpec):
         name_template='%s_merge',
         desc='output image file name',
         argstr='-prefix %s',
-        name_source='in_file')
+        name_source='in_files')
     doall = traits.Bool(
         desc='apply options to all sub-bricks in dataset', argstr='-doall')
     blurfwhm = traits.Int(
@@ -2635,9 +2635,10 @@ class ZcatInputSpec(AFNICommandInputSpec):
         mandatory=True,
         copyfile=False)
     out_file = File(
-        name_template='zcat',
+        name_template='%s_zcat',
         desc='output dataset prefix name (default \'zcat\')',
-        argstr='-prefix %s')
+        argstr='-prefix %s',
+        name_source='in_files')
     datum = traits.Enum(
         'byte',
         'short',
diff --git a/nipype/interfaces/base/core.py b/nipype/interfaces/base/core.py
@@ -170,12 +170,14 @@ class BaseInterface(Interface):
     references_ = []
     resource_monitor = True  # Enabled for this interface IFF enabled in the config
 
-    def __init__(self, from_file=None, resource_monitor=None, **inputs):
+    def __init__(self, from_file=None, resource_monitor=None,
+                 ignore_exception=False, **inputs):
         if not self.input_spec:
             raise Exception(
                 'No input_spec in class: %s' % self.__class__.__name__)
 
         self.inputs = self.input_spec(**inputs)
+        self.ignore_exception = ignore_exception
 
         if resource_monitor is not None:
             self.resource_monitor = resource_monitor
@@ -470,7 +472,6 @@ def run(self, cwd=None, **inputs):
         os.chdir(cwd)  # Change to the interface wd
 
         enable_rm = config.resource_monitor and self.resource_monitor
-        force_raise = not getattr(self.inputs, 'ignore_exception', False)
         self.inputs.trait_set(**inputs)
         self._check_mandatory_inputs()
         self._check_version_requirements(self.inputs)
@@ -530,7 +531,7 @@ def run(self, cwd=None, **inputs):
             runtime.traceback_args = ('\n'.join(
                 ['%s' % arg for arg in exc_args]), )
 
-            if force_raise:
+            if not self.ignore_exception:
                 raise
         finally:
             # This needs to be done always
diff --git a/nipype/interfaces/base/traits_extension.py b/nipype/interfaces/base/traits_extension.py
@@ -217,6 +217,16 @@ def __init__(self,
         super(ImageFile, self).__init__(value, filter, auto_set, entries,
                                         exists, **metadata)
 
+    def info(self):
+        existing = 'n existing' if self.exists else ''
+        comma = ',' if self.exists and not self.allow_compressed else ''
+        uncompressed = ' uncompressed' if not self.allow_compressed else ''
+        with_ext = ' (valid extensions: [{}])'.format(
+            ', '.join(self.grab_exts())) if self.types else ''
+        return 'a{existing}{comma}{uncompressed} file{with_ext}'.format(
+            existing=existing, comma=comma, uncompressed=uncompressed,
+            with_ext=with_ext)
+
     def grab_exts(self):
         # TODO: file type validation
         exts = []
@@ -243,11 +253,11 @@ def validate(self, object, name, value):
         """
         validated_value = super(ImageFile, self).validate(object, name, value)
         if validated_value and self.types:
-            self._exts = self.grab_exts()
-            if not any(validated_value.endswith(x) for x in self._exts):
+            _exts = self.grab_exts()
+            if not any(validated_value.endswith(x) for x in _exts):
                 raise TraitError(
                     args="{} is not included in allowed types: {}".format(
-                        validated_value, ', '.join(self._exts)))
+                        validated_value, ', '.join(_exts)))
         return validated_value
 
 
@@ -322,15 +332,11 @@ def validate(self, object, name, value):
 
         newvalue = value
 
+        inner_trait = self.inner_traits()[0]
         if not isinstance(value, list) \
-            or (self.inner_traits() and
-                isinstance(self.inner_traits()[0].trait_type,
-                           traits.List) and not
-                isinstance(self.inner_traits()[0].trait_type,
-                           InputMultiPath) and
-                isinstance(value, list) and
-                value and not
-                isinstance(value[0], list)):
+            or (isinstance(inner_trait.trait_type, traits.List) and
+                not isinstance(inner_trait.trait_type, InputMultiPath) and
+                not isinstance(value[0], list)):
             newvalue = [value]
         value = super(MultiPath, self).validate(object, name, newvalue)
 
diff --git a/nipype/interfaces/spm/preprocess.py b/nipype/interfaces/spm/preprocess.py
@@ -125,7 +125,8 @@ def _list_outputs(self):
 
 class RealignInputSpec(SPMCommandInputSpec):
     in_files = InputMultiPath(
-        ImageFileSPM(exists=True),
+        traits.Either(ImageFileSPM(exists=True),
+                      traits.List(ImageFileSPM(exists=True))),
         field='data',
         mandatory=True,
         copyfile=True,
diff --git a/nipype/pipeline/plugins/base.py b/nipype/pipeline/plugins/base.py
@@ -7,13 +7,13 @@
                         absolute_import)
 from builtins import range, object, open
 
+import sys
 from copy import deepcopy
 from glob import glob
 import os
 import shutil
-import sys
 from time import sleep, time
-from traceback import format_exc
+from traceback import format_exception
 
 import numpy as np
 import scipy.sparse as ssp
@@ -159,7 +159,7 @@ def run(self, graph, config, updatehash=False):
                             graph,
                             result={
                                 'result': None,
-                                'traceback': format_exc()
+                                'traceback': '\n'.join(format_exception(*sys.exc_info()))
                             }))
                 else:
                     if result:
@@ -244,7 +244,7 @@ def _submit_mapnode(self, jobid):
         mapnodesubids = self.procs[jobid].get_subnodes()
         numnodes = len(mapnodesubids)
         logger.debug('Adding %d jobs for mapnode %s', numnodes,
-                     self.procs[jobid]._id)
+                     self.procs[jobid])
         for i in range(numnodes):
             self.mapnodesubids[self.depidx.shape[0] + i] = jobid
         self.procs.extend(mapnodesubids)
@@ -274,7 +274,7 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                 slots = None
             else:
                 slots = max(0, self.max_jobs - num_jobs)
-            logger.debug('Slots available: %s' % slots)
+            logger.debug('Slots available: %s', slots)
             if (num_jobs >= self.max_jobs) or (slots == 0):
                 break
 
@@ -303,14 +303,14 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                     self.proc_done[jobid] = True
                     self.proc_pending[jobid] = True
                     # Send job to task manager and add to pending tasks
-                    logger.info('Submitting: %s ID: %d' %
-                                (self.procs[jobid]._id, jobid))
+                    logger.info('Submitting: %s ID: %d',
+                                self.procs[jobid], jobid)
                     if self._status_callback:
                         self._status_callback(self.procs[jobid], 'start')
 
                     if not self._local_hash_check(jobid, graph):
                         if self.procs[jobid].run_without_submitting:
-                            logger.debug('Running node %s on master thread' %
+                            logger.debug('Running node %s on master thread',
                                          self.procs[jobid])
                             try:
                                 self.procs[jobid].run()
@@ -327,8 +327,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                                 self.proc_pending[jobid] = False
                             else:
                                 self.pending_tasks.insert(0, (tid, jobid))
-                    logger.info('Finished submitting: %s ID: %d' %
-                                (self.procs[jobid]._id, jobid))
+                    logger.info('Finished submitting: %s ID: %d',
+                                self.procs[jobid], jobid)
             else:
                 break
 
@@ -337,22 +337,38 @@ def _local_hash_check(self, jobid, graph):
                 self.procs[jobid].config['execution']['local_hash_check']):
             return False
 
-        cached, updated = self.procs[jobid].is_cached()
+        try:
+            cached, updated = self.procs[jobid].is_cached()
+        except Exception:
+            logger.warning(
+                'Error while checking node hash, forcing re-run. '
+                'Although this error may not prevent the workflow from running, '
+                'it could indicate a major problem. Please report a new issue '
+                'at https://github.com/nipy/nipype/issues adding the following '
+                'information:\n\n\tNode: %s\n\tInterface: %s.%s\n\tTraceback:\n%s',
+                self.procs[jobid],
+                self.procs[jobid].interface.__module__,
+                self.procs[jobid].interface.__class__.__name__,
+                '\n'.join(format_exception(*sys.exc_info()))
+            )
+            return False
+
         logger.debug('Checking hash "%s" locally: cached=%s, updated=%s.',
-                    self.procs[jobid].fullname, cached, updated)
+                     self.procs[jobid], cached, updated)
         overwrite = self.procs[jobid].overwrite
-        always_run = self.procs[jobid]._interface.always_run
+        always_run = self.procs[jobid].interface.always_run
 
         if cached and updated and (overwrite is False or
                                    overwrite is None and not always_run):
             logger.debug('Skipping cached node %s with ID %s.',
-                         self.procs[jobid]._id, jobid)
+                         self.procs[jobid], jobid)
             try:
                 self._task_finished_cb(jobid, cached=True)
                 self._remove_node_dirs()
             except Exception:
-                logger.debug('Error skipping cached node %s (%s).',
-                             self.procs[jobid]._id, jobid)
+                logger.debug('Error skipping cached node %s (%s).\n\n%s',
+                             self.procs[jobid], jobid,
+                             '\n'.join(format_exception(*sys.exc_info())))
                 self._clean_queue(jobid, graph)
                 self.proc_pending[jobid] = False
             return True
@@ -364,7 +380,7 @@ def _task_finished_cb(self, jobid, cached=False):
         This is called when a job is completed.
         """
         logger.info('[Job %d] %s (%s).', jobid, 'Cached'
-                    if cached else 'Completed', self.procs[jobid].fullname)
+                    if cached else 'Completed', self.procs[jobid])
         if self._status_callback:
             self._status_callback(self.procs[jobid], 'end')
         # Update job and worker queues
@@ -481,7 +497,7 @@ def _get_result(self, taskid):
                                      taskid, timeout, node_dir))
                 raise IOError(error_message)
             except IOError as e:
-                result_data['traceback'] = format_exc()
+                result_data['traceback'] = '\n'.join(format_exception(*sys.exc_info()))
         else:
             results_file = glob(os.path.join(node_dir, 'result_*.pklz'))[0]
             result_data = loadpkl(results_file)
diff --git a/nipype/pipeline/plugins/slurmgraph.py b/nipype/pipeline/plugins/slurmgraph.py
@@ -44,6 +44,7 @@ class SLURMGraphPlugin(GraphPluginBase):
     _template = "#!/bin/bash"
 
     def __init__(self, **kwargs):
+        self._sbatch_args = ''
         if 'plugin_args' in kwargs and kwargs['plugin_args']:
             if 'retry_timeout' in kwargs['plugin_args']:
                 self._retry_timeout = kwargs['plugin_args']['retry_timeout']
diff --git a/nipype/utils/filemanip.py b/nipype/utils/filemanip.py
diff --git a/nipype/workflows/smri/ants/ANTSBuildTemplate.py b/nipype/workflows/smri/ants/ANTSBuildTemplate.py
diff --git a/nipype/workflows/smri/ants/antsRegistrationBuildTemplate.py b/nipype/workflows/smri/ants/antsRegistrationBuildTemplate.py
