| """ |
| 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 |
|
|
