Cleaned up some of the code to PEP8 and checked for errors

Author: pintohutch

nipype/interfaces/afni/__init__.py

@@ -8,8 +8,8 @@
 
 from .base import Info
 from .preprocess import (To3D, Refit, Resample, TStat, Automask, Volreg, Merge,
-                         ZCutUp, Calc, TShift, Warp, Detrend, Despike,
-                         Copy, Fourier, Allineate, Maskave, SkullStrip, TCat, Fim,
+                         ZCutUp, Calc, TShift, Warp, Detrend, Despike, Copy,
+                         Fourier, Allineate, Maskave, SkullStrip, TCat, Fim,
                          BlurInMask, Autobox, TCorrMap, Bandpass, Retroicor,
                          TCorrelate, TCorr1D, BrickStat, ROIStats, AutoTcorrelate,
                          AFNItoNIFTI, Eval, Means)

nipype/interfaces/afni/preprocess.py

@@ -180,7 +180,6 @@ class RefitInputSpec(CommandLineInputSpec):
                         ' template type, e.g. TLRC, MNI, ORIG')
 
 
-
 class Refit(CommandLine):
     """Changes some of the information inside a 3D dataset's header
 

nipype/interfaces/base.py

@@ -1032,7 +1032,6 @@ def run(self, **inputs):
         self._check_mandatory_inputs()
         self._check_version_requirements(self.inputs)
         interface = self.__class__
-
         # initialize provenance tracking
         env = deepcopy(dict(os.environ))
         runtime = Bunch(cwd=os.getcwd(),
@@ -1207,8 +1206,18 @@ def _read(self, drain):
 
 # Get number of threads for process
 def _get_num_threads(proc):
-    '''
-    '''
+    """Function to get the number of threads a process is using
+
+    Parameters
+    ----------
+    proc : psutil.Process instance
+        the process to evaluate thead usage of
+
+    Returns
+    -------
+    num_threads : int
+        the number of threads that the process is using
+    """
 
     # Import packages
     import psutil
@@ -1223,14 +1232,32 @@ def _get_num_threads(proc):
     except psutil.NoSuchProcess:
         pass
 
+    # Return the number of threads found
     return num_threads
 
 
 # Get max resources used for process
 def _get_max_resources_used(proc, mem_mb, num_threads, poll=False):
-    '''
-    docstring
-    '''
+    """Function to get the RAM and threads usage of a process
+
+    Paramters
+    ---------
+    proc : subprocess.Popen instance
+        the process to profile
+    mem_mb : float
+        the high memory watermark so far during process execution (in MB)
+    num_threads: int
+        the high thread watermark so far during process execution
+    poll : boolean
+        whether to poll the process or not
+
+    Returns
+    -------
+    mem_mb : float
+        the new high memory watermark of process (MB)
+    num_threads : float
+        the new high thread watermark of process
+    """
 
     # Import packages
     from memory_profiler import _get_memory
@@ -1264,9 +1291,8 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
         import psutil
         runtime_profile = True
     except ImportError as exc:
-        logger.info('Unable to import packages needed for runtime '\
-                    'profiling. Turning off runtime profiler.\n'\
-                    'Error: %s' % exc)
+        logger.info('Unable to import packages needed for runtime profiling. '\
+                    'Turning off runtime profiler.\nError: %s' % exc)
         runtime_profile = False
 
     # Init variables
@@ -1305,7 +1331,6 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     # Init variables for memory profiling
     mem_mb = -1
     num_threads = -1
-    interval = 1
 
     if output == 'stream':
         streams = [Stream('stdout', proc.stdout), Stream('stderr', proc.stderr)]

nipype/interfaces/tests/test_runtime_profiler.py

@@ -14,6 +14,7 @@
 # UseResources inputspec
 class UseResourcesInputSpec(CommandLineInputSpec):
     '''
+    use_resources cmd interface inputspec
     '''
 
     # Init attributes
@@ -26,6 +27,7 @@ class UseResourcesInputSpec(CommandLineInputSpec):
 # UseResources interface
 class UseResources(CommandLine):
     '''
+    use_resources cmd interface
     '''
 
     # Import packages

nipype/interfaces/tests/use_resources

@@ -19,6 +19,7 @@ def use_gb_ram(num_gb):
     # Eat 1 GB of memory for 1 second
     gb_str = ' ' * int(num_gb*1024.0**3)
 
+    # Spin CPU
     ctr = 0
     while ctr < 100e6:
         ctr+= 1

nipype/pipeline/engine/nodes.py

@@ -52,7 +52,6 @@
 
 from ... import config, logging
 logger = logging.getLogger('workflow')
-
 from ...interfaces.base import (traits, InputMultiPath, CommandLine,
                                 Undefined, TraitedSpec, DynamicTraitedSpec,
                                 Bunch, InterfaceResult, md5, Interface,

nipype/pipeline/engine/tests/test_engine.py

@@ -715,7 +715,7 @@ def func1(in1):
     w1.config['execution'] = {'stop_on_first_crash': 'true',
                               'local_hash_check': 'true',
                               'crashdump_dir': wd,
-                              'poll_sleep_duration' : 2}
+                              'poll_sleep_duration': 2}
 
     # test output of num_subnodes method when serial is default (False)
     yield assert_equal, n1.num_subnodes(), len(n1.inputs.in1)

nipype/pipeline/plugins/base.py

@@ -20,6 +20,7 @@
 import numpy as np
 import scipy.sparse as ssp
 
+
 from ...utils.filemanip import savepkl, loadpkl
 from ...utils.misc import str2bool
 from ..engine.utils import (nx, dfs_preorder, topological_sort)

nipype/pipeline/plugins/callback_log.py

@@ -1,9 +1,17 @@
+# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
+# vi: set ft=python sts=4 ts=4 sw=4 et:
+"""Callback logger for recording workflow and node run stats
+"""
+
+# Import packages
 import datetime
 import logging
 
+# Log node stats function
 def log_nodes_cb(node, status, result=None):
-    '''
-    '''
+    """Function to record node run statistics to a log file as json
+    dictionaries
+    """
 
     # Init variables
     logger = logging.getLogger('callback')

nipype/pipeline/plugins/multiproc.py

@@ -10,6 +10,7 @@
 from multiprocessing import Process, Pool, cpu_count, pool
 from traceback import format_exception
 import sys
+
 import numpy as np
 from copy import deepcopy
 from ..engine import MapNode
@@ -24,7 +25,20 @@
 
 # Run node
 def run_node(node, updatehash):
-    """docstring
+    """Function to execute node.run(), catch and log any errors and
+    return the result dictionary
+
+    Parameters
+    ----------
+    node : nipype Node instance
+        the node to run
+    updatehash : boolean
+        flag for updating hash
+
+    Returns
+    -------
+    result : dictionary
+        dictionary containing the node runtime results and stats
     """
 
     # Import packages
