STM32F4-Base/scripts/stack_report.py

#!/usr/bin/env python3
"""Parse the Keil static call graph (*.htm) and report per-task stack budgets.

UV4 / armlink writes "MDK-ARM/STM32F407-Demo/STM32F407-Demo.htm" on every build
("Static Call Graph"). Every function entry carries:

    <STRONG><a name="[7a]"></a>StartDefaultTask</STRONG> (Thumb, 508 bytes, Stack size 560 bytes, ...)
    [Stack]<UL><LI>Max Depth = 2776<LI>Call Chain = StartDefaultTask => ... => SPI_WaitFlagStateUntilTimeout</UL>

i.e. the function's own frame ("Stack size N bytes") plus the worst-case
cumulative depth of its whole subtree ("Max Depth = N") and the chain that
reaches it. That is a build-time stack observability source: no target needed.

This tool turns it into:
  1. a per-task budget table (static worst-case depth vs configured stack),
  2. the worst call chain of every task,
  3. a frame-size ranking (which functions are the width hogs),
  4. optionally, runtime high-water-mark data overlaid from a serial log.

Runtime lines are produced by stack_probe_poll() in Src/freertos.c:
    [STACK] <task> peak_used=<B>/<cfg> free_min=<B> (<pct>%)
    [STACK] heap free=<B> min_ever=<B>

Usage:
    python scripts/stack_report.py [-i HTML] [-s SERIAL_LOG] [-o OUTPUT]
                                   [--isr-margin 1024] [--min-free-pct 30]

Exit code: 0 = all tasks pass, 1 = at least one task fails, 2 = parse error.
"""
import argparse
import os
import pathlib
import re
import sys

REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
HTM_DEFAULT = REPO_ROOT / "MDK-ARM" / "STM32F407-Demo" / "STM32F407-Demo.htm"
FREERTOS_C = REPO_ROOT / "Src" / "freertos.c"

# Entry function of each task -> FreeRTOS task name (attribute prefix in freertos.c).
ROOTS = {
    "StartDefaultTask": "defaultTask",
    "StartNetTask": "netTask",
    "StartFtpTask": "ftpTask",
    "StartUiTask": "uiTask",
    "start_test_task": "testTask",
    "time_sync_task": "timeSyncTask",
    "adc_task_func": "adcTask",
    "iec104_serv_task": "iec104Task",
    "protect_task": "protectTask",
}

# Runtime lines use the short labels emitted by stack_probe_poll() in freertos.c.
RUNTIME_ALIAS = {
    "defaultTask": "default",
    "netTask": "net",
    "ftpTask": "ftp",
    "uiTask": "ui",
    "testTask": "test",
    "timeSyncTask": "timesync",
    "adcTask": "adc",
    "iec104Task": "iec104",
    "protectTask": "protect",
}

FUNC_RE = re.compile(
    r'<STRONG><a name="\[[0-9a-f]+\]"></a>([^<]+)</STRONG>\s*\(([^)]*)\)', re.I)
FRAME_RE = re.compile(r'Stack size\s+(\d+)\s+bytes', re.I)
DEPTH_RE = re.compile(r'Max Depth\s*=\s*(\d+)', re.I)
CHAIN_RE = re.compile(r'Call Chain\s*=\s*(.*?)</UL>', re.I | re.S)
UPDATED_RE = re.compile(r'Last Updated:\s*(.*?)$', re.M)
MAXSTACK_RE = re.compile(r'Maximum Stack Usage\s*=\s*(\d+)\s*bytes', re.I)
ATTR_RE = re.compile(r'osThreadAttr_t\s+(\w+?)_attributes\s*=\s*\{(.*?)\};', re.S)
STACKSIZE_RE = re.compile(r'\.stack_size\s*=\s*(\d+)\s*(?:\*\s*(\d+))?')
RUNTIME_RE = re.compile(
    r'\[STACK\]\s+(\w+)\s+peak_used=(\d+)/(\d+)\s+free_min=(\d+)\s+\((\d+)%\)')
HEAP_RE = re.compile(r'\[STACK\]\s+heap\s+free=(\d+)\s+min_ever=(\d+)')
TAG_RE = re.compile(r'<[^>]+>')


