Coverage for src / tracekit / core / performance.py: 100%
82 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Performance timing and monitoring infrastructure.
3This module provides automatic performance timing for operations,
4metric collection, and performance summary reporting.
7Example:
8 >>> from tracekit.core.performance import timed, get_performance_summary
9 >>> @timed(threshold=0.1)
10 ... def slow_operation(data):
11 ... result = expensive_computation(data)
12 ... return result
13 >>> summary = get_performance_summary()
15References:
16 - Python time.perf_counter() for high-resolution timing
17 - Performance monitoring best practices
18"""
20from __future__ import annotations
22import functools
23import time
24from collections import defaultdict
25from collections.abc import Callable
26from dataclasses import dataclass, field
27from datetime import UTC, datetime
28from typing import TYPE_CHECKING, Any, TypeVar
30import numpy as np
32if TYPE_CHECKING:
33 from numpy.typing import NDArray
36@dataclass
37class PerformanceRecord:
38 """Record of a timed operation.
40 Attributes:
41 operation: Name of the operation (typically function name).
42 duration: Duration in seconds.
43 timestamp: ISO 8601 timestamp of when operation completed.
44 metadata: Optional metadata about the operation.
46 References:
47 LOG-006: Automatic Performance Timing
48 """
50 operation: str
51 duration: float
52 timestamp: str
53 metadata: dict[str, Any] = field(default_factory=dict)
56class PerformanceCollector:
57 """Collects and aggregates performance timing data.
59 Thread-safe collector for recording and analyzing operation timings.
60 Provides summary statistics per operation type.
62 Example:
63 >>> collector = PerformanceCollector()
64 >>> collector.record("fft_computation", 1.23, samples=1000000)
65 >>> collector.record("fft_computation", 1.45, samples=2000000)
66 >>> summary = collector.get_summary()
67 >>> print(summary["fft_computation"]["mean"])
68 1.34
70 References:
71 LOG-006: Automatic Performance Timing
72 """
74 def __init__(self) -> None:
75 """Initialize the performance collector."""
76 self.records: list[PerformanceRecord] = []
78 def record(
79 self,
80 operation: str,
81 duration: float,
82 **metadata: Any,
83 ) -> None:
84 """Record a performance measurement.
86 Args:
87 operation: Name of the operation being timed.
88 duration: Duration in seconds (use time.perf_counter()).
89 **metadata: Additional metadata about the operation.
91 Example:
92 >>> collector.record("load_trace", 0.5, file_size_mb=100)
93 >>> collector.record("compute_fft", 2.3, samples=10000000)
95 References:
96 LOG-006: Automatic Performance Timing
97 """
98 self.records.append(
99 PerformanceRecord(
100 operation=operation,
101 duration=duration,
102 timestamp=datetime.now(UTC).isoformat(),
103 metadata=metadata,
104 )
105 )
107 def get_summary(self) -> dict[str, dict[str, float]]:
108 """Get summary statistics per operation.
110 Returns:
111 Dictionary mapping operation names to statistics:
112 - count: Number of times operation was performed
113 - mean: Mean duration in seconds
114 - std: Standard deviation of duration
115 - min: Minimum duration
116 - max: Maximum duration
117 - total: Total time spent in operation
119 Example:
120 >>> summary = collector.get_summary()
121 >>> fft_stats = summary["fft_computation"]
122 >>> print(f"FFT avg: {fft_stats['mean']:.3f}s")
124 References:
125 LOG-006: Automatic Performance Timing
126 """
127 by_op: dict[str, list[float]] = defaultdict(list)
128 for record in self.records:
129 by_op[record.operation].append(record.duration)
131 summary: dict[str, dict[str, float]] = {}
132 for op, durations in by_op.items():
133 arr: NDArray[np.float64] = np.array(durations)
134 summary[op] = {
135 "count": float(len(arr)),
136 "mean": float(np.mean(arr)),
137 "std": float(np.std(arr)),
138 "min": float(np.min(arr)),
139 "max": float(np.max(arr)),
140 "total": float(np.sum(arr)),
141 }
142 return summary
144 def clear(self) -> None:
145 """Clear all performance records.
147 Useful for resetting measurements between test runs or
148 analysis sessions.
150 Example:
151 >>> collector.clear()
152 """
153 self.records.clear()
155 def get_records(
156 self,
157 operation: str | None = None,
158 since: datetime | None = None,
159 ) -> list[PerformanceRecord]:
160 """Get performance records with optional filtering.
162 Args:
163 operation: Filter by operation name, or None for all.
164 since: Filter records after this timestamp, or None for all.
166 Returns:
167 List of matching performance records.
169 Example:
170 >>> from datetime import datetime, timezone, timedelta
171 >>> one_hour_ago = datetime.now(timezone.utc) - timedelta(hours=1)
172 >>> recent = collector.get_records(since=one_hour_ago)
173 """
174 records = self.records
176 if operation is not None:
177 records = [r for r in records if r.operation == operation]
179 if since is not None:
180 since_iso = since.isoformat()
181 records = [r for r in records if r.timestamp >= since_iso]
183 return records
186# Global performance collector
187_global_collector = PerformanceCollector()
190F = TypeVar("F", bound=Callable[..., Any])
193def timed(
194 threshold: float | None = None,
195 log_level: str = "DEBUG",
196 collect: bool = True,
197) -> Callable[[F], F]:
198 """Decorator to time and log function execution.
200 Automatically times function execution using time.perf_counter()
201 and optionally logs if duration exceeds threshold.
203 Args:
204 threshold: Only log if duration exceeds this (seconds). None logs all.
205 log_level: Log level for timing messages (DEBUG, INFO, WARNING, etc.).
206 collect: Whether to collect timing in global PerformanceCollector.
208 Returns:
209 Decorated function.
211 Example:
212 >>> @timed(threshold=0.1)
213 ... def slow_function(data):
214 ... time.sleep(0.2)
215 ... return process(data)
217 >>> @timed(log_level="INFO")
218 ... def important_operation():
219 ... return compute_result()
221 References:
222 LOG-006: Automatic Performance Timing
223 """
225 def decorator(func: F) -> F:
226 @functools.wraps(func)
227 def wrapper(*args: Any, **kwargs: Any) -> Any:
228 start = time.perf_counter()
229 try:
230 return func(*args, **kwargs)
231 finally:
232 duration = time.perf_counter() - start
234 # Record in global collector
235 if collect:
236 _global_collector.record(func.__qualname__, duration)
238 # Log if threshold met
239 if threshold is None or duration >= threshold:
240 # Lazy import to avoid circular dependency
241 from tracekit.core.logging import get_logger
243 logger = get_logger(func.__module__ or "tracekit")
244 log_func = getattr(logger, log_level.lower())
245 log_func(
246 f"Function {func.__qualname__} completed",
247 extra={"duration_seconds": round(duration, 6)},
248 )
250 return wrapper # type: ignore[return-value]
252 return decorator
255def get_performance_summary() -> dict[str, dict[str, float]]:
256 """Get summary of collected performance data from global collector.
258 Returns:
259 Dictionary of performance statistics per operation.
261 Example:
262 >>> summary = get_performance_summary()
263 >>> for op, stats in summary.items():
264 ... print(f"{op}: {stats['mean']:.3f}s avg, {stats['count']:.0f} calls")
266 References:
267 LOG-006: Automatic Performance Timing
268 """
269 return _global_collector.get_summary()
272def clear_performance_data() -> None:
273 """Clear all collected performance data from global collector.
275 Example:
276 >>> clear_performance_data()
277 """
278 _global_collector.clear()
281def get_performance_records(
282 operation: str | None = None,
283 since: datetime | None = None,
284) -> list[PerformanceRecord]:
285 """Get performance records from global collector.
287 Args:
288 operation: Filter by operation name, or None for all.
289 since: Filter records after this timestamp, or None for all.
291 Returns:
292 List of matching performance records.
294 Example:
295 >>> records = get_performance_records(operation="fft_computation")
296 >>> for record in records:
297 ... print(f"{record.timestamp}: {record.duration:.3f}s")
298 """
299 return _global_collector.get_records(operation=operation, since=since)
302class PerformanceContext:
303 """Context manager for timing a block of code.
305 Example:
306 >>> with PerformanceContext("data_loading") as ctx:
307 ... data = load_large_file()
308 >>> print(f"Loading took {ctx.duration:.3f}s")
310 References:
311 LOG-006: Automatic Performance Timing
312 """
314 def __init__(
315 self,
316 operation: str,
317 log_threshold: float | None = None,
318 collect: bool = True,
319 ):
320 """Initialize performance context.
322 Args:
323 operation: Name of the operation being timed.
324 log_threshold: Log if duration exceeds threshold (seconds).
325 collect: Whether to collect in global collector.
326 """
327 self.operation = operation
328 self.log_threshold = log_threshold
329 self.collect = collect
330 self.start_time: float = 0.0
331 self.duration: float = 0.0
333 def __enter__(self) -> PerformanceContext:
334 """Start timing."""
335 self.start_time = time.perf_counter()
336 return self
338 def __exit__(self, *args: Any) -> None:
339 """Stop timing and record."""
340 self.duration = time.perf_counter() - self.start_time
342 if self.collect:
343 _global_collector.record(self.operation, self.duration)
345 if self.log_threshold is None or self.duration >= self.log_threshold:
346 from tracekit.core.logging import get_logger
348 logger = get_logger("tracekit.performance")
349 logger.debug(
350 f"Operation {self.operation} completed",
351 extra={"duration_seconds": round(self.duration, 6)},
352 )