Print real-time sampling statistics.
(self)
| 188 | ) |
| 189 | |
| 190 | def _print_realtime_stats(self): |
| 191 | """Print real-time sampling statistics.""" |
| 192 | if len(self.sample_intervals) < 2: |
| 193 | return |
| 194 | |
| 195 | # Calculate statistics on the Hz values (deque automatically maintains rolling window) |
| 196 | hz_values = list(self.sample_intervals) |
| 197 | mean_hz = statistics.mean(hz_values) |
| 198 | min_hz = min(hz_values) |
| 199 | max_hz = max(hz_values) |
| 200 | |
| 201 | # Calculate microseconds per sample for all metrics (1/Hz * 1,000,000) |
| 202 | mean_us_per_sample = (1.0 / mean_hz) * 1_000_000 if mean_hz > 0 else 0 |
| 203 | min_us_per_sample = ( |
| 204 | (1.0 / max_hz) * 1_000_000 if max_hz > 0 else 0 |
| 205 | ) # Min time = Max Hz |
| 206 | max_us_per_sample = ( |
| 207 | (1.0 / min_hz) * 1_000_000 if min_hz > 0 else 0 |
| 208 | ) # Max time = Min Hz |
| 209 | |
| 210 | # Build cache stats string if stats collection is enabled |
| 211 | cache_stats_str = "" |
| 212 | if self.collect_stats: |
| 213 | try: |
| 214 | stats = self.unwinder.get_stats() |
| 215 | hits = stats.get('frame_cache_hits', 0) |
| 216 | partial = stats.get('frame_cache_partial_hits', 0) |
| 217 | misses = stats.get('frame_cache_misses', 0) |
| 218 | total = hits + partial + misses |
| 219 | if total > 0: |
| 220 | hit_pct = (hits + partial) / total * 100 |
| 221 | cache_stats_str = f" {ANSIColors.MAGENTA}Cache: {fmt(hit_pct)}% ({hits}+{partial}/{misses}){ANSIColors.RESET}" |
| 222 | except RuntimeError: |
| 223 | pass |
| 224 | |
| 225 | # Clear line and print stats |
| 226 | print( |
| 227 | f"\r\033[K{ANSIColors.BOLD_BLUE}Stats:{ANSIColors.RESET} " |
| 228 | f"{ANSIColors.YELLOW}{fmt(mean_hz)}Hz ({fmt(mean_us_per_sample)}µs){ANSIColors.RESET} " |
| 229 | f"{ANSIColors.GREEN}Min: {fmt(min_hz)}Hz{ANSIColors.RESET} " |
| 230 | f"{ANSIColors.RED}Max: {fmt(max_hz)}Hz{ANSIColors.RESET} " |
| 231 | f"{ANSIColors.CYAN}N={self.total_samples}{ANSIColors.RESET}" |
| 232 | f"{cache_stats_str}", |
| 233 | end="", |
| 234 | flush=True, |
| 235 | ) |
| 236 | |
| 237 | def _print_unwinder_stats(self): |
| 238 | """Print unwinder statistics including cache performance.""" |