Function progress_wrapper

pandarallel/progress_bars.py:185–218

Wrap the function to apply in a function which monitor the part of work already done.

(
    user_defined_function: Callable,
    master_workers_queue: multiprocessing.Queue,
    index: int,
    chunk_size: int,
)

def progress_wrapper(
    user_defined_function: Callable,
    master_workers_queue: multiprocessing.Queue,
    index: int,
    chunk_size: int,
) -> Callable:
    """Wrap the function to apply in a function which monitor the part of work already
    done.
    """
    counter = count()
    state = ProgressState(chunk_size)

    def closure(*user_defined_function_args, **user_defined_functions_kwargs):
        iteration = next(counter)

        if iteration == state.next_put_iteration:
            time_now = time_ns()
            master_workers_queue.put_nowait((index, WorkerStatus.Running, iteration))

            delta_t = time_now - state.last_put_time
            delta_i = iteration - state.last_put_iteration

            state.next_put_iteration += (
                max(int((delta_i / delta_t) * INTERVAL_NS), 1) if delta_t != 0 else 1
            )

            state.last_put_iteration = iteration
            state.last_put_time = time_now

        return user_defined_function(
            *user_defined_function_args, **user_defined_functions_kwargs
        )

    return closure