From 5a64bf77372f7a6899e8814bad5918ea36a5c5aa Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 26 Aug 2025 05:55:50 +0000 Subject: [PATCH 1/5] Initial plan From bc135ef31658ec901288f0fa28f8ddf4c93c3ba8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 26 Aug 2025 06:03:54 +0000 Subject: [PATCH 2/5] Implement Timer multiple runs enhancement with runs parameter Co-authored-by: mmcky <8263752+mmcky@users.noreply.github.com> --- quantecon/util/tests/test_timing.py | 102 +++++++++++++++++++ quantecon/util/timing.py | 152 ++++++++++++++++++++++++---- 2 files changed, 233 insertions(+), 21 deletions(-) diff --git a/quantecon/util/tests/test_timing.py b/quantecon/util/tests/test_timing.py index 6d454072..4d886f3a 100644 --- a/quantecon/util/tests/test_timing.py +++ b/quantecon/util/tests/test_timing.py @@ -147,3 +147,105 @@ 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_multiple_runs_basic(self): + """Test basic multiple runs functionality.""" + def test_func(): + time.sleep(self.sleep_time) + + timer = Timer(runs=3, silent=True) + timer.timeit(test_func) + + # Check that we have results + assert timer.elapsed is not None + assert isinstance(timer.elapsed, list) + assert len(timer.elapsed) == 3 + assert timer.minimum is not None + assert timer.maximum is not None + assert timer.average is not None + + # Check timing accuracy + for run_time in timer.elapsed: + assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) + + # Check statistics + assert_allclose(timer.average, self.sleep_time, atol=0.05, rtol=2) + assert timer.minimum <= timer.average <= timer.maximum + + def test_multiple_runs_with_args(self): + """Test multiple runs with function arguments.""" + def test_func_with_args(sleep_time, multiplier=1): + time.sleep(sleep_time * multiplier) + + timer = Timer(runs=2, silent=True) + timer.timeit(test_func_with_args, self.sleep_time, multiplier=2) + + expected_time = self.sleep_time * 2 + assert len(timer.elapsed) == 2 + for run_time in timer.elapsed: + assert_allclose(run_time, expected_time, atol=0.05, rtol=2) + + def test_multiple_runs_validation(self): + """Test validation for multiple runs mode.""" + # Test invalid runs parameter + try: + Timer(runs=0) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "runs must be a positive integer" in str(e) + + try: + Timer(runs=-1) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "runs must be a positive integer" in str(e) + + try: + Timer(runs="invalid") + assert False, "Should have raised ValueError" + except ValueError as e: + assert "runs must be a positive integer" in str(e) + + def test_timeit_single_run_error(self): + """Test that timeit() raises error when runs=1.""" + timer = Timer(runs=1, silent=True) + + def dummy_func(): + pass + + try: + timer.timeit(dummy_func) + assert False, "Should have raised RuntimeError" + except RuntimeError as e: + assert "timeit() is only available when runs > 1" in str(e) + + def test_multiple_runs_different_units(self): + """Test multiple runs with different time units.""" + def test_func(): + time.sleep(self.sleep_time) + + # Test milliseconds + timer_ms = Timer(runs=2, unit="milliseconds", silent=True) + timer_ms.timeit(test_func) + + # Times should still be stored in seconds internally + for run_time in timer_ms.elapsed: + assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) + + # Test microseconds + timer_us = Timer(runs=2, unit="microseconds", silent=True) + timer_us.timeit(test_func) + + for run_time in timer_us.elapsed: + assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) + + def test_multiple_runs_message(self): + """Test multiple runs with custom message.""" + def test_func(): + time.sleep(self.sleep_time) + + timer = Timer(runs=2, message="Test operation", silent=True) + timer.timeit(test_func) + + assert len(timer.elapsed) == 2 + assert timer.average is not None diff --git a/quantecon/util/timing.py b/quantecon/util/timing.py index fed189fa..cacf0051 100644 --- a/quantecon/util/timing.py +++ b/quantecon/util/timing.py @@ -196,58 +196,168 @@ class Timer: Unit to display timing in. Options: "seconds", "milliseconds", "microseconds" silent : bool, optional(default=False) If True, suppress printing of timing results. + runs : int, optional(default=1) + Number of runs to execute. If > 1, enables multiple runs mode. Attributes ---------- - elapsed : float - The elapsed time in seconds. Available after exiting the context. + elapsed : float or list + For single run (runs=1): elapsed time in seconds. + For multiple runs (runs>1): list of elapsed times for each run. + minimum : float + Minimum elapsed time (only available when runs > 1). + maximum : float + Maximum elapsed time (only available when runs > 1). + average : float + Average elapsed time (only available when runs > 1). Examples -------- - Basic usage: + Basic usage (single run): >>> with Timer(): ... # some code ... pass 0.00 seconds elapsed - With custom message and precision: - >>> with Timer("Computing results", precision=4): - ... # some code - ... pass - Computing results: 0.0001 seconds elapsed + Multiple runs with callable: + >>> def my_function(): + ... time.sleep(0.01) + >>> timer = Timer(runs=3, silent=True) + >>> timer.timeit(my_function) + >>> print(f"Average: {timer.average:.4f}s") + Average: 0.0101s - Store elapsed time for comparison: - >>> timer = Timer(silent=True) - >>> with timer: - ... # some code - ... pass - >>> print(f"Method took {timer.elapsed:.6f} seconds") - Method took 0.000123 seconds + Multiple runs with context manager and manual timing: + >>> with Timer(runs=3) as timer: + ... for i in range(3): + ... with timer.time_run(): + ... time.sleep(0.01) + Run 1/3: 0.01 seconds + Run 2/3: 0.01 seconds + Run 3/3: 0.01 seconds + Average: 0.01 seconds, Min: 0.01 seconds, Max: 0.01 seconds """ - def __init__(self, message="", precision=2, unit="seconds", silent=False): + def __init__(self, message="", precision=2, unit="seconds", silent=False, runs=1): self.message = message self.precision = precision self.unit = unit.lower() self.silent = silent + self.runs = runs self.elapsed = None + self.minimum = None + self.maximum = None + self.average = None self._start_time = None + self._run_times = [] # Validate unit valid_units = ["seconds", "milliseconds", "microseconds"] if self.unit not in valid_units: raise ValueError(f"unit must be one of {valid_units}") + + # Validate runs + if not isinstance(runs, int) or runs < 1: + raise ValueError("runs must be a positive integer") def __enter__(self): - self._start_time = time.time() - return self + if self.runs == 1: + # Single run mode - start timing immediately + self._start_time = time.time() + return self + else: + # Multiple runs mode - don't start timing until timeit() is called + return self def __exit__(self, exc_type, exc_val, exc_tb): - end_time = time.time() - self.elapsed = end_time - self._start_time + if self.runs == 1: + # Single run mode - record elapsed time + end_time = time.time() + self.elapsed = end_time - self._start_time + + if not self.silent: + self._print_elapsed() + # For multiple runs mode, __exit__ does nothing since timing + # is handled in timeit() method + + def timeit(self, func, *args, **kwargs): + """ + Execute a callable multiple times and collect timing statistics. + Only available when runs > 1. + + Parameters + ---------- + func : callable + Function to execute multiple times + *args, **kwargs + Arguments to pass to func + + Returns + ------- + None + Results are stored in elapsed, minimum, maximum, average attributes + """ + if self.runs == 1: + raise RuntimeError("timeit() is only available when runs > 1. Use context manager for single runs.") + + run_times = [] + for i in range(self.runs): + start_time = time.time() + func(*args, **kwargs) + end_time = time.time() + run_time = end_time - start_time + run_times.append(run_time) + + if not self.silent: + self._print_single_run(i + 1, run_time) + + # Store results + self.elapsed = run_times + self.minimum = min(run_times) + self.maximum = max(run_times) + self.average = sum(run_times) / len(run_times) if not self.silent: - self._print_elapsed() + self._print_multiple_runs_summary() + + def _print_single_run(self, run_number, run_time): + """Print timing for a single run in multiple runs mode.""" + # Convert to requested unit + if self.unit == "milliseconds": + elapsed_display = run_time * 1000 + unit_str = "ms" + elif self.unit == "microseconds": + elapsed_display = run_time * 1000000 + unit_str = "μs" + else: # seconds + elapsed_display = run_time + unit_str = "seconds" + + print(f"Run {run_number}/{self.runs}: {elapsed_display:.{self.precision}f} {unit_str}") + + def _print_multiple_runs_summary(self): + """Print summary statistics for multiple runs.""" + # Convert to requested unit + if self.unit == "milliseconds": + avg_display = self.average * 1000 + min_display = self.minimum * 1000 + max_display = self.maximum * 1000 + unit_str = "ms" + elif self.unit == "microseconds": + avg_display = self.average * 1000000 + min_display = self.minimum * 1000000 + max_display = self.maximum * 1000000 + unit_str = "μs" + else: # seconds + avg_display = self.average + min_display = self.minimum + max_display = self.maximum + unit_str = "seconds" + + prefix = f"{self.message}: " if self.message else "" + print(f"{prefix}Average: {avg_display:.{self.precision}f} {unit_str}, " + f"Min: {min_display:.{self.precision}f} {unit_str}, " + f"Max: {max_display:.{self.precision}f} {unit_str}") def _print_elapsed(self): """Print the elapsed time with appropriate formatting.""" From c0fac9d43b61850faa2c96136263266636533811 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 27 Aug 2025 00:10:54 +0000 Subject: [PATCH 3/5] Add time_run() context manager for multiple runs mode - Implement time_run() method that returns a context manager for timing individual runs - Allow usage pattern: with Timer(runs=5): for i in range(5): with timer.time_run(): ... - Enhanced __exit__ to compute statistics when all runs are completed in context manager mode - Add comprehensive tests for new time_run functionality - Maintain full backward compatibility with existing timeit() method and single run mode Co-authored-by: mmcky <8263752+mmcky@users.noreply.github.com> --- quantecon/util/tests/test_timing.py | 53 ++++++++++++++++++++++++++ quantecon/util/timing.py | 58 ++++++++++++++++++++++++++++- 2 files changed, 109 insertions(+), 2 deletions(-) diff --git a/quantecon/util/tests/test_timing.py b/quantecon/util/tests/test_timing.py index 4d886f3a..848a8667 100644 --- a/quantecon/util/tests/test_timing.py +++ b/quantecon/util/tests/test_timing.py @@ -249,3 +249,56 @@ def test_func(): assert len(timer.elapsed) == 2 assert timer.average is not None + + def test_time_run_context_manager(self): + """Test time_run() context manager for manual timing of multiple runs.""" + timer = Timer(runs=3, silent=True) + + with timer: + for i in range(3): + with timer.time_run(): + time.sleep(self.sleep_time) + + # Check that we collected all runs + assert timer.elapsed is not None + assert isinstance(timer.elapsed, list) + assert len(timer.elapsed) == 3 + assert timer.minimum is not None + assert timer.maximum is not None + assert timer.average is not None + + # Check timing accuracy + for run_time in timer.elapsed: + assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) + + # Check statistics + assert_allclose(timer.average, self.sleep_time, atol=0.05, rtol=2) + assert timer.minimum <= timer.average <= timer.maximum + + def test_time_run_single_run_error(self): + """Test that time_run() raises error when runs=1.""" + timer = Timer(runs=1, silent=True) + + try: + timer.time_run() + assert False, "Should have raised RuntimeError" + except RuntimeError as e: + assert "time_run() is only available when runs > 1" in str(e) + + def test_time_run_partial_runs(self): + """Test time_run() with partial completion of runs.""" + timer = Timer(runs=5, silent=True) + + with timer: + # Only complete 3 out of 5 runs + for i in range(3): + with timer.time_run(): + time.sleep(self.sleep_time) + + # Should have recorded the 3 completed runs + assert len(timer._run_times) == 3 + # But final statistics should not be set since not all runs completed + assert timer.elapsed is None + assert timer.minimum is None + assert timer.maximum is None + assert timer.average is None diff --git a/quantecon/util/timing.py b/quantecon/util/timing.py index cacf0051..6bc25049 100644 --- a/quantecon/util/timing.py +++ b/quantecon/util/timing.py @@ -6,6 +6,28 @@ import numpy as np +class _RunTimer: + """Internal context manager for individual run timing.""" + + def __init__(self, parent_timer): + self.parent = parent_timer + self._start_time = None + + def __enter__(self): + self._start_time = time.time() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + end_time = time.time() + run_time = end_time - self._start_time + run_number = len(self.parent._run_times) + 1 + + self.parent._run_times.append(run_time) + + if not self.parent.silent: + self.parent._print_single_run(run_number, run_time) + + class __Timer__: """Computes elapsed time, between tic, tac, and toc. @@ -277,8 +299,16 @@ def __exit__(self, exc_type, exc_val, exc_tb): if not self.silent: self._print_elapsed() - # For multiple runs mode, __exit__ does nothing since timing - # is handled in timeit() method + else: + # Multiple runs mode - print summary if runs were completed + if len(self._run_times) == self.runs: + self.elapsed = self._run_times.copy() + self.minimum = min(self._run_times) + self.maximum = max(self._run_times) + self.average = sum(self._run_times) / len(self._run_times) + + if not self.silent: + self._print_multiple_runs_summary() def timeit(self, func, *args, **kwargs): """ @@ -320,6 +350,30 @@ def timeit(self, func, *args, **kwargs): if not self.silent: self._print_multiple_runs_summary() + def time_run(self): + """ + Context manager for timing individual runs in multiple runs mode. + Only available when runs > 1. + + Returns + ------- + _RunTimer + A context manager for timing a single run + + Examples + -------- + >>> with Timer(runs=3, silent=True) as timer: + ... for i in range(3): + ... with timer.time_run(): + ... time.sleep(0.01) + >>> len(timer.elapsed) + 3 + """ + if self.runs == 1: + raise RuntimeError("time_run() is only available when runs > 1. Use the Timer context manager directly for single runs.") + + return _RunTimer(self) + def _print_single_run(self, run_number, run_time): """Print timing for a single run in multiple runs mode.""" # Convert to requested unit From c880606bac9db6160400a9cc6449128efad82647 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 27 Aug 2025 00:32:05 +0000 Subject: [PATCH 4/5] Simplify Timer API by removing time_run() method - Remove _RunTimer class and time_run() method as requested - Context manager now only supports single runs (runs=1) - Multiple runs only supported via timeit() method - Update tests to reflect simplified API - Maintain full backward compatibility for existing usage patterns Co-authored-by: mmcky <8263752+mmcky@users.noreply.github.com> --- quantecon/util/tests/test_timing.py | 53 +++----------------- quantecon/util/timing.py | 78 ++++------------------------- 2 files changed, 16 insertions(+), 115 deletions(-) diff --git a/quantecon/util/tests/test_timing.py b/quantecon/util/tests/test_timing.py index 848a8667..5be10e48 100644 --- a/quantecon/util/tests/test_timing.py +++ b/quantecon/util/tests/test_timing.py @@ -250,55 +250,14 @@ def test_func(): assert len(timer.elapsed) == 2 assert timer.average is not None - def test_time_run_context_manager(self): - """Test time_run() context manager for manual timing of multiple runs.""" + def test_context_manager_multiple_runs_error(self): + """Test that context manager usage raises error when runs > 1.""" timer = Timer(runs=3, silent=True) - with timer: - for i in range(3): - with timer.time_run(): - time.sleep(self.sleep_time) - - # Check that we collected all runs - assert timer.elapsed is not None - assert isinstance(timer.elapsed, list) - assert len(timer.elapsed) == 3 - assert timer.minimum is not None - assert timer.maximum is not None - assert timer.average is not None - - # Check timing accuracy - for run_time in timer.elapsed: - assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) - - # Check statistics - assert_allclose(timer.average, self.sleep_time, atol=0.05, rtol=2) - assert timer.minimum <= timer.average <= timer.maximum - - def test_time_run_single_run_error(self): - """Test that time_run() raises error when runs=1.""" - timer = Timer(runs=1, silent=True) - try: - timer.time_run() + with timer: + time.sleep(self.sleep_time) assert False, "Should have raised RuntimeError" except RuntimeError as e: - assert "time_run() is only available when runs > 1" in str(e) - - def test_time_run_partial_runs(self): - """Test time_run() with partial completion of runs.""" - timer = Timer(runs=5, silent=True) - - with timer: - # Only complete 3 out of 5 runs - for i in range(3): - with timer.time_run(): - time.sleep(self.sleep_time) - - # Should have recorded the 3 completed runs - assert len(timer._run_times) == 3 - # But final statistics should not be set since not all runs completed - assert timer.elapsed is None - assert timer.minimum is None - assert timer.maximum is None - assert timer.average is None + assert "Context manager usage is only supported for single runs" in str(e) + diff --git a/quantecon/util/timing.py b/quantecon/util/timing.py index 6bc25049..a15ba9df 100644 --- a/quantecon/util/timing.py +++ b/quantecon/util/timing.py @@ -6,27 +6,6 @@ import numpy as np -class _RunTimer: - """Internal context manager for individual run timing.""" - - def __init__(self, parent_timer): - self.parent = parent_timer - self._start_time = None - - def __enter__(self): - self._start_time = time.time() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - end_time = time.time() - run_time = end_time - self._start_time - run_number = len(self.parent._run_times) + 1 - - self.parent._run_times.append(run_time) - - if not self.parent.silent: - self.parent._print_single_run(run_number, run_time) - class __Timer__: """Computes elapsed time, between tic, tac, and toc. @@ -249,11 +228,11 @@ class Timer: >>> print(f"Average: {timer.average:.4f}s") Average: 0.0101s - Multiple runs with context manager and manual timing: - >>> with Timer(runs=3) as timer: - ... for i in range(3): - ... with timer.time_run(): - ... time.sleep(0.01) + Multiple runs with callable: + >>> def my_function(): + ... time.sleep(0.01) + >>> timer = Timer(runs=3) + >>> timer.timeit(my_function) Run 1/3: 0.01 seconds Run 2/3: 0.01 seconds Run 3/3: 0.01 seconds @@ -271,7 +250,6 @@ def __init__(self, message="", precision=2, unit="seconds", silent=False, runs=1 self.maximum = None self.average = None self._start_time = None - self._run_times = [] # Validate unit valid_units = ["seconds", "milliseconds", "microseconds"] @@ -283,13 +261,8 @@ def __init__(self, message="", precision=2, unit="seconds", silent=False, runs=1 raise ValueError("runs must be a positive integer") def __enter__(self): - if self.runs == 1: - # Single run mode - start timing immediately - self._start_time = time.time() - return self - else: - # Multiple runs mode - don't start timing until timeit() is called - return self + self._start_time = time.time() + return self def __exit__(self, exc_type, exc_val, exc_tb): if self.runs == 1: @@ -300,15 +273,8 @@ def __exit__(self, exc_type, exc_val, exc_tb): if not self.silent: self._print_elapsed() else: - # Multiple runs mode - print summary if runs were completed - if len(self._run_times) == self.runs: - self.elapsed = self._run_times.copy() - self.minimum = min(self._run_times) - self.maximum = max(self._run_times) - self.average = sum(self._run_times) / len(self._run_times) - - if not self.silent: - self._print_multiple_runs_summary() + # Multiple runs mode - context manager not supported, only timeit() + raise RuntimeError("Context manager usage is only supported for single runs (runs=1). For multiple runs, use the timeit() method.") def timeit(self, func, *args, **kwargs): """ @@ -349,31 +315,7 @@ def timeit(self, func, *args, **kwargs): if not self.silent: self._print_multiple_runs_summary() - - def time_run(self): - """ - Context manager for timing individual runs in multiple runs mode. - Only available when runs > 1. - - Returns - ------- - _RunTimer - A context manager for timing a single run - - Examples - -------- - >>> with Timer(runs=3, silent=True) as timer: - ... for i in range(3): - ... with timer.time_run(): - ... time.sleep(0.01) - >>> len(timer.elapsed) - 3 - """ - if self.runs == 1: - raise RuntimeError("time_run() is only available when runs > 1. Use the Timer context manager directly for single runs.") - - return _RunTimer(self) - + def _print_single_run(self, run_number, run_time): """Print timing for a single run in multiple runs mode.""" # Convert to requested unit From 8bf6f45fe048d1e97720e0ace35ceb6e8399d5f3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 27 Aug 2025 01:26:02 +0000 Subject: [PATCH 5/5] Implement timeit function as requested - revert Timer to original simple form and add separate timeit function Co-authored-by: mmcky <8263752+mmcky@users.noreply.github.com> --- quantecon/__init__.py | 2 +- quantecon/util/__init__.py | 2 +- quantecon/util/tests/test_timing.py | 159 ++++++++-------- quantecon/util/timing.py | 286 ++++++++++++++++------------ 4 files changed, 242 insertions(+), 207 deletions(-) diff --git a/quantecon/__init__.py b/quantecon/__init__.py index 8a08b0b9..3cb0aa98 100644 --- a/quantecon/__init__.py +++ b/quantecon/__init__.py @@ -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 diff --git a/quantecon/util/__init__.py b/quantecon/util/__init__.py index 453e76c9..8ecb232d 100644 --- a/quantecon/util/__init__.py +++ b/quantecon/util/__init__.py @@ -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 diff --git a/quantecon/util/tests/test_timing.py b/quantecon/util/tests/test_timing.py index 5be10e48..ee026a6a 100644 --- a/quantecon/util/tests/test_timing.py +++ b/quantecon/util/tests/test_timing.py @@ -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: @@ -148,116 +148,115 @@ def test_timer_exception_handling(self): assert timer.elapsed is not None assert_allclose(timer.elapsed, self.sleep_time, atol=0.05, rtol=2) - def test_multiple_runs_basic(self): - """Test basic multiple runs functionality.""" + def test_timeit_basic(self): + """Test basic timeit functionality.""" def test_func(): time.sleep(self.sleep_time) - timer = Timer(runs=3, silent=True) - timer.timeit(test_func) + result = timeit(test_func, runs=3, silent=True) # Check that we have results - assert timer.elapsed is not None - assert isinstance(timer.elapsed, list) - assert len(timer.elapsed) == 3 - assert timer.minimum is not None - assert timer.maximum is not None - assert timer.average is not None + 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 timer.elapsed: + for run_time in result['elapsed']: assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) - - # Check statistics - assert_allclose(timer.average, self.sleep_time, atol=0.05, rtol=2) - assert timer.minimum <= timer.average <= timer.maximum - - def test_multiple_runs_with_args(self): - """Test multiple runs with function arguments.""" + + 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) - timer = Timer(runs=2, silent=True) - timer.timeit(test_func_with_args, self.sleep_time, multiplier=2) + # 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) - expected_time = self.sleep_time * 2 - assert len(timer.elapsed) == 2 - for run_time in timer.elapsed: - assert_allclose(run_time, expected_time, atol=0.05, rtol=2) - - def test_multiple_runs_validation(self): - """Test validation for multiple runs mode.""" + # 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: - Timer(runs=0) + 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: - Timer(runs=-1) - assert False, "Should have raised ValueError" + 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: - Timer(runs="invalid") + timeit("not a function", runs=1) assert False, "Should have raised ValueError" except ValueError as e: - assert "runs must be a positive integer" in str(e) - - def test_timeit_single_run_error(self): - """Test that timeit() raises error when runs=1.""" - timer = Timer(runs=1, silent=True) - - def dummy_func(): - pass - - try: - timer.timeit(dummy_func) - assert False, "Should have raised RuntimeError" - except RuntimeError as e: - assert "timeit() is only available when runs > 1" in str(e) - - def test_multiple_runs_different_units(self): - """Test multiple runs with different time units.""" + 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) - # Test milliseconds - timer_ms = Timer(runs=2, unit="milliseconds", silent=True) - timer_ms.timeit(test_func) - - # Times should still be stored in seconds internally - for run_time in timer_ms.elapsed: - assert_allclose(run_time, self.sleep_time, atol=0.05, rtol=2) + 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 - timer_us = Timer(runs=2, unit="microseconds", silent=True) - timer_us.timeit(test_func) + result_us = timeit(test_func, runs=2, unit="microseconds", silent=True) + assert len(result_us['elapsed']) == 2 - for run_time in timer_us.elapsed: + # 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_multiple_runs_message(self): - """Test multiple runs with custom message.""" + + def test_timeit_stats_only(self): + """Test timeit with stats_only option.""" def test_func(): time.sleep(self.sleep_time) - - timer = Timer(runs=2, message="Test operation", silent=True) - timer.timeit(test_func) - - assert len(timer.elapsed) == 2 - assert timer.average is not None - - def test_context_manager_multiple_runs_error(self): - """Test that context manager usage raises error when runs > 1.""" - timer = Timer(runs=3, silent=True) - + + # 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: - with timer: - time.sleep(self.sleep_time) - assert False, "Should have raised RuntimeError" - except RuntimeError as e: - assert "Context manager usage is only supported for single runs" in str(e) + 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) diff --git a/quantecon/util/timing.py b/quantecon/util/timing.py index a15ba9df..1b724792 100644 --- a/quantecon/util/timing.py +++ b/quantecon/util/timing.py @@ -6,7 +6,6 @@ import numpy as np - class __Timer__: """Computes elapsed time, between tic, tac, and toc. @@ -197,163 +196,58 @@ class Timer: Unit to display timing in. Options: "seconds", "milliseconds", "microseconds" silent : bool, optional(default=False) If True, suppress printing of timing results. - runs : int, optional(default=1) - Number of runs to execute. If > 1, enables multiple runs mode. Attributes ---------- - elapsed : float or list - For single run (runs=1): elapsed time in seconds. - For multiple runs (runs>1): list of elapsed times for each run. - minimum : float - Minimum elapsed time (only available when runs > 1). - maximum : float - Maximum elapsed time (only available when runs > 1). - average : float - Average elapsed time (only available when runs > 1). + elapsed : float + The elapsed time in seconds. Available after exiting the context. Examples -------- - Basic usage (single run): + Basic usage: >>> with Timer(): ... # some code ... pass 0.00 seconds elapsed - Multiple runs with callable: - >>> def my_function(): - ... time.sleep(0.01) - >>> timer = Timer(runs=3, silent=True) - >>> timer.timeit(my_function) - >>> print(f"Average: {timer.average:.4f}s") - Average: 0.0101s + With custom message and precision: + >>> with Timer("Computing results", precision=4): + ... # some code + ... pass + Computing results: 0.0001 seconds elapsed - Multiple runs with callable: - >>> def my_function(): - ... time.sleep(0.01) - >>> timer = Timer(runs=3) - >>> timer.timeit(my_function) - Run 1/3: 0.01 seconds - Run 2/3: 0.01 seconds - Run 3/3: 0.01 seconds - Average: 0.01 seconds, Min: 0.01 seconds, Max: 0.01 seconds + Store elapsed time for comparison: + >>> timer = Timer(silent=True) + >>> with timer: + ... # some code + ... pass + >>> print(f"Method took {timer.elapsed:.6f} seconds") + Method took 0.000123 seconds """ - def __init__(self, message="", precision=2, unit="seconds", silent=False, runs=1): + def __init__(self, message="", precision=2, unit="seconds", silent=False): self.message = message self.precision = precision self.unit = unit.lower() self.silent = silent - self.runs = runs self.elapsed = None - self.minimum = None - self.maximum = None - self.average = None self._start_time = None # Validate unit valid_units = ["seconds", "milliseconds", "microseconds"] if self.unit not in valid_units: raise ValueError(f"unit must be one of {valid_units}") - - # Validate runs - if not isinstance(runs, int) or runs < 1: - raise ValueError("runs must be a positive integer") def __enter__(self): self._start_time = time.time() return self def __exit__(self, exc_type, exc_val, exc_tb): - if self.runs == 1: - # Single run mode - record elapsed time - end_time = time.time() - self.elapsed = end_time - self._start_time - - if not self.silent: - self._print_elapsed() - else: - # Multiple runs mode - context manager not supported, only timeit() - raise RuntimeError("Context manager usage is only supported for single runs (runs=1). For multiple runs, use the timeit() method.") - - def timeit(self, func, *args, **kwargs): - """ - Execute a callable multiple times and collect timing statistics. - Only available when runs > 1. - - Parameters - ---------- - func : callable - Function to execute multiple times - *args, **kwargs - Arguments to pass to func - - Returns - ------- - None - Results are stored in elapsed, minimum, maximum, average attributes - """ - if self.runs == 1: - raise RuntimeError("timeit() is only available when runs > 1. Use context manager for single runs.") - - run_times = [] - for i in range(self.runs): - start_time = time.time() - func(*args, **kwargs) - end_time = time.time() - run_time = end_time - start_time - run_times.append(run_time) - - if not self.silent: - self._print_single_run(i + 1, run_time) - - # Store results - self.elapsed = run_times - self.minimum = min(run_times) - self.maximum = max(run_times) - self.average = sum(run_times) / len(run_times) + end_time = time.time() + self.elapsed = end_time - self._start_time if not self.silent: - self._print_multiple_runs_summary() - - def _print_single_run(self, run_number, run_time): - """Print timing for a single run in multiple runs mode.""" - # Convert to requested unit - if self.unit == "milliseconds": - elapsed_display = run_time * 1000 - unit_str = "ms" - elif self.unit == "microseconds": - elapsed_display = run_time * 1000000 - unit_str = "μs" - else: # seconds - elapsed_display = run_time - unit_str = "seconds" - - print(f"Run {run_number}/{self.runs}: {elapsed_display:.{self.precision}f} {unit_str}") - - def _print_multiple_runs_summary(self): - """Print summary statistics for multiple runs.""" - # Convert to requested unit - if self.unit == "milliseconds": - avg_display = self.average * 1000 - min_display = self.minimum * 1000 - max_display = self.maximum * 1000 - unit_str = "ms" - elif self.unit == "microseconds": - avg_display = self.average * 1000000 - min_display = self.minimum * 1000000 - max_display = self.maximum * 1000000 - unit_str = "μs" - else: # seconds - avg_display = self.average - min_display = self.minimum - max_display = self.maximum - unit_str = "seconds" - - prefix = f"{self.message}: " if self.message else "" - print(f"{prefix}Average: {avg_display:.{self.precision}f} {unit_str}, " - f"Min: {min_display:.{self.precision}f} {unit_str}, " - f"Max: {max_display:.{self.precision}f} {unit_str}") + self._print_elapsed() def _print_elapsed(self): """Print the elapsed time with appropriate formatting.""" @@ -377,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()