Coverage for src / tracekit / api / profiling.py: 98%
142 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Performance profiling for signal analysis operations.
3This module provides profiling utilities for measuring and analyzing
4performance of signal processing operations.
5"""
7from __future__ import annotations
9import functools
10import logging
11import statistics
12import time
13from contextlib import contextmanager
14from dataclasses import dataclass, field
15from typing import TYPE_CHECKING, Any
17if TYPE_CHECKING:
18 from collections.abc import Callable, Iterator
20logger = logging.getLogger(__name__)
22__all__ = [
23 "OperationProfile",
24 "ProfileReport",
25 "Profiler",
26 "profile",
27]
30@dataclass
31class OperationProfile:
32 """Profile data for a single operation.
34 Attributes:
35 name: Operation name
36 calls: Number of calls
37 total_time: Total time in seconds
38 min_time: Minimum time
39 max_time: Maximum time
40 times: List of individual times
41 memory_peak: Peak memory usage (bytes)
42 input_size: Input data size
44 References:
45 API-012: Performance Profiling API
46 """
48 name: str
49 calls: int = 0
50 total_time: float = 0.0
51 min_time: float = float("inf")
52 max_time: float = 0.0
53 times: list[float] = field(default_factory=list)
54 memory_peak: int = 0
55 input_size: int = 0
57 @property
58 def mean_time(self) -> float:
59 """Average time per call."""
60 return self.total_time / self.calls if self.calls > 0 else 0.0
62 @property
63 def std_time(self) -> float:
64 """Standard deviation of times."""
65 if len(self.times) < 2:
66 return 0.0
67 return statistics.stdev(self.times)
69 @property
70 def throughput(self) -> float:
71 """Throughput in items per second."""
72 if self.total_time > 0 and self.input_size > 0:
73 return (self.input_size * self.calls) / self.total_time
74 return 0.0
76 def record(self, elapsed: float, size: int = 0) -> None:
77 """Record a timing.
79 Args:
80 elapsed: Elapsed time in seconds
81 size: Input size
82 """
83 self.calls += 1
84 self.total_time += elapsed
85 self.min_time = min(self.min_time, elapsed)
86 self.max_time = max(self.max_time, elapsed)
87 self.times.append(elapsed)
88 if size > 0:
89 self.input_size = size
91 def to_dict(self) -> dict[str, Any]:
92 """Convert to dictionary."""
93 return {
94 "name": self.name,
95 "calls": self.calls,
96 "total_time": self.total_time,
97 "mean_time": self.mean_time,
98 "min_time": self.min_time if self.min_time != float("inf") else 0,
99 "max_time": self.max_time,
100 "std_time": self.std_time,
101 "throughput": self.throughput,
102 }
105@dataclass
106class ProfileReport:
107 """Complete profiling report.
109 Attributes:
110 profiles: Dictionary of operation profiles
111 start_time: Report start time
112 end_time: Report end time
113 total_operations: Total number of operations
115 References:
116 API-012: Performance Profiling API
117 """
119 profiles: dict[str, OperationProfile] = field(default_factory=dict)
120 start_time: float = 0.0
121 end_time: float = 0.0
122 total_operations: int = 0
124 @property
125 def total_time(self) -> float:
126 """Total profiled time."""
127 return sum(p.total_time for p in self.profiles.values())
129 @property
130 def wall_time(self) -> float:
131 """Wall clock time."""
132 return self.end_time - self.start_time if self.end_time > 0 else 0.0
134 def get_slowest(self, n: int = 5) -> list[OperationProfile]:
135 """Get slowest operations.
137 Args:
138 n: Number of operations
140 Returns:
141 List of slowest operation profiles
142 """
143 sorted_profiles = sorted(self.profiles.values(), key=lambda p: p.total_time, reverse=True)
144 return sorted_profiles[:n]
146 def get_most_called(self, n: int = 5) -> list[OperationProfile]:
147 """Get most frequently called operations.
149 Args:
150 n: Number of operations
152 Returns:
153 List of most called operation profiles
154 """
155 sorted_profiles = sorted(self.profiles.values(), key=lambda p: p.calls, reverse=True)
156 return sorted_profiles[:n]
158 def summary(self) -> str:
159 """Generate text summary.
161 Returns:
162 Summary string
163 """
164 lines = [
165 "Performance Profile Report",
166 "=" * 50,
167 f"Total operations: {self.total_operations}",
168 f"Total profiled time: {self.total_time:.4f}s",
169 f"Wall clock time: {self.wall_time:.4f}s",
170 "",
171 "Slowest Operations:",
172 "-" * 30,
173 ]
175 for profile in self.get_slowest():
176 lines.append(
177 f" {profile.name}: "
178 f"{profile.total_time:.4f}s ({profile.calls} calls, "
179 f"{profile.mean_time * 1000:.2f}ms avg)"
180 )
182 return "\n".join(lines)
184 def to_dict(self) -> dict[str, Any]:
185 """Convert to dictionary."""
186 return {
187 "total_time": self.total_time,
188 "wall_time": self.wall_time,
189 "total_operations": self.total_operations,
190 "profiles": {name: profile.to_dict() for name, profile in self.profiles.items()},
191 }
194class Profiler:
195 """Performance profiler for signal analysis operations.
197 Tracks timing and performance metrics for operations.
199 Example:
200 >>> profiler = Profiler()
201 >>> with profiler.profile("fft"):
202 ... result = np.fft.fft(data)
203 >>> report = profiler.report()
204 >>> print(report.summary())
206 References:
207 API-012: Performance Profiling API
208 """
210 _instance: Profiler | None = None
212 def __init__(self) -> None:
213 """Initialize profiler."""
214 self._profiles: dict[str, OperationProfile] = {}
215 self._start_time: float = 0.0
216 self._enabled: bool = True
217 self._stack: list[str] = []
219 @classmethod
220 def get_instance(cls) -> Profiler:
221 """Get global profiler instance."""
222 if cls._instance is None:
223 cls._instance = cls()
224 return cls._instance
226 def enable(self) -> None:
227 """Enable profiling."""
228 self._enabled = True
230 def disable(self) -> None:
231 """Disable profiling."""
232 self._enabled = False
234 def reset(self) -> None:
235 """Reset all profiles."""
236 self._profiles.clear()
237 self._start_time = 0.0
239 @contextmanager
240 def profile(self, name: str, input_size: int = 0) -> Iterator[None]:
241 """Context manager for profiling code block.
243 Args:
244 name: Operation name
245 input_size: Input data size
247 Yields:
248 None
250 Example:
251 >>> with profiler.profile("fft"):
252 ... result = compute_fft(data)
253 """
254 if not self._enabled:
255 yield
256 return
258 if self._start_time == 0:
259 self._start_time = time.perf_counter()
261 if name not in self._profiles:
262 self._profiles[name] = OperationProfile(name)
264 self._stack.append(name)
265 start = time.perf_counter()
267 try:
268 yield
269 finally:
270 elapsed = time.perf_counter() - start
271 self._profiles[name].record(elapsed, input_size)
272 self._stack.pop()
274 def record(self, name: str, elapsed: float, input_size: int = 0) -> None:
275 """Manually record a timing.
277 Args:
278 name: Operation name
279 elapsed: Elapsed time
280 input_size: Input size
281 """
282 if not self._enabled:
283 return
285 if name not in self._profiles: 285 ↛ 288line 285 didn't jump to line 288 because the condition on line 285 was always true
286 self._profiles[name] = OperationProfile(name)
288 self._profiles[name].record(elapsed, input_size)
290 def get_profile(self, name: str) -> OperationProfile | None:
291 """Get profile for operation.
293 Args:
294 name: Operation name
296 Returns:
297 Operation profile or None
298 """
299 return self._profiles.get(name)
301 def report(self) -> ProfileReport:
302 """Generate profiling report.
304 Returns:
305 Profile report
306 """
307 return ProfileReport(
308 profiles=self._profiles.copy(),
309 start_time=self._start_time,
310 end_time=time.perf_counter(),
311 total_operations=sum(p.calls for p in self._profiles.values()),
312 )
315def profile(name: str | None = None, input_size_arg: str | None = None) -> Callable: # type: ignore[type-arg]
316 """Decorator for profiling functions.
318 Args:
319 name: Profile name (defaults to function name)
320 input_size_arg: Argument name for input size
322 Returns:
323 Decorated function
325 Example:
326 >>> @profile()
327 >>> def compute_fft(data, nfft=None):
328 ... return np.fft.fft(data, n=nfft)
330 References:
331 API-012: Performance Profiling API
332 """
334 def decorator(func: Callable) -> Callable: # type: ignore[type-arg]
335 profile_name = name or func.__name__
337 @functools.wraps(func)
338 def wrapper(*args: Any, **kwargs: Any) -> Any:
339 profiler = Profiler.get_instance()
341 # Determine input size
342 input_size = 0
343 if input_size_arg:
344 if input_size_arg in kwargs: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 data = kwargs[input_size_arg]
346 elif args:
347 data = args[0]
348 else:
349 data = None
351 if hasattr(data, "__len__"):
352 input_size = len(data)
354 with profiler.profile(profile_name, input_size):
355 return func(*args, **kwargs)
357 return wrapper
359 return decorator
362# Convenience functions
363def get_profiler() -> Profiler:
364 """Get global profiler instance.
366 Returns:
367 Global Profiler instance
369 References:
370 API-012: Performance Profiling API
371 """
372 return Profiler.get_instance()
375def enable_profiling() -> None:
376 """Enable global profiling."""
377 get_profiler().enable()
380def disable_profiling() -> None:
381 """Disable global profiling."""
382 get_profiler().disable()
385def reset_profiling() -> None:
386 """Reset global profiler."""
387 get_profiler().reset()
390def get_profile_report() -> ProfileReport:
391 """Get global profile report.
393 Returns:
394 Profile report
395 """
396 return get_profiler().report()