Progression module

progression.progress.ESC_MOVE_LINE_DOWN(n)[source]
progression.progress.ESC_MOVE_LINE_UP(n)[source]
progression.progress.ESC_SEQ_to_HTML(s)[source]
progression.progress.FloatValue(val=0.0)[source]
class progression.progress.Loop(func, args=(), interval=1, verbose=None, sigint='stop', sigterm='stop', auto_kill_on_last_resort=False, raise_error=True)[source]

Bases: object

class to run a function periodically an seperate process.

In case the called function returns True, the loop will stop. Otherwise a time interval given by interval will be slept before another execution is triggered.

The shared memory variable _run (accessible via the class property run) also determines if the function if executed another time. If set to False the execution stops.

For safe cleanup (and in order to catch any Errors) it is advisable to instantiate this class using ‘with’ statement as follows:

with Loop(**kwargs) as my_loop:
my_loop.start() ...

this will guarantee you that the spawned loop process is down when exiting the ‘with’ scope.

The only circumstance where the process is still running is when you set auto_kill_on_last_resort to False and answer the question to send SIGKILL with no.

getpid()[source]
is_alive()[source]
join(timeout)[source]

calls join for the spawned process with given timeout

pause()[source]
resume()[source]
run

makes the shared memory boolean accessible as class attribute

Set run False, the loop will stop repeating. Calling start, will set run True and start the loop again as a new process.

start()[source]

uses multiprocess Process to call _wrapper_func in subprocess

stop()[source]

stops the process triggered by start

Setting the shared memory boolean run to false, which should prevent the loop from repeating. Call __cleanup to make sure the process stopped. After that we could trigger start() again.

exception progression.progress.LoopExceptionError[source]

Bases: RuntimeError

exception progression.progress.LoopInterruptError[source]

Bases: Exception

class progression.progress.MultiLineFormatter(fmt=None, datefmt=None, style='%')[source]

Bases: logging.Formatter

pads a multiline log message with spaces such that

<HEAD> msg_line1
msg_line2
...
format(record)[source]
class progression.progress.PipeFromProgressToIPythonHTMLWidget[source]

Bases: object

progression.progress.PipeHandler

alias of PipeToPrint

class progression.progress.PipeToPrint[source]

Bases: object

class progression.progress.Progress(count, max_count=None, prepend=None, width='auto', speed_calc_cycles=10, interval=1, verbose=None, sigint='stop', sigterm='stop', info_line=None)[source]

Bases: progression.progress.Loop

Abstract Progress Loop

Uses Loop class to implement a repeating function to display the progress of multiples processes.

In the simplest case, given only a list of shared memory values ‘count’ (NOTE: a single value will be automatically mapped to a one element list), the total elapses time (TET) and a speed estimation are calculated for each process and passed to the display function show_stat.

This functions needs to be implemented in a subclass. It is intended to show the progress of ONE process in one line.

When max_count is given (in general as list of shared memory values, a single shared memory value will be mapped to a one element list) the time to go TTG will also be calculated and passed tow show_stat.

Information about the terminal width will be retrieved when setting width=’auto’. If supported by the terminal emulator the width in characters of the terminal emulator will be stored in width and also passed to show_stat. Otherwise, a default width of 80 characters will be chosen.

Also you can specify a fixed width by setting width to the desired number.

NOTE: in order to achieve multiline progress special escape sequences are used which may not be supported by all terminal emulators.

example:

c1 = UnsignedIntValue(val=0) c2 = UnsignedIntValue(val=0)

c = [c1, c2] prepend = [‘c1: ‘, ‘c2: ‘] with ProgressCounter(count=c, prepend=prepend) as sc:

sc.start() while True:

i = np.random.randint(0,2) with c[i].get_lock():

c[i].value += 1
if c[0].value == 1000:
break

time.sleep(0.01)

using start() within the ‘with’ statement ensures that the subprocess which is started by start() in order to show the progress repeatedly will be terminated on context exit. Otherwise one has to make sure that at some point the stop() routine is called. When dealing with signal handling and exception this might become tricky, so that the use of the ‘with’ statement is highly encouraged.

reset(i=None)[source]

convenient function to reset progress information

i [None, int] - None: reset all, int: reset process indexed by i

static show_stat(count_value, max_count_value, prepend, speed, tet, ttg, width, **kwargs)[source]

re implement this function in a subclass

count_value - current value of progress

max_count_value - maximum value the progress can take

prepend - some extra string to be put for example in front of the progress display

speed - estimated speed in counts per second (use for example humanize_speed to get readable information in string format)

tet - total elapsed time in seconds (use for example humanize_time to get readable information in string format)

ttg - time to go in seconds (use for example humanize_time to get readable information in string format)

static show_stat_wrapper(count, last_count, start_time, max_count, speed_calc_cycles, width, q, last_speed, prepend, show_stat_function, add_args, i, lock)[source]
static show_stat_wrapper_multi(count, last_count, start_time, max_count, speed_calc_cycles, width, q, last_speed, prepend, show_stat_function, len_, add_args, lock, info_line, no_move_up=False)[source]

call the static method show_stat_wrapper for each process

start()[source]
stop()[source]

