ubelt/util_cmd.py
def cmd(command, shell=False, detach=False, verbose=0, tee=None, cwd=None,
        env=None, tee_backend='auto', verbout=None, **kwargs):
    """
    Executes a command in a subprocess.

    The advantage of this wrapper around subprocess is that
    (1) you control if the subprocess prints to stdout,
    (2) the text written to stdout and stderr is returned for parsing,
    (3) cross platform behavior that lets you specify the command as a string
    or tuple regardless of whether or not shell=True.
    (4) ability to detach, return the process object and allow the process to
    run in the background (eventually we may return a Future object instead).

    Args:
        command (str or Sequence): bash-like command string or tuple of
            executable and args

        shell (bool): if True, process is run in shell, defaults to False.

        detach (bool): if True, process is detached and run in background,
            defaults to False.

        verbose (int): verbosity mode. Can be 0, 1, 2, or 3. Defaults to 0.

        tee (bool, optional): if True, simultaneously writes to stdout while
            capturing output from the command. If not specified, defaults to
            True if verbose > 0.  If detech is True, then this argument is
            ignored.

        cwd (PathLike, optional): path to run command

        env (str, optional): environment passed to Popen

        tee_backend (str, optional): backend for tee output.
            Valid choices are: "auto", "select" (POSIX only), and "thread".

        **kwargs: only used to support deprecated arguments

    Returns:
        dict: info - information about command status.
            if detach is False `info` contains captured standard out,
            standard error, and the return code
            if detach is False `info` contains a reference to the process.

    Notes:
        Inputs can either be text or tuple based. On UNIX we ensure conversion
        to text if shell=True, and to tuple if shell=False. On windows, the
        input is always text based.  See [3] for a potential cross-platform
        shlex solution for windows.

    CommandLine:
        python -m ubelt.util_cmd cmd
        python -c "import ubelt as ub; ub.cmd('ping localhost -c 2', verbose=2)"

    References:
        [1] https://stackoverflow.com/questions/11495783/redirect-subprocess-stderr-to-stdout
        [2] https://stackoverflow.com/questions/7729336/how-can-i-print-and-display-subprocess-stdout-and-stderr-output-without-distorti
        [3] https://stackoverflow.com/questions/33560364/python-windows-parsing-command-lines-with-shlex

    Example:
        >>> info = cmd(('echo', 'simple cmdline interface'), verbose=1)
        simple cmdline interface
        >>> assert info['ret'] == 0
        >>> assert info['out'].strip() == 'simple cmdline interface'
        >>> assert info['err'].strip() == ''

    Doctest:
        >>> info = cmd('echo str noshell', verbose=0)
        >>> assert info['out'].strip() == 'str noshell'

    Doctest:
        >>> # windows echo will output extra single quotes
        >>> info = cmd(('echo', 'tuple noshell'), verbose=0)
        >>> assert info['out'].strip().strip("'") == 'tuple noshell'

    Doctest:
        >>> # Note this command is formatted to work on win32 and unix
        >>> info = cmd('echo str&&echo shell', verbose=0, shell=True)
        >>> assert info['out'].strip() == 'str' + chr(10) + 'shell'

    Doctest:
        >>> info = cmd(('echo', 'tuple shell'), verbose=0, shell=True)
        >>> assert info['out'].strip().strip("'") == 'tuple shell'

    Doctest:
        >>> import ubelt as ub
        >>> from os.path import join, exists
        >>> fpath1 = join(ub.get_app_cache_dir('ubelt'), 'cmdout1.txt')
        >>> fpath2 = join(ub.get_app_cache_dir('ubelt'), 'cmdout2.txt')
        >>> ub.delete(fpath1)
        >>> ub.delete(fpath2)
        >>> info1 = ub.cmd(('touch', fpath1), detach=True)
        >>> info2 = ub.cmd('echo writing2 > ' + fpath2, shell=True, detach=True)
        >>> while not exists(fpath1):
        ...     pass
        >>> while not exists(fpath2):
        ...     pass
        >>> assert ub.readfrom(fpath1) == ''
        >>> assert ub.readfrom(fpath2).strip() == 'writing2'
        >>> info1['proc'].wait()
        >>> info2['proc'].wait()
    """
    if kwargs:  # nocover
        if 'verbout' in kwargs:
            warnings.warn(
                '`verbout` is deprecated and will be removed. '
                'Use `tee` instead', DeprecationWarning)
            tee = kwargs.pop('verbout')

        if 'detatch' in kwargs:
            warnings.warn(
                '`detatch` is deprecated (misspelled) and will be removed. '
                'Use `detach` instead', DeprecationWarning)
            detach = kwargs.pop('detatch')

        if kwargs:
            raise ValueError('Unknown kwargs: {}'.format(list(kwargs.keys)))

    # Determine if command is specified as text or a tuple
    if isinstance(command, six.string_types):
        command_text = command
        command_tup = None
    else:
        import pipes
        command_tup = command
        command_text = ' '.join(list(map(pipes.quote, command_tup)))

    if shell or sys.platform.startswith('win32'):
        # When shell=True, args is sent to the shell (e.g. bin/sh) as text
        args = command_text
    else:
        # When shell=False, args is a list of executable and arguments
        if command_tup is None:
            # parse this out of the string
            # NOTE: perhaps use the solution from [3] here?
            import shlex
            command_tup = shlex.split(command_text)
            # command_tup = shlex.split(command_text, posix=not WIN32)
        args = command_tup

    if tee is None:
        tee = verbose > 0
    if verbose > 1:
        import os
        import platform
        import getpass
        from ubelt import util_path
        if verbose > 2:
            try:
                print('┌─── START CMD ───')
            except Exception:  # nocover
                print('+=== START CMD ===')
        cwd_ = os.getcwd() if cwd is None else cwd
        compname = platform.node()
        username = getpass.getuser()
        cwd_ = util_path.compressuser(cwd_)
        ps1 = '[ubelt.cmd] {}@{}:{}$ '.format(username, compname, cwd_)
        print(ps1 + command_text)

    # Create a new process to execute the command
    def make_proc():
        # delay the creation of the process until we validate all args
        import subprocess
        proc = subprocess.Popen(args, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, shell=shell,
                                universal_newlines=True, cwd=cwd, env=env)
        return proc

    if detach:
        info = {'proc': make_proc(), 'command': command_text}
        if verbose > 0:  # nocover
            print('...detaching')
    else:
        if tee:
            # we need to tee output and start threads if tee is False?
            stdout, stderr = sys.stdout, sys.stderr
            proc, logged_out, logged_err = _tee_output(make_proc, stdout, stderr,
                                                       backend=tee_backend)

            try:
                out = ''.join(logged_out)
            except UnicodeDecodeError:  # nocover
                out = '\n'.join(_.decode('utf-8') for _ in logged_out)
            try:
                err = ''.join(logged_err)
            except UnicodeDecodeError:  # nocover
                err = '\n'.join(_.decode('utf-8') for _ in logged_err)
            (out_, err_) = proc.communicate()
        else:
            proc = make_proc()
            (out, err) = proc.communicate()
        # calling wait means that the process will terminate and it is safe to
        # return a reference to the process object.
        ret = proc.wait()
        info = {
            'out': out,
            'err': err,
            'ret': ret,
            'proc': proc,
            'cwd': cwd,
            'command': command_text
        }
        if verbose > 2:
            # https://en.wikipedia.org/wiki/Box-drawing_character
            try:
                print('└─── END CMD ───')
            except Exception:  # nocover
                print('L___ END CMD ___')
    return info
