Merge branch 'master' into enh/versioning

* master:
  fix: deprecation warning does not reset trait to Undefined - will set new value
  add support for template files
  API fix for somaworkflow
  PEP8
  Further refactoring
  Reafactoring - less redundant code between SGE and PBS.
  Respect custom execution parameters in graph based models.
  DOC: Added docstring for MpiCommandLine with example usage.
  ENH: Provide base interface for MPI capable command line software.

Author: satra

nipype/interfaces/base.py

@@ -268,6 +268,7 @@ def __init__(self, interface, runtime, inputs=None, outputs=None):
     def version(self):
         return self._version
 
+
 class BaseTraitedSpec(traits.HasTraits):
     """Provide a few methods necessary to support nipype interface api
 
@@ -374,7 +375,6 @@ def _deprecated_warn(self, obj, name, old, new):
                                   self.__class__.__name__.split('InputSpec')[0])
             msg2 = ('Will be removed or raise an error as of release %s') % \
                                                            trait_spec.deprecated
-            self.trait_set(trait_change_notify=False, **{'%s' % name: Undefined})
             if trait_spec.new_name:
                 if trait_spec.new_name not in self.copyable_trait_names():
                     raise TraitError(msg1 + ' Replacement trait %s not found' %
@@ -387,7 +387,12 @@ def _deprecated_warn(self, obj, name, old, new):
                 raise TraitError(msg)
             else:
                 warn(msg)
-
+                if trait_spec.new_name:
+                    warn('Unsetting %s and setting %s.' % (name,
+                                                           trait_spec.new_name))
+                    self.trait_set(trait_change_notify=False,
+                                   **{'%s' % name: Undefined,
+                                      '%s' % trait_spec.new_name: new})
 
     def _hash_infile(self, adict, key):
         """ Inject file hashes into adict[key]"""
@@ -1328,6 +1333,46 @@ def _gen_filename(self, name):
     def _gen_outfilename(self):
         raise NotImplementedError
 
+class MpiCommandLineInputSpec(CommandLineInputSpec):
+    use_mpi = traits.Bool(False, 
+                          desc="Whether or not to run the command with mpiexec",
+                          usedefault=True)
+    n_procs = traits.Int(desc="Num processors to specify to mpiexec. Do not "
+                         "specify if this is managed externally (e.g. through "
+                         "SGE)")
+    
+
+class MpiCommandLine(CommandLine):
+    '''Implements functionality to interact with command line programs 
+    that can be run with MPI (i.e. using 'mpiexec'). 
+
+    Examples
+    --------
+    >>> from nipype.interfaces.base import MpiCommandLine
+    >>> mpi_cli = MpiCommandLine(command='my_mpi_prog')
+    >>> mpi_cli.inputs.args = '-v'
+    >>> mpi_cli.cmdline
+    'my_mpi_prog -v'
+    
+    >>> mpi_cli.inputs.use_mpi = True
+    >>> mpi_cli.inputs.n_procs = 8
+    >>> mpi_cli.cmdline
+    
+    'mpiexec -n 8 my_mpi_prog -v'
+    '''
+    input_spec = MpiCommandLineInputSpec
+    
+    @property
+    def cmdline(self):
+        """Adds 'mpiexec' to begining of command"""
+        result = []
+        if self.inputs.use_mpi:
+            result.append('mpiexec')
+            if self.inputs.n_procs: 
+                result.append('-n %d' % self.inputs.n_procs)
+        result.append(super(MpiCommandLine, self).cmdline)
+        return ' '.join(result)
+
 class SEMLikeCommandLine(CommandLine):
     """By default in SEM derived interface all outputs have corresponding inputs.
     However, some SEM commands create outputs that are not defined in the XML.

nipype/interfaces/tests/test_base.py

@@ -139,7 +139,7 @@ class DeprecationSpec1(nib.TraitedSpec):
     set_foo = lambda : setattr(spec_instance, 'foo', 1)
     yield assert_raises, nib.TraitError, set_foo
     class DeprecationSpec1numeric(nib.TraitedSpec):
-        foo = nib.traits.Int(deprecated=0.1)
+        foo = nib.traits.Int(deprecated='0.1')
     spec_instance = DeprecationSpec1numeric()
     set_foo = lambda : setattr(spec_instance, 'foo', 1)
     yield assert_raises, nib.TraitError, set_foo
@@ -158,6 +158,18 @@ class DeprecationSpec3(nib.TraitedSpec):
     except nib.TraitError:
         not_raised = False
     yield assert_true, not_raised
+    class DeprecationSpec3(nib.TraitedSpec):
+        foo = nib.traits.Int(deprecated='1000', new_name='bar')
+        bar = nib.traits.Int()
+    spec_instance = DeprecationSpec3()
+    not_raised = True
+    try:
+        spec_instance.foo = 1
+    except nib.TraitError:
+        not_raised = False
+    yield assert_true, not_raised
+    yield assert_equal, spec_instance.foo, Undefined
+    yield assert_equal, spec_instance.bar, 1
 
 def checknose():
     """check version of nose for known incompatability"""

nipype/pipeline/plugins/base.py

@@ -218,7 +218,7 @@ def run(self, graph, config, updatehash=False):
         # setup polling - TODO: change to threaded model
         notrun = []
         while np.any(self.proc_done == False) | \
-              np.any(self.proc_pending == True):
+                    np.any(self.proc_pending == True):
             toappend = []
             # trigger callbacks for any pending results
             while self.pending_tasks:
@@ -297,11 +297,12 @@ def _submit_mapnode(self, jobid):
         self.procs.extend(mapnodesubids)
         self.depidx = ssp.vstack((self.depidx,
                                   ssp.lil_matrix(np.zeros((numnodes,
-                                                    self.depidx.shape[1])))),
+                                                           self.depidx.shape[1])))),
                                  'lil')
         self.depidx = ssp.hstack((self.depidx,
-                                  ssp.lil_matrix(np.zeros((self.depidx.shape[0],
-                                                           numnodes)))),
+                                  ssp.lil_matrix(
+                                      np.zeros((self.depidx.shape[0],
+                                                numnodes)))),
                                  'lil')
         self.depidx[-numnodes:, jobid] = 1
         self.proc_done = np.concatenate((self.proc_done,
@@ -315,7 +316,7 @@ def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
         """
         while np.any(self.proc_done == False):
             # Check to see if a job is available
-            jobids = np.flatnonzero((self.proc_done == False) & \
+            jobids = np.flatnonzero((self.proc_done == False) &
                                     (self.depidx.sum(axis=0) == 0).__array__())
             if len(jobids) > 0:
                 # send all available jobs
@@ -336,20 +337,21 @@ def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
                     self.proc_done[jobid] = True
                     self.proc_pending[jobid] = True
                     # Send job to task manager and add to pending tasks
-                    logger.info('Executing: %s ID: %d' % \
-                                    (self.procs[jobid]._id, jobid))
+                    logger.info('Executing: %s ID: %d' %
+                               (self.procs[jobid]._id, jobid))
                     if self._status_callback:
                         self._status_callback(self.procs[jobid], 'start')
                     continue_with_submission = True
                     if str2bool(self.procs[jobid].config['execution']['local_hash_check']):
                         logger.debug('checking hash locally')
                         try:
-                            hash_exists, _, _, _ = self.procs[jobid].hash_exists()
+                            hash_exists, _, _, _ = self.procs[
+                                jobid].hash_exists()
                             logger.debug('Hash exists %s' % str(hash_exists))
                             if (hash_exists and
-                            (self.procs[jobid].overwrite == False or
-                             (self.procs[jobid].overwrite == None and
-                              not self.procs[jobid]._interface.always_run))):
+                               (self.procs[jobid].overwrite == False or
+                               (self.procs[jobid].overwrite == None and
+                                    not self.procs[jobid]._interface.always_run))):
                                 continue_with_submission = False
                                 self._task_finished_cb(jobid)
                                 self._remove_node_dirs()
@@ -385,7 +387,7 @@ def _task_finished_cb(self, jobid):
 
         This is called when a job is completed.
         """
-        logger.info('[Job finished] jobname: %s jobid: %d' % \
+        logger.info('[Job finished] jobname: %s jobid: %d' %
                     (self.procs[jobid]._id, jobid))
         if self._status_callback:
             self._status_callback(self.procs[jobid], 'end')
@@ -431,7 +433,7 @@ def _remove_node_dirs(self):
                     self.refidx[idx, idx] = -1
                     outdir = self.procs[idx]._output_directory()
                     logger.info(('[node dependencies finished] '
-                                 'removing node: %s from directory %s') % \
+                                 'removing node: %s from directory %s') %
                                 (self.procs[idx]._id, outdir))
                     shutil.rmtree(outdir)
 
@@ -562,9 +564,26 @@ def run(self, graph, config, updatehash=False):
                                            store_exception=False))
             dependencies[idx] = [nodes.index(prevnode) for prevnode in
                                  graph.predecessors(node)]
