Throttle

class throttle.Throttle(interval)

This class offers synchronous and asynchronous mechanisms to accurately rate-limit the execution of some iterative code. It can be used explicitly to wait between each time interval or with generator functions to iterate through intervals.

Example: Video recorder at 24 fps. Each iteration in the loop starts precisely every 1/24 seconds (if the iterations don’t last longer).

>>> rate = 24   # fps
>>> throttle = Throttle(interval=(1 / rate))
>>> iters = 0
>>> t_start = perf_counter()
>>> for i in throttle.loop(24):
...     # Take, process and save image
...     iters += 1
>>> total_time = round(perf_counter() - t_start, 2)
>>> print('iters: {}, total_time: {}'.format(iters, total_time))
iters: 24, total_time: 1.0

__init__(interval)

Returns a Throttle instance. Time reference will be set the first time a timing function is called.

Parameters:interval – Interval value for timing functions, in seconds.

aloop(max_ticks=None, duration=None)

Returns an asynchronous generator yielding every time an interval has elapsed.

You can specify a maximum number of iterations or a maximum duration for the generator, but not both. If none is set, the generator will run until you call break, return or raise an exception.

Parameters:

• max_ticks – Maximum number of intervals the generator will wait for.
• duration – Seconds of loop duration. The generator will loop for ceil(duration / self.interval) iterations, so the elapsed time may be longer than duration, but never shorter.

Returns:

Yields the number of intervals elapsed since the function was called.

await_next()

Waits asynchronously until the end of the current interval. Note that this function can return immediately if the next interval to wait has already elapsed. This happens when reusing a Throttle instance without calling restart() first.

elapsed(auto_reset=True, exact=True)

Checks if the interval has elapsed.

Parameters:

• auto_reset – If True, the time reference will be updated if the the interval has elapsed.
• exact – If True, the time reference will be increased with the interval value, which will result in deterministic and accurate intervals. If False, the time reference will be updated to the current time. exact has no effect if auto_reset is False.

Returns:

True if the interval has elapsed, otherwise False.

loop(max_ticks=None, duration=None)

Returns a synchronous generator yielding every time an interval has elapsed.

You can specify a maximum number of iterations or a maximum duration for the generator, but not both. If none is set, the generator will run until you call break, return or raise an exception.

Parameters:

• max_ticks – Maximum number of intervals the generator will wait for.
• duration – Seconds of loop duration. The generator will loop for ceil(duration / self.interval) iterations, so the elapsed time may be longer than duration, but never shorter.

Returns:

Yields the number of intervals elapsed since the function was called.

restart()

Reset the instance deleting its time reference, which will be set the next time a timing function is called.

wait_next()

Blocks until the end of the current interval. Note that this function can return immediately if the next interval to wait has already elapsed. This happens when reusing a Throttle instance without calling restart() first.

throttle.throttle(limit=1, interval=1.0, wait=False, on_fail=None)

Decorator to limit the number of calls to a synchronous function in an interval of time. It ensures that the decorated function is not called more than limit times in the same time interval. If limit is reached, it can return a custom result or sleep until the next time interval. Do not use this function to decorate an asynchronous function (use athrottle() instead).

Parameters:

• limit – Maximum number of calls allowed to the decorated function in each time interval.
• interval – Seconds of every time period.
• wait – If True, it will block when limit is reached until the next interval, and then it will call the function.
• on_fail – Value, object or function to return if the call limit is reached. If on_fail is a function, the decorator will return the result of the call to this function. Note that on_fail only makes sense if wait is False.

Returns:

Result of the call to the decorated function, or on_fail if limit is reached (and wait is False).

throttle.athrottle(limit=1, interval=1.0, wait=False, on_fail=None)

Decorator to limit the number of calls to a synchronous or asynchronous function in an interval of time. It ensures that the decorated function is not called more than limit times in the same time interval. If limit is reached, it can return a custom result or sleep until the next time interval. Do not use this function to decorate a synchronous function (use throttle() instead).

Parameters:

• limit – Maximum number of calls allowed to the decorated function in each time interval.
• interval – Seconds of every time period.
• wait – If True, it will asynchronously wait when limit is reached until the next interval, and then it will call the function.
• on_fail – Value, object or function to return if the call limit is reached. If on_fail is a function, the decorator will return the result of the call to this function (if this function is asynchronous, it will await for its result). Note that on_fail only makes sense if wait is False.

Returns:

Result of the call to the decorated function, or on_fail if limit is reached (and wait is False).