Merged resource_multiproc into s3_multiproc

Author: pintohutch

nipype/__init__.py

@@ -20,7 +20,7 @@
 try:
     import faulthandler
     faulthandler.enable()
-except:
+except (ImportError,IOError) as e:
     pass
 
 

nipype/algorithms/metrics.py

@@ -105,12 +105,12 @@ def _eucl_min(self, nii1, nii2):
                 set2_coordinates.T[point2, :])
 
     def _eucl_cog(self, nii1, nii2):
-        origdata1 = nii1.get_data().astype(np.bool)
-        cog_t = np.array(center_of_mass(origdata1)).reshape(-1, 1)
+        origdata1 = np.logical_and(nii1.get_data() != 0, np.logical_not(np.isnan(nii1.get_data())))
+        cog_t = np.array(center_of_mass(origdata1.copy())).reshape(-1, 1)
         cog_t = np.vstack((cog_t, np.array([1])))
         cog_t_coor = np.dot(nii1.affine, cog_t)[:3, :]
 
-        origdata2 = nii2.get_data().astype(np.bool)
+        origdata2 = np.logical_and(nii2.get_data() != 0, np.logical_not(np.isnan(nii2.get_data())))
         (labeled_data, n_labels) = label(origdata2)
 
         cogs = np.ones((4, n_labels))
@@ -181,8 +181,9 @@ def _eucl_max(self, nii1, nii2):
         return np.max(mins)
 
     def _run_interface(self, runtime):
-        nii1 = nb.load(self.inputs.volume1)
-        nii2 = nb.load(self.inputs.volume2)
+        # there is a bug in some scipy ndimage methods that gets tripped by memory mapped objects
+        nii1 = nb.load(self.inputs.volume1, mmap=False)
+        nii2 = nb.load(self.inputs.volume2, mmap=False)
 
         if self.inputs.method == "eucl_min":
             self._distance, self._point1, self._point2 = self._eucl_min(

nipype/interfaces/ants/base.py

@@ -12,7 +12,7 @@
 # -Using -1 gives primary responsibilty to ITKv4 to do the correct
 #  thread limitings.
 # -Using 1 takes a very conservative approach to avoid overloading
-#  the computer (when running MultiProc) by forcing everything to
+#  the computer (when running ResourceMultiProc) by forcing everything to
 #  single threaded.  This can be a severe penalty for registration
 #  performance.
 LOCAL_DEFAULT_NUMBER_OF_THREADS = 1

nipype/interfaces/base.py

@@ -1297,6 +1297,7 @@ 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)]
@@ -1346,7 +1347,7 @@ def _process(drain=0):
             try:
                 stderr = stderr.decode()
             except UnicodeDecodeError:
-                stdout = stdout.decode("ISO-8859-1")
+                stderr = stderr.decode("ISO-8859-1")
 
         result['stdout'] = str(stdout).split('\n')
         result['stderr'] = str(stderr).split('\n')

nipype/interfaces/io.py

@@ -134,54 +134,7 @@ def _add_output_traits(self, base):
         return base
 
 
-# Class to track percentage of S3 file upload
-class ProgressPercentage(object):
-    '''
-    Callable class instsance (via __call__ method) that displays
-    upload percentage of a file to S3
-    '''
-
-    def __init__(self, filename):
-        '''
-        '''
-
-        # Import packages
-        import threading
-
-        # Initialize data attributes
-        self._filename = filename
-        self._size = float(os.path.getsize(filename))
-        self._seen_so_far = 0
-        self._lock = threading.Lock()
-
-    def __call__(self, bytes_amount):
-        '''
-        '''
-
-        # Import packages
-        import sys
-
-        # With the lock on, print upload status
-        with self._lock:
-            self._seen_so_far += bytes_amount
-            if self._size != 0:
-                percentage = (self._seen_so_far / self._size) * 100
-            else:
-                percentage = 0
-            progress_str = '%d / %d (%.2f%%)\r'\
-                           % (self._seen_so_far, self._size, percentage)
-
-            # Write to stdout
-            sys.stdout.write(progress_str)
-            sys.stdout.flush()
-
-
-# DataSink inputs
 class DataSinkInputSpec(DynamicTraitedSpec, BaseInterfaceInputSpec):
