Merge pull request #2777 from oesteban/enh/process-spawn-2

[MAINT] Move ``interfaces.base.run_command`` to ``nipype.utils.subprocess``

Author: effigies

nipype/interfaces/base/core.py

@@ -17,14 +17,11 @@
 
 from builtins import object, open, str, bytes
 
-import gc
 from copy import deepcopy
 from datetime import datetime as dt
-import errno
 import os
 import re
 import platform
-import select
 import subprocess as sp
 import shlex
 import sys
@@ -35,16 +32,16 @@
 from ... import config, logging, LooseVersion
 from ...utils.provenance import write_provenance
 from ...utils.misc import trim, str2bool, rgetcwd
-from ...utils.filemanip import (FileNotFoundError, split_filename, read_stream,
-                                which, get_dependencies, canonicalize_env as
-                                _canonicalize_env)
+from ...utils.filemanip import (FileNotFoundError, split_filename,
+                                which, get_dependencies)
+from ...utils.subprocess import run_command
 
 from ...external.due import due
 
 from .traits_extension import traits, isdefined, TraitError
 from .specs import (BaseInterfaceInputSpec, CommandLineInputSpec,
                     StdOutCommandLineInputSpec, MpiCommandLineInputSpec)
-from .support import (Bunch, Stream, InterfaceResult, NipypeInterfaceError)
+from .support import (Bunch, InterfaceResult, NipypeInterfaceError)
 
 from future import standard_library
 standard_library.install_aliases()
@@ -732,135 +729,6 @@ def _list_outputs(self):
         return self._results
 
 
-def run_command(runtime, output=None, timeout=0.01):
-    """Run a command, read stdout and stderr, prefix with timestamp.
-
-    The returned runtime contains a merged stdout+stderr log with timestamps
-    """
-
-    # Init variables
-    cmdline = runtime.cmdline
-    env = _canonicalize_env(runtime.environ)
-
-    errfile = None
-    outfile = None
-    stdout = sp.PIPE
-    stderr = sp.PIPE
-
-    if output == 'file':
-        outfile = os.path.join(runtime.cwd, 'output.nipype')
-        stdout = open(outfile, 'wb')  # t=='text'===default
-        stderr = sp.STDOUT
-    elif output == 'file_split':
-        outfile = os.path.join(runtime.cwd, 'stdout.nipype')
-        stdout = open(outfile, 'wb')
-        errfile = os.path.join(runtime.cwd, 'stderr.nipype')
-        stderr = open(errfile, 'wb')
-    elif output == 'file_stdout':
-        outfile = os.path.join(runtime.cwd, 'stdout.nipype')
-        stdout = open(outfile, 'wb')
-    elif output == 'file_stderr':
-        errfile = os.path.join(runtime.cwd, 'stderr.nipype')
-        stderr = open(errfile, 'wb')
-
-    proc = sp.Popen(
-        cmdline,
-        stdout=stdout,
-        stderr=stderr,
-        shell=True,
-        cwd=runtime.cwd,
-        env=env,
-        close_fds=(not sys.platform.startswith('win')),
-    )
-
-    result = {
-        'stdout': [],
-        'stderr': [],
-        'merged': [],
-    }
-
-    if output == 'stream':
-        streams = [
-            Stream('stdout', proc.stdout),
-            Stream('stderr', proc.stderr)
-        ]
-
-        def _process(drain=0):
-            try:
-                res = select.select(streams, [], [], timeout)
-            except select.error as e:
-                iflogger.info(e)
-                if e[0] == errno.EINTR:
-                    return
-                else:
-                    raise
-            else:
-                for stream in res[0]:
-                    stream.read(drain)
-
-        while proc.returncode is None:
-            proc.poll()
-            _process()
-
-        _process(drain=1)
-
-        # collect results, merge and return
-        result = {}
-        temp = []
-        for stream in streams:
-            rows = stream._rows
-            temp += rows
-            result[stream._name] = [r[2] for r in rows]
-        temp.sort()
-        result['merged'] = [r[1] for r in temp]
-
-    if output.startswith('file'):
-        proc.wait()
-        if outfile is not None:
-            stdout.flush()
-            stdout.close()
-            with open(outfile, 'rb') as ofh:
-                stdoutstr = ofh.read()
-            result['stdout'] = read_stream(stdoutstr, logger=iflogger)
-            del stdoutstr
-
-        if errfile is not None:
-            stderr.flush()
-            stderr.close()
-            with open(errfile, 'rb') as efh:
-                stderrstr = efh.read()
-            result['stderr'] = read_stream(stderrstr, logger=iflogger)
-            del stderrstr
-
-        if output == 'file':
-            result['merged'] = result['stdout']
-            result['stdout'] = []
-    else:
-        stdout, stderr = proc.communicate()
-        if output == 'allatonce':  # Discard stdout and stderr otherwise
-            result['stdout'] = read_stream(stdout, logger=iflogger)
-            result['stderr'] = read_stream(stderr, logger=iflogger)
-
-    runtime.returncode = proc.returncode
-    try:
-        proc.terminate()  # Ensure we are done
-    except OSError as error:
-        # Python 2 raises when the process is already gone
-        if error.errno != errno.ESRCH:
-            raise
-
-    # Dereference & force GC for a cleanup
-    del proc
-    del stdout
-    del stderr
-    gc.collect()
-
-    runtime.stderr = '\n'.join(result['stderr'])
-    runtime.stdout = '\n'.join(result['stdout'])
-    runtime.merged = '\n'.join(result['merged'])
-    return runtime
-
-
 class CommandLine(BaseInterface):
     """Implements functionality to interact with command line programs
     class must be instantiated with a command argument

nipype/interfaces/base/support.py

@@ -9,14 +9,11 @@
 """
 from __future__ import (print_function, division, unicode_literals,
                         absolute_import)
