progressbar.utils module

Color stripping, delta coalescing, and stdout/stderr redirection.

The redirection machinery (WrappingIO, StreamWrapper, and the module-level streams singleton constructed at the bottom of this module) is what lets print() calls and logging output appear as normal lines above a redrawing progress bar instead of corrupting its line. See the “Print while a bar is running” how-to for the full picture.

class progressbar.utils.AttributeDict

Bases: dict[str, T], Generic[T]

A dict that can be accessed with .attribute.

Note

Double-underscore names (e.g. __orig_class__, set by typing.Generic on subscripted instances) are routed to real instance attributes instead of dict entries, keeping runtime metadata like that out of the mapping’s contents. See __setattr__/__delattr__.

>>> attrs = AttributeDict(spam=123)

# Reading

>>> attrs['spam']
123
>>> attrs.spam
123

# Read after update using attribute

>>> attrs.spam = 456
>>> attrs['spam']
456
>>> attrs.spam
456

# Read after update using dict access

>>> attrs['spam'] = 123
>>> attrs['spam']
123
>>> attrs.spam
123

# Read after update using dict access

>>> del attrs.spam
>>> attrs['spam']
Traceback (most recent call last):
...
KeyError: 'spam'
>>> attrs.spam
Traceback (most recent call last):
...
AttributeError: No such attribute: spam
>>> del attrs.spam
Traceback (most recent call last):
...
AttributeError: No such attribute: spam

class progressbar.utils.StreamWrapper

Bases: object

Wrap sys.stdout/sys.stderr for output, logging, and a bar to share.

Almost always used via the module-level streams singleton (constructed once at the bottom of this module) rather than instantiated directly. Each wrap_*/unwrap_* pair is refcounted, so nested or concurrent bars that both request redirection share one wrapper and it’s only undone once the last one finishes.

Capture the current sys.stdout/sys.stderr as “real”.

Note

This runs once, at construction, and streams (below) is constructed at import time. Anything that reassigns sys.stdout/sys.stderr after progressbar.utils is first imported will not be picked up: original_stdout/ original_stderr keep pointing at whatever was installed at that moment, not whatever is live later. This has been a repeat source of bugs.

capturing: int = 0

excepthook(exc_type: type[BaseException], exc_value: BaseException, exc_traceback: TracebackType | None) → None

Run the original excepthook, then flush buffered output.

Installed as sys.excepthook while stdout or stderr is wrapped (see wrap_excepthook), so an uncaught exception’s traceback, written via the original hook, is followed by whatever output was still buffered, instead of that text getting lost or appearing in the wrong order relative to the traceback.

flush() → None

Flush buffered captured output on both wrapped streams.

If writing the buffered text to a stream’s target raises io.UnsupportedOperation (as happens for some non-seekable streams), that stream’s redirection disables itself: wrapped_stdout/wrapped_stderr is reset to 0 so this method stops attempting to flush it on future calls, and a warning is logged. sys.stdout/sys.stderr are left installed as-is: only further flush attempts are skipped, not the wrapping itself.

listeners: set[_ProgressListener]

logging_handlers: list[tuple[StreamHandler[IO], IO]]

needs_clear() → bool

Return whether either wrapped stream has buffered output.

Uses getattr with a False default so this is safe to call whether or not stdout/stderr are currently wrapped, since a plain, unwrapped stream has no needs_clear attribute.

Returns:

Whether a bar’s next redraw should erase its line first, so buffered print()/logging output can be flushed above it.

original_excepthook: Callable[[type[BaseException], BaseException, TracebackType | None], None]

start_capturing(bar: _ProgressListener | None = None) → None

Turn capturing on for bar and bump the shared refcount.

Parameters:

bar – Registered as a listener so it’s notified (update()) when captured output completes a line. Omit to just bump the refcount without listening.

stderr: TextIO | WrappingIO

stdout: TextIO | WrappingIO

stop_capturing(bar: _ProgressListener | None = None) → None

Unregister bar and drop the shared capturing refcount.

Parameters:

bar – The listener to remove, if it was registered via start_capturing.

unwrap(stdout: bool = False, stderr: bool = False) → None

Unwrap stdout and/or stderr, per the given flags.

unwrap_excepthook() → None

Restore the original sys.excepthook, if currently wrapped.

unwrap_logging() → None

Undo one wrap_logging() call, restoring at refcount zero.

Only the call that brings wrapped_logging to zero actually restores anything: it pops every entry wrap_logging recorded in logging_handlers and puts each handler’s original stream back.

unwrap_stderr() → None

Undo one wrap_stderr() call, restoring at refcount zero.

Mirrors unwrap_stdout(): only the call that brings wrapped_stderr to zero restores sys.stderr, and the shared excepthook is unwrapped once stdout is back to its original too.

unwrap_stdout() → None

Undo one wrap_stdout() call, restoring at refcount zero.

Only the call that brings wrapped_stdout to zero actually restores sys.stdout, and self.stdout alongside it, so needs_clear() and update_capturing() don’t keep reading a wrapper that’s no longer installed. Also unwraps the shared excepthook once stderr is back to its original too, since it’s shared between the two.

update_capturing() → None

Propagate the capturing refcount to the wrapped streams.

Flushes immediately once the refcount drops to zero or below, so whatever’s left in the buffer reaches the terminal as soon as the last bar stops capturing, rather than sitting there until something else happens to trigger a flush.

wrap(stdout: bool = False, stderr: bool = False) → None

Wrap stdout and/or stderr, per the given flags.

wrap_excepthook() → None

Install the shared excepthook that flushes buffered output.

A no-op if already wrapped: wrap_stdout() and wrap_stderr() both call this unconditionally, and either one may already have wrapped it.

wrap_logging() → None

Retarget every StreamHandler in the logger tree to the wrapper.

