mirror of
https://github.com/python/cpython.git
synced 2026-01-17 12:50:02 +00:00
1090 lines
38 KiB
Python
1090 lines
38 KiB
Python
"""Widget classes for the live profiling collector UI."""
|
|
|
|
import curses
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
|
|
from .constants import (
|
|
TOP_FUNCTIONS_DISPLAY_COUNT,
|
|
MIN_FUNC_NAME_WIDTH,
|
|
MAX_FUNC_NAME_WIDTH,
|
|
WIDTH_THRESHOLD_SAMPLE_PCT,
|
|
WIDTH_THRESHOLD_TOTTIME,
|
|
WIDTH_THRESHOLD_CUMUL_PCT,
|
|
WIDTH_THRESHOLD_CUMTIME,
|
|
MICROSECONDS_PER_SECOND,
|
|
DISPLAY_UPDATE_INTERVAL_SEC,
|
|
MIN_BAR_WIDTH,
|
|
MAX_SAMPLE_RATE_BAR_WIDTH,
|
|
MAX_EFFICIENCY_BAR_WIDTH,
|
|
MIN_SAMPLE_RATE_FOR_SCALING,
|
|
FOOTER_LINES,
|
|
FINISHED_BANNER_EXTRA_LINES,
|
|
OPCODE_PANEL_HEIGHT,
|
|
)
|
|
from ..constants import (
|
|
THREAD_STATUS_HAS_GIL,
|
|
THREAD_STATUS_ON_CPU,
|
|
THREAD_STATUS_UNKNOWN,
|
|
THREAD_STATUS_GIL_REQUESTED,
|
|
PROFILING_MODE_CPU,
|
|
PROFILING_MODE_GIL,
|
|
PROFILING_MODE_WALL,
|
|
)
|
|
from ..opcode_utils import get_opcode_info, format_opcode
|
|
|
|
|
|
class Widget(ABC):
|
|
"""Base class for UI widgets."""
|
|
|
|
def __init__(self, display, colors):
|
|
"""
|
|
Initialize widget.
|
|
|
|
Args:
|
|
display: DisplayInterface implementation
|
|
colors: Dictionary of color attributes
|
|
"""
|
|
self.display = display
|
|
self.colors = colors
|
|
|
|
@abstractmethod
|
|
def render(self, line, width, **kwargs):
|
|
"""
|
|
Render the widget starting at the given line.
|
|
|
|
Args:
|
|
line: Starting line number
|
|
width: Available width
|
|
**kwargs: Additional rendering parameters
|
|
|
|
Returns:
|
|
Next available line number after rendering
|
|
"""
|
|
pass
|
|
|
|
def add_str(self, line, col, text, attr=0):
|
|
"""Add a string to the display at the specified position."""
|
|
self.display.add_str(line, col, text, attr)
|
|
|
|
|
|
class ProgressBarWidget(Widget):
|
|
"""Reusable progress bar widget."""
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""Render is not used for progress bars - use render_bar instead."""
|
|
raise NotImplementedError("Use render_bar method instead")
|
|
|
|
def render_bar(
|
|
self, filled, total, max_width, fill_char="█", empty_char="░"
|
|
):
|
|
"""
|
|
Render a progress bar and return the bar string and its length.
|
|
|
|
Args:
|
|
filled: Current filled amount
|
|
total: Total amount (max value)
|
|
max_width: Maximum width for the bar
|
|
fill_char: Character to use for filled portion
|
|
empty_char: Character to use for empty portion
|
|
|
|
Returns:
|
|
Tuple of (bar_string, bar_length)
|
|
"""
|
|
bar_width = min(max_width, max_width)
|
|
normalized = min(filled / max(total, 1), 1.0)
|
|
bar_fill = int(normalized * bar_width)
|
|
|
|
bar = "["
|
|
for i in range(bar_width):
|
|
if i < bar_fill:
|
|
bar += fill_char
|
|
else:
|
|
bar += empty_char
|
|
bar += "]"
|
|
return bar, len(bar)
|
|
|
|
|
|
class HeaderWidget(Widget):
|
|
"""Widget for rendering the header section (lines 0-8)."""
|
|
|
|
def __init__(self, display, colors, collector):
|
|
"""
|
|
Initialize header widget.
|
|
|
|
Args:
|
|
display: DisplayInterface implementation
|
|
colors: Dictionary of color attributes
|
|
collector: Reference to LiveStatsCollector for accessing stats
|
|
"""
|
|
super().__init__(display, colors)
|
|
self.collector = collector
|
|
self.progress_bar = ProgressBarWidget(display, colors)
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""
|
|
Render the complete header section.
|
|
|
|
Args:
|
|
line: Starting line number
|
|
width: Available width
|
|
kwargs: Must contain 'elapsed' key
|
|
|
|
Returns:
|
|
Next available line number
|
|
"""
|
|
elapsed = kwargs["elapsed"]
|
|
|
|
line = self.draw_header_info(line, width, elapsed)
|
|
line = self.draw_sample_stats(line, width, elapsed)
|
|
line = self.draw_efficiency_bar(line, width)
|
|
line = self.draw_thread_status(line, width)
|
|
line = self.draw_function_stats(
|
|
line, width, kwargs.get("stats_list", [])
|
|
)
|
|
line = self.draw_top_functions(
|
|
line, width, kwargs.get("stats_list", [])
|
|
)
|
|
|
|
# Show prominent finished banner if profiling is complete
|
|
if self.collector.finished:
|
|
line = self.draw_finished_banner(line, width)
|
|
|
|
# Separator
|
|
A_DIM = self.display.get_attr("A_DIM")
|
|
separator = "─" * (width - 1)
|
|
self.add_str(line, 0, separator[: width - 1], A_DIM)
|
|
line += 1
|
|
|
|
return line
|
|
|
|
def format_uptime(self, elapsed):
|
|
"""Format elapsed time as uptime string."""
|
|
uptime_sec = int(elapsed)
|
|
hours = uptime_sec // 3600
|
|
minutes = (uptime_sec % 3600) // 60
|
|
seconds = uptime_sec % 60
|
|
if hours > 0:
|
|
return f"{hours}h{minutes:02d}m{seconds:02d}s"
|
|
else:
|
|
return f"{minutes}m{seconds:02d}s"
|
|
|
|
def draw_header_info(self, line, width, elapsed):
|
|
"""Draw the header information line with PID, uptime, time, and interval."""
|
|
# Draw title
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
title = "Tachyon Profiler"
|
|
self.add_str(line, 0, title, A_BOLD | self.colors["cyan"])
|
|
line += 1
|
|
|
|
current_time = self.collector.current_time_display
|
|
uptime = self.format_uptime(elapsed)
|
|
|
|
# Calculate display refresh rate
|
|
refresh_hz = (
|
|
1.0 / self.collector.display_update_interval_sec if self.collector.display_update_interval_sec > 0 else 0
|
|
)
|
|
|
|
# Get current view mode and thread display
|
|
if self.collector.view_mode == "ALL":
|
|
thread_name = "ALL"
|
|
thread_color = self.colors["green"]
|
|
else:
|
|
# PER_THREAD mode
|
|
if self.collector.current_thread_index < len(
|
|
self.collector.thread_ids
|
|
):
|
|
thread_id = self.collector.thread_ids[
|
|
self.collector.current_thread_index
|
|
]
|
|
num_threads = len(self.collector.thread_ids)
|
|
thread_name = f"{thread_id} ({self.collector.current_thread_index + 1}/{num_threads})"
|
|
thread_color = self.colors["magenta"]
|
|
else:
|
|
thread_name = "ALL"
|
|
thread_color = self.colors["green"]
|
|
|
|
header_parts = [
|
|
("PID: ", curses.A_BOLD),
|
|
(f"{self.collector.pid}", self.colors["cyan"]),
|
|
(" │ ", curses.A_DIM),
|
|
("Thread: ", curses.A_BOLD),
|
|
(thread_name, thread_color),
|
|
(" │ ", curses.A_DIM),
|
|
("Uptime: ", curses.A_BOLD),
|
|
(uptime, self.colors["green"]),
|
|
(" │ ", curses.A_DIM),
|
|
("Time: ", curses.A_BOLD),
|
|
(current_time, self.colors["yellow"]),
|
|
(" │ ", curses.A_DIM),
|
|
("Interval: ", curses.A_BOLD),
|
|
(
|
|
f"{self.collector.sample_interval_usec}µs",
|
|
self.colors["magenta"],
|
|
),
|
|
(" │ ", curses.A_DIM),
|
|
("Display: ", curses.A_BOLD),
|
|
(f"{refresh_hz:.1f}Hz", self.colors["cyan"]),
|
|
]
|
|
|
|
col = 0
|
|
for text, attr in header_parts:
|
|
if col < width - 1:
|
|
self.add_str(line, col, text, attr)
|
|
col += len(text)
|
|
return line + 1
|
|
|
|
def format_rate_with_units(self, rate_hz):
|
|
"""Format a rate in Hz with appropriate units (Hz, KHz, MHz)."""
|
|
if rate_hz >= MICROSECONDS_PER_SECOND:
|
|
return f"{rate_hz / MICROSECONDS_PER_SECOND:.1f}MHz"
|
|
elif rate_hz >= 1_000:
|
|
return f"{rate_hz / 1_000:.1f}KHz"
|
|
else:
|
|
return f"{rate_hz:.1f}Hz"
|
|
|
|
def draw_sample_stats(self, line, width, elapsed):
|
|
"""Draw sample statistics with visual progress bar."""
|
|
sample_rate = (
|
|
self.collector.total_samples / elapsed if elapsed > 0 else 0
|
|
)
|
|
|
|
# Update max sample rate
|
|
if sample_rate > self.collector.max_sample_rate:
|
|
self.collector.max_sample_rate = sample_rate
|
|
|
|
col = 0
|
|
self.add_str(line, col, "Samples: ", curses.A_BOLD)
|
|
col += 9
|
|
self.add_str(
|
|
line,
|
|
col,
|
|
f"{self.collector.total_samples:>8}",
|
|
self.colors["cyan"],
|
|
)
|
|
col += 8
|
|
self.add_str(
|
|
line, col, f" total ({sample_rate:>7.1f}/s) ", curses.A_NORMAL
|
|
)
|
|
col += 23
|
|
|
|
# Draw sample rate bar
|
|
target_rate = (
|
|
MICROSECONDS_PER_SECOND / self.collector.sample_interval_usec
|
|
)
|
|
|
|
# Show current/target ratio with percentage
|
|
if sample_rate > 0 and target_rate > 0:
|
|
percentage = min((sample_rate / target_rate) * 100, 100)
|
|
current_formatted = self.format_rate_with_units(sample_rate)
|
|
target_formatted = self.format_rate_with_units(target_rate)
|
|
|
|
if percentage >= 99.5: # Show 100% when very close
|
|
rate_label = f" {current_formatted}/{target_formatted} (100%)"
|
|
else:
|
|
rate_label = f" {current_formatted}/{target_formatted} ({percentage:>4.1f}%)"
|
|
else:
|
|
target_formatted = self.format_rate_with_units(target_rate)
|
|
rate_label = f" target: {target_formatted}"
|
|
|
|
available_width = width - col - len(rate_label) - 3
|
|
|
|
if available_width >= MIN_BAR_WIDTH:
|
|
bar_width = min(MAX_SAMPLE_RATE_BAR_WIDTH, available_width)
|
|
# Use target rate as the reference, with a minimum for scaling
|
|
reference_rate = max(target_rate, MIN_SAMPLE_RATE_FOR_SCALING)
|
|
normalized_rate = min(sample_rate / reference_rate, 1.0)
|
|
bar_fill = int(normalized_rate * bar_width)
|
|
|
|
bar = "["
|
|
for i in range(bar_width):
|
|
bar += "█" if i < bar_fill else "░"
|
|
bar += "]"
|
|
self.add_str(line, col, bar, self.colors["green"])
|
|
col += len(bar)
|
|
|
|
if col + len(rate_label) < width - 1:
|
|
self.add_str(line, col + 1, rate_label, curses.A_DIM)
|
|
return line + 1
|
|
|
|
def draw_efficiency_bar(self, line, width):
|
|
"""Draw sample efficiency bar showing success/failure rates."""
|
|
# total_samples = successful_samples + failed_samples, so percentages add to 100%
|
|
total = max(1, self.collector.total_samples)
|
|
success_pct = (self.collector.successful_samples / total) * 100
|
|
failed_pct = (self.collector.failed_samples / total) * 100
|
|
|
|
col = 0
|
|
self.add_str(line, col, "Efficiency:", curses.A_BOLD)
|
|
col += 11
|
|
|
|
label = f" {success_pct:>5.2f}% good, {failed_pct:>5.2f}% failed"
|
|
available_width = width - col - len(label) - 3
|
|
|
|
if available_width >= MIN_BAR_WIDTH:
|
|
bar_width = min(MAX_EFFICIENCY_BAR_WIDTH, available_width)
|
|
success_fill = int((self.collector.successful_samples / total) * bar_width)
|
|
failed_fill = bar_width - success_fill
|
|
|
|
self.add_str(line, col, "[", curses.A_NORMAL)
|
|
col += 1
|
|
if success_fill > 0:
|
|
self.add_str(
|
|
line, col, "█" * success_fill, self.colors["green"]
|
|
)
|
|
col += success_fill
|
|
if failed_fill > 0:
|
|
self.add_str(line, col, "█" * failed_fill, self.colors["red"])
|
|
col += failed_fill
|
|
self.add_str(line, col, "]", curses.A_NORMAL)
|
|
col += 1
|
|
|
|
self.add_str(line, col + 1, label, curses.A_NORMAL)
|
|
return line + 1
|
|
|
|
def _add_percentage_stat(
|
|
self, line, col, value, label, color, add_separator=False
|
|
):
|
|
"""Add a percentage stat to the display.
|
|
|
|
Args:
|
|
line: Line number
|
|
col: Starting column
|
|
value: Percentage value
|
|
label: Label text
|
|
color: Color attribute
|
|
add_separator: Whether to add separator before the stat
|
|
|
|
Returns:
|
|
Updated column position
|
|
"""
|
|
if add_separator:
|
|
self.add_str(line, col, " │ ", curses.A_DIM)
|
|
col += 3
|
|
|
|
self.add_str(line, col, f"{value:>4.1f}", color)
|
|
col += 4
|
|
self.add_str(line, col, f"% {label}", curses.A_NORMAL)
|
|
col += len(label) + 2
|
|
|
|
return col
|
|
|
|
def draw_thread_status(self, line, width):
|
|
"""Draw thread status statistics and GC information."""
|
|
# Get status counts for current view mode
|
|
thread_data = self.collector._get_current_thread_data()
|
|
status_counts = thread_data.as_status_dict() if thread_data else self.collector.thread_status_counts
|
|
|
|
# Calculate percentages
|
|
total_threads = max(1, status_counts["total"])
|
|
pct_on_gil = (status_counts["has_gil"] / total_threads) * 100
|
|
pct_off_gil = 100.0 - pct_on_gil
|
|
pct_gil_requested = (status_counts["gil_requested"] / total_threads) * 100
|
|
pct_exception = (status_counts.get("has_exception", 0) / total_threads) * 100
|
|
|
|
# Get GC percentage based on view mode
|
|
if thread_data:
|
|
total_samples = max(1, thread_data.sample_count)
|
|
pct_gc = (thread_data.gc_frame_samples / total_samples) * 100
|
|
else:
|
|
# Use total_samples for GC percentage since gc_frame_samples is tracked
|
|
# across ALL samples (via thread status), not just successful ones
|
|
total_samples = max(1, self.collector.total_samples)
|
|
pct_gc = (self.collector.gc_frame_samples / total_samples) * 100
|
|
|
|
col = 0
|
|
self.add_str(line, col, "Threads: ", curses.A_BOLD)
|
|
col += 11
|
|
|
|
# Show GIL stats only if mode is not GIL (GIL mode filters to only GIL holders)
|
|
if self.collector.mode != PROFILING_MODE_GIL:
|
|
col = self._add_percentage_stat(
|
|
line, col, pct_on_gil, "on gil", self.colors["green"]
|
|
)
|
|
col = self._add_percentage_stat(
|
|
line,
|
|
col,
|
|
pct_off_gil,
|
|
"off gil",
|
|
self.colors["red"],
|
|
add_separator=True,
|
|
)
|
|
|
|
# Show "waiting for gil" only if mode is not GIL
|
|
if self.collector.mode != PROFILING_MODE_GIL and col < width - 30:
|
|
col = self._add_percentage_stat(
|
|
line,
|
|
col,
|
|
pct_gil_requested,
|
|
"waiting for gil",
|
|
self.colors["yellow"],
|
|
add_separator=True,
|
|
)
|
|
|
|
# Show exception stats
|
|
if col < width - 15:
|
|
col = self._add_percentage_stat(
|
|
line,
|
|
col,
|
|
pct_exception,
|
|
"exc",
|
|
self.colors["red"],
|
|
add_separator=(col > 11),
|
|
)
|
|
|
|
# Always show GC stats
|
|
if col < width - 15:
|
|
col = self._add_percentage_stat(
|
|
line,
|
|
col,
|
|
pct_gc,
|
|
"GC",
|
|
self.colors["magenta"],
|
|
add_separator=(col > 11),
|
|
)
|
|
|
|
return line + 1
|
|
|
|
def draw_function_stats(self, line, width, stats_list):
|
|
"""Draw function statistics summary."""
|
|
result_set = self.collector._get_current_result_source()
|
|
total_funcs = len(result_set)
|
|
funcs_shown = len(stats_list)
|
|
executing_funcs = sum(
|
|
1 for f in result_set.values() if f.get("direct_calls", 0) > 0
|
|
)
|
|
stack_only = total_funcs - executing_funcs
|
|
|
|
col = 0
|
|
self.add_str(line, col, "Functions: ", curses.A_BOLD)
|
|
col += 11
|
|
self.add_str(line, col, f"{total_funcs:>5}", self.colors["cyan"])
|
|
col += 5
|
|
self.add_str(line, col, " total", curses.A_NORMAL)
|
|
col += 6
|
|
|
|
if col < width - 25:
|
|
self.add_str(line, col, " │ ", curses.A_DIM)
|
|
col += 3
|
|
self.add_str(
|
|
line, col, f"{executing_funcs:>5}", self.colors["green"]
|
|
)
|
|
col += 5
|
|
self.add_str(line, col, " exec", curses.A_NORMAL)
|
|
col += 5
|
|
|
|
if col < width - 25:
|
|
self.add_str(line, col, " │ ", curses.A_DIM)
|
|
col += 3
|
|
self.add_str(line, col, f"{stack_only:>5}", self.colors["yellow"])
|
|
col += 5
|
|
self.add_str(line, col, " stack", curses.A_NORMAL)
|
|
col += 6
|
|
|
|
if col < width - 20:
|
|
self.add_str(line, col, " │ ", curses.A_DIM)
|
|
col += 3
|
|
self.add_str(
|
|
line, col, f"{funcs_shown:>5}", self.colors["magenta"]
|
|
)
|
|
col += 5
|
|
self.add_str(line, col, " shown", curses.A_NORMAL)
|
|
return line + 1
|
|
|
|
def draw_top_functions(self, line, width, stats_list):
|
|
"""Draw top N hottest functions."""
|
|
col = 0
|
|
self.add_str(
|
|
line,
|
|
col,
|
|
f"Top {TOP_FUNCTIONS_DISPLAY_COUNT}: ",
|
|
curses.A_BOLD,
|
|
)
|
|
col += 11
|
|
|
|
top_by_samples = sorted(
|
|
stats_list, key=lambda x: x["direct_calls"], reverse=True
|
|
)
|
|
emojis = ["🥇", "🥈", "🥉"]
|
|
medal_colors = [
|
|
self.colors["red"],
|
|
self.colors["yellow"],
|
|
self.colors["green"],
|
|
]
|
|
|
|
displayed = 0
|
|
for func_data in top_by_samples:
|
|
if displayed >= TOP_FUNCTIONS_DISPLAY_COUNT:
|
|
break
|
|
if col >= width - 20:
|
|
break
|
|
if func_data["direct_calls"] == 0:
|
|
continue
|
|
|
|
func_name = func_data["func"][2]
|
|
func_pct = func_data["sample_pct"]
|
|
|
|
# Medal emoji
|
|
if col + 3 < width - 15:
|
|
self.add_str(
|
|
line, col, emojis[displayed] + " ", medal_colors[displayed]
|
|
)
|
|
col += 3
|
|
|
|
# Function name (truncate to fit)
|
|
available_for_name = width - col - 15
|
|
max_name_len = min(25, max(5, available_for_name))
|
|
if len(func_name) > max_name_len:
|
|
func_name = func_name[: max_name_len - 3] + "..."
|
|
|
|
if col + len(func_name) < width - 10:
|
|
self.add_str(line, col, func_name, medal_colors[displayed])
|
|
col += len(func_name)
|
|
|
|
pct_str = (
|
|
f" ({func_pct:.1f}%)"
|
|
if func_pct >= 0.1
|
|
else f" ({func_data['direct_calls']})"
|
|
)
|
|
self.add_str(line, col, pct_str, curses.A_DIM)
|
|
col += len(pct_str)
|
|
|
|
displayed += 1
|
|
|
|
if displayed < 3 and col < width - 30:
|
|
self.add_str(line, col, " │ ", curses.A_DIM)
|
|
col += 3
|
|
|
|
if displayed == 0 and col < width - 25:
|
|
self.add_str(line, col, "(collecting samples...)", curses.A_DIM)
|
|
|
|
return line + 1
|
|
|
|
def draw_finished_banner(self, line, width):
|
|
"""Draw a prominent banner when profiling is finished."""
|
|
A_REVERSE = self.display.get_attr("A_REVERSE")
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
|
|
# Add blank line for separation
|
|
line += 1
|
|
|
|
# Create the banner message
|
|
message = " ✓ PROFILING COMPLETE - Final Results Below - Press 'q' to Quit "
|
|
|
|
# Center the message and fill the width with reverse video
|
|
if len(message) < width - 1:
|
|
padding_total = width - len(message) - 1
|
|
padding_left = padding_total // 2
|
|
padding_right = padding_total - padding_left
|
|
full_message = " " * padding_left + message + " " * padding_right
|
|
else:
|
|
full_message = message[: width - 1]
|
|
|
|
# Draw the banner with reverse video and bold
|
|
self.add_str(
|
|
line, 0, full_message, A_REVERSE | A_BOLD | self.colors["green"]
|
|
)
|
|
line += 1
|
|
|
|
# Add blank line for separation
|
|
line += 1
|
|
|
|
return line
|
|
|
|
|
|
class TableWidget(Widget):
|
|
"""Widget for rendering column headers and data rows."""
|
|
|
|
def __init__(self, display, colors, collector):
|
|
"""
|
|
Initialize table widget.
|
|
|
|
Args:
|
|
display: DisplayInterface implementation
|
|
colors: Dictionary of color attributes
|
|
collector: Reference to LiveStatsCollector for accessing stats
|
|
"""
|
|
super().__init__(display, colors)
|
|
self.collector = collector
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""
|
|
Render column headers and data rows.
|
|
|
|
Args:
|
|
line: Starting line number
|
|
width: Available width
|
|
kwargs: Must contain 'height' and 'stats_list' keys
|
|
|
|
Returns:
|
|
Next available line number
|
|
"""
|
|
height = kwargs["height"]
|
|
stats_list = kwargs["stats_list"]
|
|
|
|
# Draw column headers
|
|
line, show_sample_pct, show_tottime, show_cumul_pct, show_cumtime = (
|
|
self.draw_column_headers(line, width)
|
|
)
|
|
column_flags = (
|
|
show_sample_pct,
|
|
show_tottime,
|
|
show_cumul_pct,
|
|
show_cumtime,
|
|
)
|
|
|
|
# Draw data rows
|
|
line = self.draw_stats_rows(
|
|
line, height, width, stats_list, column_flags
|
|
)
|
|
|
|
return line
|
|
|
|
def draw_column_headers(self, line, width):
|
|
"""Draw column headers with sort indicators."""
|
|
# Determine which columns to show based on width
|
|
show_sample_pct = width >= WIDTH_THRESHOLD_SAMPLE_PCT
|
|
show_tottime = width >= WIDTH_THRESHOLD_TOTTIME
|
|
show_cumul_pct = width >= WIDTH_THRESHOLD_CUMUL_PCT
|
|
show_cumtime = width >= WIDTH_THRESHOLD_CUMTIME
|
|
|
|
sorted_header = self.colors["sorted_header"]
|
|
normal_header = self.colors["normal_header"]
|
|
|
|
# Determine which column is sorted
|
|
sort_col = {
|
|
"nsamples": 0,
|
|
"sample_pct": 1,
|
|
"tottime": 2,
|
|
"cumul_pct": 3,
|
|
"cumtime": 4,
|
|
}.get(self.collector.sort_by, -1)
|
|
|
|
# Build the full header line first, then draw it
|
|
# This avoids gaps between columns when using reverse video
|
|
header_parts = []
|
|
col = 0
|
|
|
|
# Column 0: nsamples
|
|
text = f"{'▼nsamples' if sort_col == 0 else 'nsamples':>13} "
|
|
header_parts.append((col, text, sorted_header if sort_col == 0 else normal_header))
|
|
col += 15
|
|
|
|
# Column 1: sample %
|
|
if show_sample_pct:
|
|
text = f"{'▼%' if sort_col == 1 else '%':>5} "
|
|
header_parts.append((col, text, sorted_header if sort_col == 1 else normal_header))
|
|
col += 7
|
|
|
|
# Column 2: tottime
|
|
if show_tottime:
|
|
text = f"{'▼tottime' if sort_col == 2 else 'tottime':>10} "
|
|
header_parts.append((col, text, sorted_header if sort_col == 2 else normal_header))
|
|
col += 12
|
|
|
|
# Column 3: cumul %
|
|
if show_cumul_pct:
|
|
text = f"{'▼%' if sort_col == 3 else '%':>5} "
|
|
header_parts.append((col, text, sorted_header if sort_col == 3 else normal_header))
|
|
col += 7
|
|
|
|
# Column 4: cumtime
|
|
if show_cumtime:
|
|
text = f"{'▼cumtime' if sort_col == 4 else 'cumtime':>10} "
|
|
header_parts.append((col, text, sorted_header if sort_col == 4 else normal_header))
|
|
col += 12
|
|
|
|
# Remaining headers
|
|
if col < width - 15:
|
|
remaining_space = width - col - 1
|
|
func_width = min(
|
|
MAX_FUNC_NAME_WIDTH,
|
|
max(MIN_FUNC_NAME_WIDTH, remaining_space // 2),
|
|
)
|
|
text = f"{'function':<{func_width}} "
|
|
header_parts.append((col, text, normal_header))
|
|
col += func_width + 2
|
|
|
|
if col < width - 10:
|
|
file_text = "file:line"
|
|
padding = width - col - len(file_text)
|
|
text = file_text + " " * max(0, padding)
|
|
header_parts.append((col, text, normal_header))
|
|
|
|
# Draw full-width background first
|
|
self.add_str(line, 0, " " * (width - 1), normal_header)
|
|
|
|
# Draw each header part on top
|
|
for col_pos, text, attr in header_parts:
|
|
self.add_str(line, col_pos, text.rstrip(), attr)
|
|
|
|
return (
|
|
line + 1,
|
|
show_sample_pct,
|
|
show_tottime,
|
|
show_cumul_pct,
|
|
show_cumtime,
|
|
)
|
|
|
|
def draw_stats_rows(self, line, height, width, stats_list, column_flags):
|
|
"""Draw the statistics data rows."""
|
|
show_sample_pct, show_tottime, show_cumul_pct, show_cumtime = (
|
|
column_flags
|
|
)
|
|
|
|
# Get color attributes
|
|
color_file = self.colors.get("color_file", curses.A_NORMAL)
|
|
color_func = self.colors.get("color_func", curses.A_NORMAL)
|
|
|
|
# Get trend tracker for color decisions
|
|
trend_tracker = self.collector._trend_tracker
|
|
|
|
# Check if opcode mode is enabled for row selection highlighting
|
|
show_opcodes = getattr(self.collector, 'show_opcodes', False)
|
|
selected_row = getattr(self.collector, 'selected_row', 0)
|
|
scroll_offset = getattr(self.collector, 'scroll_offset', 0) if show_opcodes else 0
|
|
A_REVERSE = self.display.get_attr("A_REVERSE")
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
|
|
# Reserve space for opcode panel when enabled
|
|
opcode_panel_height = OPCODE_PANEL_HEIGHT if show_opcodes else 0
|
|
|
|
# Apply scroll offset when in opcode mode
|
|
display_stats = stats_list[scroll_offset:] if show_opcodes else stats_list
|
|
|
|
for row_idx, stat in enumerate(display_stats):
|
|
if line >= height - FOOTER_LINES - opcode_panel_height:
|
|
break
|
|
|
|
func = stat["func"]
|
|
direct_calls = stat["direct_calls"]
|
|
cumulative_calls = stat["cumulative_calls"]
|
|
total_time = stat["total_time"]
|
|
cumulative_time = stat["cumulative_time"]
|
|
sample_pct = stat["sample_pct"]
|
|
cum_pct = stat["cumul_pct"]
|
|
trends = stat.get("trends", {})
|
|
|
|
# Check if this row is selected
|
|
is_selected = show_opcodes and row_idx == selected_row
|
|
|
|
# Helper function to get trend color
|
|
def get_trend_color(column_name):
|
|
if is_selected:
|
|
return A_REVERSE | A_BOLD
|
|
trend = trends.get(column_name, "stable")
|
|
if trend_tracker is not None and trend_tracker.enabled:
|
|
return trend_tracker.get_color(trend)
|
|
return curses.A_NORMAL
|
|
|
|
filename, lineno, funcname = func[0], func[1], func[2]
|
|
samples_str = f"{direct_calls}/{cumulative_calls}"
|
|
col = 0
|
|
|
|
# Fill entire row with reverse video background for selected row
|
|
if is_selected:
|
|
self.add_str(line, 0, " " * (width - 1), A_REVERSE | A_BOLD)
|
|
|
|
# Show selection indicator when opcode panel is enabled
|
|
if show_opcodes:
|
|
if is_selected:
|
|
self.add_str(line, col, "►", A_REVERSE | A_BOLD)
|
|
else:
|
|
self.add_str(line, col, " ", curses.A_NORMAL)
|
|
col += 2
|
|
|
|
# Samples column - apply trend color based on nsamples trend
|
|
nsamples_color = get_trend_color("nsamples")
|
|
self.add_str(line, col, f"{samples_str:>13} ", nsamples_color)
|
|
col += 15
|
|
|
|
# Sample % column
|
|
if show_sample_pct:
|
|
sample_pct_color = get_trend_color("sample_pct")
|
|
self.add_str(line, col, f"{sample_pct:>5.1f} ", sample_pct_color)
|
|
col += 7
|
|
|
|
# Total time column
|
|
if show_tottime:
|
|
tottime_color = get_trend_color("tottime")
|
|
self.add_str(line, col, f"{total_time:>10.3f} ", tottime_color)
|
|
col += 12
|
|
|
|
# Cumul % column
|
|
if show_cumul_pct:
|
|
cumul_pct_color = get_trend_color("cumul_pct")
|
|
self.add_str(line, col, f"{cum_pct:>5.1f} ", cumul_pct_color)
|
|
col += 7
|
|
|
|
# Cumul time column
|
|
if show_cumtime:
|
|
cumtime_color = get_trend_color("cumtime")
|
|
self.add_str(line, col, f"{cumulative_time:>10.3f} ", cumtime_color)
|
|
col += 12
|
|
|
|
# Function name column
|
|
if col < width - 15:
|
|
remaining_space = width - col - 1
|
|
func_width = min(
|
|
MAX_FUNC_NAME_WIDTH,
|
|
max(MIN_FUNC_NAME_WIDTH, remaining_space // 2),
|
|
)
|
|
|
|
func_display = funcname
|
|
if len(funcname) > func_width:
|
|
func_display = funcname[: func_width - 3] + "..."
|
|
func_display = f"{func_display:<{func_width}}"
|
|
func_color = A_REVERSE | A_BOLD if is_selected else color_func
|
|
self.add_str(line, col, func_display, func_color)
|
|
col += func_width + 2
|
|
|
|
# File:line column
|
|
if col < width - 10:
|
|
simplified_path = self.collector.simplify_path(filename)
|
|
file_line = f"{simplified_path}:{lineno}"
|
|
remaining_width = width - col - 1
|
|
file_color = A_REVERSE | A_BOLD if is_selected else color_file
|
|
self.add_str(
|
|
line, col, file_line[:remaining_width], file_color
|
|
)
|
|
|
|
line += 1
|
|
|
|
return line
|
|
|
|
|
|
class FooterWidget(Widget):
|
|
"""Widget for rendering the footer section (legend and controls)."""
|
|
|
|
def __init__(self, display, colors, collector):
|
|
"""
|
|
Initialize footer widget.
|
|
|
|
Args:
|
|
display: DisplayInterface implementation
|
|
colors: Dictionary of color attributes
|
|
collector: Reference to LiveStatsCollector for accessing state
|
|
"""
|
|
super().__init__(display, colors)
|
|
self.collector = collector
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""
|
|
Render the footer at the specified position.
|
|
|
|
Args:
|
|
line: Starting line number (should be height - 2)
|
|
width: Available width
|
|
|
|
Returns:
|
|
Next available line number
|
|
"""
|
|
A_DIM = self.display.get_attr("A_DIM")
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
|
|
# Legend line
|
|
legend = "nsamples: direct/cumulative (direct=executing, cumulative=on stack)"
|
|
self.add_str(line, 0, legend[: width - 1], A_DIM)
|
|
line += 1
|
|
|
|
# Controls line with status
|
|
sort_names = {
|
|
"tottime": "Total Time",
|
|
"nsamples": "Direct Samples",
|
|
"cumtime": "Cumulative Time",
|
|
"sample_pct": "Sample %",
|
|
"cumul_pct": "Cumulative %",
|
|
}
|
|
sort_display = sort_names.get(
|
|
self.collector.sort_by, self.collector.sort_by
|
|
)
|
|
|
|
# Build status indicators
|
|
status = []
|
|
if self.collector.finished:
|
|
status.append("[PROFILING FINISHED - Press 'q' to quit]")
|
|
elif self.collector.paused:
|
|
status.append("[PAUSED]")
|
|
if self.collector.filter_pattern:
|
|
status.append(
|
|
f"[Filter: {self.collector.filter_pattern} (c to clear)]"
|
|
)
|
|
# Show trend colors status if disabled
|
|
if self.collector._trend_tracker is not None and not self.collector._trend_tracker.enabled:
|
|
status.append("[Trend colors: OFF]")
|
|
status_str = " ".join(status) + " " if status else ""
|
|
|
|
if self.collector.finished:
|
|
footer = f"{status_str}"
|
|
else:
|
|
footer = f"{status_str}Sort: {sort_display} | 't':mode 'x':trends ←→:thread 'h':help 'q':quit"
|
|
self.add_str(
|
|
line,
|
|
0,
|
|
footer[: width - 1],
|
|
A_BOLD
|
|
if (self.collector.paused or self.collector.finished)
|
|
else A_DIM,
|
|
)
|
|
|
|
return line + 1
|
|
|
|
def render_filter_input_prompt(self, line, width):
|
|
"""Draw the filter input prompt at the bottom of the screen."""
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
A_REVERSE = self.display.get_attr("A_REVERSE")
|
|
|
|
# Draw prompt on last line
|
|
prompt = f"Function filter: {self.collector.filter_input_buffer}_"
|
|
self.add_str(line, 0, prompt[: width - 1], A_REVERSE | A_BOLD)
|
|
|
|
|
|
class HelpWidget(Widget):
|
|
"""Widget for rendering the help screen overlay."""
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""
|
|
Render the help screen.
|
|
|
|
Args:
|
|
line: Starting line number (ignored, help is centered)
|
|
width: Available width
|
|
kwargs: Must contain 'height' key
|
|
|
|
Returns:
|
|
Next available line number (not used for overlays)
|
|
"""
|
|
height = kwargs["height"]
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
A_NORMAL = self.display.get_attr("A_NORMAL")
|
|
|
|
help_lines = [
|
|
("Tachyon Profiler - Interactive Commands", A_BOLD),
|
|
("", A_NORMAL),
|
|
("Navigation & Display:", A_BOLD),
|
|
(" s - Cycle through sort modes (forward)", A_NORMAL),
|
|
(" S - Cycle through sort modes (backward)", A_NORMAL),
|
|
(" t - Toggle view mode (ALL / per-thread)", A_NORMAL),
|
|
(" x - Toggle trend colors (on/off)", A_NORMAL),
|
|
(" j/k or ↑/↓ - Select next/previous function (--opcodes)", A_NORMAL),
|
|
(" ← / → - Cycle through threads", A_NORMAL),
|
|
(" + - Faster display refresh rate", A_NORMAL),
|
|
(" - - Slower display refresh rate", A_NORMAL),
|
|
("", A_NORMAL),
|
|
("Control:", A_BOLD),
|
|
(" p - Freeze display (snapshot)", A_NORMAL),
|
|
(" r - Reset all statistics", A_NORMAL),
|
|
("", A_NORMAL),
|
|
("Filtering:", A_BOLD),
|
|
(" / - Enter function filter (substring)", A_NORMAL),
|
|
(" c - Clear filter", A_NORMAL),
|
|
(" ESC - Cancel filter input", A_NORMAL),
|
|
("", A_NORMAL),
|
|
("Other:", A_BOLD),
|
|
(" h or ? - Show/hide this help", A_NORMAL),
|
|
(" q - Quit profiler", A_NORMAL),
|
|
("", A_NORMAL),
|
|
("Press any key to close this help screen", A_BOLD),
|
|
]
|
|
|
|
start_line = (height - len(help_lines)) // 2
|
|
for i, (text, attr) in enumerate(help_lines):
|
|
if start_line + i < height - 1:
|
|
col = 2 # Left-align with small margin
|
|
self.add_str(start_line + i, col, text[: width - 3], attr)
|
|
|
|
return line # Not used for overlays
|
|
|
|
|
|
class OpcodePanel(Widget):
|
|
"""Widget for displaying opcode statistics for a selected function."""
|
|
|
|
def __init__(self, display, colors, collector):
|
|
super().__init__(display, colors)
|
|
self.collector = collector
|
|
|
|
def render(self, line, width, **kwargs):
|
|
"""Render opcode statistics panel.
|
|
|
|
Args:
|
|
line: Starting line number
|
|
width: Available width
|
|
kwargs: Must contain 'stats_list', 'height'
|
|
|
|
Returns:
|
|
Next available line number
|
|
"""
|
|
stats_list = kwargs.get("stats_list", [])
|
|
height = kwargs.get("height", 24)
|
|
selected_row = self.collector.selected_row
|
|
scroll_offset = getattr(self.collector, 'scroll_offset', 0)
|
|
|
|
A_BOLD = self.display.get_attr("A_BOLD")
|
|
A_NORMAL = self.display.get_attr("A_NORMAL")
|
|
color_cyan = self.colors.get("color_cyan", A_NORMAL)
|
|
color_yellow = self.colors.get("color_yellow", A_NORMAL)
|
|
color_magenta = self.colors.get("color_magenta", A_NORMAL)
|
|
|
|
# Get the selected function from stats_list (accounting for scroll)
|
|
actual_index = scroll_offset + selected_row
|
|
if not stats_list or actual_index >= len(stats_list):
|
|
self.add_str(line, 0, "No function selected (use j/k to select)", A_NORMAL)
|
|
return line + 1
|
|
|
|
selected_stat = stats_list[actual_index]
|
|
func = selected_stat["func"]
|
|
filename, lineno, funcname = func
|
|
|
|
# Get opcode stats for this function
|
|
opcode_stats = self.collector.opcode_stats.get(func, {})
|
|
|
|
if not opcode_stats:
|
|
self.add_str(line, 0, f"No opcode data for {funcname}() (requires --opcodes)", A_NORMAL)
|
|
return line + 1
|
|
|
|
# Sort opcodes by count
|
|
sorted_opcodes = sorted(opcode_stats.items(), key=lambda x: -x[1])
|
|
total_opcode_samples = sum(opcode_stats.values())
|
|
|
|
# Draw header
|
|
header = f"─── Opcodes for {funcname}() "
|
|
header += "─" * max(0, width - len(header) - 1)
|
|
self.add_str(line, 0, header[:width-1], color_cyan | A_BOLD)
|
|
line += 1
|
|
|
|
# Calculate max samples for bar scaling
|
|
max_count = sorted_opcodes[0][1] if sorted_opcodes else 1
|
|
|
|
# Draw opcode rows (limit to available space)
|
|
max_rows = min(8, height - line - 3) # Leave room for footer
|
|
bar_width = 20
|
|
|
|
for i, (opcode_num, count) in enumerate(sorted_opcodes[:max_rows]):
|
|
if line >= height - 3:
|
|
break
|
|
|
|
opcode_info = get_opcode_info(opcode_num)
|
|
is_specialized = opcode_info["is_specialized"]
|
|
name_display = format_opcode(opcode_num)
|
|
|
|
pct = (count / total_opcode_samples * 100) if total_opcode_samples > 0 else 0
|
|
|
|
# Draw bar
|
|
bar_fill = int((count / max_count) * bar_width) if max_count > 0 else 0
|
|
bar = "█" * bar_fill + "░" * (bar_width - bar_fill)
|
|
|
|
# Format: [████████░░░░] LOAD_ATTR 45.2% (1234)
|
|
# Specialized opcodes shown in magenta, base opcodes in yellow
|
|
name_color = color_magenta if is_specialized else color_yellow
|
|
|
|
row_text = f"[{bar}] {name_display:<35} {pct:>5.1f}% ({count:>6})"
|
|
self.add_str(line, 2, row_text[:width-3], name_color)
|
|
line += 1
|
|
|
|
# Show "..." if more opcodes exist
|
|
if len(sorted_opcodes) > max_rows:
|
|
remaining = len(sorted_opcodes) - max_rows
|
|
self.add_str(line, 2, f"... and {remaining} more opcodes", A_NORMAL)
|
|
line += 1
|
|
|
|
return line
|