path: root/.claude/skills/etm-trace/scripts/etm_profile.py

#!/usr/bin/env python3
"""Analyze an etm_capture.py output dir: hot functions, coverage, itrace digest,
ISR episode timing, and optimization hints.

Input:  code_profile.txt (Ozone Export.CodeProfile text report), and optionally
        itrace.csv (Export.Trace raw instruction history) with --elf for
        address->function mapping (arm-none-eabi-nm).
Output: markdown report on stdout.

--isr SYM[,SYM..]  episode timing for an interrupt handler: first symbol is the
        entry anchor (its first instruction marks each ISR entry), all symbols
        form the body address set. Example: --isr OTG_HS_IRQHandler,dcd_int_handler
        Durations are calibrated against SysTick_Handler beats (1 ms apart), so
        the itrace must be captured with timestamps enabled.
"""

import argparse
import bisect
import csv
import os
import re
import statistics
import subprocess
import sys


def parse_profile(path):
    """Parse the two report sections. Function rows are indented 2 spaces under
    a non-indented module row; columns are '|'-separated."""
    lines = open(path, errors="replace").read().splitlines()
    try:
        cov_start = lines.index("Code Coverage Summary")
        prof_start = lines.index("Code Profile Summary")
    except ValueError:
        sys.exit(f"error: {path} is not an Ozone code-profile text report")

    def rows(section):
        module = None
        for ln in section:
            if "|" not in ln:
                continue
            cells = ln.split("|")
            name = cells[0].rstrip()
            if (not name or name.startswith("Module/Function")
                    or set(name.strip()) <= {"-", "+"}):
                continue
            if not name.startswith("  "):
                module = name.strip()
                continue
            yield module, name.strip(), cells[1:]

    def num(s):
        s = s.strip().replace(" ", "")
        return int(s) if s else 0

    cov_pat = re.compile(r"^\s*([\d ]+)/\s*([\d ]+)\s+([\d.]+)%")
    funcs, totals = {}, {}
    for module, name, cells in rows(lines[prof_start:]):
        run, fetch = (num(cells[0]) if cells else 0), (num(cells[1]) if len(cells) > 1 else 0)
        if name == "Total":
            totals["run"], totals["fetch"] = run, fetch
        elif name == "[Unaccounted]":
            totals["unaccounted"] = fetch
        elif name in funcs and funcs[name]["module"] != module:
            # same-named static from another module: keep both rows distinct
            funcs[f"{name} [{module}]"] = {"module": module, "run": run,
                                           "fetch": fetch}
        else:
            funcs[name] = {"module": module, "run": run, "fetch": fetch}
    for module, name, cells in rows(lines[cov_start:prof_start]):
        m_src = cov_pat.match(cells[0]) if cells else None
        m_inst = cov_pat.match(cells[1]) if len(cells) > 1 else None
        if name == "Total" and m_inst and m_src:
            totals["src_cov"] = num(m_src.group(1)), num(m_src.group(2))
            totals["inst_cov"] = num(m_inst.group(1)), num(m_inst.group(2))
        elif m_inst:
            # match the de-collided key when a same-named static from another
            # module was renamed during the profile pass
            key = name if (name in funcs and funcs[name]["module"] == module) \
                else f"{name} [{module}]"
            if key in funcs:
                funcs[key]["inst_pct"] = float(m_inst.group(3))
    return funcs, totals


def load_symbols(elf):
    """Sorted (addr, size, name) from nm; for mapping itrace addresses."""
    nm = "arm-none-eabi-nm"
    out = subprocess.run([nm, "-S", "--defined-only", "-C", elf],
                         capture_output=True, text=True)
    if out.returncode != 0:
        sys.exit(f"error: {nm} failed on {elf}: {out.stderr.strip()}")
    syms = []
    for ln in out.stdout.splitlines():
        parts = ln.split(maxsplit=3)
        if len(parts) == 4 and parts[2].lower() in ("t", "w"):
            syms.append((int(parts[0], 16) & ~1, int(parts[1], 16), parts[3]))
        elif len(parts) == 3 and parts[1].lower() in ("t", "w"):
            # sizeless symbol (e.g. weak asm stub): assume a 2-byte body
            syms.append((int(parts[0], 16) & ~1, 2, parts[2]))
    return sorted(syms)


