[FIX, WIP] Encoding problems in SGE plugin

Fixes #1625

Author: oesteban

nipype/pipeline/plugins/sge.py

@@ -3,7 +3,7 @@
 """
 from __future__ import print_function, division, unicode_literals, absolute_import
 
-from builtins import object
+from builtins import object, str, bytes
 
 import os
 import pwd
@@ -18,13 +18,13 @@
 from ...interfaces.base import CommandLine
 from .base import (SGELikeBatchManagerBase, logger, iflogger, logging)
 
-DEBUGGING_PREFIX = str(int(random.uniform(100, 999)))
+DEBUGGING_PREFIX = '{0!d}'.format(random.uniform(100, 999))
 
 
 def sge_debug_print(message):
     """  Needed for debugging on big jobs.  Once this is fully vetted, it can be removed.
     """
-    logger.debug(DEBUGGING_PREFIX + " " + "=!" * 3 + "  " + message)
+    logger.debug('%s ' + '=!' * 3 + '  %s', DEBUGGING_PREFIX, message)
     # print DEBUGGING_PREFIX + " " + "=!" * 3 + "  " + message
 
 
@@ -40,8 +40,7 @@ def __init__(self, job_num, job_queue_state, job_time, job_queue_name, job_slots
         self._job_num = int(
             job_num)      # The primary unique identifier for this job, must be an integer!
         # self._jobOwn  = None           # Who owns this job
-        self._job_queue_state = str(
-            job_queue_state)     # ["running","zombie",...??]
+        self._job_queue_state = '%s' % job_queue_state     # ["running","zombie",...??]
         # self._jobActionState = str(jobActionState)  # ['r','qw','S',...??]
         self._job_time = job_time               # The job start time
         self._job_info_creation_time = time.time(
@@ -51,12 +50,10 @@ def __init__(self, job_num, job_queue_state, job_time, job_queue_name, job_slots
         self._qsub_command_line = qsub_command_line
 
     def __repr__(self):
-        return str(self._job_num).ljust(8) \
-            + str(self._job_queue_state).ljust(12) \
-            + str(self._job_slots).ljust(3) \
-            + time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(self._job_time)).ljust(20) \
-            + str(self._job_queue_name).ljust(8) \
-            + str(self._qsub_command_line)
+        return '{:<8d}{:12}{:<3d}{:20}{:8}{}'.format(
+            self._job_num, self._queue_state, self._job_slots,
+            time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(self._job_time)),
+            self._job_queue_name, self._qsub_command_line)
 
     def is_initializing(self):
         return self._job_queue_state == "initializing"
@@ -150,12 +147,13 @@ def _qacct_verified_complete(taskid):
         while qacct_retries > 0:
             qacct_retries -= 1
             try:
+                strtaskid = '{0!d}'.format(taskid)
                 proc = subprocess.Popen(
-                    [this_command, '-o', pwd.getpwuid(os.getuid())[0], '-j', str(taskid)],
+                    [this_command, '-o', pwd.getpwuid(os.getuid())[0], '-j', strtaskid],
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE)
                 qacct_result, _ = proc.communicate()
-                if qacct_result.find(str(taskid)):
+                if qacct_result.find(strtaskid):
                     is_complete = True
                 sge_debug_print(
                     "NOTE: qacct for jobs\n{0}".format(qacct_result))
@@ -276,7 +274,7 @@ def _run_qstat(self, reason_for_qstat, force_instant=True):
     def print_dictionary(self):
         """For debugging"""
         for vv in list(self._task_dictionary.values()):
-            sge_debug_print(str(vv))
+            sge_debug_print('%s' % vv)
 
     def is_job_pending(self, task_id):
         task_id = int(task_id)  # Ensure that it is an integer
@@ -398,7 +396,7 @@ def _submit_batchtask(self, scriptfile, node):
         oldlevel = iflogger.level
         iflogger.setLevel(logging.getLevelName('CRITICAL'))
         tries = 0
-        result = list()
+        result = None
         while True:
             try:
                 result = cmd.run()
@@ -409,9 +407,8 @@ def _submit_batchtask(self, scriptfile, node):
                         self._retry_timeout)  # sleep 2 seconds and try again.
                 else:
                     iflogger.setLevel(oldlevel)
-                    raise RuntimeError('\n'.join((('Could not submit sge task'
-                                                   ' for node %s') % node._id,
-                                                  str(e))))
+                    raise RuntimeError(
+                        'Could not submit sge task for node %s\n%s' % (node._id, e))
             else:
                 break
         iflogger.setLevel(oldlevel)
@@ -421,6 +418,6 @@ def _submit_batchtask(self, scriptfile, node):
                               lines[-1]).groups()[0])
         self._pending[taskid] = node.output_dir()
         self._refQstatSubstitute.add_startup_job(taskid, cmd.cmdline)
-        logger.debug('submitted sge task: %d for node %s with %s' %
-                     (taskid, node._id, cmd.cmdline))
+        logger.debug('submitted sge task: %d for node %s with %s',
+                     taskid, node._id, cmd.cmdline)
         return taskid