Was created a class that inherit from ContextDecorator
and implement methods __enter__()
and __exit__()
to
use as Python ContextManager
Based in:
Official References:
Was created a class that inherit from ContextDecorator
and implement methods __enter__()
and __exit__()
to
use as Python ContextManager
Based in:
Official References:
class CaptureOutput(ContextDecorator): | |
""" | |
Get the console output from a c/c++ shared library, and | |
save it into a variable | |
References: | |
- Based in stackoverflow answer: `Capture the stdout from a c++ shared library`_. | |
Official References | |
- Context Manager decorator official reference: ContextDecorator_. | |
- With statement official reference: WithStatement_. | |
- Context Manager official reference: ContextManager_. | |
.. _Capture the stdout from a c++ shared library: https://stackoverflow.com/a/24277852 | |
.. _ContextDecorator: https://docs.python.org/3/library/contextlib.html#contextlib.ContextDecorator | |
.. _WithStatement: https://docs.python.org/3/reference/compound_stmts.html#with | |
.. _ContextManager: https://docs.python.org/3/library/stdtypes.html#typecontextmanager | |
""" | |
_thread: Thread | |
def __init__(self, stream=sys.stdout): | |
""" | |
:param stream: sys.stdout or sys.stderr from sys package/module | |
:Example: | |
# Wrap the c/c++ shared library call into a Context manager | |
with CaptureOutput() as cm: | |
mylib.print() | |
captured = cm.captured_output | |
# Warning: Don't use any output Python function here (e.g print(captured)) | |
""" | |
self.stream_fileno = stream.fileno() | |
self.stream_save = 0 | |
self.stdout_pipe = (0, 0) | |
self.captured_output = '' | |
def __enter__(self): | |
# Create pipe and dup2() the write end of it on top of stdout, saving a copy | |
# of the old stdout | |
self.stream_save = os.dup(self.stream_fileno) | |
self.stdout_pipe = os.pipe() | |
os.dup2(self.stdout_pipe[1], self.stream_fileno) | |
os.close(self.stdout_pipe[1]) | |
self._thread = Thread(target=self._drain_pipe) | |
self._thread.start() | |
return self | |
def __exit__(self, type, value, traceback): | |
# Close the write end of the pipe to unblock the reader thread and trigger it | |
# to exit | |
os.close(self.stream_fileno) | |
self._thread.join() | |
# Clean up the pipe and restore the original stdout | |
os.close(self.stdout_pipe[0]) | |
os.dup2(self.stream_save, self.stream_fileno) | |
os.close(self.stream_save) | |
return False | |
def _drain_pipe(self): | |
while True: | |
data = os.read(self.stdout_pipe[0], 1024) | |
if not data: | |
break | |
self.captured_output += str(data) |