-        self._submit_graph(pyfiles, dependencies)
+        self._submit_graph(pyfiles, dependencies, nodes)
+
+    def _get_args(self, node, keywords):
+        values = ()
+        for keyword in keywords:
+            value = getattr(self, "_" + keyword)
+            if keyword == "template" and os.path.isfile(value):
+                value = open(value).read()
+            if hasattr(node, "plugin_args") and isinstance(node.plugin_args, dict) and keyword in node.plugin_args:
+                    if keyword == "template" and os.path.isfile(node.plugin_args[keyword]):
+                        tmp_value = open(node.plugin_args[keyword]).read()
+                    if 'overwrite' in node.plugin_args and node.plugin_args['overwrite']:
+                        value = tmp_value
+                    else:
+                        value += tmp_value
+            else:
+                values += (value, )
+        return values
 
-    def _submit_graph(self, pyfiles, dependencies):
+    def _submit_graph(self, pyfiles, dependencies, nodes):
         """
         pyfiles: list of files corresponding to a topological sort
         dependencies: dictionary of dependencies based on the toplogical sort

nipype/pipeline/plugins/dagman.py

@@ -49,7 +49,7 @@ def __init__(self, **kwargs):
                 self._dagman_args = plugin_args['dagman_args']
         super(CondorDAGManPlugin, self).__init__(**kwargs)
 
-    def _submit_graph(self, pyfiles, dependencies):
+    def _submit_graph(self, pyfiles, dependencies, nodes):
         # location of all scripts, place dagman output in here too
         batch_dir, _ = os.path.split(pyfiles[0])
         # DAG description filename
@@ -58,27 +58,30 @@ def _submit_graph(self, pyfiles, dependencies):
             # loop over all scripts, create submit files, and define them
             # as jobs in the DAG
             for idx, pyscript in enumerate(pyfiles):
+                node = nodes[idx]
+                template, submit_specs = self._get_args(
+                    node, ["template", "submit_specs"])
                 # XXX redundant with previous value? or could it change between
                 # scripts?
                 batch_dir, name = os.path.split(pyscript)
                 name = '.'.join(name.split('.')[:-1])
                 submitspec = '\n'.join(
-                                (self._template,
-                                 'executable = %s' % sys.executable,
-                                 'arguments = %s' % pyscript,
-                                 'output = %s' % os.path.join(batch_dir,
-                                                             '%s.out' % name),
-                                 'error = %s' % os.path.join(batch_dir,
-                                                             '%s.err' % name),
-                                 'log = %s' % os.path.join(batch_dir,
-                                                           '%s.log' % name),
-                                 'getenv = True',
-                                 self._submit_specs,
-                                 'queue'
-                                 ))
+                    (template,
+                     'executable = %s' % sys.executable,
+                     'arguments = %s' % pyscript,
+                     'output = %s' % os.path.join(batch_dir,
+                                                  '%s.out' % name),
+                     'error = %s' % os.path.join(batch_dir,
+                                                 '%s.err' % name),
+                     'log = %s' % os.path.join(batch_dir,
+                                               '%s.log' % name),
+                     'getenv = True',
+                     submit_specs,
+                     'queue'
+                     ))
                 # write submit spec for this job
                 submitfile = os.path.join(batch_dir,
-                                               '%s.submit' % name)
+                                          '%s.submit' % name)
                 with open(submitfile, 'wt') as submitfileprt:
                     submitfileprt.writelines(submitspec)
                     submitfileprt.close()
@@ -98,4 +101,3 @@ def _submit_graph(self, pyfiles, dependencies):
                                                     self._dagman_args)
         cmd.run()
         logger.info('submitted all jobs to Condor DAGMan')
-

nipype/pipeline/plugins/pbsgraph.py

@@ -7,9 +7,10 @@
 from .base import (GraphPluginBase, logger)
 
 from ...interfaces.base import CommandLine
+from .sgegraph import SGEGraphPlugin
 
 
-class PBSGraphPlugin(GraphPluginBase):
+class PBSGraphPlugin(SGEGraphPlugin):
     """Execute using PBS/Torque
 
     The plugin_args input to run can be used to control the SGE execution.
@@ -20,31 +21,23 @@ class PBSGraphPlugin(GraphPluginBase):
                   qsub call
 
     """
-
-    def __init__(self, **kwargs):
-        self._template = """
+    _template = """
 #PBS -V
-        """
-        self._qsub_args = None
-        if 'plugin_args' in kwargs:
-            plugin_args = kwargs['plugin_args']
-            if 'template' in plugin_args:
-                self._template = plugin_args['template']
-                if os.path.isfile(self._template):
-                    self._template = open(self._template).read()
-            if 'qsub_args' in plugin_args:
-                self._qsub_args = plugin_args['qsub_args']
-        super(PBSGraphPlugin, self).__init__(**kwargs)
+"""
 
-    def _submit_graph(self, pyfiles, dependencies):
+    def _submit_graph(self, pyfiles, dependencies, nodes):
         batch_dir, _ = os.path.split(pyfiles[0])
         submitjobsfile = os.path.join(batch_dir, 'submit_jobs.sh')
         with open(submitjobsfile, 'wt') as fp:
             fp.writelines('#!/usr/bin/env sh\n')
             for idx, pyscript in enumerate(pyfiles):
+                node = nodes[idx]
+                template, qsub_args = self._get_args(
+                    node, ["template", "qsub_args"])
+
                 batch_dir, name = os.path.split(pyscript)
                 name = '.'.join(name.split('.')[:-1])
-                batchscript = '\n'.join((self._template,
+                batchscript = '\n'.join((template,
                                          '%s %s' % (sys.executable, pyscript)))
                 batchscriptfile = os.path.join(batch_dir,
                                                'batchscript_%s.sh' % name)
@@ -53,14 +46,14 @@ def _submit_graph(self, pyfiles, dependencies):
                     batchfp.close()
                 deps = ''
                 if idx in dependencies:
-                    values = ['$job%05d' % jobid for jobid in dependencies[idx]]
+                    values = ['$job%05d' %
+                              jobid for jobid in dependencies[idx]]
                     if len(values):
                         deps = '-W depend=afterok:%s' % ':'.join(values)
                 fp.writelines('job%05d=`qsub %s %s %s`\n' % (idx, deps,
-                                                             self._qsub_args,
+                                                             qsub_args,
                                                              batchscriptfile))
         cmd = CommandLine('sh', environ=os.environ.data)
         cmd.inputs.args = '%s' % submitjobsfile
         cmd.run()
         logger.info('submitted all jobs to queue')
-