2

Right now I use this to catch the output of a Python function and store it in a variable:

import io
from contextlib import redirect_stdout

def catch_output(func):
    result = io.StringIO()
    with redirect_stdout(result):
        func()
    return result.getvalue()

output = catch_output(my_func)

This works fine, but it also mutes the console until the func call finished. Does anybody know if I can write/pipe the live output of the func to the console and store it in a variable at the same time?

5
  • Have you tried threads? Commented Jan 10, 2017 at 22:00
  • Googling tee stream in Python should yield some reasonable solutions. Commented Jan 10, 2017 at 22:06
  • to add more to @justin_shapiro 's comment, you have at least two things that need to happen at the same time.. you have a function that is running, and printing text to a file buffer (redirected from stdout) and something running at the same time to periodically copy that buffer to some sort of storage and also print it out to the real stdout. Commented Jan 10, 2017 at 22:07
  • What operating system are you using? Commented Jan 10, 2017 at 23:00
  • Thanks for the comments! I'm using Linux. Commented Jan 10, 2017 at 23:41

3 Answers 3

4

You can redirect stdout to a custom file-like object that forwards writes to multiple files:

import contextlib
import io
import sys


class TeeIO:
    def __init__(self, original, target):
        self.original = original
        self.target = target

    def write(self, b):
        self.original.write(b)
        self.target.write(b)


@contextlib.contextmanager
def tee_stdout(target):
    tee = TeeIO(sys.stdout, target)
    with contextlib.redirect_stdout(tee):
        yield


buf = io.StringIO()
with tee_stdout(buf):
    print("foo")
print(buf.getvalue())
Sign up to request clarification or add additional context in comments.

3 Comments

So simple, yet couldn't find this anywhere else. Thank you!
Will this fail if I use other functions on the output, like seek, for instance? Is there a reason TeeIO isn't derived from something like StringIO? It's confusing to understand what ends up happening here.
Or flush, also
2

This is what I ended up using. I thought I leave this here for people who have a hard time with classes and oop, like me.

import sys
import io
from contextlib import redirect_stdout


def get_multi_writer(streams):
    writer = type('obj', (object,), {})
    writer.write = lambda s: [stream.write(s) for stream in streams]
    return writer


def catch_output(func, args, kwargs):
    streams = [sys.stdout, io.StringIO()]
    with redirect_stdout(get_multi_writer(streams)):
        func(*args, **kwargs)
    return streams[1].getvalue()


print(catch_output(my_func, [], {}))

Comments

1

As per the suggestions from the comments I've made and example turning our function into a thread so we can simultaneously check for output from that function periodically and copy it to the real stdout.

import sys
import time
import threading
from cStringIO import StringIO

def foo(n):
    for x in range(n):
        time.sleep(1) #intense computation
        print('test: {}'.format(n))


#i'm using python 2.7 so I don't have contextlib.redirect_stdout
realstdout = sys.stdout
sys.stdout = StringIO()

t = threading.Thread(target=foo, args=(10,))
t.start()

lastpos = 0 #last cursor position in file

while True:
    t.join(.1) #wait .1 sec for thread to complete

    if sys.stdout.tell() != lastpos: #data has been written to stdout
        sys.stdout.seek(lastpos) #go back to our last position
        realstdout.write(sys.stdout.read()) #read the data to the real stdout
        lastpos = sys.stdout.tell() #update lastpos

    if not t.is_alive(): #when we're done
        break

sys.stdout.seek(0) #seek back to beginning of file
output = sys.stdout.read() #copy to a usable variable
sys.stdout = realstdout #reset stdout

1 Comment

node: I use python 2.7 so there may be some semantics that are different, and a more elegant way to do this. Most notably the differences in how unicode is handled

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.