general cleanup

Author: oesteban

nipype/pipeline/engine/nodes.py

@@ -327,7 +327,7 @@ def run(self, updatehash=False):
         makedirs(outdir, exist_ok=True)
         os.chdir(outdir)
 
-        logger.info('[Node] Executing "%s" (%s)', self.fullname, outdir)
+        logger.info('[Node] Setting-up "%s" in "%s".', self.fullname, outdir)
         hash_info = self.hash_exists(updatehash=updatehash)
         hash_exists, hashvalue, hashfile, hashed_inputs = hash_info
         force_run = self.overwrite or (self.overwrite is None and self._interface.always_run)
@@ -346,7 +346,7 @@ def run(self, updatehash=False):
                 savepkl(node_file, self)
 
             self._run_interface(execute=False, updatehash=updatehash)
-            logger.info('[Node] Cached "%s" (%s)\n', self.fullname, outdir)
+            logger.info('[Node] Cached "%s".', self.fullname)
             os.chdir(cwd)
             return self.result
 
@@ -412,7 +412,7 @@ def run(self, updatehash=False):
         # Tear-up
         shutil.move(hashfile_unfinished, hashfile)
         self.write_report(report_type='postexec', cwd=outdir)
-        logger.info('[Node] Completed "%s" (%s)', self.fullname, outdir)
+        logger.info('[Node] Finished "%s".', self.fullname)
         os.chdir(cwd)
         return self._result
 
@@ -623,7 +623,7 @@ def _run_command(self, execute, copyfiles=True):
             if copyfiles:
                 self._copyfiles_to_wd(cwd, execute)
 
-            message = 'Running node "%s" ("%s.%s")'
+            message = '[Node] Running "%s" ("%s.%s")'
             if issubclass(self._interface.__class__, CommandLine):
                 try:
                     cmd = self._interface.cmdline

nipype/pipeline/engine/workflows.py