@@ -45,7 +59,7 @@ def run_node(node, updatehash):
             result['runtime_threads'] = retval.runtime.get('runtime_threads')
     except:
         etype, eval, etr = sys.exc_info()
-        result['traceback'] = format_exception(etype,eval,etr)
+        result['traceback'] = format_exception(etype, eval, etr)
         result['result'] = node.result
 
     # Return the result dictionary
@@ -125,15 +139,13 @@ def _wait(self):
             semaphore_singleton.semaphore.acquire()
         semaphore_singleton.semaphore.release()
 
-
     def _get_result(self, taskid):
         if taskid not in self._taskresult:
             raise RuntimeError('Multiproc task %d not found' % taskid)
         if not self._taskresult[taskid].ready():
             return None
         return self._taskresult[taskid].get()
 
-
     def _report_crash(self, node, result=None):
         if result and result['traceback']:
             node._result = result['result']
@@ -167,7 +179,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
         executing_now = []
 
         # Check to see if a job is available
-        jobids = np.flatnonzero((self.proc_pending == True) & (self.depidx.sum(axis=0) == 0).__array__())
+        jobids = np.flatnonzero((self.proc_pending == True) & \
+                                (self.depidx.sum(axis=0) == 0).__array__())
 
         #check available system resources by summing all threads and memory used
         busy_memory = 0
@@ -181,22 +194,29 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
 
 
         #check all jobs without dependency not run
-        jobids = np.flatnonzero((self.proc_done == False) & (self.depidx.sum(axis=0) == 0).__array__())
+        jobids = np.flatnonzero((self.proc_done == False) & \
+                                (self.depidx.sum(axis=0) == 0).__array__())
 
 
         #sort jobs ready to run first by memory and then by number of threads
         #The most resource consuming jobs run first
-        jobids = sorted(jobids, key=lambda item: (self.procs[item]._interface.estimated_memory, self.procs[item]._interface.num_threads))
+        jobids = sorted(jobids,
+                        key=lambda item: (self.procs[item]._interface.estimated_memory,
+                                          self.procs[item]._interface.num_threads))
 
-        logger.debug('Free memory: %d, Free processors: %d', free_memory, free_processors)
+        logger.debug('Free memory: %d, Free processors: %d',
+                     free_memory, free_processors)
 
 
         #while have enough memory and processors for first job
         #submit first job on the list
         for jobid in jobids:
-            logger.debug('Next Job: %d, memory: %d, threads: %d' %(jobid, self.procs[jobid]._interface.estimated_memory, self.procs[jobid]._interface.num_threads))
+            logger.debug('Next Job: %d, memory: %d, threads: %d' \
+                         % (jobid, self.procs[jobid]._interface.estimated_memory,
+                            self.procs[jobid]._interface.num_threads))
 
-            if self.procs[jobid]._interface.estimated_memory <= free_memory and self.procs[jobid]._interface.num_threads <= free_processors:
+            if self.procs[jobid]._interface.estimated_memory <= free_memory and \
+               self.procs[jobid]._interface.num_threads <= free_processors:
                 logger.info('Executing: %s ID: %d' %(self.procs[jobid]._id, jobid))
                 executing_now.append(self.procs[jobid])
 
@@ -228,7 +248,9 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                         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))):
+                        if (hash_exists and (self.procs[jobid].overwrite == False or \
+                                             (self.procs[jobid].overwrite == None and \
+                                              not self.procs[jobid]._interface.always_run))):
                             self._task_finished_cb(jobid)
                             self._remove_node_dirs()
                             continue
@@ -239,7 +261,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                 logger.debug('Finished checking hash')
 
                 if self.procs[jobid].run_without_submitting:
-                    logger.debug('Running node %s on master thread' %self.procs[jobid])
+                    logger.debug('Running node %s on master thread' \
+                                 % self.procs[jobid])
                     try:
                         self.procs[jobid].run()
                     except Exception:
@@ -249,7 +272,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
 
                 else:
                     logger.debug('submitting %s' % str(jobid))
-                    tid = self._submit_job(deepcopy(self.procs[jobid]), updatehash=updatehash)
+                    tid = self._submit_job(deepcopy(self.procs[jobid]),
+                                           updatehash=updatehash)
                     if tid is None:
                         self.proc_done[jobid] = False
                         self.proc_pending[jobid] = False