Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion quantecon/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,4 +52,4 @@
#<-
from ._rank_nullspace import rank_est, nullspace
from ._robustlq import RBLQ
from .util import searchsorted, fetch_nb_dependencies, tic, tac, toc, Timer
from .util import searchsorted, fetch_nb_dependencies, tic, tac, toc, Timer, timeit
2 changes: 1 addition & 1 deletion quantecon/util/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,4 +6,4 @@
from .array import searchsorted
from .notebooks import fetch_nb_dependencies
from .random import check_random_state, rng_integers
from .timing import tic, tac, toc, loop_timer, Timer
from .timing import tic, tac, toc, loop_timer, Timer, timeit
115 changes: 114 additions & 1 deletion quantecon/util/tests/test_timing.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

import time
from numpy.testing import assert_allclose, assert_
from quantecon.util import tic, tac, toc, loop_timer, Timer
from quantecon.util import tic, tac, toc, loop_timer, Timer, timeit


class TestTicTacToc:
Expand Down Expand Up @@ -147,3 +147,116 @@ def test_timer_exception_handling(self):
# Timer should still record elapsed time
assert timer.elapsed is not None
assert_allclose(timer.elapsed, self.sleep_time, atol=0.05, rtol=2)

def test_timeit_basic(self):
"""Test basic timeit functionality."""
def test_func():
time.sleep(self.sleep_time)

result = timeit(test_func, runs=3, silent=True)

# Check that we have results
assert 'elapsed' in result
assert 'average' in result
assert 'minimum' in result
assert 'maximum' in result
assert isinstance(result['elapsed'], list)
assert len(result['elapsed']) == 3

# Check timing accuracy
for run_time in result['elapsed']:
assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2)

assert_allclose(result['average'], self.sleep_time, atol=0.05, rtol=2)
assert result['minimum'] <= result['average'] <= result['maximum']

def test_timeit_lambda_function(self):
"""Test timeit with lambda functions for arguments."""
def test_func_with_args(sleep_time, multiplier=1):
time.sleep(sleep_time * multiplier)

# Use lambda to bind arguments
func_with_args = lambda: test_func_with_args(self.sleep_time, 0.5)
result = timeit(func_with_args, runs=2, silent=True)

# Check results
assert len(result['elapsed']) == 2
for run_time in result['elapsed']:
assert_allclose(run_time, self.sleep_time * 0.5, atol=0.05, rtol=2)

def test_timeit_validation(self):
"""Test validation for timeit function."""
def test_func():
time.sleep(self.sleep_time)

# Test invalid runs parameter
try:
timeit(test_func, runs=0)
assert False, "Should have raised ValueError"
except ValueError as e:
assert "runs must be a positive integer" in str(e)

try:
timeit(test_func, runs=-1)
assert False, "Should have raised ValueError"
except ValueError as e:
assert "runs must be a positive integer" in str(e)

# Test invalid function
try:
timeit("not a function", runs=1)
assert False, "Should have raised ValueError"
except ValueError as e:
assert "func must be callable" in str(e)

def test_timeit_single_run(self):
"""Test that timeit works with single run."""
def test_func():
time.sleep(self.sleep_time)

result = timeit(test_func, runs=1, silent=True)

assert len(result['elapsed']) == 1
assert result['average'] == result['elapsed'][0]
assert result['minimum'] == result['elapsed'][0]
assert result['maximum'] == result['elapsed'][0]

def test_timeit_different_units(self):
"""Test timeit with different time units."""
def test_func():
time.sleep(self.sleep_time)

# Test milliseconds (silent mode to avoid output during tests)
result_ms = timeit(test_func, runs=2, unit="milliseconds", silent=True)
assert len(result_ms['elapsed']) == 2

# Test microseconds
result_us = timeit(test_func, runs=2, unit="microseconds", silent=True)
assert len(result_us['elapsed']) == 2

# All results should be in seconds regardless of display unit
for run_time in result_ms['elapsed']:
assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2)
for run_time in result_us['elapsed']:
assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2)

def test_timeit_stats_only(self):
"""Test timeit with stats_only option."""
def test_func():
time.sleep(self.sleep_time)

# This test is mainly to ensure stats_only doesn't crash
result = timeit(test_func, runs=2, stats_only=True, silent=True)
assert len(result['elapsed']) == 2

def test_timeit_invalid_timer_kwargs(self):
"""Test that invalid timer kwargs are rejected."""
def test_func():
time.sleep(self.sleep_time)

try:
timeit(test_func, runs=1, invalid_param="test")
assert False, "Should have raised ValueError"
except ValueError as e:
assert "Unknown timer parameters" in str(e)

