import os
class Silence:
"""Context manager which uses low-level file descriptors to suppress
output to stdout/stderr, optionally redirecting to the named file(s).
>>> import sys, numpy.f2py
>>> # build a test fortran extension module with F2PY
...
>>> with open('hellofortran.f', 'w') as f:
... f.write('''\
... integer function foo (n)
... integer n
... print *, "Hello from Fortran!"
... print *, "n = ", n
... foo = n
... end
... ''')
...
>>> sys.argv = ['f2py', '-c', '-m', 'hellofortran', 'hellofortran.f']
>>> with Silence():
... # assuming this succeeds, since output is suppressed
... numpy.f2py.main()
...
>>> import hellofortran
>>> foo = hellofortran.foo(1)
Hello from Fortran!
n = 1
>>> print "Before silence"
Before silence
>>> with Silence(stdout='output.txt', mode='w'):
... print "Hello from Python!"
... bar = hellofortran.foo(2)
... with Silence():
... print "This will fall on deaf ears"
... baz = hellofortran.foo(3)
... print "Goodbye from Python!"
...
...
>>> print "After silence"
After silence
>>> # ... do some other stuff ...
...
>>> with Silence(stderr='output.txt', mode='a'):
... # appending to existing file
... print >> sys.stderr, "Hello from stderr"
... print "Stdout redirected to os.devnull"
...
...
>>> # check the redirected output
...
>>> with open('output.txt', 'r') as f:
... print "=== contents of 'output.txt' ==="
... print f.read()
... print "================================"
...
=== contents of 'output.txt' ===
Hello from Python!
Hello from Fortran!
n = 2
Goodbye from Python!
Hello from stderr
================================
>>> foo, bar, baz
(1, 2, 3)
>>>
"""
def __init__(self, stdout=os.devnull, stderr=os.devnull, mode='w'):
self.outfiles = stdout, stderr
self.combine = (stdout == stderr)
self.mode = mode
def __enter__(self):
import sys
self.sys = sys
# save previous stdout/stderr
self.saved_streams = saved_streams = sys.__stdout__, sys.__stderr__
self.fds = fds = [s.fileno() for s in saved_streams]
self.saved_fds = map(os.dup, fds)
# flush any pending output
for s in saved_streams: s.flush()
# open surrogate files
if self.combine:
null_streams = [open(self.outfiles[0], self.mode, 0)] * 2
if self.outfiles[0] != os.devnull:
# disable buffering so output is merged immediately
sys.stdout, sys.stderr = map(os.fdopen, fds, ['w']*2, [0]*2)
else: null_streams = [open(f, self.mode, 0) for f in self.outfiles]
self.null_fds = null_fds = [s.fileno() for s in null_streams]
self.null_streams = null_streams
# overwrite file objects and low-level file descriptors
map(os.dup2, null_fds, fds)
def __exit__(self, *args):
sys = self.sys
# flush any pending output
for s in self.saved_streams: s.flush()
# restore original streams and file descriptors
map(os.dup2, self.saved_fds, self.fds)
sys.stdout, sys.stderr = self.saved_streams
# clean up
for s in self.null_streams: s.close()
for fd in self.saved_fds: os.close(fd)
return False
Diff to Previous Revision
--- revision 2 2011-02-12 00:03:21
+++ revision 3 2012-01-26 02:14:25
@@ -104,4 +104,5 @@
sys.stdout, sys.stderr = self.saved_streams
# clean up
for s in self.null_streams: s.close()
+ for fd in self.saved_fds: os.close(fd)
return False
