Merge remote-tracking branch 'origin/fix/use_current_prov' into fix/use_current_prov

* origin/fix/use_current_prov:
  fix: string decoding for prov literal and few pipeline job submission info statements

Author: satra

nipype/pipeline/plugins/base.py

@@ -342,7 +342,10 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                                     (self.depidx.sum(axis=0) == 0).__array__())
             if len(jobids) > 0:
                 # send all available jobs
-                logger.info('Submitting %d jobs' % len(jobids[:slots]))
+                if slots:
+                    logger.info('Pending[%d] Submitting[%d] jobs Slots[%d]' % (num_jobs, len(jobids[:slots]), slots))
+                else:
+                    logger.info('Pending[%d] Submitting[%d] jobs Slots[inf]' % (num_jobs, len(jobids)))
                 for jobid in jobids[:slots]:
                     if isinstance(self.procs[jobid], MapNode):
                         try:
@@ -401,6 +404,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 executing: %s ID: %d' %
+                                (self.procs[jobid]._id, jobid))
             else:
                 break
 
@@ -507,9 +512,6 @@ def _get_result(self, taskid):
         timed_out = True
         while (time() - t) < timeout:
             try:
-                logger.debug(os.listdir(os.path.realpath(os.path.join(node_dir,
-                                                                      '..'))))
-                logger.debug(os.listdir(node_dir))
                 glob(os.path.join(node_dir, 'result_*.pklz')).pop()
                 timed_out = False
                 break
@@ -633,9 +635,6 @@ def _get_result(self, taskid):
             return None
         node_dir = self._pending[taskid]
 
-        logger.debug(os.listdir(os.path.realpath(os.path.join(node_dir,
-                                                              '..'))))
-        logger.debug(os.listdir(node_dir))
         glob(os.path.join(node_dir, 'result_*.pklz')).pop()
 
         results_file = glob(os.path.join(node_dir, 'result_*.pklz'))[0]

nipype/utils/provenance.py

@@ -119,7 +119,7 @@ def safe_encode(x, as_literal=True):
     if x is None:
         value = "Unknown"
         if as_literal:
-            return pm.Literal(value, pm.XSD['string'])
+            return pm.Literal(value.decode('utf-8'), pm.XSD['string'])
         else:
             return value
     try:
@@ -131,15 +131,15 @@ def safe_encode(x, as_literal=True):
                 try:
                     return pm.URIRef(value)
                 except AttributeError:
-                    return pm.Literal(value, pm.XSD['anyURI'])
+                    return pm.Literal(value.decode('utf-8'), pm.XSD['anyURI'])
             else:
                 if len(x) > max_text_len:
                     value = x[:max_text_len - 13] + ['...Clipped...']
                 else:
                     value = x
                 if not as_literal:
                     return value
-                return pm.Literal(value, pm.XSD['string'])
+                return pm.Literal(value.decode('utf-8'), pm.XSD['string'])
         if isinstance(x, int):
             if not as_literal:
                 return x
@@ -158,7 +158,7 @@ def safe_encode(x, as_literal=True):
                     outdict[key] = encoded_value
             if not as_literal:
                 return simplejson.dumps(outdict)
-            return pm.Literal(simplejson.dumps(outdict), pm.XSD['string'])
+            return pm.Literal(simplejson.dumps(outdict).decode('utf-8'), pm.XSD['string'])
         if isinstance(x, list):
             try:
                 nptype = np.array(x).dtype
@@ -176,7 +176,7 @@ def safe_encode(x, as_literal=True):
                 outlist = x
             if not as_literal:
                 return simplejson.dumps(outlist)
-            return pm.Literal(simplejson.dumps(outlist), pm.XSD['string'])
+            return pm.Literal(simplejson.dumps(outlist).decode('utf-8'), pm.XSD['string'])
         if not as_literal:
             return dumps(x)
         return pm.Literal(dumps(x), nipype_ns['pickle'])
@@ -185,7 +185,7 @@ def safe_encode(x, as_literal=True):
         value = "Could not encode: " + str(e)
         if not as_literal:
             return value
-        return pm.Literal(value, pm.XSD['string'])
+        return pm.Literal(value.decode('utf-8'), pm.XSD['string'])
 
 
 def prov_encode(graph, value, create_container=True):