-from builtins import range, object, str
+from builtins import object, str
 
 import os
 from copy import deepcopy
 
-import datetime
-import locale
-
 from ... import logging
 from ...utils.misc import is_container
 from ...utils.filemanip import md5, to_str, hash_infile
@@ -238,58 +235,6 @@ def version(self):
         return self._version
 
 
-class Stream(object):
-    """Function to capture stdout and stderr streams with timestamps
-
-    stackoverflow.com/questions/4984549/merge-and-sync-stdout-and-stderr/5188359
-    """
-
-    def __init__(self, name, impl):
-        self._name = name
-        self._impl = impl
-        self._buf = ''
-        self._rows = []
-        self._lastidx = 0
-        self.default_encoding = locale.getdefaultlocale()[1] or 'UTF-8'
-
-    def fileno(self):
-        "Pass-through for file descriptor."
-        return self._impl.fileno()
-
-    def read(self, drain=0):
-        "Read from the file descriptor. If 'drain' set, read until EOF."
-        while self._read(drain) is not None:
-            if not drain:
-                break
-
-    def _read(self, drain):
-        "Read from the file descriptor"
-        fd = self.fileno()
-        buf = os.read(fd, 4096).decode(self.default_encoding)
-        if not buf and not self._buf:
-            return None
-        if '\n' not in buf:
-            if not drain:
-                self._buf += buf
-                return []
-
-        # prepend any data previously read, then split into lines and format
-        buf = self._buf + buf
-        if '\n' in buf:
-            tmp, rest = buf.rsplit('\n', 1)
-        else:
-            tmp = buf
-            rest = None
-        self._buf = rest
-        now = datetime.datetime.now().isoformat()
-        rows = tmp.split('\n')
-        self._rows += [(now, '%s %s:%s' % (self._name, now, r), r)
-                       for r in rows]
-        for idx in range(self._lastidx, len(self._rows)):
-            iflogger.info(self._rows[idx][1])
-        self._lastidx = len(self._rows)
-
-
 def load_template(name):
     """
     Deprecated stub for backwards compatibility,