def addr_to_func(syms, addr):
    i = bisect.bisect_right(syms, (addr, 1 << 62, "")) - 1
    if i >= 0 and syms[i][0] <= addr < syms[i][0] + syms[i][1]:
        return syms[i][2]
    return None


def sym_ranges(syms, names):
    """(lo, hi) address ranges for the named symbols (base name match);
    None if any symbol is missing (caller degrades gracefully)."""
    out = []
    for want in names:
        for a, sz, n in syms:
            if n == want or n.split("(")[0] == want:
                out.append((a, a + sz))
                break
        else:
            print(f"note: symbol '{want}' not found in ELF")
            return None
    return out


def iter_itrace(path):
    """Yield (t_raw, addr) chronologically-reversed (file order: newest first).
    Also returns the timestamp unit from the header via generator .send? No -
    caller reads unit separately with itrace_unit()."""
    with open(path, newline="", errors="replace") as f:
        rd = csv.reader(f)
        hdr = next(rd, None) or []
        # --no-timestamps captures drop the Timestamp column entirely: locate
        # the Address column from the header and yield t=None for such rows
        # (consumers count instructions but skip time math)
        has_ts = any("Timestamp" in c for c in hdr)
        try:
            addr_i = next(i for i, c in enumerate(hdr) if "Address" in c)
        except StopIteration:
            addr_i = 1 if has_ts else 0
        for row in rd:
            if not row or len(row) <= addr_i:
                continue
            try:
                addr = int(row[addr_i], 16)
            except ValueError:
                continue
            if has_ts and row[0] != "PC":
                try:
                    yield float(row[0]), addr
                    continue
                except ValueError:
                    pass
            yield None, addr


def itrace_unit(path):
    hdr = open(path, errors="replace").readline()
    m = re.search(r"Timestamp\[([^\]]+)\]", hdr)
    return m.group(1) if m else "?"


def time_by_func(path, syms):
    """Per-function raw-time and instruction attribution from the itrace.
    Rows are newest-first; the gap to the next (older) row is attributed to the
    older instruction's function. Outlier gaps (trace-block boundaries) are
    capped so one discontinuity cannot skew a function. Shares are scale-free."""
    sample = []
    t_prev = None
    for t, _ in iter_itrace(path):
        if t is None:
            continue
        if t_prev is not None and t_prev - t > 0:
            sample.append(t_prev - t)
            if len(sample) >= 200000:
                break
        t_prev = t
    cap = 10000 * statistics.median(sample) if sample else float("inf")
    t_prev = None
    tf, cf = {}, {}
    n = 0
    for t, a in iter_itrace(path):
        n += 1
        fn = addr_to_func(syms, a)
        if fn:
            cf[fn] = cf.get(fn, 0) + 1
        if t is None:
            continue
        if t_prev is not None:
            d = t_prev - t
            if 0 < d < cap and fn:
                tf[fn] = tf.get(fn, 0) + d
        t_prev = t
    return tf, cf, n