@@ -47,11 +47,11 @@
 from ...utils.filemanip import (save_json, FileNotFoundError,
                                 filename_to_list, list_to_filename,
                                 copyfiles, fnames_presuffix, loadpkl,
-                                split_filename, load_json, savepkl,
+                                split_filename, load_json, makedirs, savepkl,
                                 write_rst_header, write_rst_dict,
                                 write_rst_list, to_str)
 from .utils import (generate_expanded_graph, modify_paths,
-                    export_graph, make_output_dir, write_workflow_prov,
+                    export_graph, write_workflow_prov,
                     write_workflow_resources,
                     clean_working_directory, format_dot, topological_sort,
                     get_print_name, merge_dict, evaluate_connect_function,
@@ -424,7 +424,7 @@ def write_graph(self, dotfilename='graph.dot', graph2use='hierarchical',
                     base_dir = op.join(base_dir, self.name)
             else:
                 base_dir = os.getcwd()
-        base_dir = make_output_dir(base_dir)
+        base_dir = makedirs(base_dir)
         if graph2use in ['hierarchical', 'colored']:
             if self.name[:1].isdigit(): # these graphs break if int
                 raise ValueError('{} graph failed, workflow name cannot begin '

nipype/pipeline/plugins/base.py

@@ -339,7 +339,7 @@ def _local_hash_check(self, jobid, graph):
             logger.debug('Skipping cached node %s with ID %s.',
                          self.procs[jobid]._id, jobid)
             try:
-                self._task_finished_cb(jobid)
+                self._task_finished_cb(jobid, cached=True)
                 self._remove_node_dirs()
             except Exception:
                 logger.debug('Error skipping cached node %s (%s).',
@@ -349,13 +349,14 @@ def _local_hash_check(self, jobid, graph):
             return True
         return False
 
-    def _task_finished_cb(self, jobid):
+    def _task_finished_cb(self, jobid, cached=False):
         """ Extract outputs and assign to inputs of dependent tasks
 
         This is called when a job is completed.
         """
-        logger.info('[Job finished] jobname: %s jobid: %d' %
-                    (self.procs[jobid]._id, jobid))
+        logger.info('[Job %d] %s (%s).', jobid,
+                    'Cached' if cached else 'Completed',
+                    self.procs[jobid].fullname)
         if self._status_callback:
             self._status_callback(self.procs[jobid], 'end')
         # Update job and worker queues

nipype/pipeline/plugins/multiproc.py

@@ -12,10 +12,11 @@
 from multiprocessing import Process, Pool, cpu_count, pool
 from traceback import format_exception
 import sys
+from textwrap import indent
+from logging import INFO
 
 from copy import deepcopy
 import numpy as np
-
 from ... import logging
 from ...utils.profiler import get_system_total_memory_gb
 from ..engine import MapNode
@@ -126,8 +127,8 @@ def __init__(self, plugin_args=None):
         self.raise_insufficient = self.plugin_args.get('raise_insufficient', True)
 
         # Instantiate different thread pools for non-daemon processes
-        logger.debug('MultiProcPlugin starting in "%sdaemon" mode (n_procs=%d, mem_gb=%0.2f)',
-                     'non' if non_daemon else '', self.processors, self.memory_gb)
+        logger.debug('[MultiProc] Starting in "%sdaemon" mode (n_procs=%d, mem_gb=%0.2f)',
+                     'non' * non_daemon, self.processors, self.memory_gb)
 
         NipypePool = NonDaemonPool if non_daemon else Pool
         try:
@@ -158,7 +159,7 @@ def _submit_job(self, node, updatehash=False):
             run_node, (node, updatehash, self._taskid),
             callback=self._async_callback)
 
-        logger.debug('MultiProc submitted task %s (taskid=%d).',
+        logger.debug('[MultiProc] Submitted task %s (taskid=%d).',
                      node.fullname, self._taskid)
         return self._taskid
 
@@ -214,9 +215,17 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
         stats = (len(self.pending_tasks), len(jobids), free_memory_gb,
                  self.memory_gb, free_processors, self.processors)
         if self._stats != stats:
-            logger.info('Currently running %d tasks, and %d jobs ready. Free '
-                        'memory (GB): %0.2f/%0.2f, Free processors: %d/%d',
-                        *stats)
+            tasks_list_msg = ''
+            if logger.level <= INFO:
+                running_tasks = ['  * %s' % self.procs[jobid].fullname
+                                 for _, jobid in self.pending_tasks]
+                if running_tasks:
+                    tasks_list_msg = '\nCurrently running:\n'
+                    tasks_list_msg += '\n'.join(running_tasks)
+                    tasks_list_msg = indent(tasks_list_msg, ' ' * 21)
+            logger.info('[MultiProc] Running %d tasks, and %d jobs ready. Free '
+                        'memory (GB): %0.2f/%0.2f, Free processors: %d/%d.%s',
+                        *stats, tasks_list_msg)
             self._stats = stats
 
         if free_memory_gb < 0.01 or free_processors == 0:

nipype/utils/filemanip.py

@@ -699,16 +699,19 @@ def emptydirs(path):
         return True
 
     for el in pathconts:
-        try:
-            shutil.rmtree(el)
-        except OSError as ex:
-            elcont = os.listdir(el)
-            if ex.errno == errno.ENOTEMPTY and not elcont:
-                fmlogger.warning(
-                    'An exception was raised trying to remove old %s, but the path '
-                    'seems empty. Is it an NFS mount?. Passing the exception.', el)
-            elif ex.errno == errno.ENOTEMPTY and elcont:
-                fmlogger.debug('Folder %s contents (%d items).', el, len(elcont))
-                raise ex
-            else:
-                raise ex
+        if os.path.isfile(el):
+            os.remove(el)
+        else:
+            try:
+                shutil.rmtree(el)
+            except OSError as ex:
+                elcont = os.listdir(el)
+                if ex.errno == errno.ENOTEMPTY and not elcont:
+                    fmlogger.warning(
+                        'An exception was raised trying to remove old %s, but the path '
+                        'seems empty. Is it an NFS mount?. Passing the exception.', el)
+                elif ex.errno == errno.ENOTEMPTY and elcont:
+                    fmlogger.debug('Folder %s contents (%d items).', el, len(elcont))
+                    raise ex
+                else:
+                    raise ex