pydvl.utils.progress

Progress

Progress(iterable: Iterable[T], is_done: StoppingCriterion, **kwargs)

Bases: Generic[T]

Displays an optional progress bar for an iterable, using StoppingCriterion.completion for the progress.

PARAMETER	DESCRIPTION
iterable	The iterable to wrap. TYPE: Iterable[T]
is_done	The stopping criterion. TYPE: StoppingCriterion
kwargs	Additional keyword arguments passed to tqdm. - total: The total number of items in the iterable (Default: 100) - unit: The unit of the progress bar. (Default: %) - desc: Description of the progress bar. (Default: str(is_done)) - bar_format: Format of the progress bar. (Default is a percentage bar) - plus anything else that tqdm accepts DEFAULT: {}

Source code in src/pydvl/utils/progress.py

def __init__(
    self,
    iterable: Iterable[T],
    is_done: StoppingCriterion,
    **kwargs,
) -> None:
    self.iterable = iterable
    self.is_done = is_done
    self.total = kwargs.pop("total", 100)
    self.unit = kwargs.pop("unit", "%")
    self.desc = kwargs.pop("desc", str(is_done))
    self.bar_format = "{desc}: {percentage:0.2f}%|{bar}| [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
    self.kwargs = kwargs
    self.pbar: tqdm | None = None

repeat_indices

repeat_indices(
    indices: Collection[int],
    result: "ValuationResult",
    done: "StoppingCriterion",
    **kwargs
) -> Iterator[int]

Helper function to cycle indefinitely over a collection of indices until the stopping criterion is satisfied while displaying progress.

PARAMETER	DESCRIPTION
indices	Collection of indices that will be cycled until done. TYPE: Collection[int]
result	Object containing the current results. TYPE: 'ValuationResult'
done	Stopping criterion. TYPE: 'StoppingCriterion'
kwargs	Keyword arguments passed to tqdm. DEFAULT: {}

Source code in src/pydvl/utils/progress.py

def repeat_indices(
    indices: Collection[int],
    result: "ValuationResult",
    done: "StoppingCriterion",
    **kwargs,
) -> Iterator[int]:
    """Helper function to cycle indefinitely over a collection of indices
    until the stopping criterion is satisfied while displaying progress.

    Args:
        indices: Collection of indices that will be cycled until done.
        result: Object containing the current results.
        done: Stopping criterion.
        kwargs: Keyword arguments passed to tqdm.
    """
    with tqdm(total=100, unit="%", **kwargs) as pbar:
        it = takewhile(lambda _: not done(result), cycle(indices))
        for i in it:
            yield i
            pbar.update(100 * done.completion() - pbar.n)
            pbar.refresh()

log_duration

log_duration(_func=None, *, log_level=logging.DEBUG)

Decorator to log execution time of a function with a configurable logging level. It can be used with or without specifying a log level.

Source code in src/pydvl/utils/progress.py

def log_duration(_func=None, *, log_level=logging.DEBUG):
    """
    Decorator to log execution time of a function with a configurable logging level.
    It can be used with or without specifying a log level.
    """

    def decorator_log_duration(func):
        @wraps(func)
        def wrapper_log_duration(*args, **kwargs):
            func_name = func.__qualname__
            logger.log(log_level, f"Function '{func_name}' is starting.")
            start_time = time()
            result = func(*args, **kwargs)
            duration = time() - start_time
            logger.log(
                log_level,
                f"Function '{func_name}' completed. " f"Duration: {duration:.2f} sec",
            )
            return result

        return wrapper_log_duration

    if _func is None:
        # If log_duration was called without arguments, return decorator
        return decorator_log_duration
    else:
        # If log_duration was called with a function, apply decorator directly
        return decorator_log_duration(_func)