def isr_report(itrace, elf, isr_arg, top):
    syms = load_symbols(elf)
    names = [s.strip() for s in isr_arg.split(",") if s.strip()]
    body = sym_ranges(syms, names)
    tick = sym_ranges(syms, ["SysTick_Handler"])
    if not body or not tick:
        print("\n## ISR timing: skipped (missing symbols above - needs the ISR "
              "symbol(s) and SysTick_Handler for calibration)")
        return
    entry = body[0][0]
    unit = itrace_unit(itrace)

    usb_rows, tick_rows, tmin, tmax = [], [], None, None
    for t, a in iter_itrace(itrace):
        if t is None:
            continue  # --no-timestamps capture: the <20-rows message below applies
        tmin = t if tmin is None else min(tmin, t)
        tmax = t if tmax is None else max(tmax, t)
        if any(lo <= a < hi for lo, hi in body):
            usb_rows.append((t, a))
        # independent, not elif: when the ISR under test IS SysTick_Handler,
        # its rows must still feed the calibration
        if any(lo <= a < hi for lo, hi in tick):
            tick_rows.append((t, a))
    usb_rows.reverse()
    tick_rows.reverse()
    if len(tick_rows) < 20:
        print(f"\n## ISR timing: not enough SysTick beats for calibration "
              f"({len(tick_rows)} rows) - capture with timestamps enabled")
        return

    # rough raw-units-per-1ms from the large mode of consecutive tick deltas;
    # threshold from the median, not the max - one trace-overflow gap would
    # otherwise inflate the cut and leave only outliers in the sample
    deltas = [b[0] - a[0] for a, b in zip(tick_rows, tick_rows[1:])]
    big = [d for d in deltas if d > 10 * statistics.median(deltas)]
    raw_ms = statistics.median(big) if big else statistics.median(deltas)
    gap = 0.03 * raw_ms       # 30 us in raw units
    edge = 0.05 * raw_ms

    def episodes(rows, g):
        out, cur = [], []
        for t, a in rows:
            if cur and t - cur[-1][0] > g:
                out.append(cur)
                cur = []
            cur.append((t, a))
        if cur:
            out.append(cur)
        return out

    starts = [e[0][0] for e in episodes(tick_rows, 0.1 * raw_ms)]

    def local_scale(t):
        i = bisect.bisect_left(starts, t)
        if 0 < i < len(starts):
            sp = starts[i] - starts[i - 1]
            if 0 < sp < 3 * raw_ms:
                return 1e-3 / sp
        return 1e-3 / raw_ms

    eps = []
    for cluster in episodes(usb_rows, gap):
        cur = None
        for t, a in cluster:
            if a == entry:
                if cur:
                    eps.append(cur)
                cur = [(t, a)]
            elif cur:
                cur.append((t, a))
        if cur:
            eps.append(cur)
    eps = [e for e in eps if e[0][0] - tmin > edge and tmax - e[-1][0] > edge
          and len(e) >= 10]
    print(f"\n## ISR timing: {names[0]} (+{len(names) - 1} body syms), "
          f"unit '{unit}', {len(starts)} SysTick beats")
    if not eps:
        print("- no complete episodes in window (wrong symbols? window missed "
              "the traffic phase?)")
        return
    stats = sorted(((e[-1][0] - e[0][0]) * local_scale(e[0][0]), len(e), e[0][0])
                   for e in eps)
    cal = [s[0] for s in stats]
    print(f"- episodes: {len(cal)}  |  fastest {min(cal) * 1e6:.2f} us, "
          f"median {statistics.median(cal) * 1e6:.2f} us, "
          f"avg {statistics.mean(cal) * 1e6:.2f} us, "
          f"worst {max(cal) * 1e6:.2f} us")
    insts = [s[1] for s in stats]
    print(f"- instructions/episode: min {min(insts)}, avg "
          f"{statistics.mean(insts):.0f}, max {max(insts)}")
    print("- worst episodes (duration, instructions, raw start):")
    for c, n, t0 in stats[-5:][::-1]:
        print(f"  - {c * 1e6:8.2f} us  {n:5d} instr  t={t0:.6f}")
    print("- caveats: timestamps are interpolated between packets (sub-us "
          "values are approximate); episodes may merge if trace overflow "
          "dropped an entry")


def short(name):
    return re.sub(r"\(.*\)$", "()", name)


def branch_bias(insts_csv, funcs, top):
    """One-sided conditional branches in executed functions (profile_insts.csv):
    a conditional fetched N times but taken/executed 0 or N times = a branch
    that never varied -> hot always-true assert, dead path, or an invariant
    that could hoist out of a loop (UM08025 SS5.19)."""
    def n(v):
        v = (v or "").replace(" ", "")
        return int(v) if v.lstrip("-").isdigit() else 0
    biased = []
    with open(insts_csv, newline="", errors="replace") as f:
        for r in csv.DictReader(f):
            if (r.get("Is Conditional") or "0").strip() != "1":
                continue
            fetched = n(r.get("Times Fetched"))
            executed = n(r.get("Times Executed"))
            if fetched < 1000:            # only hot conditionals matter
                continue
            if executed == 0 or executed == fetched:
                biased.append((fetched, r.get("Function", "?"),
                               r.get("Address", ""), r.get("AsmCode", "").strip(),
                               "always-taken" if executed == fetched else "never-taken"))
    if not biased:
        return
    biased.sort(reverse=True)
    print(f"\n## One-sided hot branches (profile_insts.csv)\n")
    print("Conditionals that never varied - candidates to hoist/remove "
          "(UM08025 §5.19):")
    for fetched, fn, addr, asm, kind in biased[:top]:
        print(f"- `{short(fn)}` @{addr} {kind} ({fetched:,}x): `{asm[:50]}`")