Refcounted like wrap_stdout/wrap_stderr: only the first call actually walks the logger tree and rewrites handlers, so nested/concurrent redirection doesn’t fight over the same handlers or lose track of what to restore. Each handler is visited once (deduplicated by id(), via _wrap_logging_handler) because the same handler object can be attached to more than one logger in the tree, and each retargeted handler is recorded in logging_handlers so unwrap_logging can put it back later.

wrap_stderr() → WrappingIO

Install a WrappingIO over sys.stderr, or share it.

See wrap_stdout for the refcounting behavior, mirrored here for sys.stderr.

Returns:

The installed WrappingIO, the same instance on every call until it’s fully unwrapped.

wrap_stdout() → WrappingIO

Install a WrappingIO over sys.stdout, or share it.

Refcounted: only the first call actually replaces sys.stdout; later calls just bump wrapped_stdout so nested/concurrent bars share one wrapper, and it takes a matching number of unwrap_stdout() calls to restore the original stream. Also wraps sys.excepthook, since a traceback printed while capturing would otherwise bypass the buffer.

Returns:

The installed WrappingIO, the same instance on every call until it’s fully unwrapped.

wrapped_excepthook: int = 0

wrapped_logging: int = 0

wrapped_stderr: int = 0

wrapped_stdout: int = 0

class progressbar.utils.WrappingIO(target: IO, capturing: bool = False, listeners: set[_ProgressListener] | None = None)

Bases: object

sys.stdout/sys.stderr replacement installed while capturing.

Buffers writes in memory instead of passing them straight through while capturing is on, so a bar can erase its own line, flush the buffer above it, and redraw. See StreamWrapper.wrap_stdout/ wrap_stderr for how one gets installed.

Wrap target so writes can be buffered while capturing.

Parameters:

• target – The real stream writes are buffered for, and eventually flushed through to.

• capturing – Start in buffering mode immediately instead of passing writes straight through.

• listeners – Bars to notify (update()) whenever a buffered write completes a line. Typically shared with StreamWrapper.listeners by the caller, so every wrapped stream notifies the same bars.

buffer: StringIO

capturing: bool

close() → None

Flush the buffer and close target.

fileno() → int

Return target’s file descriptor.

flush() → None

Flush the in-memory buffer, a no-op for io.StringIO.

Only satisfies the file-like flush() protocol. It does not write buffered text through to target. Use _flush() (or StreamWrapper.flush(), which calls it) for that.

flush_target() → None

Flush target itself, if it’s open and flushable.

isatty() → bool

Return whether target is a tty.

listeners: set[_ProgressListener]

needs_clear: bool = False

read(n: int = -1) → str

Read up to n characters from target.

readable() → bool

Return whether target supports reading.

readline(limit: int = -1) → str

Read a single line (up to limit characters) from target.

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

Read all lines from target.

seek(offset: int, whence: int = 0) → int

Seek target to offset, relative to whence.

seekable() → bool

Return whether target supports seeking.

target: IO

tell() → int

Return target’s current stream position.

truncate(size: int | None = None) → int

Truncate target to size.

writable() → bool

Return whether target supports writing.

write(value: str) → int

Write value, buffering it in memory while capturing.

While capturing is on, value is appended to buffer instead of reaching target. If the buffered text now contains a newline, needs_clear is set and every listener’s update() is called, so a capturing bar redraws promptly instead of waiting for its next scheduled update. While not capturing, value is written straight through to target and target is flushed on every newline, so unbuffered output still appears live.

Returns:

The number of characters written (buffered or passed through), mirroring the return value of a normal file write().

writelines(lines: Iterable[str]) → None

Write lines to target.

progressbar.utils.deltas_to_seconds(*deltas: timedelta | float | int | None, default: type[ValueError] = <class 'ValueError'>) → float

progressbar.utils.deltas_to_seconds(*deltas: timedelta | float | int | None, default: T) → float | T

Coalesce timedeltas and second counts to a single seconds float.

Returns the first argument in deltas that isn’t None, converted to seconds as a float. Raises (or returns default, if given) only when every argument is None.

>>> deltas_to_seconds(datetime.timedelta(seconds=1, milliseconds=234))
1.234
>>> deltas_to_seconds(123)
123.0
>>> deltas_to_seconds(1.234)
1.234
>>> deltas_to_seconds(None, 1.234)
1.234
>>> deltas_to_seconds(0, 1.234)
0.0
>>> deltas_to_seconds()
Traceback (most recent call last):
...
ValueError: No valid deltas passed to `deltas_to_seconds`
>>> deltas_to_seconds(None)
Traceback (most recent call last):
...
ValueError: No valid deltas passed to `deltas_to_seconds`
>>> deltas_to_seconds(default=0.0)
0.0

progressbar.utils.len_color(value: str | bytes) → int

Return the length of value without ANSI escape codes.

>>> len_color(b'[1234]abc')
3
>>> len_color('[1234]abc')
3
>>> len_color('[1234]abc')
3

progressbar.utils.no_color(value: StringT) → StringT

Return the value without ANSI escape codes.

>>> no_color(b'[1234]abc')
b'abc'
>>> str(no_color('[1234]abc'))
'abc'
>>> str(no_color('[1234]abc'))
'abc'
>>> no_color(123)
Traceback (most recent call last):
...
TypeError: `value` must be a string or bytes, got 123

progressbar.utils.streams = <progressbar.utils.StreamWrapper object>

Process-global, constructed once at import. The only place that knows the real sys.stdout/sys.stderr versus whatever is currently installed in their place. Every bar that redirects goes through this one shared instance rather than each keeping its own. Mutating it (wrapping/unwrapping) affects every bar and every plain print() in the process. bar.py’s StdRedirectMixin is its only real consumer.