graphtage.progress¶

A module for printing status messages and progress bars to the command line.

progress classes¶

StatusWriter¶

class graphtage.progress.StatusWriter(out_stream: Optional[TextIO] = None, quiet: bool = False)¶

Bases: typing.IO

A writer compatible with the graphtage.printer.Writer protocol that can print status.

See StatusWriter.tqdm() and StatusWriter.trange(). If StatusWriter.status_stream is either sys.stdout or sys.stderr, then bytes printed to this writer will be buffered. For each full line buffered, a call to tqdm.write() will be made.

A status writer whose lifetime is not controlled by instantiation in a with block must be manually flushed with StatusWriter.flush(final=True) after its final write, or else the last line written may be lost.

__init__(out_stream: Optional[TextIO] = None, quiet: bool = False)¶

Initializes a status writer.

Parameters

out_stream – An optional stream to which to write. If omitted this defaults to sys.stdout.
quiet – Whether or not tqdm status messages and progress should be suppressed.

close() → None ¶

property closed¶

fileno() → int ¶

flush(final=False)¶

Flushes this writer.

If final is True, any extra bytes will be flushed along with a final newline.

isatty() → bool ¶

property mode¶

property name¶

quiet = None¶: Whether or not tqdm status messages and progress should be suppressed.

read(n: int = Ellipsis) → AnyStr¶

readable() → bool ¶

readline(limit: int = Ellipsis) → AnyStr¶

readlines(hint: int = Ellipsis) → List[AnyStr]¶

seek(offset: int, whence: int = Ellipsis) → int ¶

seekable() → bool ¶

status_stream: TextIO = None¶: The status stream to which to print.

tell() → int ¶

tqdm(*args, **kwargs) → tqdm.std.tqdm¶: Returns a tqdm.tqdm object.

trange(*args, **kwargs) → tqdm.std.trange¶: Returns a tqdm.trange object.

truncate(size: Optional[int] = Ellipsis) → int ¶

writable() → bool ¶

write(text: str) → int ¶

write_raw = None¶

If True, this writer will not buffer output and use tqdm.write().

This defaults to:

self.write_raw = self.quiet or (
    out_stream.fileno() != sys.stderr.fileno() and out_stream.fileno() != sys.stdout.fileno()
)

writelines(lines: Iterable[AnyStr]) → None ¶