Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |
| 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 | |