| """Heatmap collector for Python profiling with line-level execution heat visualization.""" | |
| import base64 | |
| import collections | |
| import html | |
| import importlib.resources | |
| import json | |
| import math | |
| import os | |
| import platform | |
| import sys | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Dict, List, Tuple | |
| from ._css_utils import get_combined_css | |
| from ._format_utils import fmt | |
| from .collector import normalize_location, extract_lineno | |
| from .opcode_utils import get_opcode_info, format_opcode | |
| from .stack_collector import StackTraceCollector | |
| from .module_utils import extract_module_name, get_python_path_info | |
| # ============================================================================ | |
| # Data Classes | |
| # ============================================================================ | |
| @dataclass | |
| class FileStats: | |
| """Statistics for a single profiled file.""" | |
| filename: str | |
| module_name: str | |
| module_type: str | |
| total_samples: int | |
| total_self_samples: int | |
| num_lines: int | |
| max_samples: int | |
| max_self_samples: int | |
| percentage: float = 0.0 | |
| @dataclass | |
| class TreeNode: | |
| """Node in the hierarchical file tree structure.""" | |
| files: List[FileStats] = field(default_factory=list) | |
| samples: int = 0 | |
| count: int = 0 | |
| children: Dict[str, 'TreeNode'] = field(default_factory=dict) | |
| # ============================================================================ | |
| # Helper Classes | |
| # ============================================================================ | |
| class _TemplateLoader: | |
| """Loads and caches HTML/CSS/JS templates for heatmap generation.""" | |
| def __init__(self): | |
| """Load all templates and assets once.""" | |
| self.index_template = None | |
| self.file_template = None | |
| self.index_css = None | |
| self.index_js = None | |
| self.file_css = None | |
| self.file_js = None | |
| self.logo_html = None | |
| self._load_templates() | |
| def _load_templates(self): | |
| """Load all template files from _heatmap_assets.""" | |
| try: | |
| template_dir = importlib.resources.files(__package__) | |
| assets_dir = template_dir / "_heatmap_assets" | |
| # Load HTML templates | |
| self.index_template = (assets_dir / "heatmap_index_template.html").read_text(encoding="utf-8") | |
| self.file_template = (assets_dir / "heatmap_pyfile_template.html").read_text(encoding="utf-8") | |
| # Load CSS (same file used for both index and file pages) | |
| css_content = get_combined_css("heatmap") | |
| self.index_css = css_content | |
| self.file_css = css_content | |
| # Load JS | |
| base_js = (template_dir / "_shared_assets" / "base.js").read_text(encoding="utf-8") | |
| heatmap_shared_js = (assets_dir / "heatmap_shared.js").read_text(encoding="utf-8") | |
| shared_js = f"{base_js}\n{heatmap_shared_js}" | |
| self.index_js = f"{shared_js}\n{(assets_dir / 'heatmap_index.js').read_text(encoding='utf-8')}" | |
| self.file_js = f"{shared_js}\n{(assets_dir / 'heatmap.js').read_text(encoding='utf-8')}" | |
| # Load Tachyon logo | |
| logo_dir = template_dir / "_assets" | |
| try: | |
| png_path = logo_dir / "tachyon-logo.png" | |
| b64_logo = base64.b64encode(png_path.read_bytes()).decode("ascii") | |
| self.logo_html = f'<img src="data:image/png;base64,{b64_logo}" alt="Tachyon logo" class="python-logo"/>' | |
| except (FileNotFoundError, IOError) as e: | |
| self.logo_html = '<div class="python-logo-placeholder"></div>' | |
| print(f"Warning: Could not load Tachyon logo: {e}") | |
| except (FileNotFoundError, IOError) as e: | |
| raise RuntimeError(f"Failed to load heatmap template files: {e}") from e | |
| class _TreeBuilder: | |
| """Builds hierarchical tree structure from file statistics.""" | |
| @staticmethod | |
| def build_file_tree(file_stats: List[FileStats]) -> Dict[str, TreeNode]: | |
| """Build hierarchical tree grouped by module type, then by module structure. | |
| Args: | |
| file_stats: List of FileStats objects | |
| Returns: | |
| Dictionary mapping module types to their tree roots | |
| """ | |
| # Group by module type first | |
| type_groups = {'stdlib': [], 'site-packages': [], 'project': [], 'other': []} | |
| for stat in file_stats: | |
| type_groups[stat.module_type].append(stat) | |
| # Build tree for each type | |
| trees = {} | |
| for module_type, stats in type_groups.items(): | |
| if not stats: | |
| continue | |
| root_node = TreeNode() | |
| for stat in stats: | |
| module_name = stat.module_name | |
| parts = module_name.split('.') | |
| # Navigate/create tree structure | |
| current_node = root_node | |
| for i, part in enumerate(parts): | |
| if i == len(parts) - 1: | |
| # Last part - store the file | |
| current_node.files.append(stat) | |
| else: | |
| # Intermediate part - create or navigate | |
| if part not in current_node.children: | |
| current_node.children[part] = TreeNode() | |
| current_node = current_node.children[part] | |
| # Calculate aggregate stats for this type's tree | |
| _TreeBuilder._calculate_node_stats(root_node) | |
| trees[module_type] = root_node | |
| return trees | |
| @staticmethod | |
| def _calculate_node_stats(node: TreeNode) -> Tuple[int, int]: | |
| """Recursively calculate aggregate statistics for tree nodes. | |
| Args: | |
| node: TreeNode to calculate stats for | |
| Returns: | |
| Tuple of (total_samples, file_count) | |
| """ | |
| total_samples = 0 | |
| file_count = 0 | |
| # Count files at this level | |
| for file_stat in node.files: | |
| total_samples += file_stat.total_samples | |
| file_count += 1 | |
| # Recursively process children | |
| for child in node.children.values(): | |
| child_samples, child_count = _TreeBuilder._calculate_node_stats(child) | |
| total_samples += child_samples | |
| file_count += child_count | |
| node.samples = total_samples | |
| node.count = file_count | |
| return total_samples, file_count | |
| class _HtmlRenderer: | |
| """Renders hierarchical tree structures as HTML.""" | |
| def __init__(self, file_index: Dict[str, str]): | |
| """Initialize renderer with file index. | |
| Args: | |
| file_index: Mapping from filenames to HTML file names | |
| """ | |
| self.file_index = file_index | |
| self.heatmap_bar_height = 16 | |
| def render_hierarchical_html(self, trees: Dict[str, TreeNode]) -> str: | |
| """Build hierarchical HTML with type sections and collapsible module folders. | |
| Args: | |
| trees: Dictionary mapping module types to tree roots | |
| Returns: | |
| Complete HTML string for all sections | |
| """ | |
| type_names = { | |
| 'stdlib': '📚 Standard Library', | |
| 'site-packages': '📦 Site Packages', | |
| 'project': '🏗️ Project Files', | |
| 'other': '📄 Other Files' | |
| } | |
| sections = [] | |
| for module_type in ['project', 'stdlib', 'site-packages', 'other']: | |
| if module_type not in trees: | |
| continue | |
| tree = trees[module_type] | |
| # Project starts expanded, others start collapsed | |
| is_collapsed = module_type in {'stdlib', 'site-packages', 'other'} | |
| icon = '▶' if is_collapsed else '▼' | |
| content_style = ' style="display: none;"' if is_collapsed else '' | |
| file_word = "file" if tree.count == 1 else "files" | |
| sample_word = "sample" if tree.samples == 1 else "samples" | |
| section_html = f''' | |
| <div class="type-section"> | |
| <div class="type-header" onclick="toggleTypeSection(this)"> | |
| <span class="type-icon">{icon}</span> | |
| <span class="type-title">{type_names[module_type]}</span> | |
| <span class="type-stats">({tree.count} {file_word}, {tree.samples:n} {sample_word})</span> | |
| </div> | |
| <div class="type-content"{content_style}> | |
| ''' | |
| # Render root folders | |
| root_folders = sorted(tree.children.items(), | |
| key=lambda x: x[1].samples, reverse=True) | |
| for folder_name, folder_node in root_folders: | |
| section_html += self._render_folder(folder_node, folder_name, level=1) | |
| # Render root files (files not in any module) | |
| if tree.files: | |
| sorted_files = sorted(tree.files, key=lambda x: x.total_samples, reverse=True) | |
| section_html += ' <div class="files-list">\n' | |
| for stat in sorted_files: | |
| section_html += self._render_file_item(stat, indent=' ') | |
| section_html += ' </div>\n' | |
| section_html += ' </div>\n</div>\n' | |
| sections.append(section_html) | |
| return '\n'.join(sections) | |
| def _render_folder(self, node: TreeNode, name: str, level: int = 1) -> str: | |
| """Render a single folder node recursively. | |
| Args: | |
| node: TreeNode to render | |
| name: Display name for the folder | |
| level: Nesting level for indentation | |
| Returns: | |
| HTML string for this folder and its contents | |
| """ | |
| indent = ' ' * level | |
| parts = [] | |
| # Render folder header (collapsed by default) | |
| file_word = "file" if node.count == 1 else "files" | |
| sample_word = "sample" if node.samples == 1 else "samples" | |
| parts.append(f'{indent}<div class="folder-node collapsed" data-level="{level}">') | |
| parts.append(f'{indent} <div class="folder-header" onclick="toggleFolder(this)">') | |
| parts.append(f'{indent} <span class="folder-icon">▶</span>') | |
| parts.append(f'{indent} <span class="folder-name">📁 {html.escape(name)}</span>') | |
| parts.append(f'{indent} <span class="folder-stats">' | |
| f'({node.count} {file_word}, {node.samples:n} {sample_word})</span>') | |
| parts.append(f'{indent} </div>') | |
| parts.append(f'{indent} <div class="folder-content" style="display: none;">') | |
| # Render sub-folders sorted by sample count | |
| subfolders = sorted(node.children.items(), | |
| key=lambda x: x[1].samples, reverse=True) | |
| for subfolder_name, subfolder_node in subfolders: | |
| parts.append(self._render_folder(subfolder_node, subfolder_name, level + 1)) | |
| # Render files in this folder | |
| if node.files: | |
| sorted_files = sorted(node.files, key=lambda x: x.total_samples, reverse=True) | |
| parts.append(f'{indent} <div class="files-list">') | |
| for stat in sorted_files: | |
| parts.append(self._render_file_item(stat, indent=f'{indent} ')) | |
| parts.append(f'{indent} </div>') | |
| parts.append(f'{indent} </div>') | |
| parts.append(f'{indent}</div>') | |
| return '\n'.join(parts) | |
| def _render_file_item(self, stat: FileStats, indent: str = '') -> str: | |
| """Render a single file item with heatmap bar. | |
| Args: | |
| stat: FileStats object | |
| indent: Indentation string | |
| Returns: | |
| HTML string for file item | |
| """ | |
| full_path = html.escape(stat.filename) | |
| module_name = html.escape(stat.module_name) | |
| intensity = stat.percentage / 100.0 | |
| bar_width = min(stat.percentage, 100) | |
| html_file = self.file_index[stat.filename] | |
| s = "" if stat.total_samples == 1 else "s" | |
| return (f'{indent}<div class="file-item">\n' | |
| f'{indent} <a href="{html_file}" class="file-link" title="{full_path}">📄 {module_name}</a>\n' | |
| f'{indent} <span class="file-samples">{stat.total_samples:n} sample{s}</span>\n' | |
| f'{indent} <div class="heatmap-bar-container"><div class="heatmap-bar" style="width: {bar_width}px; height: {self.heatmap_bar_height}px;" data-intensity="{intensity:.3f}"></div></div>\n' | |
| f'{indent}</div>\n') | |
| # ============================================================================ | |
| # Main Collector Class | |
| # ============================================================================ | |
| class HeatmapCollector(StackTraceCollector): | |
| """Collector that generates coverage.py-style heatmap HTML output with line intensity. | |
| This collector creates detailed HTML reports showing which lines of code | |
| were executed most frequently during profiling, similar to coverage.py | |
| but showing execution "heat" rather than just coverage. | |
| """ | |
| # File naming and formatting constants | |
| FILE_INDEX_FORMAT = "file_{:04d}.html" | |
| def __init__(self, *args, **kwargs): | |
| """Initialize the heatmap collector with data structures for analysis.""" | |
| super().__init__(*args, **kwargs) | |
| # Sample counting data structures | |
| self.line_samples = collections.Counter() | |
| self.file_samples = collections.defaultdict(collections.Counter) | |
| self.line_self_samples = collections.Counter() | |
| self.file_self_samples = collections.defaultdict(collections.Counter) | |
| # Call graph data structures for navigation (sets for O(1) deduplication) | |
| self.call_graph = collections.defaultdict(set) | |
| self.callers_graph = collections.defaultdict(set) | |
| self.function_definitions = {} | |
| # Map each sampled line to its function for proper caller lookup | |
| # (filename, lineno) -> funcname | |
| self.line_to_function = {} | |
| # Edge counting for call path analysis | |
| self.edge_samples = collections.Counter() | |
| # Bytecode-level tracking data structures | |
| # Track samples per (file, lineno) -> {opcode: {'count': N, 'locations': set()}} | |
| # Locations are deduplicated via set to minimize memory usage | |
| self.line_opcodes = collections.defaultdict(dict) | |
| # Statistics and metadata | |
| self._total_samples = 0 | |
| self._path_info = get_python_path_info() | |
| self.stats = {} | |
| # Opcode collection flag | |
| self.opcodes_enabled = False | |
| # Template loader (loads all templates once) | |
| self._template_loader = _TemplateLoader() | |
| # File index (populated during export) | |
| self.file_index = {} | |
| # Reusable set for deduplicating line locations within a single sample. | |
| # This avoids over-counting recursive functions in cumulative stats. | |
| self._seen_lines = set() | |
| def set_stats(self, sample_interval_usec, duration_sec, sample_rate, error_rate=None, missed_samples=None, **kwargs): | |
| """Set profiling statistics to include in heatmap output. | |
| Args: | |
| sample_interval_usec: Sampling interval in microseconds | |
| duration_sec: Total profiling duration in seconds | |
| sample_rate: Effective sampling rate | |
| error_rate: Optional error rate during profiling | |
| missed_samples: Optional percentage of missed samples | |
| **kwargs: Additional statistics to include | |
| """ | |
| self.stats = { | |
| "sample_interval_usec": sample_interval_usec, | |
| "duration_sec": duration_sec, | |
| "sample_rate": sample_rate, | |
| "error_rate": error_rate, | |
| "missed_samples": missed_samples, | |
| "python_version": sys.version, | |
| "python_implementation": platform.python_implementation(), | |
| "platform": platform.platform(), | |
| } | |
| self.stats.update(kwargs) | |
| def process_frames(self, frames, thread_id, weight=1): | |
| """Process stack frames and count samples per line. | |
| Args: | |
| frames: List of (filename, location, funcname, opcode) tuples in | |
| leaf-to-root order. location is (lineno, end_lineno, col_offset, end_col_offset). | |
| opcode is None if not gathered. | |
| thread_id: Thread ID for this stack trace | |
| weight: Number of samples this stack represents (for batched RLE) | |
| """ | |
| self._total_samples += weight | |
| self._seen_lines.clear() | |
| for i, (filename, location, funcname, opcode) in enumerate(frames): | |
| # Normalize location to 4-tuple format | |
| lineno, end_lineno, col_offset, end_col_offset = normalize_location(location) | |
| if not self._is_valid_frame(filename, lineno): | |
| continue | |
| # frames[0] is the leaf - where execution is actually happening | |
| is_leaf = (i == 0) | |
| line_key = (filename, lineno) | |
| count_cumulative = line_key not in self._seen_lines | |
| if count_cumulative: | |
| self._seen_lines.add(line_key) | |
| self._record_line_sample(filename, lineno, funcname, is_leaf=is_leaf, | |
| count_cumulative=count_cumulative, weight=weight) | |
| if opcode is not None: | |
| # Set opcodes_enabled flag when we first encounter opcode data | |
| self.opcodes_enabled = True | |
| self._record_bytecode_sample(filename, lineno, opcode, | |
| end_lineno, col_offset, end_col_offset, | |
| weight=weight) | |
| # Build call graph for adjacent frames (relationships are deduplicated anyway) | |
| if i + 1 < len(frames): | |
| next_frame = frames[i + 1] | |
| next_lineno = extract_lineno(next_frame[1]) | |
| self._record_call_relationship( | |
| (filename, lineno, funcname), | |
| (next_frame[0], next_lineno, next_frame[2]), | |
| weight=weight, | |
| ) | |
| def _is_valid_frame(self, filename, lineno): | |
| """Check if a frame should be included in the heatmap.""" | |
| # Skip internal or invalid files | |
| if not filename or filename.startswith('<') or filename.startswith('['): | |
| return False | |
| # Skip invalid frames with corrupted filename data | |
| if filename == "__init__" and lineno == 0: | |
| return False | |
| return True | |
| def _record_line_sample(self, filename, lineno, funcname, is_leaf=False, | |
| count_cumulative=True, weight=1): | |
| """Record a sample for a specific line.""" | |
| # Track cumulative samples (all occurrences in stack) | |
| if count_cumulative: | |
| self.line_samples[(filename, lineno)] += weight | |
| self.file_samples[filename][lineno] += weight | |
| # Track self/leaf samples (only when at top of stack) | |
| if is_leaf: | |
| self.line_self_samples[(filename, lineno)] += weight | |
| self.file_self_samples[filename][lineno] += weight | |
| # Record function definition location | |
| if funcname and (filename, funcname) not in self.function_definitions: | |
| self.function_definitions[(filename, funcname)] = lineno | |
| # Map this line to its function for caller/callee navigation | |
| if funcname: | |
| self.line_to_function[(filename, lineno)] = funcname | |
| def _record_bytecode_sample(self, filename, lineno, opcode, | |
| end_lineno=None, col_offset=None, end_col_offset=None, | |
| weight=1): | |
| """Record a sample for a specific bytecode instruction. | |
| Args: | |
| filename: Source filename | |
| lineno: Line number | |
| opcode: Opcode number being executed | |
| end_lineno: End line number (may be -1 if not available) | |
| col_offset: Column offset in UTF-8 bytes (may be -1 if not available) | |
| end_col_offset: End column offset in UTF-8 bytes (may be -1 if not available) | |
| weight: Number of samples this represents (for batched RLE) | |
| """ | |
| key = (filename, lineno) | |
| # Initialize opcode entry if needed - use set for location deduplication | |
| if opcode not in self.line_opcodes[key]: | |
| self.line_opcodes[key][opcode] = {'count': 0, 'locations': set()} | |
| self.line_opcodes[key][opcode]['count'] += weight | |
| # Store unique location info if column offset is available (not -1) | |
| if col_offset is not None and col_offset >= 0: | |
| # Use tuple as set key for deduplication | |
| loc_key = (end_lineno, col_offset, end_col_offset) | |
| self.line_opcodes[key][opcode]['locations'].add(loc_key) | |
| def _get_bytecode_data_for_line(self, filename, lineno): | |
| """Get bytecode disassembly data for instructions on a specific line. | |
| Args: | |
| filename: Source filename | |
| lineno: Line number | |
| Returns: | |
| List of dicts with instruction info, sorted by samples descending | |
| """ | |
| key = (filename, lineno) | |
| opcode_data = self.line_opcodes.get(key, {}) | |
| result = [] | |
| for opcode, data in opcode_data.items(): | |
| info = get_opcode_info(opcode) | |
| # Handle both old format (int count) and new format (dict with count/locations) | |
| if isinstance(data, dict): | |
| count = data.get('count', 0) | |
| raw_locations = data.get('locations', set()) | |
| # Convert set of tuples to list of dicts for JSON serialization | |
| if isinstance(raw_locations, set): | |
| locations = [ | |
| {'end_lineno': loc[0], 'col_offset': loc[1], 'end_col_offset': loc[2]} | |
| for loc in raw_locations | |
| ] | |
| else: | |
| locations = raw_locations | |
| else: | |
| count = data | |
| locations = [] | |
| result.append({ | |
| 'opcode': opcode, | |
| 'opname': format_opcode(opcode), | |
| 'base_opname': info['base_opname'], | |
| 'is_specialized': info['is_specialized'], | |
| 'samples': count, | |
| 'locations': locations, | |
| }) | |
| # Sort by samples descending, then by opcode number | |
| result.sort(key=lambda x: (-x['samples'], x['opcode'])) | |
| return result | |
| def _record_call_relationship(self, callee_frame, caller_frame, weight=1): | |
| """Record caller/callee relationship between adjacent frames.""" | |
| callee_filename, callee_lineno, callee_funcname = callee_frame | |
| caller_filename, caller_lineno, caller_funcname = caller_frame | |
| # Skip internal files for call graph | |
| if callee_filename.startswith('<') or callee_filename.startswith('['): | |
| return | |
| # Get the callee's function definition line | |
| callee_def_line = self.function_definitions.get( | |
| (callee_filename, callee_funcname), callee_lineno | |
| ) | |
| # Record caller -> callee relationship (set handles deduplication) | |
| caller_key = (caller_filename, caller_lineno) | |
| callee_info = (callee_filename, callee_def_line, callee_funcname) | |
| self.call_graph[caller_key].add(callee_info) | |
| # Record callee <- caller relationship (set handles deduplication) | |
| callee_key = (callee_filename, callee_def_line) | |
| caller_info = (caller_filename, caller_lineno, caller_funcname) | |
| self.callers_graph[callee_key].add(caller_info) | |
| # Count this call edge for path analysis | |
| edge_key = (caller_key, callee_key) | |
| self.edge_samples[edge_key] += weight | |
| def export(self, output_path): | |
| """Export heatmap data as HTML files in a directory. | |
| Args: | |
| output_path: Path where to create the heatmap output directory | |
| """ | |
| if not self.file_samples: | |
| print("Warning: No heatmap data to export") | |
| return False | |
| try: | |
| output_dir = self._prepare_output_directory(output_path) | |
| file_stats = self._calculate_file_stats() | |
| self._create_file_index(file_stats) | |
| # Generate individual file reports | |
| self._generate_file_reports(output_dir, file_stats) | |
| # Generate index page | |
| self._generate_index_html(output_dir / 'index.html', file_stats) | |
| self._print_export_summary(output_dir, file_stats) | |
| return True | |
| except Exception as e: | |
| print(f"Error: Failed to export heatmap: {e}") | |
| raise | |
| def _prepare_output_directory(self, output_path): | |
| """Create output directory for heatmap files.""" | |
| output_dir = Path(output_path) | |
| if output_dir.suffix == '.html': | |
| output_dir = output_dir.with_suffix('') | |
| try: | |
| output_dir.mkdir(exist_ok=True, parents=True) | |
| except (IOError, OSError) as e: | |
| raise RuntimeError(f"Failed to create output directory {output_dir}: {e}") from e | |
| return output_dir | |
| def _create_file_index(self, file_stats: List[FileStats]): | |
| """Create mapping from filenames to HTML file names.""" | |
| self.file_index = { | |
| stat.filename: self.FILE_INDEX_FORMAT.format(i) | |
| for i, stat in enumerate(file_stats) | |
| } | |
| def _generate_file_reports(self, output_dir, file_stats: List[FileStats]): | |
| """Generate HTML report for each source file.""" | |
| for stat in file_stats: | |
| file_path = output_dir / self.file_index[stat.filename] | |
| line_counts = self.file_samples[stat.filename] | |
| valid_line_counts = {line: count for line, count in line_counts.items() if line >= 0} | |
| self_counts = self.file_self_samples.get(stat.filename, {}) | |
| valid_self_counts = {line: count for line, count in self_counts.items() if line >= 0} | |
| self._generate_file_html( | |
| file_path, | |
| stat.filename, | |
| valid_line_counts, | |
| valid_self_counts, | |
| stat | |
| ) | |
| def _print_export_summary(self, output_dir, file_stats: List[FileStats]): | |
| """Print summary of exported heatmap.""" | |
| print(f"Heatmap output written to {output_dir}/") | |
| print(f" - Index: {output_dir / 'index.html'}") | |
| s = "" if len(file_stats) == 1 else "s" | |
| print(f" - {len(file_stats)} source file{s} analyzed") | |
| def _calculate_file_stats(self) -> List[FileStats]: | |
| """Calculate statistics for each file. | |
| Returns: | |
| List of FileStats objects sorted by total samples | |
| """ | |
| file_stats = [] | |
| for filename, line_counts in self.file_samples.items(): | |
| # Skip special frames | |
| if filename in ('~', '...', '.') or filename.startswith('<') or filename.startswith('['): | |
| continue | |
| # Filter out lines with -1 (special frames) | |
| valid_line_counts = {line: count for line, count in line_counts.items() if line >= 0} | |
| if not valid_line_counts: | |
| continue | |
| # Get self samples for this file | |
| self_line_counts = self.file_self_samples.get(filename, {}) | |
| valid_self_counts = {line: count for line, count in self_line_counts.items() if line >= 0} | |
| total_samples = sum(valid_line_counts.values()) | |
| total_self_samples = sum(valid_self_counts.values()) | |
| num_lines = len(valid_line_counts) | |
| max_samples = max(valid_line_counts.values()) | |
| max_self_samples = max(valid_self_counts.values()) if valid_self_counts else 0 | |
| module_name, module_type = extract_module_name(filename, self._path_info) | |
| file_stats.append(FileStats( | |
| filename=filename, | |
| module_name=module_name, | |
| module_type=module_type, | |
| total_samples=total_samples, | |
| total_self_samples=total_self_samples, | |
| num_lines=num_lines, | |
| max_samples=max_samples, | |
| max_self_samples=max_self_samples, | |
| percentage=0.0 | |
| )) | |
| # Sort by total samples and calculate percentages | |
| file_stats.sort(key=lambda x: x.total_samples, reverse=True) | |
| if file_stats: | |
| max_total = file_stats[0].total_samples | |
| for stat in file_stats: | |
| stat.percentage = (stat.total_samples / max_total * 100) if max_total > 0 else 0 | |
| return file_stats | |
| def _generate_index_html(self, index_path: Path, file_stats: List[FileStats]): | |
| """Generate index.html with list of all profiled files.""" | |
| # Build hierarchical tree | |
| tree = _TreeBuilder.build_file_tree(file_stats) | |
| # Render tree as HTML | |
| renderer = _HtmlRenderer(self.file_index) | |
| sections_html = renderer.render_hierarchical_html(tree) | |
| # Format error rate and missed samples with bar classes | |
| error_rate = self.stats.get('error_rate') | |
| if error_rate is not None: | |
| error_rate_str = f"{fmt(error_rate)}%" | |
| error_rate_width = min(error_rate, 100) | |
| # Determine bar color class based on rate | |
| if error_rate < 5: | |
| error_rate_class = "good" | |
| elif error_rate < 15: | |
| error_rate_class = "warning" | |
| else: | |
| error_rate_class = "error" | |
| else: | |
| error_rate_str = "N/A" | |
| error_rate_width = 0 | |
| error_rate_class = "good" | |
| missed_samples = self.stats.get('missed_samples') | |
| if missed_samples is not None: | |
| missed_samples_str = f"{fmt(missed_samples)}%" | |
| missed_samples_width = min(missed_samples, 100) | |
| if missed_samples < 5: | |
| missed_samples_class = "good" | |
| elif missed_samples < 15: | |
| missed_samples_class = "warning" | |
| else: | |
| missed_samples_class = "error" | |
| else: | |
| missed_samples_str = "N/A" | |
| missed_samples_width = 0 | |
| missed_samples_class = "good" | |
| # Populate template | |
| replacements = { | |
| "<!-- INLINE_CSS -->": f"<style>\n{self._template_loader.index_css}\n</style>", | |
| "<!-- INLINE_JS -->": f"<script>\n{self._template_loader.index_js}\n</script>", | |
| "<!-- PYTHON_LOGO -->": self._template_loader.logo_html, | |
| "<!-- PYTHON_VERSION -->": f"{sys.version_info.major}.{sys.version_info.minor}", | |
| "<!-- NUM_FILES -->": f"{len(file_stats):n}", | |
| "<!-- TOTAL_SAMPLES -->": f"{self._total_samples:n}", | |
| "<!-- DURATION -->": fmt(self.stats.get('duration_sec', 0)), | |
| "<!-- SAMPLE_RATE -->": fmt(self.stats.get('sample_rate', 0)), | |
| "<!-- ERROR_RATE -->": error_rate_str, | |
| "<!-- ERROR_RATE_WIDTH -->": str(error_rate_width), | |
| "<!-- ERROR_RATE_CLASS -->": error_rate_class, | |
| "<!-- MISSED_SAMPLES -->": missed_samples_str, | |
| "<!-- MISSED_SAMPLES_WIDTH -->": str(missed_samples_width), | |
| "<!-- MISSED_SAMPLES_CLASS -->": missed_samples_class, | |
| "<!-- SECTIONS_HTML -->": sections_html, | |
| } | |
| html_content = self._template_loader.index_template | |
| for placeholder, value in replacements.items(): | |
| html_content = html_content.replace(placeholder, value) | |
| try: | |
| index_path.write_text(html_content, encoding='utf-8') | |
| except (IOError, OSError) as e: | |
| raise RuntimeError(f"Failed to write index file {index_path}: {e}") from e | |
| def _generate_file_html(self, output_path: Path, filename: str, | |
| line_counts: Dict[int, int], self_counts: Dict[int, int], | |
| file_stat: FileStats): | |
| """Generate HTML for a single source file with heatmap coloring.""" | |
| source_lines = [f"# Source file not available: {filename}"] | |
| try: | |
| path = Path(filename) | |
| if path.is_file(): | |
| source_lines = path.read_text( | |
| encoding='utf-8', errors='replace').splitlines() | |
| except (IOError, OSError): | |
| pass | |
| # Generate HTML for each line | |
| max_samples = max(line_counts.values()) if line_counts else 1 | |
| max_self_samples = max(self_counts.values()) if self_counts else 1 | |
| code_lines_html = [ | |
| self._build_line_html(line_num, line_content, line_counts, self_counts, | |
| max_samples, max_self_samples, filename) | |
| for line_num, line_content in enumerate(source_lines, start=1) | |
| ] | |
| # Populate template | |
| replacements = { | |
| "<!-- FILENAME -->": html.escape(filename), | |
| "<!-- TOTAL_SAMPLES -->": f"{file_stat.total_samples:n}", | |
| "<!-- TOTAL_SELF_SAMPLES -->": f"{file_stat.total_self_samples:n}", | |
| "<!-- NUM_LINES -->": f"{file_stat.num_lines:n}", | |
| "<!-- PERCENTAGE -->": fmt(file_stat.percentage, 2), | |
| "<!-- MAX_SAMPLES -->": f"{file_stat.max_samples:n}", | |
| "<!-- MAX_SELF_SAMPLES -->": f"{file_stat.max_self_samples:n}", | |
| "<!-- CODE_LINES -->": ''.join(code_lines_html), | |
| "<!-- INLINE_CSS -->": f"<style>\n{self._template_loader.file_css}\n</style>", | |
| "<!-- INLINE_JS -->": f"<script>\n{self._template_loader.file_js}\n</script>", | |
| "<!-- PYTHON_LOGO -->": self._template_loader.logo_html, | |
| "<!-- PYTHON_VERSION -->": f"{sys.version_info.major}.{sys.version_info.minor}", | |
| } | |
| html_content = self._template_loader.file_template | |
| for placeholder, value in replacements.items(): | |
| html_content = html_content.replace(placeholder, value) | |
| try: | |
| output_path.write_text(html_content, encoding='utf-8') | |
| except (IOError, OSError) as e: | |
| raise RuntimeError(f"Failed to write file {output_path}: {e}") from e | |
| def _build_line_html(self, line_num: int, line_content: str, | |
| line_counts: Dict[int, int], self_counts: Dict[int, int], | |
| max_samples: int, max_self_samples: int, filename: str) -> str: | |
| """Build HTML for a single line of source code.""" | |
| cumulative_samples = line_counts.get(line_num, 0) | |
| self_samples = self_counts.get(line_num, 0) | |
| # Calculate colors for both self and cumulative modes | |
| if cumulative_samples > 0: | |
| log_cumulative = math.log(cumulative_samples + 1) | |
| log_max = math.log(max_samples + 1) | |
| cumulative_intensity = log_cumulative / log_max if log_max > 0 else 0 | |
| if self_samples > 0 and max_self_samples > 0: | |
| log_self = math.log(self_samples + 1) | |
| log_max_self = math.log(max_self_samples + 1) | |
| self_intensity = log_self / log_max_self if log_max_self > 0 else 0 | |
| else: | |
| self_intensity = 0 | |
| self_display = f"{self_samples:n}" if self_samples > 0 else "" | |
| cumulative_display = f"{cumulative_samples:n}" | |
| tooltip = f"Self: {self_samples:n}, Total: {cumulative_samples:n}" | |
| else: | |
| cumulative_intensity = 0 | |
| self_intensity = 0 | |
| self_display = "" | |
| cumulative_display = "" | |
| tooltip = "" | |
| # Get bytecode data for this line (if any) | |
| bytecode_data = self._get_bytecode_data_for_line(filename, line_num) | |
| has_bytecode = len(bytecode_data) > 0 and cumulative_samples > 0 | |
| # Build bytecode toggle button if data is available | |
| bytecode_btn_html = '' | |
| bytecode_panel_html = '' | |
| if has_bytecode: | |
| bytecode_json = html.escape(json.dumps(bytecode_data)) | |
| # Calculate specialization percentage | |
| total_samples = sum(d['samples'] for d in bytecode_data) | |
| specialized_samples = sum(d['samples'] for d in bytecode_data if d['is_specialized']) | |
| spec_pct = int(100 * specialized_samples / total_samples) if total_samples > 0 else 0 | |
| bytecode_btn_html = ( | |
| f'<button class="bytecode-toggle" data-bytecode=\'{bytecode_json}\' ' | |
| f'data-spec-pct="{spec_pct}" ' | |
| f'onclick="toggleBytecode(this)" title="Show bytecode">▶</button>' | |
| ) | |
| # Wrapper contains columns + content panel | |
| bytecode_panel_html = ( | |
| f' <div class="bytecode-wrapper" id="bytecode-wrapper-{line_num}">\n' | |
| f' <div class="bytecode-columns">' | |
| f'<div class="line-number"></div>' | |
| f'<div class="line-samples-self"></div>' | |
| f'<div class="line-samples-cumulative"></div>' | |
| f'</div>\n' | |
| f' <div class="bytecode-panel" id="bytecode-{line_num}"></div>\n' | |
| f' </div>\n' | |
| ) | |
| elif self.opcodes_enabled: | |
| # Add invisible spacer to maintain consistent indentation when opcodes are enabled | |
| bytecode_btn_html = '<div class="bytecode-spacer"></div>' | |
| # Get navigation buttons | |
| nav_buttons_html = self._build_navigation_buttons(filename, line_num) | |
| # Build line HTML with instruction highlights if available | |
| line_html = self._render_source_with_highlights(line_content, line_num, | |
| filename, bytecode_data) | |
| title_attr = f' title="{html.escape(tooltip)}"' if tooltip else "" | |
| # Specialization color for toggle mode (green gradient based on spec %) | |
| spec_color_attr = '' | |
| if has_bytecode: | |
| spec_color = self._format_specialization_color(spec_pct) | |
| spec_color_attr = f'data-spec-color="{spec_color}" ' | |
| return ( | |
| f' <div class="code-line" ' | |
| f'data-self-intensity="{self_intensity:.3f}" ' | |
| f'data-cumulative-intensity="{cumulative_intensity:.3f}" ' | |
| f'{spec_color_attr}' | |
| f'id="line-{line_num}"{title_attr}>\n' | |
| f' <div class="line-number">{line_num}</div>\n' | |
| f' <div class="line-samples-self">{self_display}</div>\n' | |
| f' <div class="line-samples-cumulative">{cumulative_display}</div>\n' | |
| f' {bytecode_btn_html}\n' | |
| f' <div class="line-content">{line_html}</div>\n' | |
| f' {nav_buttons_html}\n' | |
| f' </div>\n' | |
| f'{bytecode_panel_html}' | |
| ) | |
| def _render_source_with_highlights(self, line_content: str, line_num: int, | |
| filename: str, bytecode_data: list) -> str: | |
| """Render source line with instruction highlight spans. | |
| Simple: collect ranges with sample counts, assign each byte position to | |
| smallest covering range, then emit spans for contiguous runs with sample data. | |
| """ | |
| content = line_content.rstrip('\n') | |
| if not content: | |
| return '' | |
| # Collect all (start, end) -> {samples, opcodes} mapping from instructions | |
| # Multiple instructions may share the same range, so we sum samples and collect opcodes | |
| range_data = {} | |
| for instr in bytecode_data: | |
| samples = instr.get('samples', 0) | |
| opname = instr.get('opname', '') | |
| for loc in instr.get('locations', []): | |
| if loc.get('end_lineno', line_num) == line_num: | |
| start, end = loc.get('col_offset', -1), loc.get('end_col_offset', -1) | |
| if start >= 0 and end >= 0: | |
| key = (start, end) | |
| if key not in range_data: | |
| range_data[key] = {'samples': 0, 'opcodes': []} | |
| range_data[key]['samples'] += samples | |
| if opname and opname not in range_data[key]['opcodes']: | |
| range_data[key]['opcodes'].append(opname) | |
| if not range_data: | |
| return html.escape(content) | |
| # For each byte position, find the smallest covering range | |
| byte_to_range = {} | |
| for (start, end) in range_data.keys(): | |
| for pos in range(start, end): | |
| if pos not in byte_to_range: | |
| byte_to_range[pos] = (start, end) | |
| else: | |
| # Keep smaller range | |
| old_start, old_end = byte_to_range[pos] | |
| if (end - start) < (old_end - old_start): | |
| byte_to_range[pos] = (start, end) | |
| # Calculate totals for percentage and intensity | |
| total_line_samples = sum(d['samples'] for d in range_data.values()) | |
| max_range_samples = max(d['samples'] for d in range_data.values()) if range_data else 1 | |
| # Render character by character | |
| result = [] | |
| byte_offset = 0 | |
| char_idx = 0 | |
| current_range = None | |
| span_chars = [] | |
| def flush_span(): | |
| nonlocal span_chars, current_range | |
| if span_chars: | |
| text = html.escape(''.join(span_chars)) | |
| if current_range: | |
| data = range_data.get(current_range, {'samples': 0, 'opcodes': []}) | |
| samples = data['samples'] | |
| opcodes = ', '.join(data['opcodes'][:3]) # Top 3 opcodes | |
| if len(data['opcodes']) > 3: | |
| opcodes += f" +{len(data['opcodes']) - 3} more" | |
| pct = int(100 * samples / total_line_samples) if total_line_samples > 0 else 0 | |
| result.append(f'<span class="instr-span" ' | |
| f'data-col-start="{current_range[0]}" ' | |
| f'data-col-end="{current_range[1]}" ' | |
| f'data-samples="{samples}" ' | |
| f'data-max-samples="{max_range_samples}" ' | |
| f'data-pct="{pct}" ' | |
| f'data-opcodes="{html.escape(opcodes)}">{text}</span>') | |
| else: | |
| result.append(text) | |
| span_chars = [] | |
| while char_idx < len(content): | |
| char = content[char_idx] | |
| char_bytes = len(char.encode('utf-8')) | |
| char_range = byte_to_range.get(byte_offset) | |
| if char_range != current_range: | |
| flush_span() | |
| current_range = char_range | |
| span_chars.append(char) | |
| byte_offset += char_bytes | |
| char_idx += 1 | |
| flush_span() | |
| return ''.join(result) | |
| def _format_specialization_color(self, spec_pct: int) -> str: | |
| """Format specialization color based on percentage. | |
| Uses a gradient from gray (0%) through orange (50%) to green (100%). | |
| """ | |
| # Normalize to 0-1 | |
| ratio = spec_pct / 100.0 | |
| if ratio >= 0.5: | |
| # Orange to green (50-100%) | |
| t = (ratio - 0.5) * 2 # 0 to 1 | |
| r = int(255 * (1 - t)) # 255 -> 0 | |
| g = int(180 + 75 * t) # 180 -> 255 | |
| b = int(50 * (1 - t)) # 50 -> 0 | |
| else: | |
| # Gray to orange (0-50%) | |
| t = ratio * 2 # 0 to 1 | |
| r = int(158 + 97 * t) # 158 -> 255 | |
| g = int(158 + 22 * t) # 158 -> 180 | |
| b = int(158 - 108 * t) # 158 -> 50 | |
| alpha = 0.15 + 0.25 * ratio # 0.15 to 0.4 | |
| return f"rgba({r}, {g}, {b}, {alpha})" | |
| def _build_navigation_buttons(self, filename: str, line_num: int) -> str: | |
| """Build navigation buttons for callers/callees. | |
| - Callers: All lines in a function show who calls this function | |
| - Callees: Only actual call site lines show what they call | |
| """ | |
| line_key = (filename, line_num) | |
| funcname = self.line_to_function.get(line_key) | |
| # Get callers: look up by function definition line, not current line | |
| # This ensures all lines in a function show who calls this function | |
| if funcname: | |
| func_def_line = self.function_definitions.get((filename, funcname), line_num) | |
| func_def_key = (filename, func_def_line) | |
| caller_list = self._deduplicate_by_function(self.callers_graph.get(func_def_key, set())) | |
| else: | |
| caller_list = self._deduplicate_by_function(self.callers_graph.get(line_key, set())) | |
| # Get callees: only show for actual call site lines (not every line in function) | |
| callee_list = self._deduplicate_by_function(self.call_graph.get(line_key, set())) | |
| # Get edge counts for each caller/callee | |
| # For callers, use the function definition key for edge lookup | |
| if funcname: | |
| func_def_line = self.function_definitions.get((filename, funcname), line_num) | |
| caller_edge_key = (filename, func_def_line) | |
| else: | |
| caller_edge_key = line_key | |
| callers_with_counts = self._get_edge_counts(caller_edge_key, caller_list, is_caller=True) | |
| # For callees, use the actual line key since that's where the call happens | |
| callees_with_counts = self._get_edge_counts(line_key, callee_list, is_caller=False) | |
| # Build navigation buttons with counts | |
| caller_btn = self._create_navigation_button(callers_with_counts, 'caller', '▲') | |
| callee_btn = self._create_navigation_button(callees_with_counts, 'callee', '▼') | |
| if caller_btn or callee_btn: | |
| return f'<div class="line-nav-buttons">{caller_btn}{callee_btn}</div>' | |
| return '' | |
| def _get_edge_counts(self, line_key: Tuple[str, int], | |
| items: List[Tuple[str, int, str]], | |
| is_caller: bool) -> List[Tuple[str, int, str, int]]: | |
| """Get sample counts for each caller/callee edge.""" | |
| result = [] | |
| for file, line, func in items: | |
| edge_line_key = (file, line) | |
| if is_caller: | |
| edge_key = (edge_line_key, line_key) | |
| else: | |
| edge_key = (line_key, edge_line_key) | |
| count = self.edge_samples.get(edge_key, 0) | |
| result.append((file, line, func, count)) | |
| result.sort(key=lambda x: x[3], reverse=True) | |
| return result | |
| def _deduplicate_by_function(self, items) -> List[Tuple[str, int, str]]: | |
| """Remove duplicate entries based on (file, function) key. | |
| Args: | |
| items: Iterable of (file, line, func) tuples (set or list) | |
| """ | |
| seen = {} | |
| result = [] | |
| for file, line, func in items: | |
| key = (file, func) | |
| if key not in seen: | |
| seen[key] = True | |
| result.append((file, line, func)) | |
| return result | |
| def _create_navigation_button(self, items_with_counts: List[Tuple[str, int, str, int]], | |
| btn_class: str, arrow: str) -> str: | |
| """Create HTML for a navigation button with sample counts.""" | |
| # Filter valid items | |
| valid_items = [(f, l, fn, cnt) for f, l, fn, cnt in items_with_counts | |
| if f in self.file_index and l > 0] | |
| if not valid_items: | |
| return "" | |
| if len(valid_items) == 1: | |
| file, line, func, count = valid_items[0] | |
| target_html = self.file_index[file] | |
| nav_data = json.dumps({'link': f"{target_html}#line-{line}", 'func': func}) | |
| title = f"Go to {btn_class}: {html.escape(func)} ({count:n} samples)" | |
| return f'<button class="nav-btn {btn_class}" data-nav=\'{html.escape(nav_data)}\' title="{title}">{arrow}</button>' | |
| # Multiple items - create menu | |
| total_samples = sum(cnt for _, _, _, cnt in valid_items) | |
| items_data = [ | |
| { | |
| 'file': os.path.basename(file), | |
| 'func': func, | |
| 'count': count, | |
| 'link': f"{self.file_index[file]}#line-{line}" | |
| } | |
| for file, line, func, count in valid_items | |
| ] | |
| items_json = html.escape(json.dumps(items_data)) | |
| title = f"{len(items_data)} {btn_class}s ({total_samples:n} samples)" | |
| return f'<button class="nav-btn {btn_class}" data-nav-multi=\'{items_json}\' title="{title}">{arrow}</button>' |