def _clean(text):
    """Strip HTML markup/entities from a chain fragment."""
    text = TAG_RE.sub('', text)
    text = text.replace('&rArr;', '->').replace('&gt;', '>').replace('&lt;', '<')
    text = text.replace('&nbsp;', ' ').replace('&amp;', '&')
    return ' '.join(text.split())


def parse_callgraph(path):
    """Return (meta, functions); functions maps name -> frame/depth/chain bytes."""
    text = pathlib.Path(path).read_text(encoding='utf-8', errors='replace')
    meta = {"max_stack": 0, "updated": ""}
    m = MAXSTACK_RE.search(text)
    if m:
        meta["max_stack"] = int(m.group(1))
    m = UPDATED_RE.search(text)
    if m:
        meta["updated"] = m.group(1).strip()

    functions = {}
    matches = list(FUNC_RE.finditer(text))
    for i, hit in enumerate(matches):
        name = hit.group(1).strip()
        head = hit.group(2)
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        body = text[hit.end():end]
        fm = FRAME_RE.search(head)
        dm = DEPTH_RE.search(body)
        cm = CHAIN_RE.search(body)
        frame = int(fm.group(1)) if fm else 0
        depth = int(dm.group(1)) if dm else frame
        chain = _clean(cm.group(1)) if cm else name
        # Same name may appear once per object; keep the deepest entry.
        old = functions.get(name)
        if old is None or depth > old["depth"]:
            functions[name] = {"frame": frame, "depth": depth, "chain": chain}
    return meta, functions


def parse_task_stacks(path):
    """Parse configured stack sizes (bytes) from Src/freertos.c attributes."""
    text = pathlib.Path(path).read_text(encoding='utf-8', errors='replace')
    stacks = {}
    for attr_name, body in ATTR_RE.findall(text):
        m = STACKSIZE_RE.search(body)
        if not m:
            continue
        size = int(m.group(1)) * (int(m.group(2)) if m.group(2) else 1)
        stacks[attr_name] = size
    return stacks