-    '''
-    '''
-
-    # Init inputspec data attributes
     base_directory = Directory(
         desc='Path to the base directory for storing data.')
     container = traits.Str(
@@ -193,11 +146,11 @@ class DataSinkInputSpec(DynamicTraitedSpec, BaseInterfaceInputSpec):
                                    desc=('List of 2-tuples reflecting string '
                                          'to substitute and string to replace '
                                          'it with'))
-    regexp_substitutions = \
-        InputMultiPath(traits.Tuple(traits.Str, traits.Str),
-                       desc=('List of 2-tuples reflecting a pair of a '\
-                             'Python regexp pattern and a replacement '\
-                             'string. Invoked after string `substitutions`'))
+    regexp_substitutions = InputMultiPath(traits.Tuple(traits.Str, traits.Str),
+                                          desc=('List of 2-tuples reflecting a pair '
+                                                'of a Python regexp pattern and a '
+                                                'replacement string. Invoked after '
+                                                'string `substitutions`'))
 
     _outputs = traits.Dict(traits.Str, value={}, usedefault=True)
     remove_dest_dir = traits.Bool(False, usedefault=True,
@@ -218,7 +171,6 @@ class DataSinkInputSpec(DynamicTraitedSpec, BaseInterfaceInputSpec):
 
     # Set call-able inputs attributes
     def __setattr__(self, key, value):
-
         if key not in self.copyable_trait_names():
             if not isdefined(value):
                 super(DataSinkInputSpec, self).__setattr__(key, value)
@@ -229,14 +181,12 @@ def __setattr__(self, key, value):
             super(DataSinkInputSpec, self).__setattr__(key, value)
 
 
-# DataSink outputs
 class DataSinkOutputSpec(TraitedSpec):
 
     # Init out file
     out_file = traits.Any(desc='datasink output')
 
 
-# Custom DataSink class
 class DataSink(IOBase):
     """ Generic datasink module to store structured outputs
 
@@ -298,12 +248,9 @@ class DataSink(IOBase):
         >>> ds.run()  # doctest: +SKIP
 
     """
-
-    # Give obj .inputs and .outputs
     input_spec = DataSinkInputSpec
     output_spec = DataSinkOutputSpec
 
-    # Initialization method to set up datasink
     def __init__(self, infields=None, force_run=True, **kwargs):
         """
         Parameters
@@ -325,7 +272,6 @@ def __init__(self, infields=None, force_run=True, **kwargs):
         if force_run:
             self._always_run = True
 
-    # Get destination paths
     def _get_dst(self, src):
         # If path is directory with trailing os.path.sep,
         # then remove that for a more robust behavior
@@ -349,7 +295,6 @@ def _get_dst(self, src):
             dst = dst[1:]
         return dst
 
-    # Substitute paths in substitutions dictionary parameter
     def _substitute(self, pathstr):
         pathstr_ = pathstr
         if isdefined(self.inputs.substitutions):
@@ -639,9 +584,6 @@ def _upload_to_s3(self, bucket, src, dst):
     def _list_outputs(self):
         """Execute this module.
         """
-
-        # Init variables
-        iflogger = logging.getLogger('interface')
         outputs = self.output_spec().get()
         out_files = []
         # Use hardlink
@@ -706,23 +648,17 @@ def _list_outputs(self):
             iflogger.debug("key: %s files: %s" % (key, str(files)))
             files = filename_to_list(files)
             tempoutdir = outdir
-            if s3_flag:
-                s3tempoutdir = s3dir
             for d in key.split('.'):
                 if d[0] == '@':
                     continue
                 tempoutdir = os.path.join(tempoutdir, d)
-                if s3_flag:
-                    s3tempoutdir = os.path.join(s3tempoutdir, d)
 
             # flattening list
             if isinstance(files, list):
                 if isinstance(files[0], list):
                     files = [item for sublist in files for item in sublist]
 
-            # Iterate through passed-in source files
             for src in filename_to_list(files):
-                # Format src and dst files
                 src = os.path.abspath(src)
                 if not os.path.isfile(src):
                     src = os.path.join(src, '')
@@ -749,22 +685,12 @@ def _list_outputs(self):
                                 pass
                             else:
                                 raise(inst)
-                    # If src is a file, copy it to dst
-                    if os.path.isfile(src):
-                        iflogger.debug('copyfile: %s %s' % (src, dst))
-                        copyfile(src, dst, copy=True, hashmethod='content',
-                                 use_hardlink=use_hardlink)
-                        out_files.append(dst)
-                    # If src is a directory, copy entire contents to dst dir
-                    elif os.path.isdir(src):
-                        if os.path.exists(dst) and self.inputs.remove_dest_dir:
-                            iflogger.debug('removing: %s' % dst)
-                            shutil.rmtree(dst)
-                        iflogger.debug('copydir: %s %s' % (src, dst))
-                        copytree(src, dst)
-                        out_files.append(dst)
-
-        # Return outputs dictionary
+                    if os.path.exists(dst) and self.inputs.remove_dest_dir:
+                        iflogger.debug("removing: %s" % dst)
+                        shutil.rmtree(dst)
+                    iflogger.debug("copydir: %s %s" % (src, dst))
+                    copytree(src, dst)
+                    out_files.append(dst)
         outputs['out_file'] = out_files
 
         return outputs

nipype/interfaces/tests/test_io.py

@@ -18,7 +18,6 @@
 import nipype.interfaces.io as nio
 from nipype.interfaces.base import Undefined
 
-# Check for boto
 noboto = False
 try:
     import boto
@@ -113,7 +112,7 @@ def test_selectfiles_valueerror():
     yield assert_raises, ValueError, sf.run
 
 
-@skipif(noboto)
+@skip
 def test_s3datagrabber_communication():
     dg = nio.S3DataGrabber(
         infields=['subj_id', 'run_num'], outfields=['func', 'struct'])

nipype/interfaces/utility.py

@@ -450,14 +450,14 @@ def _run_interface(self, runtime):
                 args[name] = value
 
         # Record memory of function_handle
-        #try:
-        #    import memory_profiler
-        #    proc = (function_handle, (), args)
-        #    mem_mb, out = memory_profiler.memory_usage(proc=proc, retval=True, include_children=True, max_usage=True)
-        #    setattr(runtime, 'cmd_memory', mem_mb[0]/1024.0)
+        try:
+            import memory_profiler
+            proc = (function_handle, (), args)
+            mem_mb, out = memory_profiler.memory_usage(proc=proc, retval=True, include_children=True, max_usage=True)
+            setattr(runtime, 'cmd_memory', mem_mb[0]/1024.0)
         # If no memory_profiler package, run without recording memory
-        #except:
-        out = function_handle(**args)
+        except ImportError:
+            out = function_handle(**args)
 
         if len(self._output_names) == 1:
             self._out[self._output_names[0]] = out

nipype/pipeline/engine/tests/test_engine.py

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

nipype/pipeline/plugins/callback_log.py

@@ -0,0 +1,39 @@
+import datetime
+import logging
+
+def log_nodes_cb(node, status, result=None):
+    logger = logging.getLogger('callback')
+    try:
+        node_mem = result['node_memory']
+        cmd_mem = result['cmd_memory']
+        run_seconds = result['run_seconds']
+        cmd_threads = result['cmd_threads']
+    except Exception as exc:
+        node_mem = cmd_mem = run_seconds = cmd_threads = 'N/A'
+    if status == 'start':
+        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
+        node._id + '"' + ',"start":' + '"' +str(datetime.datetime.now()) +\
+        '"' + ',"estimated_memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
+        + str(node._interface.num_threads) + '}'
+
+        logger.debug(message)
+
+    elif status == 'end':
+        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
+        node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +  \
+        '"' + ',"estimated_memory":' + '"'+ str(node._interface.estimated_memory) + '"'+ \
+        ',"num_threads":' + '"'+ str(node._interface.num_threads) + '"'+ \
+        ',"cmd-level_threads":' + '"'+ str(cmd_threads) + '"'+ \
+        ',"node-level_memory":' + '"'+ str(node_mem) + '"'+ \
+        ',"cmd-level_memory":' + '"'+ str(cmd_mem) + '"' + \
+        ',"run_seconds":' + '"'+ str(run_seconds) + '"'+ '}'
+
+        logger.debug(message)
+
+    else:
+        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
+        node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
+        '"' + ',"estimated_memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
+        + str(node._interface.num_threads) + ',"error":"True"}'
+
+        logger.debug(message)

nipype/pipeline/plugins/multiproc.py

@@ -35,6 +35,7 @@ def run_node(node, updatehash, runtime_profile=False):
     # Init variables
     result = dict(result=None, traceback=None)
     runtime_profile = False
+
     # If we're profiling the run
     if runtime_profile:
         try: