A context manager to temporarily redirect stdout or stderr

    e.g.:

    with stdchannel_redirected(sys.stderr, os.devnull):
        if compiler.has_function('clock_gettime', libraries=['rt']):
            libraries.append('rt')
504990
def stdchannel_redirected(stdchannel, dest_filename, fake=False):
    
    if fake:
        yield
        return
    oldstdchannel = dest_file = None
    try:
        oldstdchannel = os.dup(stdchannel.fileno())
        dest_file = open(dest_filename, 'w')
        os.dup2(dest_file.fileno(), stdchannel.fileno())

        yield
    finally:
        if oldstdchannel is not None:
            os.dup2(oldstdchannel, stdchannel.fileno())
        if dest_file is not None:
            dest_file.close()
654299
def output_stderr(self, text):
        ""
        binary_stderr.write(b''.join([
            self._red,
            b't=%07d' % (time.time() - self._t0),
            self._reset,
            b' ',
            text,
        ]))
        binary_stderr.flush()
66294
def shell_out(cmd, stderr=STDOUT, cwd=None):
    """"""
    if cwd is None:
        from os import getcwd
        cwd = getcwd()  # TODO do I need to normalize this on Windows

    out = check_output(cmd, cwd=cwd, stderr=stderr, universal_newlines=True)
    return _clean_output(out)
1237490
def tempput(local_path=None, remote_path=None, use_sudo=False,
            mirror_local_mode=False, mode=None):
    """"""
    import warnings
    warnings.simplefilter('ignore', RuntimeWarning)
    if remote_path is None:
        remote_path = os.tempnam()
    put(local_path, remote_path, use_sudo, mirror_local_mode, mode)
    yield remote_path
    run("rm '{}'".format(remote_path))
1046251
def _capture_stdout(callable: Callable[..., CaptureResult], *args, **kwargs) -> CaptureResult:
        
        stream = StringIO()
        with redirect_stdout(stream):
            return_value = callable(*args, **kwargs)
        assert return_value.stdout is None, "stdout appears to have already been captured"
        return CaptureResult(return_value=return_value.return_value, stdout=stream.getvalue(),
                             stderr=return_value.stderr, exception=return_value.exception)
1306247
function handleStdoutStderr(settings) {
  if (settings.options.stdio !== "ignore") {
    settings.scriptHandler.stdout.on("data", function (stdout) {
      if (typeof settings.stdoutFunction === "function") {
        settings.stdoutFunction(stdout.toString("utf8"));
      }
    });

    if (typeof settings.stderrFunction === "function") {
      settings.scriptHandler.stderr.on("data", function (stderr) {
        settings.stderrFunction(stderr.toString("utf8"));
      });
    }
  }
}
287107
def stderr_avail(self):
        """"""
        data = self.interpreter.stderr_write.empty_queue()
        if data:
            self.write(data, error=True)
            self.flush(error=True)
366284
def stderr(self):
        
        stream = self._log_file_handle if self._log_file_handle else osutils.stderr()
        return StreamWriter(stream, self._is_debugging)
1053856
def redirect_stdout(self, enabled=True, log_level=logging.INFO):
        
        if enabled:
            if self.__stdout_wrapper:
                self.__stdout_wrapper.update_log_level(log_level=log_level)
            else:
                self.__stdout_wrapper = StdOutWrapper(logger=self, log_level=log_level)

            self.__stdout_stream = self.__stdout_wrapper
        else:
            self.__stdout_stream = _original_stdout

        # Assign the new stream to sys.stdout
        sys.stdout = self.__stdout_stream
1056493
def captureOutput(func, *args, **kwargs):
    
    stdout = sys.stdout
    sys.stdout = StringIO.StringIO()
    stderr = sys.stderr
    sys.stderr = StringIO.StringIO()
    try:
        func(*args, **kwargs)
        stdoutOutput = sys.stdout.getvalue()
        stderrOutput = sys.stderr.getvalue()
    finally:
        sys.stdout.close()
        sys.stdout = stdout
        sys.stderr.close()
        sys.stderr = stderr
    return stdoutOutput, stderrOutput
1098525
def redirect_output!
      if log_path = options[:log_path]
        # if the log directory doesn't exist, create it
        FileUtils.mkdir_p File.dirname(options[:log_path]), :mode => 0755
        # touch the log file to create it
        FileUtils.touch log_path
        # Set permissions on the log file
        File.chmod(0644, log_path)
        # Reopen $stdout (NOT +STDOUT+) to start writing to the log file
        $stdout.reopen(log_path, 'a')
        # Redirect $stderr to $stdout
        $stderr.reopen $stdout
        $stdout.sync = true
      else # redirect to /dev/null
        # We're not bothering to sync if we're dumping to /dev/null
        # because /dev/null doesn't care about buffered output
        $stdin.reopen '/dev/null'
        $stdout.reopen '/dev/null', 'a'
        $stderr.reopen $stdout
      end
      log_path = options[:log_path] ? options[:log_path] : '/dev/null'
    end
1273364
def tee_output_fd():
    """"""
    with NamedTemporaryFile(mode='w+') as target:
        original_stdout_fd = 1
        original_stderr_fd = 2
        target_fd = target.fileno()

        # Save a copy of the original stdout and stderr file descriptors
        saved_stdout_fd = os.dup(original_stdout_fd)
        saved_stderr_fd = os.dup(original_stderr_fd)

        try:
            # we call os.setsid to move process to a new process group
            # this is done to avoid receiving KeyboardInterrupts (see #149)
            # in Python 3 we could just pass start_new_session=True
            tee_stdout = subprocess.Popen(
                ['tee', '-a', target.name], preexec_fn=os.setsid,
                stdin=subprocess.PIPE, stdout=1)
            tee_stderr = subprocess.Popen(
                ['tee', '-a', target.name], preexec_fn=os.setsid,
                stdin=subprocess.PIPE, stdout=2)
        except (FileNotFoundError, OSError, AttributeError):
            # No tee found in this operating system. Trying to use a python
            # implementation of tee. However this is slow and error-prone.
            tee_stdout = subprocess.Popen(
                [sys.executable, "-m", "sacred.pytee"],
                stdin=subprocess.PIPE, stderr=target_fd)
            tee_stderr = subprocess.Popen(
                [sys.executable, "-m", "sacred.pytee"],
                stdin=subprocess.PIPE, stdout=target_fd)

        flush()
        os.dup2(tee_stdout.stdin.fileno(), original_stdout_fd)
        os.dup2(tee_stderr.stdin.fileno(), original_stderr_fd)
        out = CapturedStdout(target)

        try:
            yield out  # let the caller do their printing
        finally:
            flush()

            # then redirect stdout back to the saved fd
            tee_stdout.stdin.close()
            tee_stderr.stdin.close()

            # restore original fds
            os.dup2(saved_stdout_fd, original_stdout_fd)
            os.dup2(saved_stderr_fd, original_stderr_fd)

            # wait for completion of the tee processes with timeout
            # implemented using a timer because timeout support is py3 only
            def kill_tees():
                tee_stdout.kill()
                tee_stderr.kill()

            tee_timer = Timer(1, kill_tees)
            try:
                tee_timer.start()
                tee_stdout.wait()
                tee_stderr.wait()
            finally:
                tee_timer.cancel()

            os.close(saved_stdout_fd)
            os.close(saved_stderr_fd)
            out.finalize()
99082
def raw_cmd(self, *args, **kwargs):
        ''''''
        cmds = [self.adb_path()] + self._host_port_args + list(args)
        kwargs['stdout'] = kwargs.get('stdout', subprocess.PIPE)
        kwargs['stderr'] = kwargs.get('stderr', subprocess.PIPE)
        # if os.name != "nt":
        #     cmd_line = [" ".join(cmd_line)]
        cmds = [strutils.decode(v) for v in cmds]
        return subprocess.Popen(cmds, **kwargs)
797750
def error(*args):
    """"""
    if sys.stdin.isatty():
        print('ERROR:', *args, file=sys.stderr)
    else:
        notify_error(*args)
147392
def get_stream_handler(stream=sys.stderr, level=logging.INFO,
                       log_format=log_formats.easy_read):
    
    sh = logging.StreamHandler(stream)
    sh.setLevel(level)
    sh.setFormatter(logging.Formatter(log_format))
    return sh
1304344
def file_or_stderr(filename, *, mode="a", encoding="utf-8"):
    
    if filename is not None:
        return open(filename, mode, encoding=encoding)

    @contextmanager
    def stderr_wrapper():
        yield sys.stderr

    return stderr_wrapper()
990035
def redirect_stderr(self, enabled=True, log_level=logging.ERROR):
        
        if enabled:
            if self.__stderr_wrapper:
                self.__stderr_wrapper.update_log_level(log_level=log_level)
            else:
                self.__stderr_wrapper = StdErrWrapper(logger=self, log_level=log_level)

            self.__stderr_stream = self.__stderr_wrapper
        else:
            self.__stderr_stream = _original_stderr

        # Assign the new stream to sys.stderr
        sys.stderr = self.__stderr_stream