progressbar.terminal.stream module

typing.TextIO wrappers that intercept or redirect stream writes.

TextIOOutputWrapper is a pass-through base that delegates every TextIO operation to a wrapped stream unchanged. Concrete wrappers subclass it and override only the operation(s) they need to change (typically write). Two concrete wrappers:

  • LineOffsetStreamWrapper writes a fixed number of lines above the current cursor position instead of at it, used by ProgressBar’s line_offset= argument.

  • LastLineStream discards everything but the most recently written line, used by MultiBar to capture a bar’s rendered output without letting it reach the terminal directly.

Every non-underscore name here is re-exported by progressbar.terminal (from .stream import *, no __all__). LineOffsetStreamWrapper is also part of the top-level progressbar public API.

class progressbar.terminal.stream.LastLineStream(stream: TextIO)[source]

Bases: TextIOOutputWrapper

Discards everything but the most recently written line.

MultiBar rebinds a bar’s fd to one of these so the bar’s own redraws never reach the terminal directly. The multibar instead reads .line back out and places it on that bar’s row of the combined frame.

Store the stream to delegate to.

line: str = ''

The most recently written line. Only write, writelines and truncate touch this. Every other inherited method still operates on the wrapped stream.

read(_LastLineStream__n: int = -1) → str[source]

Return .line, or its first __n characters.

readable() → bool[source]

Always True: .line can always be read back.

readline(_LastLineStream__limit: int = -1) → str[source]

Return .line, or its first __limit characters.

seekable() → bool[source]

Always False: there is nothing to seek within.

truncate(_LastLineStream__size: int | None = None) → int[source]

Clear .line, or cut it to its first __size characters.

write(data: str) → int[source]

Replace .line with data.

writelines(_LastLineStream__lines: Iterable[str]) → None[source]

Keep only the last of __lines, discarding the rest.

Deliberate: MultiBar only ever wants a bar’s most recent rendered line, not its history.

class progressbar.terminal.stream.LineOffsetStreamWrapper(lines: int = 0, stream: TextIO = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>)[source]

Bases: TextIOOutputWrapper

Writes land a fixed number of lines above the cursor.

Each write moves the cursor up lines rows, writes there, then moves back down, leaving the cursor where it started. Used by ProgressBar’s line_offset= argument to draw a bar above other terminal output instead of at the current line.

Store the offset and the stream writes are redirected to.

Parameters:

  • lines – Number of lines above the current cursor position to write to.

  • stream – The underlying stream to write to.

DOWN = '\x1b[B'

ANSI “cursor down one line” (CSI B).

UP = '\x1b[F'

ANSI “cursor up one line” (CSI F).

write(data: str) → int[source]

Write data self.lines rows above the cursor.

Moves the cursor up, writes data with trailing newlines stripped (so the write itself doesn’t move the cursor), then moves back down to restore the original position.

Parameters:

data – Text to write.

Returns:

The length of data before newline-stripping, so callers can detect short writes.

class progressbar.terminal.stream.TextIOOutputWrapper(stream: TextIO)[source]

Bases: TextIO

Pass-through TextIO base that delegates to a wrapped stream.

Every operation forwards to self.stream unchanged, and subclasses override only what they need to intercept. write is left unimplemented: this class is meant to be subclassed, not used as-is.

Store the stream to delegate to.

close() → None[source]

Delegate to the wrapped stream.

fileno() → int[source]

Delegate to the wrapped stream.

flush() → None[source]

Delegate to the wrapped stream.

isatty() → bool[source]

Delegate to the wrapped stream.

read(_TextIOOutputWrapper__n: int = -1) → str[source]

Delegate to the wrapped stream.

readable() → bool[source]

Delegate to the wrapped stream.

readline(_TextIOOutputWrapper__limit: int = -1) → str[source]

Delegate to the wrapped stream.

readlines(_TextIOOutputWrapper__hint: int = -1) → list[str][source]

Delegate to the wrapped stream.

seek(_TextIOOutputWrapper__offset: int, _TextIOOutputWrapper__whence: int = 0) → int[source]

Delegate to the wrapped stream.

seekable() → bool[source]

Delegate to the wrapped stream.

tell() → int[source]

Delegate to the wrapped stream.

truncate(_TextIOOutputWrapper__size: int | None = None) → int[source]

Delegate to the wrapped stream.

writable() → bool[source]

Delegate to the wrapped stream.

writelines(_TextIOOutputWrapper__lines: Iterable[str]) → None[source]

Delegate to the wrapped stream.