AndroidRemoteController/python/py/Lib/site-packages/about_time/core.py

import time
from contextlib import contextmanager

from .human_count import HumanCount
from .human_duration import HumanDuration
from .human_throughput import HumanThroughput


def about_time(func_or_it=None, *args, **kwargs):
    """Measure timing and throughput of code blocks, with beautiful
    human friendly representations.

    There are three modes of operation: context manager, callable and
    throughput.

    1. Use it like a context manager:

    >>> with about_time() as t:
    ....    # code block.

    2. Use it with a callable:

    >>> def func(a, b): ...
    >>> t = about_time(func, 1, b=2)  # send arguments at will.

    3. Use it with an iterable or generator:

    >>> t = about_time(it)  # any iterable or generator.
    >>> for item in t:
    ....    # use item
    """

    timings = [0.0, 0.0]

    # use as a context manager.
    if func_or_it is None:
        return _context_timing(timings, Handle(timings))

    # use as a callable.
    if callable(func_or_it):
        with _context_timing(timings):
            result = func_or_it(*args, **kwargs)
        return HandleResult(timings, result)

    try:
        it = iter(func_or_it)
    except TypeError:
        raise UserWarning('param should be callable or iterable.')

    # use as a counter/throughput iterator.
    def it_closure():
        with _context_timing(timings):
            for it_closure.count, elem in enumerate(it, 1):  # iterators are iterable.
                yield elem

    it_closure.count = 0  # the count will only be updated after starting iterating.
    return HandleStats(timings, it_closure)


@contextmanager
def _context_timing(timings, handle=None):
    timings[0] = time.perf_counter()
    yield handle
    timings[1] = time.perf_counter()


class Handle(object):
    def __init__(self, timings):
        self.__timings = timings

    @property
    def duration(self) -> float:
        """Return the actual duration in seconds.
        This is dynamically updated in real time.

        Returns:
            the number of seconds.

        """
        return (self.__timings[1] or time.perf_counter()) - self.__timings[0]

    @property
    def duration_human(self) -> HumanDuration:
        """Return a beautiful representation of the duration.
        It dynamically calculates the best unit to use.

        Returns:
            the human representation.

        """
        return HumanDuration(self.duration)


class HandleResult(Handle):
    def __init__(self, timings, result):
        super(HandleResult, self).__init__(timings)
        self.__result = result

    @property
    def result(self):
        """Return the result of the callable.

        Returns:
            the result of the callable.

        """
        return self.__result


class HandleStats(Handle):
    def __init__(self, timings, it_closure):
        super(HandleStats, self).__init__(timings)
        self.__it = it_closure

    def __iter__(self):
        return self.__it()

    @property
    def count(self) -> int:
        """Return the current iteration count.
        This is dynamically updated in real time.

        Returns:
            the current iteration count.

        """
        return self.__it.count

    @property
    def count_human(self) -> HumanCount:
        """Return a beautiful representation of the current iteration count.
        This is dynamically updated in real time.

        Returns:
            the human representation.

        """
        return self.count_human_as('')

    def count_human_as(self, unit: str) -> HumanCount:
        """Return a beautiful representation of the current iteration count.
        This is dynamically updated in real time.

        Args:
            unit: what is being measured

        Returns:
            the human representation.

        """
        return HumanCount(self.count, unit)

    @property
    def throughput(self) -> float:
        """Return the current throughput in items per second.
        This is dynamically updated in real time.

        Returns:
            the number of items per second.

        """
        try:
            return self.count / self.duration
        except ZeroDivisionError:  # pragma: no cover
            return float('nan')

    @property
    def throughput_human(self) -> HumanThroughput:
        """Return a beautiful representation of the current throughput.
        It dynamically calculates the best unit to use.

        Returns:
            the human representation.

        """
        return self.throughput_human_as('')

    def throughput_human_as(self, unit: str) -> HumanThroughput:
        """Return a beautiful representation of the current throughput.
        It dynamically calculates the best unit to use.

        Args:
            unit: what is being measured

        Returns:
            the human representation.

        """
        return HumanThroughput(self.throughput, unit)