Coverage for src / local_deep_research / benchmarks / efficiency / speed_profiler.py: 14%
85 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Speed profiling tools for Local Deep Research.
4This module provides functionality for measuring execution time
5of different components and processes in the research system.
6"""
8from loguru import logger
9import time
10from contextlib import contextmanager
11from typing import Any, Callable, Dict
14class SpeedProfiler:
15 """
16 Profiler for tracking execution speed of components.
18 This class provides methods for timing operations and
19 collecting performance statistics for later analysis.
20 """
22 def __init__(self):
23 """Initialize the profiler with empty timing data."""
24 self.timings = {}
25 self.current_timers = {}
26 self.total_start_time = None
27 self.total_end_time = None
29 def start(self):
30 """Start the global profiling session."""
31 self.timings = {}
32 self.current_timers = {}
33 self.total_start_time = time.time()
35 def stop(self):
36 """Stop the global profiling session."""
37 self.total_end_time = time.time()
39 # Stop any timers that are still running
40 for name in list(self.current_timers.keys()):
41 self.stop_timer(name)
43 def start_timer(self, name: str):
44 """
45 Start a named timer.
47 Args:
48 name: Name of the timer to start
49 """
50 if name in self.current_timers:
51 logger.warning(f"Timer '{name}' is already running. Restarting.")
53 self.current_timers[name] = time.time()
55 def stop_timer(self, name: str):
56 """
57 Stop a named timer and record the elapsed time.
59 Args:
60 name: Name of the timer to stop
61 """
62 if name not in self.current_timers:
63 logger.warning(f"Timer '{name}' was not started.")
64 return
66 elapsed = time.time() - self.current_timers[name]
68 if name not in self.timings:
69 self.timings[name] = {
70 "total": elapsed,
71 "count": 1,
72 "min": elapsed,
73 "max": elapsed,
74 "starts": [self.current_timers[name]],
75 "durations": [elapsed],
76 }
77 else:
78 self.timings[name]["total"] += elapsed
79 self.timings[name]["count"] += 1
80 self.timings[name]["min"] = min(self.timings[name]["min"], elapsed)
81 self.timings[name]["max"] = max(self.timings[name]["max"], elapsed)
82 self.timings[name]["starts"].append(self.current_timers[name])
83 self.timings[name]["durations"].append(elapsed)
85 del self.current_timers[name]
87 @contextmanager
88 def timer(self, name: str):
89 """
90 Context manager for timing a block of code.
92 Args:
93 name: Name of the timer
95 Example:
96 with profiler.timer("my_operation"):
97 # Code to time
98 do_something()
99 """
100 self.start_timer(name)
101 try:
102 yield
103 finally:
104 self.stop_timer(name)
106 def get_timings(self) -> Dict[str, Any]:
107 """
108 Get all recorded timings.
110 Returns:
111 Dictionary of timing data for all measured operations
112 """
113 result = self.timings.copy()
115 # Add averages
116 for name, data in result.items():
117 if data["count"] > 0:
118 data["avg"] = data["total"] / data["count"]
120 # Add total duration
121 if (
122 self.total_start_time is not None
123 and self.total_end_time is not None
124 ):
125 result["total"] = {
126 "total": self.total_end_time - self.total_start_time,
127 "count": 1,
128 "min": self.total_end_time - self.total_start_time,
129 "max": self.total_end_time - self.total_start_time,
130 "avg": self.total_end_time - self.total_start_time,
131 "starts": [self.total_start_time],
132 "durations": [self.total_end_time - self.total_start_time],
133 }
135 return result
137 def get_summary(self) -> Dict[str, float]:
138 """
139 Get a summary of timing information.
141 Returns:
142 Dictionary with summary statistics
143 """
144 timings = self.get_timings()
145 summary = {}
147 # Total duration
148 if "total" in timings:
149 summary["total_duration"] = timings["total"]["total"]
150 elif (
151 self.total_start_time is not None
152 and self.total_end_time is not None
153 ):
154 summary["total_duration"] = (
155 self.total_end_time - self.total_start_time
156 )
157 else:
158 summary["total_duration"] = sum(
159 t["total"] for t in timings.values()
160 )
162 # Component durations
163 for name, data in timings.items():
164 if name != "total":
165 summary[f"{name}_duration"] = data["total"]
166 summary[f"{name}_percent"] = (
167 data["total"] / summary["total_duration"] * 100
168 if summary["total_duration"] > 0
169 else 0
170 )
172 # Per-operation breakdowns
173 for name, data in timings.items():
174 if data["count"] > 0:
175 summary[f"{name}_per_operation"] = data["total"] / data["count"]
177 return summary
179 def print_summary(self):
180 """Print a formatted summary of timing information."""
181 summary = self.get_summary()
182 total = summary.get("total_duration", 0)
184 print("\n===== SPEED PROFILE SUMMARY =====")
185 print(f"Total execution time: {total:.2f} seconds")
186 print("\n--- Component Breakdown ---")
188 # Print each component's timing
189 for name, data in self.timings.items():
190 if name != "total":
191 percent = data["total"] / total * 100 if total > 0 else 0
192 print(
193 f"{name}: {data['total']:.2f}s ({percent:.1f}%) - "
194 f"{data['count']} calls, avg {data['total'] / data['count']:.3f}s per call"
195 )
197 print("\n==============================")
200def time_function(func: Callable) -> Callable:
201 """
202 Decorator to time a function's execution.
204 Args:
205 func: Function to time
207 Returns:
208 Wrapped function that logs its execution time
210 Example:
211 @time_function
212 def my_slow_function():
213 # Some slow code
214 pass
215 """
217 def wrapper(*args, **kwargs):
218 start_time = time.time()
219 result = func(*args, **kwargs)
220 elapsed = time.time() - start_time
222 logger.info(f"{func.__name__} took {elapsed:.3f} seconds")
224 return result
226 return wrapper