trigger clean up by hand, needs to be done when not using context management via ‘with’ statement

  • will terminate loop process
  • show a last progress -> see the full 100% on exit
  • releases terminal reservation
class progression.progress.ProgressBar(*args, **kwargs)[source]

Bases: progression.progress.Progress

implements a progress bar similar to the one known from ‘wget’ or ‘pv’

static show_stat(count_value, max_count_value, prepend, speed, tet, ttg, width, i, **kwargs)[source]
class progression.progress.ProgressBarCounter(speed_calc_cycles_counter=5, **kwargs)[source]

Bases: progression.progress.Progress

records also the time of each reset and calculates the speed of the resets.

shows the TET since init (not effected by reset) the speed of the resets (number of finished processed per time) and the number of finished processes

after that also show a progress of each process max_count > 0 and not None -> bar max_count == None -> absolute count statistic max_count == 0 -> hide process statistic at all

get_counter_count(i=0)[source]
static show_stat(count_value, max_count_value, prepend, speed, tet, ttg, width, i, **kwargs)[source]
class progression.progress.ProgressBarCounterFancy(speed_calc_cycles_counter=5, **kwargs)[source]

Bases: progression.progress.ProgressBarCounter

static show_stat(count_value, max_count_value, prepend, speed, tet, ttg, width, i, **kwargs)[source]
class progression.progress.ProgressBarFancy(*args, **kwargs)[source]

Bases: progression.progress.Progress

implements a progress bar where the color indicates the current status similar to the bars known from ‘htop’

static full_minor_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static full_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static get_d(s1, s2, width, lp, lps)[source]
static kw_bold(s, ch_after)[source]
static reduced_1_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static reduced_2_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static reduced_3_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static reduced_4_stat(p, tet, speed, ttg, eta, ort, repl_ch, width, lp, lps)[source]
static show_stat(count_value, max_count_value, prepend, speed, tet, ttg, width, i, **kwargs)[source]
class progression.progress.SIG_handler_Loop(sigint, sigterm, log, prefix)[source]

Bases: object

class to setup signal handling for the Loop class

Note: each subprocess receives the default signal handling from it’s parent. If the signal function from the module signal is evoked within the subprocess this default behavior can be overwritten.

The init function receives a shared memory boolean object which will be set false in case of signal detection. Since the Loop class will check the state of this boolean object before each repetition, the loop will stop when a signal was receives.

set_signal(sig, handler_str)[source]
class progression.progress.StdoutPipe(conn)[source]

Bases: object

replacement for stream objects such as stdout which forwards all incoming data using the send method of a connection

example usage:

>>> import sys
>>> from multiprocessing import Pipe
>>> from progression import StdoutPipe
>>> conn_recv, conn_send = Pipe(False)
>>> sys.stdout = StdoutPipe(conn_send)
>>> print("hallo welt", end='')  # this is no going through the pipe
>>> msg = conn_recv.recv()
>>> sys.stdout = sys.__stdout__
>>> print(msg)
hallo welt
>>> assert msg == "hallo welt"
flush()[source]
write(b)[source]
progression.progress.StringValue(num_of_bytes)[source]
progression.progress.UnsignedIntValue(val=0)[source]
progression.progress.check_process_termination(proc, prefix, timeout, auto_kill_on_last_resort=False)[source]
progression.progress.choose_color_theme(name)[source]
progression.progress.choose_pipe_handler(kind='print', color_theme=None)[source]
progression.progress.codecov_subprocess_check()[source]
progression.progress.getCountKwargs(func)[source]

Returns a list [“count kwarg”, “count_max kwarg”] for a given function. Valid combinations are defined in progress.validCountKwargs.

Returns None if no keyword arguments are found.

progression.progress.get_identifier(name=None, pid=None, bold=True)[source]
progression.progress.get_terminal_size(defaultw=80)[source]

Checks various methods to determine the terminal size

Methods: - shutil.get_terminal_size (only Python3) - fcntl.ioctl - subprocess.check_output - os.environ

defaultw
: int
Default width of terminal.
width, height
: int
Width and height of the terminal. If one of them could not be found, None is return in its place.
progression.progress.get_terminal_width(default=80, name=None)[source]
progression.progress.humanize_speed(c_per_sec)[source]

convert a speed in counts per second to counts per [s, min, h, d], choosing the smallest value greater zero.

progression.progress.humanize_time(secs)[source]

convert second in to hh:mm:ss format

progression.progress.len_string_without_ESC(s)[source]
progression.progress.remove_ESC_SEQ_from_string(s)[source]
progression.progress.terminal_reserve(progress_obj, terminal_obj=None, identifier=None)[source]

Registers the terminal (stdout) for printing.

Useful to prevent multiple processes from writing progress bars to stdout.

One process (server) prints to stdout and a couple of subprocesses do not print to the same stdout, because the server has reserved it. Of course, the clients have to be nice and check with terminal_reserve first if they should (not) print. Nothing is locked.

True if reservation was successful (or if we have already reserved this tty), False if there already is a reservation from another instance.

progression.progress.terminal_unreserve(progress_obj, terminal_obj=None, verbose=0, identifier=None)[source]

Unregisters the terminal (stdout) for printing.

an instance (progress_obj) can only unreserve the tty (terminal_obj) when it also reserved it

see terminal_reserved for more information

None