ubelt.util_stream module

Functions for capturing and redirecting IO streams with optional tee-functionality.

The CaptureStdout captures all text sent to stdout and optionally prevents it from actually reaching stdout.

The TeeStringIO does the same thing but for arbitrary streams. It is how the former is implemented.

class ubelt.util_stream.TeeStringIO(redirect=None)

Bases: StringIO

An IO object that writes to itself and another IO stream.

Variables:

redirect (io.IOBase | None) – The other stream to write to.

Example

>>> import ubelt as ub
>>> import io
>>> redirect = io.StringIO()
>>> self = ub.TeeStringIO(redirect)
>>> self.write('spam')
>>> assert self.getvalue() == 'spam'
>>> assert redirect.getvalue() == 'spam'

Parameters:

redirect (io.IOBase) – The other stream to write to.

isatty()

Returns true of the redirect is a terminal.

Note

Needed for IPython.embed to work properly when this class is used to override stdout / stderr.

SeeAlso:

io.IOBase.isatty()

Returns:

bool

fileno()

Returns underlying file descriptor of the redirected IOBase object if one exists.

Returns:

the integer corresponding to the file descriptor

Return type:

int

SeeAlso:

io.IOBase.fileno()

Example

>>> import ubelt as ub
>>> dpath = ub.Path.appdir('ubelt/tests/util_stream').ensuredir()
>>> fpath = dpath / 'fileno-test.txt'
>>> with open(fpath, 'w') as file:
>>>     self = ub.TeeStringIO(file)
>>>     descriptor = self.fileno()
>>>     print(f'descriptor={descriptor}')
>>>     assert isinstance(descriptor, int)

Example

>>> # Test errors
>>> # Not sure the best way to test, this func is important for
>>> # capturing stdout when ipython embedding
>>> import io
>>> import pytest
>>> import ubelt as ub
>>> with pytest.raises(io.UnsupportedOperation):
>>>     ub.TeeStringIO(redirect=io.StringIO()).fileno()
>>> with pytest.raises(io.UnsupportedOperation):
>>>     ub.TeeStringIO(None).fileno()

property encoding

Gets the encoding of the redirect IO object

FIXME:

My complains that this violates the Liskov substitution principle because the return type can be str or None, whereas the parent class always returns a None. In the future we may raise an exception instead of returning None.

SeeAlso:

io.TextIOBase.encoding

Example

>>> import ubelt as ub
>>> redirect = io.StringIO()
>>> assert ub.TeeStringIO(redirect).encoding is None
>>> assert ub.TeeStringIO(None).encoding is None
>>> assert ub.TeeStringIO(sys.stdout).encoding is sys.stdout.encoding
>>> redirect = io.TextIOWrapper(io.StringIO())
>>> assert ub.TeeStringIO(redirect).encoding is redirect.encoding

write(msg)

Write to this and the redirected stream

Parameters:

msg (str) – the data to write

SeeAlso:

io.TextIOBase.write()

Example

>>> import ubelt as ub
>>> dpath = ub.Path.appdir('ubelt/tests/util_stream').ensuredir()
>>> fpath = dpath / 'write-test.txt'
>>> with open(fpath, 'w') as file:
>>>     self = ub.TeeStringIO(file)
>>>     n = self.write('hello world')
>>>     assert n == 11
>>> assert self.getvalue() == 'hello world'
>>> assert fpath.read_text() == 'hello world'

flush()

Flush to this and the redirected stream

SeeAlso:

io.IOBase.flush()

class ubelt.util_stream.CaptureStdout(suppress=True, enabled=True)

Bases: CaptureStream

Context manager that captures stdout and stores it in an internal stream.

Depending on the value of supress, the user can control if stdout is printed (i.e. if stdout is tee-ed or supressed) while it is being captured.

SeeAlso:

contextlib.redirect_stdout() - similar, but does not have the

ability to print stdout while it is being captured.

Variables:

• text (str | None) – internal storage for the most recent part

• parts (List[str]) – internal storage for all parts

• cap_stdout (None | TeeStringIO) – internal stream proxy

• orig_stdout (io.TextIOBase) – internal pointer to the original stdout stream

Example

>>> import ubelt as ub
>>> self = ub.CaptureStdout(suppress=True)
>>> print('dont capture the table flip (╯°□°）╯︵ ┻━┻')
>>> with self:
...     text = 'capture the heart ♥'
...     print(text)
>>> print('dont capture look of disapproval ಠ_ಠ')
>>> assert isinstance(self.text, str)
>>> assert self.text == text + '\n', 'failed capture text'

Example

>>> import ubelt as ub
>>> self = ub.CaptureStdout(suppress=False)
>>> with self:
...     print('I am captured and printed in stdout')
>>> assert self.text.strip() == 'I am captured and printed in stdout'

Example

>>> import ubelt as ub
>>> self = ub.CaptureStdout(suppress=True, enabled=False)
>>> with self:
...     print('dont capture')
>>> assert self.text is None

Parameters:

• suppress (bool) – if True, stdout is not printed while captured. Defaults to True.

• enabled (bool) – does nothing if this is False. Defaults to True.

log_part()

Log what has been captured so far

start()

stop()

Example

>>> import ubelt as ub
>>> ub.CaptureStdout(enabled=False).stop()
>>> ub.CaptureStdout(enabled=True).stop()

close()

class ubelt.util_stream.CaptureStream

Bases: object

Generic class for capturing streaming output from stdout or stderr