def parse_runtime(path):
    """Extract the last [STACK] block from a serial log: task -> peak/free."""
    text = pathlib.Path(path).read_text(encoding='utf-8', errors='replace')
    tasks = {}
    for m in RUNTIME_RE.finditer(text):
        tasks[m.group(1)] = {
            "peak": int(m.group(2)), "cfg": int(m.group(3)),
            "free": int(m.group(4)), "pct": int(m.group(5)),
        }
    heap = None
    hits = list(HEAP_RE.finditer(text))
    if hits:
        heap = {"free": int(hits[-1].group(1)), "min_ever": int(hits[-1].group(2))}
    return tasks, heap


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument('-i', '--html', default=str(HTM_DEFAULT),
                    help='static call graph HTML (default: MDK-ARM output)')
    ap.add_argument('-s', '--serial', default=None,
                    help='serial log with [STACK] lines (optional, runtime overlay)')
    ap.add_argument('-o', '--output', default=None, help='write markdown report here')
    ap.add_argument('--isr-margin', type=int, default=1024,
                    help='bytes reserved for ISR nesting (default 1024)')
    ap.add_argument('--min-free-pct', type=int, default=30,
                    help='required free margin in %% of the stack (default 30)')
    args = ap.parse_args()

    html_path = pathlib.Path(args.html)
    if not html_path.is_file():
        print(f'ERROR: call graph HTML not found: {html_path}', file=sys.stderr)
        print('       build first: .venv\\Scripts\\python.exe .dsh\\skills\\keil-build\\scripts\\keil_build.py', file=sys.stderr)
        return 2

    meta, functions = parse_callgraph(html_path)
    stacks = parse_task_stacks(FREERTOS_C)
    runtime, heap = ({}, None)
    if args.serial:
        runtime, heap = parse_runtime(args.serial)

    out = []
    add = out.append
    add('# Stack budget report')
    add('')
    # 显示用相对路径：传相对 -i 时 relative_to 会抛 ValueError（2026-10-22 实测踩到），
    # 且绝不把本机绝对路径写进入库的 data/ 快照。
    def _disp(pth):
        try:
            return pathlib.Path(pth).resolve().relative_to(REPO_ROOT)
        except (ValueError, OSError):
            return pathlib.Path(pth).name

    add(f'- Source: `{_disp(html_path)}` (Keil static call graph)')
    if meta["updated"]:
        add(f'- Linker run: {meta["updated"]}')
    add(f'- Global maximum stack usage: **{meta["max_stack"]} B** (any call tree)')
    if runtime:
        add(f'- Runtime watermark: {len(runtime)} task(s) from 串口采集日志 `{_disp(args.serial)}`')
    add('')
    add('Rule: `margin = cfg - max(static_max_depth, runtime_peak)` must be '
        f'>= max({args.isr_margin} B ISR, {args.min_free_pct}% of cfg).')
    add('')
    add('| Task | Entry | Cfg (B) | Static max (B) | Runtime peak (B) | Worst (B) | Margin (B) | Margin % | Verdict |')
    add('|---|---|---|---|---|---|---|---|---|')

    failures = 0
    rows = []
    for entry, task in ROOTS.items():
        fn = functions.get(entry)
        cfg = stacks.get(task, 0)
        if fn is None or cfg == 0:
            rows.append((task, entry, cfg, None, None, None, None, None,
                         'N/A (not in image)'))
            continue
        depth = fn["depth"]
        rt = runtime.get(task) or runtime.get(RUNTIME_ALIAS.get(task, ""))
        rt_peak = rt["peak"] if rt else 0
        # The verdict must use the worse of the two sources: the static graph is an
        # upper bound for straight-line code but a LOWER bound once indirect calls /
        # recursion are involved, and only the runtime watermark covers paths that
        # actually ran (e.g. ftpTask's file-transfer loop).
        worst = max(depth, rt_peak)
        margin = cfg - worst
        pct = (margin * 100) // cfg if cfg else 0
        need = max(args.isr_margin, (cfg * args.min_free_pct) // 100)
        ok = margin >= need
        if not ok:
            failures += 1
        rows.append((task, entry, cfg, depth, rt_peak if rt else None, worst,
                     margin, pct, 'PASS' if ok else 'FAIL'))
    for (task, entry, cfg, depth, rt_peak, worst, margin, pct, verdict) in rows:
        if depth is None:
            add(f'| {task} | `{entry}` | {cfg or "-"} | - | {rt_peak if rt_peak else "-"} '
                f'| - | - | - | {verdict} |')
        else:
            add(f'| {task} | `{entry}` | {cfg} | {depth} | {rt_peak if rt_peak else "-"} '
                f'| {worst} | {margin} | {pct}% | {verdict} |')
    add('')
    add('`Static max` = armlink worst-case call-tree depth; `Runtime peak` = measured '
        'high-water mark (boot + whatever workload ran before the sample); '
        '`Worst` = max of the two, which is what the verdict is computed from.')
    add('')

    if heap:
        add(f'Runtime heap: free={heap["free"]} B, min_ever={heap["min_ever"]} B')
        add('')

    add('## Worst call chain per task')
    add('')
    for entry, task in ROOTS.items():
        fn = functions.get(entry)
        if fn is None:
            continue
        add(f'### {task}: {fn["depth"]} B (own frame {fn["frame"]} B)')
        add('')
        add(f'```\n{fn["chain"]}\n```')
        add('')

    add('## Frame-size ranking (own frame, top 15)')
    add('')
    add('| Bytes | Function |')
    add('|---|---|')
    for name, info in sorted(functions.items(), key=lambda kv: -kv[1]["frame"])[:15]:
        add(f'| {info["frame"]} | `{name}` |')
    add('')

    report = '\n'.join(out)
    if args.output:
        # newline="\n"：避免 Windows 换行翻译污染 data/ 快照的 LF 行尾（与 .gitattributes 一致）。
        pathlib.Path(args.output).write_text(report, encoding='utf-8', newline='\n')
        print(f'written: {args.output}')
    else:
        print(report)

    return 1 if failures else 0


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