def suggestions(funcs, totals, tshare, cshare, exclude):
    """Rule-based optimization hints (after UM08025 §5.19)."""
    print("\n## Optimization hints")
    total_fetch = totals.get("fetch") or 1
    ex = [n for n in funcs if exclude and re.search(exclude, n)]
    ex_fetch = sum(funcs[n]["fetch"] for n in ex)
    if ex:
        print(f"- excluded from load ranking (--exclude): {len(ex)} functions, "
              f"{100.0 * ex_fetch / total_fetch:.1f}% of fetches")
    rest = {n: v for n, v in funcs.items() if n not in ex}
    rt = sum(v["fetch"] for v in rest.values()) or 1
    hot = sorted(rest.items(), key=lambda kv: -kv[1]["fetch"])[:5]
    print("- top load after exclusion: " + ", ".join(
        f"`{short(n)}` {100.0 * v['fetch'] / rt:.1f}%" for n, v in hot))
    polls = [(n, v) for n, v in rest.items()
             if v["fetch"] / total_fetch > 0.02 and v["run"]
             and v["fetch"] / v["run"] < 40]
    if polls:
        print("- busy-poll candidates (>2% load, <40 instr/entry - called in a "
              "tight loop; consider event-driven or rate-limiting):")
        for n, v in sorted(polls, key=lambda kv: -kv[1]["fetch"]):
            print(f"  - `{short(n)}`: {v['run']:,} calls, "
                  f"{v['fetch'] / v['run']:.0f} instr/call, "
                  f"{100.0 * v['fetch'] / total_fetch:.1f}% load")
    if tshare:
        stalls = []
        for n, ts in tshare.items():
            cs = cshare.get(n, 0)
            if ts > 0.01 and cs and ts / cs > 4:
                stalls.append((n, ts, ts / cs))
        if stalls:
            print("- stall/wait-dominated (time share >> instruction share - "
                  "waiting on hardware, consider DMA/IRQ instead of polling):")
            for n, ts, ratio in sorted(stalls, key=lambda x: -x[1])[:5]:
                print(f"  - `{short(n)}`: {100 * ts:.1f}% of time, "
                      f"{ratio:.0f}x its instruction share")