142 changes: 142 additions & 0 deletions quantecon/util/timing.py
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,148 @@ def _print_elapsed(self):
print(f"{prefix}{elapsed_display:.{self.precision}f} {unit_str} elapsed")


def timeit(func, runs=1, stats_only=False, **timer_kwargs):
"""
Execute a function multiple times and collect timing statistics.

This function provides a convenient way to time a function multiple times
and get summary statistics, using the Timer context manager internally.

Parameters
----------
func : callable
Function to execute multiple times. Function should take no arguments,
or be a partial function or lambda with arguments already bound.
runs : int, optional(default=1)
Number of runs to execute. Must be a positive integer.
stats_only : bool, optional(default=False)
If True, only display summary statistics. If False, display
individual run times followed by summary statistics.
**timer_kwargs
Keyword arguments to pass to Timer (message, precision, unit, silent).
Note: silent parameter controls all output when stats_only=False.

Returns
-------
dict
Dictionary containing timing results with keys:
- 'elapsed': list of elapsed times for each run
- 'average': average elapsed time
- 'minimum': minimum elapsed time
- 'maximum': maximum elapsed time

Examples
--------
Basic usage:
>>> def slow_function():
... time.sleep(0.01)
>>> timeit(slow_function, runs=3)
Run 1: 0.01 seconds
Run 2: 0.01 seconds
Run 3: 0.01 seconds
Average: 0.01 seconds, Minimum: 0.01 seconds, Maximum: 0.01 seconds

Summary only:
>>> timeit(slow_function, runs=3, stats_only=True)
Average: 0.01 seconds, Minimum: 0.01 seconds, Maximum: 0.01 seconds

With custom Timer options:
>>> timeit(slow_function, runs=2, unit="milliseconds", precision=1)
Run 1: 10.1 ms
Run 2: 10.0 ms
Average: 10.1 ms, Minimum: 10.0 ms, Maximum: 10.1 ms

With function arguments using lambda:
>>> add_func = lambda: expensive_computation(5, 10)
>>> timeit(add_func, runs=2)
"""
if not isinstance(runs, int) or runs < 1:
raise ValueError("runs must be a positive integer")

if not callable(func):
raise ValueError("func must be callable")

# Extract Timer parameters
timer_params = {
'message': timer_kwargs.pop('message', ''),
'precision': timer_kwargs.pop('precision', 2),
'unit': timer_kwargs.pop('unit', 'seconds'),
'silent': timer_kwargs.pop('silent', False) # Explicit silent parameter
}

# Warn about unused kwargs
if timer_kwargs:
raise ValueError(f"Unknown timer parameters: {list(timer_kwargs.keys())}")

run_times = []

# Execute the function multiple times
for i in range(runs):
# Always silence individual Timer output to avoid duplication with our run display
individual_timer_params = timer_params.copy()
individual_timer_params['silent'] = True

with Timer(**individual_timer_params) as timer:
func()
run_times.append(timer.elapsed)

# Print individual run times unless stats_only or silent
if not stats_only and not timer_params['silent']:
# Convert to requested unit for display
unit = timer_params['unit'].lower()
precision = timer_params['precision']

if unit == "milliseconds":
elapsed_display = timer.elapsed * 1000
unit_str = "ms"
elif unit == "microseconds":
elapsed_display = timer.elapsed * 1000000
unit_str = "μs"
else: # seconds
elapsed_display = timer.elapsed
unit_str = "seconds"

print(f"Run {i + 1}: {elapsed_display:.{precision}f} {unit_str}")

# Calculate statistics
average = sum(run_times) / len(run_times)
minimum = min(run_times)
maximum = max(run_times)

# Print summary statistics unless completely silent
if not timer_params['silent']:
# Convert to requested unit for display
unit = timer_params['unit'].lower()
precision = timer_params['precision']

if unit == "milliseconds":
avg_display = average * 1000
min_display = minimum * 1000
max_display = maximum * 1000
unit_str = "ms"
elif unit == "microseconds":
avg_display = average * 1000000
min_display = minimum * 1000000
max_display = maximum * 1000000
unit_str = "μs"
else: # seconds
avg_display = average
min_display = minimum
max_display = maximum
unit_str = "seconds"

print(f"Average: {avg_display:.{precision}f} {unit_str}, "
f"Minimum: {min_display:.{precision}f} {unit_str}, "
f"Maximum: {max_display:.{precision}f} {unit_str}")

return {
'elapsed': run_times,
'average': average,
'minimum': minimum,
'maximum': maximum
}


def tic():
return __timer__.tic()

Expand Down
Loading