File size: 3,027 Bytes
df4a21a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """
Timing utilities for performance measurement.
"""
import time
from contextlib import contextmanager
from typing import Dict, Generator, Optional
from app.core.logging import get_logger
logger = get_logger(__name__)
class Timer:
"""
Timer class for measuring execution time of code blocks.
Usage:
timer = Timer()
with timer.measure("inference"):
# do inference
with timer.measure("fusion"):
# do fusion
timings = timer.get_timings()
"""
def __init__(self):
self._start_time: Optional[float] = None
self._timings: Dict[str, int] = {}
self._total_start: Optional[float] = None
def start_total(self) -> None:
"""Start the total timer."""
self._total_start = time.perf_counter()
def stop_total(self) -> None:
"""Stop the total timer and record the duration."""
if self._total_start is not None:
elapsed_ms = int((time.perf_counter() - self._total_start) * 1000)
self._timings["total"] = elapsed_ms
@contextmanager
def measure(self, name: str) -> Generator[None, None, None]:
"""
Context manager to measure execution time of a block.
Args:
name: Name for this timing measurement
Yields:
None
"""
start = time.perf_counter()
try:
yield
finally:
elapsed_ms = int((time.perf_counter() - start) * 1000)
self._timings[name] = elapsed_ms
logger.debug(f"Timer [{name}]: {elapsed_ms}ms")
def record(self, name: str, duration_ms: int) -> None:
"""
Manually record a timing.
Args:
name: Name for this timing
duration_ms: Duration in milliseconds
"""
self._timings[name] = duration_ms
def get_timings(self) -> Dict[str, int]:
"""
Get all recorded timings.
Returns:
Dictionary of timing name -> milliseconds
"""
return self._timings.copy()
def get(self, name: str) -> Optional[int]:
"""
Get a specific timing.
Args:
name: Timing name
Returns:
Duration in milliseconds, or None if not recorded
"""
return self._timings.get(name)
def reset(self) -> None:
"""Reset all timings."""
self._timings.clear()
self._total_start = None
def measure_time(func):
"""
Decorator to measure function execution time.
Logs the execution time at DEBUG level.
"""
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
result = func(*args, **kwargs)
return result
finally:
elapsed_ms = int((time.perf_counter() - start) * 1000)
logger.debug(f"Function [{func.__name__}]: {elapsed_ms}ms")
return wrapper
|