def main():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("capture_dir", help="etm_capture.py output dir")
    p.add_argument("--top", type=int, default=10, help="table size")
    p.add_argument("--elf", help="firmware ELF, enables itrace analyses")
    p.add_argument("--isr", help="entry[,body..] symbols for ISR episode timing")
    p.add_argument("--exclude", help="regex of functions to exclude from the "
                   "optimization load ranking (e.g. the idle poll loop)")
    args = p.parse_args()

    profile = os.path.join(args.capture_dir, "code_profile.txt")
    itrace = os.path.join(args.capture_dir, "itrace.csv")
    funcs, totals = parse_profile(profile)
    total_fetch = totals.get("fetch") or 1
    hot = sorted(funcs.items(), key=lambda kv: kv[1]["fetch"], reverse=True)[:args.top]

    print(f"# ETM profile: {args.capture_dir}\n")
    print(f"## Top {args.top} hottest functions (instruction-fetch share)\n")
    print("| # | Function | Module | Run Count | Fetch Count | Load % |")
    print("|---|----------|--------|-----------|-------------|--------|")
    for i, (name, v) in enumerate(hot, 1):
        print(f"| {i} | `{short(name)}` | {v['module']} | {v['run']:,} "
              f"| {v['fetch']:,} | {100.0 * v['fetch'] / total_fetch:.2f}% |")
    print(f"\nTotal fetches {totals.get('fetch', 0):,} "
          f"(runs {totals.get('run', 0):,}, unaccounted {totals.get('unaccounted', 0):,})\n")

    ic, sc = totals.get("inst_cov"), totals.get("src_cov")
    print("## Coverage (NOPs excluded)\n")
    if ic:
        print(f"- instructions fully executed: {ic[0]:,} / {ic[1]:,} "
              f"({100.0 * ic[0] / ic[1]:.1f}%)")
    if sc:
        print(f"- source lines fully covered:  {sc[0]:,} / {sc[1]:,} "
              f"({100.0 * sc[0] / sc[1]:.1f}%)")
    partial = [n for n, v in funcs.items()
               if v["fetch"] > 0 and 0 < v.get("inst_pct", 100) < 100]
    print(f"- executed but only partially covered: {len(partial)} functions")
    dead = sorted(n for n, v in funcs.items()
                  if v["fetch"] == 0 and "(always inlined)" not in n)
    print(f"- never-executed out-of-line functions: {len(dead)}")
    by_mod = {}
    for n in dead:
        by_mod.setdefault(funcs[n]["module"], []).append(short(n))
    for mod in sorted(by_mod, key=str):
        print(f"  - {mod}: {', '.join('`%s`' % f for f in by_mod[mod])}")

    tshare, cshare = {}, {}
    if os.path.isfile(itrace) and args.elf:
        syms = load_symbols(args.elf)
        tf, cf, n = time_by_func(itrace, syms)
        tt, tc = sum(tf.values()) or 1, sum(cf.values()) or 1
        tshare = {k: v / tt for k, v in tf.items()}
        cshare = {k: v / tc for k, v in cf.items()}
        unit = itrace_unit(itrace)
        print(f"\n## Instruction history (itrace.csv, unit '{unit}')\n")
        if not tf:
            print(f"- {n:,} instructions, NO timestamps (--no-timestamps "
                  f"capture): time shares unavailable, top {args.top} by "
                  f"instruction share:")
            for name, cs in sorted(cshare.items(), key=lambda kv: -kv[1])[:args.top]:
                print(f"  - `{short(name)}`: {100 * cs:.1f}% instructions")
        else:
            print(f"- {n:,} instructions; top {args.top} by TIME share "
                  f"(vs instruction share):")
            for name, ts in sorted(tshare.items(), key=lambda kv: -kv[1])[:args.top]:
                print(f"  - `{short(name)}`: {100 * ts:.1f}% time, "
                      f"{100 * cshare.get(name, 0):.1f}% instructions")

    lines_csv = os.path.join(args.capture_dir, "profile_lines.csv")
    if os.path.isfile(lines_csv):
        def n(v):  # Ozone groups thousands with spaces
            v = (v or "").replace(" ", "")
            return int(v) if v.isdigit() else 0
        with open(lines_csv, newline="", errors="replace") as f:
            rows = [r for r in csv.DictReader(f)
                    if r.get("File") and n(r.get("Instructions Fetched")) > 0]
        rows.sort(key=lambda r: -n(r["Instructions Fetched"]))
        print(f"\n## Hottest source lines (profile_lines.csv)\n")
        for r in rows[:args.top]:
            src = (r.get("Content") or "").strip()
            print(f"- {os.path.basename(r['File'])}:{r['Line']}  "
                  f"{n(r['Instructions Fetched']):,} fetches  `{src[:60]}`")

    insts_csv = os.path.join(args.capture_dir, "profile_insts.csv")
    if os.path.isfile(insts_csv):
        branch_bias(insts_csv, funcs, args.top)

    if args.isr:
        if not (os.path.isfile(itrace) and args.elf):
            sys.exit("error: --isr needs itrace.csv (--trace-csv capture) and --elf")
        isr_report(itrace, args.elf, args.isr, args.top)

    suggestions(funcs, totals, tshare, cshare, args.exclude)
    return 0


if __name__ == "__main__":
    sys.exit(main())