#!/usr/bin/env python3# SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)"""Interactive perf list.""" from abc import ABC, abstractmethodimport argparsefrom dataclasses import dataclassimport mathfrom typing import Any, Dict, Optional, Tupleimport perffrom textual import onfrom textual.app import App, ComposeResultfrom textual.binding import Bindingfrom textual.containers import Horizontal, HorizontalGroup, Vertical, VerticalScrollfrom textual.css.query import NoMatchesfrom textual.command import SearchIconfrom textual.screen import ModalScreenfrom textual.widgets import Button, Footer, Header, Input, Label, Sparkline, Static, Treefrom 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  @dataclassclass Metric(TreeValue):    """A metric in the tree."""    metric_name: str    metric_pmu: 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            if self.metric_pmu and metric["PMU"] != self.metric_pmu:                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, self.metric_pmu)     def value(self, evlist: perf.evlist, evsel: perf.evsel, cpu: int, thread: int) -> float:        try:            val = evlist.compute_metric(self.metric_name, cpu, thread)            return 0 if math.isnan(val) else val        except:            # Be tolerant of failures to compute metrics on particular CPUs/threads.            return 0  @dataclassclass 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 "deprecated" in event:                            continue                        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, pmu: str = None):                for metric in sorted(perf.metrics(), key=lambda x: x["MetricName"]):                    metric_pmu = metric.get('PMU')                    if pmu and metric_pmu and metric_pmu != pmu:                        continue                    if parent in metric["MetricGroup"]:                        name = metric["MetricName"]                        display_name = name                        if metric_pmu:                            display_name += f" ({metric_pmu})"                        node.add_leaf(display_name, data=Metric(name, metric_pmu))                        child_group_name = f'{name}_group'                        if child_group_name in groups:                            display_child_group_name = child_group_name                            if metric_pmu:                                display_child_group_name += f" ({metric_pmu})"                            add_metrics_to_tree(node.add(display_child_group_name),                                                child_group_name,                                                metric_pmu)             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")

Lines 1–500 of 515. Load more