'
- html_template = html_template.replace("", logo_html)
+ return html_template.replace("", logo_html)
- d3_js = d3_path.read_text(encoding="utf-8")
- fg_css = fg_css_path.read_text(encoding="utf-8")
- fg_js = fg_js_path.read_text(encoding="utf-8")
- fg_tooltip_js = fg_tooltip_js_path.read_text(encoding="utf-8")
- html_template = html_template.replace(
- "",
- f"",
- )
- html_template = html_template.replace(
- "",
- f"",
- )
- html_template = html_template.replace(
- "",
- f"",
- )
- html_template = html_template.replace(
- "",
- f"",
+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.
+ """
+
+ 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() # (filename, lineno) -> count
+ self.file_samples = collections.defaultdict(collections.Counter) # filename -> {lineno: count}
+
+ # Call graph data structures for navigation
+ self.call_graph = collections.defaultdict(list) # (caller_file, caller_line) -> [callees]
+ self.callers_graph = collections.defaultdict(list) # (callee_file, callee_line) -> [callers]
+ self.function_definitions = {} # (filename, funcname) -> lineno
+
+ # Statistics and metadata
+ self._total_samples = 0
+ self._path_info = get_python_path_info()
+ self.stats = {}
+
+ def set_stats(self, sample_interval_usec, duration_sec, sample_rate, error_rate=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
+ **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,
+ "python_version": sys.version,
+ "python_implementation": platform.python_implementation(),
+ "platform": platform.platform(),
+ }
+ self.stats.update(kwargs)
+
+ def process_frames(self, frames, thread_id):
+ """Process stack frames and count samples per line.
+
+ Args:
+ frames: List of frame tuples (filename, lineno, funcname)
+ thread_id: Thread ID for this stack trace
+ """
+ self._total_samples += 1
+
+ # Count each line in the stack and build call graph
+ for i, frame_info in enumerate(frames):
+ filename, lineno, funcname = frame_info
+
+ if not self._is_valid_frame(filename, lineno):
+ continue
+
+ self._record_line_sample(filename, lineno, funcname)
+
+ # Build call graph for adjacent frames
+ if i + 1 < len(frames):
+ self._record_call_relationship(frames[i], frames[i + 1])
+
+ def _is_valid_frame(self, filename, lineno):
+ # Skip internal or invalid files
+ if not filename or filename.startswith('<') or filename.startswith('['):
+ return False
+
+ # Skip invalid frames with corrupted filename data
+ # These come from C code when frame info is unavailable
+ if filename == "__init__" and lineno == 0:
+ return False
+
+ return True
+
+ def _record_line_sample(self, filename, lineno, funcname):
+ self.line_samples[(filename, lineno)] += 1
+ self.file_samples[filename][lineno] += 1
+
+ # Record function definition location
+ if funcname and (filename, funcname) not in self.function_definitions:
+ self.function_definitions[(filename, funcname)] = lineno
+
+ def _record_call_relationship(self, callee_frame, caller_frame):
+ 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
)
- # Replace the placeholder with actual data
- html_content = html_template.replace(
- "{{FLAMEGRAPH_DATA}}", data_json
+ # Record caller -> callee relationship
+ caller_key = (caller_filename, caller_lineno)
+ callee_info = (callee_filename, callee_def_line, callee_funcname)
+ if callee_info not in self.call_graph[caller_key]:
+ self.call_graph[caller_key].append(callee_info)
+
+ # Record callee <- caller relationship
+ callee_key = (callee_filename, callee_def_line)
+ caller_info = (caller_filename, caller_lineno, caller_funcname)
+ if caller_info not in self.callers_graph[callee_key]:
+ self.callers_graph[callee_key].append(caller_info)
+
+ 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
+
+ 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)
+
+ except Exception as e:
+ print(f"Error: Failed to export heatmap: {e}")
+ raise
+
+ def _prepare_output_directory(self, output_path):
+ output_dir = Path(output_path)
+ if output_dir.suffix == '.html':
+ output_dir = output_dir.with_suffix('')
+
+ output_dir.mkdir(exist_ok=True)
+ return output_dir
+
+ def _create_file_index(self, file_stats):
+ self.file_index = {
+ stat['filename']: f"file_{i:04d}.html"
+ for i, stat in enumerate(file_stats)
+ }
+
+ def _generate_file_reports(self, output_dir, file_stats):
+ for stat in file_stats:
+ file_path = output_dir / self.file_index[stat['filename']]
+ self._generate_file_html(
+ file_path,
+ stat['filename'],
+ self.file_samples[stat['filename']],
+ stat
+ )
+
+ def _print_export_summary(self, output_dir, file_stats):
+ print(f"Heatmap output written to {output_dir}/")
+ print(f" - Index: {output_dir / 'index.html'}")
+ print(f" - {len(file_stats)} source file(s) analyzed")
+
+ def _calculate_file_stats(self):
+ """Calculate statistics for each file."""
+ file_stats = []
+ for filename, line_counts in self.file_samples.items():
+ total_samples = sum(line_counts.values())
+ num_lines = len(line_counts)
+ max_samples = max(line_counts.values()) if line_counts else 0
+ module_name, module_type = extract_module_name(filename, self._path_info)
+
+ file_stats.append({
+ 'filename': filename,
+ 'module_name': module_name,
+ 'module_type': module_type,
+ 'total_samples': total_samples,
+ 'num_lines': num_lines,
+ 'max_samples': max_samples,
+ 'percentage': 0 # Will be calculated after sorting
+ })
+
+ # 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, file_stats):
+ """Generate index.html with list of all profiled files."""
+ import html
+ import base64
+ import importlib.resources
+
+ # Load template and assets
+ template_dir = importlib.resources.files(__package__)
+ assets_dir = template_dir / "_assets"
+
+ template_content = (template_dir / "heatmap_index_template.html").read_text(encoding="utf-8")
+ css_content = (template_dir / "heatmap.css").read_text(encoding="utf-8")
+ js_content = (template_dir / "heatmap_index.js").read_text(encoding="utf-8")
+
+ # Load Python logo
+ png_path = assets_dir / "python-logo-only.png"
+ b64_logo = base64.b64encode(png_path.read_bytes()).decode("ascii")
+ logo_html = f''
+
+ # Build table rows
+ table_rows = self._build_table_rows(file_stats)
+
+ # Populate template
+ replacements = {
+ "": f"",
+ "": f"",
+ "": logo_html,
+ "": str(len(file_stats)),
+ "": f"{self._total_samples:,}",
+ "": f"{self.stats.get('duration_sec', 0):.1f}s",
+ "": f"{self.stats.get('sample_rate', 0):.1f}",
+ "": ''.join(table_rows),
+ }
+
+ html_content = template_content
+ for placeholder, value in replacements.items():
+ html_content = html_content.replace(placeholder, value)
+
+ index_path.write_text(html_content, encoding='utf-8')
+
+ def _build_table_rows(self, file_stats):
+ """Build HTML table rows for the file listing."""
+ import html
+
+ table_rows = []
+ for stat in file_stats:
+ full_path = html.escape(stat['filename'])
+ module_name = html.escape(stat['module_name'])
+ module_type = stat['module_type']
+ badge_class = f"badge-{module_type}"
+
+ # Calculate heatmap color for intensity visualization
+ intensity = stat['percentage'] / 100.0
+ r, g, b, alpha = self._calculate_intensity_color(intensity)
+ bg_color = f"rgba({r}, {g}, {b}, {alpha})"
+ bar_width = stat['percentage'] # Use actual percentage for bar width
+
+ html_file = self.file_index[stat['filename']]
+ row = f'''
+ \n'
+ f'
\n'
)
- return html_content
diff --git a/Lib/test/test_profiling/test_sampling_profiler.py b/Lib/test/test_profiling/test_sampling_profiler.py
index 59bc18b9bcf14d..d0754ffb33c900 100644
--- a/Lib/test/test_profiling/test_sampling_profiler.py
+++ b/Lib/test/test_profiling/test_sampling_profiler.py
@@ -18,6 +18,7 @@
from profiling.sampling.stack_collector import (
CollapsedStackCollector,
FlamegraphCollector,
+ HeatmapCollector,
)
from profiling.sampling.gecko_collector import GeckoCollector
@@ -668,6 +669,97 @@ def test_gecko_collector_export(self):
self.assertIn("func2", string_array)
self.assertIn("other_func", string_array)
+ def test_heatmap_collector_basic(self):
+ """Test basic HeatmapCollector functionality."""
+ collector = HeatmapCollector()
+
+ # Test empty state
+ self.assertEqual(len(collector.file_samples), 0)
+ self.assertEqual(len(collector.line_samples), 0)
+
+ # Test collecting sample data
+ test_frames = [
+ MockInterpreterInfo(
+ 0,
+ [MockThreadInfo(
+ 1,
+ [("file.py", 10, "func1"), ("file.py", 20, "func2")],
+ )]
+ )
+ ]
+ collector.collect(test_frames)
+
+ # Should have recorded samples for the file
+ self.assertGreater(len(collector.line_samples), 0)
+ self.assertIn("file.py", collector.file_samples)
+
+ # Check that line samples were recorded
+ file_data = collector.file_samples["file.py"]
+ self.assertGreater(len(file_data), 0)
+
+ def test_heatmap_collector_export(self):
+ """Test heatmap HTML export functionality."""
+ heatmap_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, heatmap_dir)
+
+ collector = HeatmapCollector()
+
+ # Create test data with multiple files
+ test_frames1 = [
+ MockInterpreterInfo(
+ 0,
+ [MockThreadInfo(1, [("file.py", 10, "func1"), ("file.py", 20, "func2")])],
+ )
+ ]
+ test_frames2 = [
+ MockInterpreterInfo(
+ 0,
+ [MockThreadInfo(1, [("file.py", 10, "func1"), ("file.py", 20, "func2")])],
+ )
+ ] # Same stack
+ test_frames3 = [
+ MockInterpreterInfo(0, [MockThreadInfo(1, [("other.py", 5, "other_func")])])
+ ]
+
+ collector.collect(test_frames1)
+ collector.collect(test_frames2)
+ collector.collect(test_frames3)
+
+ # Export heatmap
+ with (captured_stdout(), captured_stderr()):
+ collector.export(heatmap_dir)
+
+ # Verify index.html was created
+ index_path = os.path.join(heatmap_dir, "index.html")
+ self.assertTrue(os.path.exists(index_path))
+ self.assertGreater(os.path.getsize(index_path), 0)
+
+ # Check index contains HTML content
+ with open(index_path, "r", encoding="utf-8") as f:
+ content = f.read()
+
+ # Should be valid HTML
+ self.assertIn("", content.lower())
+ self.assertIn("{line_num}
\n'
+ f' {samples_display}
\n'
+ f' {line_html}
\n'
+ f' {nav_buttons_html}\n'
+ f'