Add opcode panel to live profiler TUI

New widget displays instruction-level stats for selected function when --opcodes is enabled. Navigation via j/k keys with scroll support. Adds per-thread opcode tracking. Updates pstats collector for new frame format.
python · pablogsal · Dec 11, 2025 · Dec 3, 2025 · Dec 3, 2025 · Dec 3, 2025
commit 7ffe4cb39e7db5813b1b29556d32011326a40841
@@ -11,7 +11,7 @@
 import time
 import _colorize
 
-from ..collector import Collector
+from ..collector import Collector, extract_lineno
 from ..constants import (
     THREAD_STATUS_HAS_GIL,
     THREAD_STATUS_ON_CPU,
@@ -41,7 +41,7 @@
     COLOR_PAIR_SORTED_HEADER,
 )
 from .display import CursesDisplay
-from .widgets import HeaderWidget, TableWidget, FooterWidget, HelpWidget
+from .widgets import HeaderWidget, TableWidget, FooterWidget, HelpWidget, OpcodePanel
 from .trend_tracker import TrendTracker
 
 
@@ -67,6 +67,11 @@ class ThreadData:
     sample_count: int = 0
     gc_frame_samples: int = 0
 
+    # Opcode statistics: {location: {opcode: count}}
+    opcode_stats: dict = field(default_factory=lambda: collections.defaultdict(
+        lambda: collections.defaultdict(int)
+    ))
+
     def increment_status_flag(self, status_flags):
         """Update status counts based on status bit flags."""
         if status_flags & THREAD_STATUS_HAS_GIL:
@@ -103,6 +108,7 @@ def __init__(
         pid=None,
         display=None,
         mode=None,
+        opcodes=False,
     ):
         """
         Initialize the live stats collector.
@@ -115,6 +121,7 @@ def __init__(
             pid: Process ID being profiled
             display: DisplayInterface implementation (None means curses will be used)
             mode: Profiling mode ('cpu', 'gil', etc.) - affects what stats are shown
+            opcodes: Whether to show opcode panel (requires --opcodes flag)
         """
         self.result = collections.defaultdict(
             lambda: dict(total_rec_calls=0, direct_calls=0, cumulative_calls=0)
@@ -152,6 +159,12 @@ def __init__(
         }
         self.gc_frame_samples = 0  # Track samples with GC frames
 
+        # Opcode statistics: {location: {opcode: count}}
+        self.opcode_stats = collections.defaultdict(lambda: collections.defaultdict(int))
+        self.show_opcodes = opcodes  # Show opcode panel when --opcodes flag is passed
+        self.selected_row = 0  # Currently selected row in table for opcode view
+        self.scroll_offset = 0  # Scroll offset for table when in opcode mode
+
         # Interactive controls state
         self.paused = False  # Pause UI updates (profiling continues)
         self.show_help = False  # Show help screen
@@ -178,6 +191,7 @@ def __init__(
         self.table_widget = None
         self.footer_widget = None
         self.help_widget = None
+        self.opcode_panel = None
 
         # Color mode
         self._can_colorize = _colorize.can_colorize()
@@ -282,18 +296,29 @@ def process_frames(self, frames, thread_id=None):
         thread_data = self._get_or_create_thread_data(thread_id) if thread_id is not None else None
 
         # Process each frame in the stack to track cumulative calls
+        # frame.location is (lineno, end_lineno, col_offset, end_col_offset), int, or None
         for frame in frames:
-            location = (frame.filename, frame.lineno, frame.funcname)
+            lineno = extract_lineno(frame.location)
+            location = (frame.filename, lineno, frame.funcname)
             self.result[location]["cumulative_calls"] += 1
             if thread_data:
                 thread_data.result[location]["cumulative_calls"] += 1
 
         # The top frame gets counted as an inline call (directly executing)
-        top_location = (frames[0].filename, frames[0].lineno, frames[0].funcname)
+        top_frame = frames[0]
+        top_lineno = extract_lineno(top_frame.location)
+        top_location = (top_frame.filename, top_lineno, top_frame.funcname)
         self.result[top_location]["direct_calls"] += 1
         if thread_data:
             thread_data.result[top_location]["direct_calls"] += 1
 
+        # Track opcode for top frame (the actively executing instruction)
+        opcode = getattr(top_frame, 'opcode', None)
+        if opcode is not None:
+            self.opcode_stats[top_location][opcode] += 1
+            if thread_data:
+                thread_data.opcode_stats[top_location][opcode] += 1
+
     def collect_failed_sample(self):
         self.failed_samples += 1
         self.total_samples += 1
@@ -431,6 +456,7 @@ def _initialize_widgets(self, colors):
             self.table_widget = TableWidget(self.display, colors, self)
             self.footer_widget = FooterWidget(self.display, colors, self)
             self.help_widget = HelpWidget(self.display, colors)
+            self.opcode_panel = OpcodePanel(self.display, colors, self)
 
     def _render_display_sections(
         self, height, width, elapsed, stats_list, colors
@@ -451,6 +477,12 @@ def _render_display_sections(
                 line, width, height=height, stats_list=stats_list
             )
 
+            # Render opcode panel if enabled
+            if self.show_opcodes:
+                line = self.opcode_panel.render(
+                    line, width, height=height, stats_list=stats_list
+                )
+
         except curses.error:
             pass
 
@@ -918,19 +950,148 @@ def _handle_input(self):
             if self._trend_tracker is not None:
                 self._trend_tracker.toggle()
 
-        elif ch == curses.KEY_LEFT or ch == curses.KEY_UP:
-            # Navigate to previous thread in PER_THREAD mode, or switch from ALL to PER_THREAD
+        elif ch == ord("j") or ch == ord("J"):
+            # Move selection down in opcode mode (with scrolling)
+            if self.show_opcodes:
+                # Use the actual displayed stats_list count, not raw result_source
+                # This matches what _prepare_display_data() produces
+                stats_list = self.build_stats_list()
+                if self.display:
+                    height, _ = self.display.get_dimensions()
+                    # Same calculation as _prepare_display_data
+                    extra_header = FINISHED_BANNER_EXTRA_LINES if self.finished else 0
+                    max_stats = max(0, height - HEADER_LINES - extra_header - FOOTER_LINES - SAFETY_MARGIN)
+                    stats_list = stats_list[:max_stats]
+                    visible_rows = max(1, height - 8 - 2 - 12)
+                else:
+                    visible_rows = self.limit
+                total_rows = len(stats_list)
+                if total_rows == 0:
+                    return
+                # Max scroll is when last item is at bottom
+                max_scroll = max(0, total_rows - visible_rows)
+                # Current absolute position
+                abs_pos = self.scroll_offset + self.selected_row
+                # Only move if not at the last item
+                if abs_pos < total_rows - 1:
+                    # Try to move selection within visible area first
+                    if self.selected_row < visible_rows - 1:
+                        self.selected_row += 1
+                    elif self.scroll_offset < max_scroll:
+                        # Scroll down
+                        self.scroll_offset += 1
+                # Clamp to valid range
+                self.scroll_offset = min(self.scroll_offset, max_scroll)
+                max_selected = min(visible_rows - 1, total_rows - self.scroll_offset - 1)
+                self.selected_row = min(self.selected_row, max(0, max_selected))
+
+        elif ch == ord("k") or ch == ord("K"):
+            # Move selection up in opcode mode (with scrolling)
+            if self.show_opcodes:
+                if self.selected_row > 0:
+                    self.selected_row -= 1
+                elif self.scroll_offset > 0:
+                    self.scroll_offset -= 1
+                # Clamp to valid range based on actual stats_list
+                stats_list = self.build_stats_list()
+                if self.display:
+                    height, _ = self.display.get_dimensions()
+                    extra_header = FINISHED_BANNER_EXTRA_LINES if self.finished else 0
+                    max_stats = max(0, height - HEADER_LINES - extra_header - FOOTER_LINES - SAFETY_MARGIN)
+                    stats_list = stats_list[:max_stats]
+                    visible_rows = max(1, height - 8 - 2 - 12)
+                else:
+                    visible_rows = self.limit
+                total_rows = len(stats_list)
+                if total_rows > 0:
+                    max_scroll = max(0, total_rows - visible_rows)
+                    self.scroll_offset = min(self.scroll_offset, max_scroll)
+                    max_selected = min(visible_rows - 1, total_rows - self.scroll_offset - 1)
+                    self.selected_row = min(self.selected_row, max(0, max_selected))
+
+        elif ch == curses.KEY_UP:
+            # Move selection up (same as 'k') when in opcode mode
+            if self.show_opcodes:
+                if self.selected_row > 0:
+                    self.selected_row -= 1
+                elif self.scroll_offset > 0:
+                    self.scroll_offset -= 1
+                # Clamp to valid range based on actual stats_list
+                stats_list = self.build_stats_list()
+                if self.display:
+                    height, _ = self.display.get_dimensions()
+                    extra_header = FINISHED_BANNER_EXTRA_LINES if self.finished else 0
+                    max_stats = max(0, height - HEADER_LINES - extra_header - FOOTER_LINES - SAFETY_MARGIN)
+                    stats_list = stats_list[:max_stats]
+                    visible_rows = max(1, height - 8 - 2 - 12)
+                else:
+                    visible_rows = self.limit
+                total_rows = len(stats_list)
+                if total_rows > 0:
+                    max_scroll = max(0, total_rows - visible_rows)
+                    self.scroll_offset = min(self.scroll_offset, max_scroll)
+                    max_selected = min(visible_rows - 1, total_rows - self.scroll_offset - 1)
+                    self.selected_row = min(self.selected_row, max(0, max_selected))
+            else:
+                # Navigate to previous thread (same as KEY_LEFT)
+                if len(self.thread_ids) > 0:
+                    if self.view_mode == "ALL":
+                        self.view_mode = "PER_THREAD"
+                        self.current_thread_index = len(self.thread_ids) - 1
+                    else:
+                        self.current_thread_index = (
+                            self.current_thread_index - 1
+                        ) % len(self.thread_ids)
+
+        elif ch == curses.KEY_DOWN:
+            # Move selection down (same as 'j') when in opcode mode
+            if self.show_opcodes:
+                stats_list = self.build_stats_list()
+                if self.display:
+                    height, _ = self.display.get_dimensions()
+                    extra_header = FINISHED_BANNER_EXTRA_LINES if self.finished else 0
+                    max_stats = max(0, height - HEADER_LINES - extra_header - FOOTER_LINES - SAFETY_MARGIN)
+                    stats_list = stats_list[:max_stats]
+                    visible_rows = max(1, height - 8 - 2 - 12)
+                else:
+                    visible_rows = self.limit
+                total_rows = len(stats_list)
+                if total_rows == 0:
+                    return
+                max_scroll = max(0, total_rows - visible_rows)
+                abs_pos = self.scroll_offset + self.selected_row
+                if abs_pos < total_rows - 1:
+                    if self.selected_row < visible_rows - 1:
+                        self.selected_row += 1
+                    elif self.scroll_offset < max_scroll:
+                        self.scroll_offset += 1
+                self.scroll_offset = min(self.scroll_offset, max_scroll)
+                max_selected = min(visible_rows - 1, total_rows - self.scroll_offset - 1)
+                self.selected_row = min(self.selected_row, max(0, max_selected))
+            else:
+                # Navigate to next thread (same as KEY_RIGHT)
+                if len(self.thread_ids) > 0:
+                    if self.view_mode == "ALL":
+                        self.view_mode = "PER_THREAD"
+                        self.current_thread_index = 0
+                    else:
+                        self.current_thread_index = (
+                            self.current_thread_index + 1
+                        ) % len(self.thread_ids)
+
+        elif ch == curses.KEY_LEFT:
+            # Navigate to previous thread
             if len(self.thread_ids) > 0:
                 if self.view_mode == "ALL":
                     self.view_mode = "PER_THREAD"
-                    self.current_thread_index = 0
+                    self.current_thread_index = len(self.thread_ids) - 1
                 else:
                     self.current_thread_index = (
                         self.current_thread_index - 1
                     ) % len(self.thread_ids)
 
-        elif ch == curses.KEY_RIGHT or ch == curses.KEY_DOWN:
-            # Navigate to next thread in PER_THREAD mode, or switch from ALL to PER_THREAD
+        elif ch == curses.KEY_RIGHT:
+            # Navigate to next thread
             if len(self.thread_ids) > 0:
                 if self.view_mode == "ALL":
                     self.view_mode = "PER_THREAD"

@@ -45,6 +45,9 @@
 # Finished banner display
 FINISHED_BANNER_EXTRA_LINES = 3  # Blank line + banner + blank line
 
+# Opcode panel display
+OPCODE_PANEL_HEIGHT = 12  # Height reserved for opcode statistics panel
+
 # Color pair IDs
 COLOR_PAIR_HEADER_BG = 4
 COLOR_PAIR_CYAN = 5