def cmd(command, shell=False, detach=False, verbose=0, tee=None, cwd=None,
        env=None, tee_backend='auto', verbout=None, **kwargs):
    """
    Executes a command in a subprocess.

    The advantage of this wrapper around subprocess is that
    (1) you control if the subprocess prints to stdout,
    (2) the text written to stdout and stderr is returned for parsing,
    (3) cross platform behavior that lets you specify the command as a string
    or tuple regardless of whether or not shell=True.
    (4) ability to detach, return the process object and allow the process to
    run in the background (eventually we may return a Future object instead).

    Args:
        command (str or Sequence): bash-like command string or tuple of
            executable and args

        shell (bool): if True, process is run in shell, defaults to False.

        detach (bool): if True, process is detached and run in background,
            defaults to False.

        verbose (int): verbosity mode. Can be 0, 1, 2, or 3. Defaults to 0.

        tee (bool, optional): if True, simultaneously writes to stdout while
            capturing output from the command. If not specified, defaults to
            True if verbose > 0.  If detech is True, then this argument is
            ignored.

        cwd (PathLike, optional): path to run command

        env (str, optional): environment passed to Popen

        tee_backend (str, optional): backend for tee output.
            Valid choices are: "auto", "select" (POSIX only), and "thread".

        **kwargs: only used to support deprecated arguments

    Returns:
        dict: info - information about command status.
            if detach is False `info` contains captured standard out,
            standard error, and the return code
            if detach is False `info` contains a reference to the process.

    Notes:
        Inputs can either be text or tuple based. On UNIX we ensure conversion
        to text if shell=True, and to tuple if shell=False. On windows, the
        input is always text based.  See [3] for a potential cross-platform
        shlex solution for windows.

    CommandLine:
        python -m ubelt.util_cmd cmd
        python -c "import ubelt as ub; ub.cmd('ping localhost -c 2', verbose=2)"

    References:
        [1] https://stackoverflow.com/questions/11495783/redirect-subprocess-stderr-to-stdout
        [2] https://stackoverflow.com/questions/7729336/how-can-i-print-and-display-subprocess-stdout-and-stderr-output-without-distorti
        [3] https://stackoverflow.com/questions/33560364/python-windows-parsing-command-lines-with-shlex

    Example:
        >>> info = cmd(('echo', 'simple cmdline interface'), verbose=1)
        simple cmdline interface
        >>> assert info['ret'] == 0
        >>> assert info['out'].strip() == 'simple cmdline interface'
        >>> assert info['err'].strip() == ''

    Doctest:
        >>> info = cmd('echo str noshell', verbose=0)
        >>> assert info['out'].strip() == 'str noshell'

    Doctest:
        >>> # windows echo will output extra single quotes
        >>> info = cmd(('echo', 'tuple noshell'), verbose=0)
        >>> assert info['out'].strip().strip("'") == 'tuple noshell'

    Doctest:
        >>> # Note this command is formatted to work on win32 and unix
        >>> info = cmd('echo str&&echo shell', verbose=0, shell=True)
        >>> assert info['out'].strip() == 'str' + chr(10) + 'shell'

    Doctest:
        >>> info = cmd(('echo', 'tuple shell'), verbose=0, shell=True)
        >>> assert info['out'].strip().strip("'") == 'tuple shell'

    Doctest:
        >>> import ubelt as ub
        >>> from os.path import join, exists
        >>> fpath1 = join(ub.get_app_cache_dir('ubelt'), 'cmdout1.txt')
        >>> fpath2 = join(ub.get_app_cache_dir('ubelt'), 'cmdout2.txt')
        >>> ub.delete(fpath1)
        >>> ub.delete(fpath2)
        >>> info1 = ub.cmd(('touch', fpath1), detach=True)
        >>> info2 = ub.cmd('echo writing2 > ' + fpath2, shell=True, detach=True)
        >>> while not exists(fpath1):
        ...     pass
        >>> while not exists(fpath2):
        ...     pass
        >>> assert ub.readfrom(fpath1) == ''
        >>> assert ub.readfrom(fpath2).strip() == 'writing2'
        >>> info1['proc'].wait()
        >>> info2['proc'].wait()
    """
    if kwargs:  # nocover
        if 'verbout' in kwargs:
            warnings.warn(
                '`verbout` is deprecated and will be removed. '
                'Use `tee` instead', DeprecationWarning)
            tee = kwargs.pop('verbout')

        if 'detatch' in kwargs:
            warnings.warn(
                '`detatch` is deprecated (misspelled) and will be removed. '
                'Use `detach` instead', DeprecationWarning)
            detach = kwargs.pop('detatch')

        if kwargs:
            raise ValueError('Unknown kwargs: {}'.format(list(kwargs.keys)))

    # Determine if command is specified as text or a tuple
    if isinstance(command, six.string_types):
        command_text = command
        command_tup = None
    else:
        import pipes
        command_tup = command
        command_text = ' '.join(list(map(pipes.quote, command_tup)))

    if shell or sys.platform.startswith('win32'):
        # When shell=True, args is sent to the shell (e.g. bin/sh) as text
        args = command_text
    else:
        # When shell=False, args is a list of executable and arguments
        if command_tup is None:
            # parse this out of the string
            # NOTE: perhaps use the solution from [3] here?
            import shlex
            command_tup = shlex.split(command_text)
            # command_tup = shlex.split(command_text, posix=not WIN32)
        args = command_tup

    if tee is None:
        tee = verbose > 0
    if verbose > 1:
        import os
        import platform
        import getpass
        from ubelt import util_path
        if verbose > 2:
            try:
                print('┌─── START CMD ───')
            except Exception:  # nocover
                print('+=== START CMD ===')
        cwd_ = os.getcwd() if cwd is None else cwd
        compname = platform.node()
        username = getpass.getuser()
        cwd_ = util_path.compressuser(cwd_)
        ps1 = '[ubelt.cmd] {}@{}:{}$ '.format(username, compname, cwd_)
        print(ps1 + command_text)

    # Create a new process to execute the command
    def make_proc():
        # delay the creation of the process until we validate all args
        import subprocess
        proc = subprocess.Popen(args, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, shell=shell,
                                universal_newlines=True, cwd=cwd, env=env)
        return proc

    if detach:
        info = {'proc': make_proc(), 'command': command_text}
        if verbose > 0:  # nocover
            print('...detaching')
    else:
        if tee:
            # we need to tee output and start threads if tee is False?
            stdout, stderr = sys.stdout, sys.stderr
            proc, logged_out, logged_err = _tee_output(make_proc, stdout, stderr,
                                                       backend=tee_backend)

            try:
                out = ''.join(logged_out)
            except UnicodeDecodeError:  # nocover
                out = '\n'.join(_.decode('utf-8') for _ in logged_out)
            try:
                err = ''.join(logged_err)
            except UnicodeDecodeError:  # nocover
                err = '\n'.join(_.decode('utf-8') for _ in logged_err)
            (out_, err_) = proc.communicate()
        else:
            proc = make_proc()
            (out, err) = proc.communicate()
        # calling wait means that the process will terminate and it is safe to
        # return a reference to the process object.
        ret = proc.wait()
        info = {
            'out': out,
            'err': err,
            'ret': ret,
            'proc': proc,
            'cwd': cwd,
            'command': command_text
        }
        if verbose > 2:
            # https://en.wikipedia.org/wiki/Box-drawing_character
            try:
                print('└─── END CMD ───')
            except Exception:  # nocover
                print('L___ END CMD ___')
    return info
