stouputils.print.output_stream module#

LINE_TOKENS_RE: Pattern[str] = re.compile('(\\n|\\r|\\x1b\\[\\d*[AB])')[source]#

newlines, carriage returns and line moves.

Type:: Splits a text around what moves a terminal cursor

REDRAW_CHECKPOINT_SECONDS: float = 60.0[source]#: Interval at which a file receives the current state of a line still being redrawn, such as a long progress bar.

class LineState( checkpoint_seconds: float = 60.0, mode: Literal['start', 'text', 'redraw', 'repeat'] = 'start', buffer: str = '', shown: bool = False, carriage: bool = False, held_repeat: str = '', since: float = 0.0, )[source]#

Bases: object

Where a non-terminal output stands in the line one thread is writing, so it receives what a terminal ends up showing.

Text is held until its line ends or the stream is flushed, like a terminal’s line-buffered stdout. Lines from several threads therefore never cut into each other. A line drawn again after a carriage return, such as a progress bar, is written once in its final state, plus every checkpoint_seconds while it lasts. A line starting with a cursor move rewrites the previous one, as stouputils does for a repeated message, and the last of such a series is written when the series ends.

>>> state = LineState()
>>> state.feed("\r 10%") + state.feed("\r100%\n") + state.feed("done\n")
'100%\ndone\n'
>>> state.feed("same\n") + state.feed("\x1b[1Asame (x2)\n") + state.feed("\x1b[1Asame (x3)\n") + state.feed("other\n")
'same\nsame (x3)\nother\n'
>>> state.feed("Processing..."), state.flush(), state.feed(" done\n")
('', 'Processing...', ' done\n')

checkpoint_seconds: float = 60.0[source]#

mode: Literal['start', 'text', 'redraw', 'repeat'] = 'start'[source]#

not started, plain text, drawn again after a carriage return, or rewriting the previous line.

Type:: What the current line is

buffer: str = ''[source]#: Text of the current line not written yet, its last drawing for a redrawn line.

shown: bool = False[source]#: Whether a flush already wrote the start of the current line.

carriage: bool = False[source]#: Whether a carriage return came since the last text, so the next text draws the line again.

held_repeat: str = ''[source]#: Last line of a series of rewrites, written once the series ends.

since: float = 0.0[source]#: time.monotonic() of the last checkpoint of a redrawn line.

feed(text: str) → str[source]#: What the output receives of the text, given the lines before it.

take(token: str) → str[source]#: What the output receives of one token: a newline, a carriage return, a line move or plain text.

draw(token: str) → str[source]#: Add text to the current line, a carriage return before it starting the drawing over.

end_line() → str[source]#: Close the current line on a newline.

flush() → str[source]#: The unwritten text of a plain line, which a terminal shows on flush, so a step announced without newline shows.

new_line() → str[source]#: A newline ending the start of the line a flush already wrote, before a drawing that cannot erase it.

release_repeat() → str[source]#: The last line of a finished series of rewrites, written once.

finish() → str[source]#: What the output still has to receive when the stream ends.

class TeeTarget(file: ~typing.IO[~typing.Any], thread: int | None = None, strip_colors: bool = True, ascii_only: bool = True, ignore_lineup: bool = True, lines: dict[int, ~stouputils.print.output_stream.LineState] = <factory>, lock: ~_thread.allocate_lock = <factory>)[source]#

Bases: object

One output of a TeeMultiOutput, with how it is written to and which thread it listens to.

file: IO[Any][source]#

strip_colors: bool = True[source]#

ascii_only: bool = True[source]#: Whether non-ASCII characters are replaced, when the output is not a terminal.

ignore_lineup: bool = True[source]#: Whether the output receives what a terminal ends up showing, when it is not a terminal itself.

lines: dict[int, LineState][source]#: The line each writing thread is at, by thread identifier.

lock: allocate_lock[source]#: Keeps two threads writing at once from mixing up lines.

terminal: bool[source]#: Whether the output is a terminal, read once since asking costs a system call.

write(obj: str) → int | None[source]#

Write the text as this output receives it.

Returns:: Characters written, or None when the output is closed and has to be dropped.

flush() → None[source]#: Flush the output, writing first the line the current thread left unfinished.

finish() → None[source]#: Write what every thread’s current line still holds back, as when the stream ends.

file_text( text: str, step: Callable[[LineState, str], str], ) → str[source]#: What a non-terminal output receives of the text, step reading it with the current thread’s line.

ascii_text(text: str) → str[source]#: The text with its non-ASCII characters replaced when ascii_only is set.

class TeeMultiOutput( *files: IO[Any], strip_colors: bool = True, ascii_only: bool = True, ignore_lineup: bool = True, )[source]#

Bases: object

File-like object that duplicates output to multiple file-like objects.

Outputs can be added and removed while it is in place, each one listening to every thread or to a single one.

Parameters:

*files – One or more file-like objects that have write and flush methods
strip_colors – Strip ANSI color codes from output sent to these files
ascii_only – Replace non-ASCII characters with their ASCII equivalents for non-stdout/stderr files
ignore_lineup – Write to non-terminal outputs what a terminal ends up showing, see LineState

>>> import sys
>>> f = open("logfile.txt", "w")
>>> sys.stdout = TeeMultiOutput(sys.stdout, f)
>>> print("Hello World")  # Output goes to both console and file
Hello World
>>> f.close()   # TeeMultiOutput will handle any future writes to closed files gracefully

targets: tuple[TeeTarget, ...][source]#: Outputs written to, the first one standing for the stream this object replaces

removable: bool[source]#: Whether LogToFile put it in place, and takes it away with the last file it holds

property files: tuple[IO[Any], ...][source]#: File-like objects written to

property encoding: str[source]#: Get the encoding of the first file, or “utf-8” as fallback. :returns: “utf-8”, “ascii”, “latin1”, etc. :rtype: The encoding, ex

write(obj: str) → int[source]#

Write the object to every output listening to the current thread.

Parameters:: obj – String to write
Returns:: Number of characters written to the first file

add_file( file: IO[Any], thread: int | None = None, strip_colors: bool = True, ascii_only: bool = True, ignore_lineup: bool = True, ) → None[source]#

Start writing to one more output.

Parameters:: thread – Thread identifier whose writes alone it receives, None for every thread.

remove_file( file: IO[Any], ) → None[source]#: Stop writing to an output, after giving it what its current line still holds back.

fileno() → int[source]#: Return the file descriptor of the first file.

isatty() → bool[source]#: Return True if the first file is a terminal/console.