| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) |
| """Interactive perf list.""" |
| |
| from abc import ABC, abstractmethod |
| import argparse |
| from dataclasses import dataclass |
| import math |
| from typing import Any, Dict, Optional, Tuple |
| import perf |
| from textual import on |
| from textual.app import App, ComposeResult |
| from textual.binding import Binding |
| from textual.containers import Horizontal, HorizontalGroup, Vertical, VerticalScroll |
| from textual.css.query import NoMatches |
| from textual.command import SearchIcon |
| from textual.screen import ModalScreen |
| from textual.widgets import Button, Footer, Header, Input, Label, Sparkline, Static, Tree |
| from textual.widgets.tree import TreeNode |
| |
| |
| def get_info(info: Dict[str, str], key: str): |
| return (info[key] + "\n") if key in info else "" |
| |
| |
| class TreeValue(ABC): |
| """Abstraction for the data of value in the tree.""" |
| |
| @abstractmethod |
| def name(self) -> str: |
| pass |
| |
| @abstractmethod |
| def description(self) -> str: |
| pass |
| |
| @abstractmethod |
| def matches(self, query: str) -> bool: |
| pass |
| |
| @abstractmethod |
| def parse(self) -> perf.evlist: |
| pass |
| |
| @abstractmethod |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: |
| pass |
| |
| |
| @dataclass |
| class Metric(TreeValue): |
| """A metric in the tree.""" |
| metric_name: str |
| |
| def name(self) -> str: |
| return self.metric_name |
| |
| def description(self) -> str: |
| """Find and format metric description.""" |
| for metric in perf.metrics(): |
| if metric["MetricName"] != self.metric_name: |
| continue |
| desc = get_info(metric, "BriefDescription") |
| desc += get_info(metric, "PublicDescription") |
| desc += get_info(metric, "MetricExpr") |
| desc += get_info(metric, "MetricThreshold") |
| return desc |
| return "description" |
| |
| def matches(self, query: str) -> bool: |
| return query in self.metric_name |
| |
| def parse(self) -> perf.evlist: |
| return perf.parse_metrics(self.metric_name) |
| |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: |
| val = evlist.compute_metric(self.metric_name, cpu, thread) |
| return 0 if math.isnan(val) else val |
| |
| |
| @dataclass |
| class PmuEvent(TreeValue): |
| """A PMU and event within the tree.""" |
| pmu: str |
| event: str |
| |
| def name(self) -> str: |
| if self.event.startswith(self.pmu) or ':' in self.event: |
| return self.event |
| else: |
| return f"{self.pmu}/{self.event}/" |
| |
| def description(self) -> str: |
| """Find and format event description for {pmu}/{event}/.""" |
| for p in perf.pmus(): |
| if p.name() != self.pmu: |
| continue |
| for info in p.events(): |
| if "name" not in info or info["name"] != self.event: |
| continue |
| |
| desc = get_info(info, "topic") |
| desc += get_info(info, "event_type_desc") |
| desc += get_info(info, "desc") |
| desc += get_info(info, "long_desc") |
| desc += get_info(info, "encoding_desc") |
| return desc |
| return "description" |
| |
| def matches(self, query: str) -> bool: |
| return query in self.pmu or query in self.event |
| |
| def parse(self) -> perf.evlist: |
| return perf.parse_events(self.name()) |
| |
| def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float: |
| return evsel.read(cpu, thread).val |
| |
| |
| class ErrorScreen(ModalScreen[bool]): |
| """Pop up dialog for errors.""" |
| |
| CSS = """ |
| ErrorScreen { |
| align: center middle; |
| } |
| """ |
| |
| def __init__(self, error: str): |
| self.error = error |
| super().__init__() |
| |
| def compose(self) -> ComposeResult: |
| yield Button(f"Error: {self.error}", variant="primary", id="error") |
| |
| def on_button_pressed(self, event: Button.Pressed) -> None: |
| self.dismiss(True) |
| |
| |
| class SearchScreen(ModalScreen[str]): |
| """Pop up dialog for search.""" |
| |
| CSS = """ |
| SearchScreen Horizontal { |
| align: center middle; |
| margin-top: 1; |
| } |
| SearchScreen Input { |
| width: 1fr; |
| } |
| """ |
| |
| def compose(self) -> ComposeResult: |
| yield Horizontal(SearchIcon(), Input(placeholder="Event name")) |
| |
| def on_input_submitted(self, event: Input.Submitted) -> None: |
| """Handle the user pressing Enter in the input field.""" |
| self.dismiss(event.value) |
| |
| |
| class Counter(HorizontalGroup): |
| """Two labels for a CPU and its counter value.""" |
| |
| CSS = """ |
| Label { |
| gutter: 1; |
| } |
| """ |
| |
| def __init__(self, cpu: int) -> None: |
| self.cpu = cpu |
| super().__init__() |
| |
| def compose(self) -> ComposeResult: |
| label = f"cpu{self.cpu}" if self.cpu >= 0 else "total" |
| yield Label(label + " ") |
| yield Label("0", id=f"counter_{label}") |
| |
| |
| class CounterSparkline(HorizontalGroup): |
| """A Sparkline for a performance counter.""" |
| |
| def __init__(self, cpu: int) -> None: |
| self.cpu = cpu |
| super().__init__() |
| |
| def compose(self) -> ComposeResult: |
| label = f"cpu{self.cpu}" if self.cpu >= 0 else "total" |
| yield Label(label) |
| yield Sparkline([], summary_function=max, id=f"sparkline_{label}") |
| |
| |
| class IListApp(App): |
| TITLE = "Interactive Perf List" |
| |
| BINDINGS = [ |
| Binding(key="s", action="search", description="Search", |
| tooltip="Search events and PMUs"), |
| Binding(key="n", action="next", description="Next", |
| tooltip="Next search result or item"), |
| Binding(key="p", action="prev", description="Previous", |
| tooltip="Previous search result or item"), |
| Binding(key="c", action="collapse", description="Collapse", |
| tooltip="Collapse the current PMU"), |
| Binding(key="^q", action="quit", description="Quit", |
| tooltip="Quit the app"), |
| ] |
| |
| CSS = """ |
| /* Make the 'total' sparkline a different color. */ |
| #sparkline_total > .sparkline--min-color { |
| color: $accent; |
| } |
| #sparkline_total > .sparkline--max-color { |
| color: $accent 30%; |
| } |
| /* |
| * Make the active_search initially not displayed with the text in |
| * the middle of the line. |
| */ |
| #active_search { |
| display: none; |
| width: 100%; |
| text-align: center; |
| } |
| """ |
| |
| def __init__(self, interval: float) -> None: |
| self.interval = interval |
| self.evlist = None |
| self.selected: Optional[TreeValue] = None |
| self.search_results: list[TreeNode[TreeValue]] = [] |
| self.cur_search_result: TreeNode[TreeValue] | None = None |
| super().__init__() |
| |
| def expand_and_select(self, node: TreeNode[Any]) -> None: |
| """Expand select a node in the tree.""" |
| if node.parent: |
| node.parent.expand() |
| if node.parent.parent: |
| node.parent.parent.expand() |
| node.expand() |
| node.tree.select_node(node) |
| node.tree.scroll_to_node(node) |
| |
| def set_searched_tree_node(self, previous: bool) -> None: |
| """Set the cur_search_result node to either the next or previous.""" |
| l = len(self.search_results) |
| |
| if l < 1: |
| tree: Tree[TreeValue] = self.query_one("#root", Tree) |
| if previous: |
| tree.action_cursor_up() |
| else: |
| tree.action_cursor_down() |
| return |
| |
| if self.cur_search_result: |
| idx = self.search_results.index(self.cur_search_result) |
| if previous: |
| idx = idx - 1 if idx > 0 else l - 1 |
| else: |
| idx = idx + 1 if idx < l - 1 else 0 |
| else: |
| idx = l - 1 if previous else 0 |
| |
| node = self.search_results[idx] |
| if node == self.cur_search_result: |
| return |
| |
| self.cur_search_result = node |
| self.expand_and_select(node) |
| |
| def action_search(self) -> None: |
| """Search was chosen.""" |
| def set_initial_focus(event: str | None) -> None: |
| """Sets the focus after the SearchScreen is dismissed.""" |
| |
| search_label = self.query_one("#active_search", Label) |
| search_label.display = True if event else False |
| if not event: |
| return |
| event = event.lower() |
| search_label.update(f'Searching for events matching "{event}"') |
| |
| tree: Tree[str] = self.query_one("#root", Tree) |
| |
| def find_search_results(event: str, node: TreeNode[str], |
| cursor_seen: bool = False, |
| match_after_cursor: Optional[TreeNode[str]] = None |
| ) -> Tuple[bool, Optional[TreeNode[str]]]: |
| """Find nodes that match the search remembering the one after the cursor.""" |
| if not cursor_seen and node == tree.cursor_node: |
| cursor_seen = True |
| if node.data and node.data.matches(event): |
| if cursor_seen and not match_after_cursor: |
| match_after_cursor = node |
| self.search_results.append(node) |
| |
| if node.children: |
| for child in node.children: |
| (cursor_seen, match_after_cursor) = \ |
| find_search_results(event, child, cursor_seen, match_after_cursor) |
| return (cursor_seen, match_after_cursor) |
| |
| self.search_results.clear() |
| (_, self.cur_search_result) = find_search_results(event, tree.root) |
| if len(self.search_results) < 1: |
| self.push_screen(ErrorScreen(f"Failed to find pmu/event or metric {event}")) |
| search_label.display = False |
| elif self.cur_search_result: |
| self.expand_and_select(self.cur_search_result) |
| else: |
| self.set_searched_tree_node(previous=False) |
| |
| self.push_screen(SearchScreen(), set_initial_focus) |
| |
| def action_next(self) -> None: |
| """Next was chosen.""" |
| self.set_searched_tree_node(previous=False) |
| |
| def action_prev(self) -> None: |
| """Previous was chosen.""" |
| self.set_searched_tree_node(previous=True) |
| |
| def action_collapse(self) -> None: |
| """Collapse the part of the tree currently on.""" |
| tree: Tree[str] = self.query_one("#root", Tree) |
| node = tree.cursor_node |
| if node and node.parent: |
| node.parent.collapse_all() |
| node.tree.scroll_to_node(node.parent) |
| |
| def update_counts(self) -> None: |
| """Called every interval to update counts.""" |
| if not self.selected or not self.evlist: |
| return |
| |
| def update_count(cpu: int, count: int): |
| # Update the raw count display. |
| counter: Label = self.query(f"#counter_cpu{cpu}" if cpu >= 0 else "#counter_total") |
| if not counter: |
| return |
| counter = counter.first(Label) |
| counter.update(str(count)) |
| |
| # Update the sparkline. |
| line: Sparkline = self.query(f"#sparkline_cpu{cpu}" if cpu >= 0 else "#sparkline_total") |
| if not line: |
| return |
| line = line.first(Sparkline) |
| # If there are more events than the width, remove the front event. |
| if len(line.data) > line.size.width: |
| line.data.pop(0) |
| line.data.append(count) |
| line.mutate_reactive(Sparkline.data) |
| |
| # Update the total and each CPU counts, assume there's just 1 evsel. |
| total = 0 |
| self.evlist.disable() |
| for evsel in self.evlist: |
| for cpu in evsel.cpus(): |
| aggr = 0 |
| for thread in evsel.threads(): |
| aggr += self.selected.value(self.evlist, evsel, cpu, thread) |
| update_count(cpu, aggr) |
| total += aggr |
| update_count(-1, total) |
| self.evlist.enable() |
| |
| def on_mount(self) -> None: |
| """When App starts set up periodic event updating.""" |
| self.update_counts() |
| self.set_interval(self.interval, self.update_counts) |
| |
| def set_selected(self, value: TreeValue) -> None: |
| """Updates the event/description and starts the counters.""" |
| try: |
| label_name = self.query_one("#event_name", Label) |
| event_description = self.query_one("#event_description", Static) |
| lines = self.query_one("#lines") |
| except NoMatches: |
| # A race with rendering, ignore the update as we can't |
| # mount the assumed output widgets. |
| return |
| |
| self.selected = value |
| |
| # Remove previous event information. |
| if self.evlist: |
| self.evlist.disable() |
| self.evlist.close() |
| old_lines = self.query(CounterSparkline) |
| for line in old_lines: |
| line.remove() |
| old_counters = self.query(Counter) |
| for counter in old_counters: |
| counter.remove() |
| |
| # Update event/metric text and description. |
| label_name.update(value.name()) |
| event_description.update(value.description()) |
| |
| # Open the event. |
| try: |
| self.evlist = value.parse() |
| if self.evlist: |
| self.evlist.open() |
| self.evlist.enable() |
| except: |
| self.evlist = None |
| |
| if not self.evlist: |
| self.push_screen(ErrorScreen(f"Failed to open {value.name()}")) |
| return |
| |
| # Add spark lines for all the CPUs. Note, must be done after |
| # open so that the evlist CPUs have been computed by propagate |
| # maps. |
| line = CounterSparkline(cpu=-1) |
| lines.mount(line) |
| for cpu in self.evlist.all_cpus(): |
| line = CounterSparkline(cpu) |
| lines.mount(line) |
| line = Counter(cpu=-1) |
| lines.mount(line) |
| for cpu in self.evlist.all_cpus(): |
| line = Counter(cpu) |
| lines.mount(line) |
| |
| def compose(self) -> ComposeResult: |
| """Draws the app.""" |
| def metric_event_tree() -> Tree: |
| """Create tree of PMUs and metricgroups with events or metrics under.""" |
| tree: Tree[TreeValue] = Tree("Root", id="root") |
| pmus = tree.root.add("PMUs") |
| for pmu in perf.pmus(): |
| pmu_name = pmu.name().lower() |
| pmu_node = pmus.add(pmu_name) |
| try: |
| for event in sorted(pmu.events(), key=lambda x: x["name"]): |
| if "name" in event: |
| e = event["name"].lower() |
| if "alias" in event: |
| pmu_node.add_leaf(f'{e} ({event["alias"]})', |
| data=PmuEvent(pmu_name, e)) |
| else: |
| pmu_node.add_leaf(e, data=PmuEvent(pmu_name, e)) |
| except: |
| # Reading events may fail with EPERM, ignore. |
| pass |
| metrics = tree.root.add("Metrics") |
| groups = set() |
| for metric in perf.metrics(): |
| groups.update(metric["MetricGroup"]) |
| |
| def add_metrics_to_tree(node: TreeNode[TreeValue], parent: str): |
| for metric in sorted(perf.metrics(), key=lambda x: x["MetricName"]): |
| if parent in metric["MetricGroup"]: |
| name = metric["MetricName"] |
| node.add_leaf(name, data=Metric(name)) |
| child_group_name = f'{name}_group' |
| if child_group_name in groups: |
| add_metrics_to_tree(node.add(child_group_name), child_group_name) |
| |
| for group in sorted(groups): |
| if group.endswith('_group'): |
| continue |
| add_metrics_to_tree(metrics.add(group), group) |
| |
| tree.root.expand() |
| return tree |
| |
| yield Header(id="header") |
| yield Horizontal(Vertical(metric_event_tree(), id="events"), |
| Vertical(Label("event name", id="event_name"), |
| Static("description", markup=False, id="event_description"), |
| )) |
| yield Label(id="active_search") |
| yield VerticalScroll(id="lines") |
| yield Footer(id="footer") |
| |
| @on(Tree.NodeSelected) |
| def on_tree_node_selected(self, event: Tree.NodeSelected[TreeValue]) -> None: |
| """Called when a tree node is selected, selecting the event.""" |
| if event.node.data: |
| self.set_selected(event.node.data) |
| |
| |
| if __name__ == "__main__": |
| ap = argparse.ArgumentParser() |
| ap.add_argument('-I', '--interval', help="Counter update interval in seconds", default=0.1) |
| args = ap.parse_args() |
| app = IListApp(float(args.interval)) |
| app.run() |