Executes a command in a subprocess.

    The advantage of this wrapper around subprocess is that
    (1) you control if the subprocess prints to stdout,
    (2) the text written to stdout and stderr is returned for parsing,
    (3) cross platform behavior that lets you specify the command as a string
    or tuple regardless of whether or not shell=True.
    (4) ability to detach, return the process object and allow the process to
    run in the background (eventually we may return a Future object instead).

    Args:
        command (str or Sequence): bash-like command string or tuple of
            executable and args

        shell (bool): if True, process is run in shell, defaults to False.

        detach (bool): if True, process is detached and run in background,
            defaults to False.

        verbose (int): verbosity mode. Can be 0, 1, 2, or 3. Defaults to 0.

        tee (bool, optional): if True, simultaneously writes to stdout while
            capturing output from the command. If not specified, defaults to
            True if verbose > 0.  If detech is True, then this argument is
            ignored.

        cwd (PathLike, optional): path to run command

        env (str, optional): environment passed to Popen

        tee_backend (str, optional): backend for tee output.
            Valid choices are: "auto", "select" (POSIX only), and "thread".

        **kwargs: only used to support deprecated arguments

    Returns:
        dict: info - information about command status.
            if detach is False `info` contains captured standard out,
            standard error, and the return code
            if detach is False `info` contains a reference to the process.

    Notes:
        Inputs can either be text or tuple based. On UNIX we ensure conversion
        to text if shell=True, and to tuple if shell=False. On windows, the
        input is always text based.  See [3] for a potential cross-platform
        shlex solution for windows.

    CommandLine:
        python -m ubelt.util_cmd cmd
        python -c "import ubelt as ub; ub.cmd('ping localhost -c 2', verbose=2)"

    References:
        [1] https://stackoverflow.com/questions/11495783/redirect-subprocess-stderr-to-stdout
        [2] https://stackoverflow.com/questions/7729336/how-can-i-print-and-display-subprocess-stdout-and-stderr-output-without-distorti
        [3] https://stackoverflow.com/questions/33560364/python-windows-parsing-command-lines-with-shlex

    Example:
        >>> info = cmd(('echo', 'simple cmdline interface'), verbose=1)
        simple cmdline interface
        >>> assert info['ret'] == 0
        >>> assert info['out'].strip() == 'simple cmdline interface'
        >>> assert info['err'].strip() == ''

    Doctest:
        >>> info = cmd('echo str noshell', verbose=0)
        >>> assert info['out'].strip() == 'str noshell'

    Doctest:
        >>> # windows echo will output extra single quotes
        >>> info = cmd(('echo', 'tuple noshell'), verbose=0)
        >>> assert info['out'].strip().strip("'") == 'tuple noshell'

    Doctest:
        >>> # Note this command is formatted to work on win32 and unix
        >>> info = cmd('echo str&&echo shell', verbose=0, shell=True)
        >>> assert info['out'].strip() == 'str' + chr(10) + 'shell'

    Doctest:
        >>> info = cmd(('echo', 'tuple shell'), verbose=0, shell=True)
        >>> assert info['out'].strip().strip("'") == 'tuple shell'

    Doctest:
        >>> import ubelt as ub
        >>> from os.path import join, exists
        >>> fpath1 = join(ub.get_app_cache_dir('ubelt'), 'cmdout1.txt')
        >>> fpath2 = join(ub.get_app_cache_dir('ubelt'), 'cmdout2.txt')
        >>> ub.delete(fpath1)
        >>> ub.delete(fpath2)
        >>> info1 = ub.cmd(('touch', fpath1), detach=True)
        >>> info2 = ub.cmd('echo writing2 > ' + fpath2, shell=True, detach=True)
        >>> while not exists(fpath1):
        ...     pass
        >>> while not exists(fpath2):
        ...     pass
        >>> assert ub.readfrom(fpath1) == ''
        >>> assert ub.readfrom(fpath2).strip() == 'writing2'
        >>> info1['proc'].wait()
        >>> info2['proc'].wait()