STM32F4-Base/scripts/ram_baseline.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RAM 预算门禁 + 趋势台账（登记簿 N-27 前半 / 审查报告 §5.5 三件套，2026-10-12）。

三件套的机械部分收在本脚本（事实源 = Keil `.map`，与 `check_mem_numbers.py` 同口径）：

  1. **92% 阈值门禁**（`--gate`，pre-commit `mem-budget` 钩子入口）：
     RW_IRAM1 / RW_CCM 任一区占比 > 92% → FAIL，并附 Top-10 RW+ZI 对象表；
     无 `.map`（未做 Keil 构建）→ 自适应跳过（与 `mem-numbers` 同口径，
     CI 盲区登记在 `docs/任务规划/测试与架构专项.md` A-5 残余）。
     `--gate` 同时校验台账受管区与 `.map` 一致（防漂移，同 `mem-numbers` 思路）。
  2. **趋势基线** `scripts/ram_baseline.tsv`（`--update` 每次 @build 后追加一行；
     `--trend` 看趋势并高亮 RW+ZI 较上一行增长 >1 KB 的对象——方案 §4.0.3
     "ZI 增量 >1KB 必须解释"的机械化）。
  3. **台账受管区**（`--write` 同步 `docs/内存管理/RAM预算台账.md` 的
     `MEM_BUDGET:BEGIN/END` 区：区域水位 + 子系统配额对照）。

子系统配额（报告 §5.5 第 3 条的规划值；RTOS 堆 2026-10-12 B5 48→56 KB）：
    网络 12 KB / 存储 12 KB / UI 16 KB / RTOS 堆 56 KB / 应用 8 KB
    ——UI 配额是 R-9/R-2 落地后的**目标态**，现状超配 = W3 债务，台账如实展示；
    配额改动 = 先登记 `docs/内存管理/RAM预算台账.md` 变更记录，再改本脚本 `QUOTAS`。

用法（仓库根）：
    python scripts/ram_baseline.py --gate
    python scripts/ram_baseline.py --write
    python scripts/ram_baseline.py --update
    python scripts/ram_baseline.py --trend [-n 10]

退出码：0 通过/跳过；1 门禁失败或受管区漂移；2 事实源解析失败。
创建日期：2026-10-12（W1-0，登记簿 N-27 前半）
"""
import argparse
import pathlib
import sys
from datetime import datetime

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
import ram_report  # noqa: E402  同目录：复用 .map 解析（regions + components，口径一致）
from check_mem_numbers import read_doc, write_doc  # noqa: E402  保真读写（BOM/CRLF）

REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
MAP_DEFAULT = REPO_ROOT / "MDK-ARM" / "STM32F407-Demo" / "STM32F407-Demo.map"
TSV_PATH = REPO_ROOT / "scripts" / "ram_baseline.tsv"
LEDGER = REPO_ROOT / "docs" / "内存管理" / "RAM预算台账.md"

BEGIN = "<!-- MEM_BUDGET:BEGIN generated by scripts/ram_baseline.py -->"
END = "<!-- MEM_BUDGET:END -->"

THRESHOLD_PCT = 92.0          # 报告 §5.5 第 1 条
GATED_REGIONS = ("RW_IRAM1", "RW_CCM")   # 门禁约束区；其余区仅展示
GROWTH_ALERT_B = 1024         # 方案 §4.0.3：增量 >1 KB 必须解释
TOP_N = 10                    # tsv 记录 / 失败附表的对象数

RTOS_HEAP = "RTOS 堆"
SUBSYSTEMS = ("网络", "存储", "UI", RTOS_HEAP, "应用", "其它")
QUOTAS = {"网络": 12 * 1024, "存储": 12 * 1024, "UI": 16 * 1024,
          RTOS_HEAP: 56 * 1024, "应用": 8 * 1024}   # RTOS 堆 48→56 KB：2026-10-12 B5（台账 §6）

# dhara 是裸对象名（nand.c/map.c/...），必须精确匹配，勿用前缀
_DHARA = {"map.o", "journal.o", "nand.o", "fctl.o", "era.o"}
# RTOS 内核静态对象：归"其它"（RTOS 配额只约束 CCM 堆本体，见台账 §2 口径表）
_KERNEL = {"tasks.o", "queue.o", "timers.o", "event_groups.o",
           "list.o", "port.o", "cmsis_os2.o", "freertos.o"}


def subsystem_of(obj):
    """对象名 → 子系统（前缀首匹配；口径表见 RAM预算台账.md §2）。"""
    n = obj.strip().lower()
    if n.startswith(("lv_", "ui_")):
        return "UI"
    if n.startswith(("ch395f", "net_", "lftpd", "ftp_task")):
        return "网络"
    if (n.startswith(("gd5f", "nand_", "ff", "diskio", "vol_lock"))
            or n in _DHARA):
        return "存储"
    if n == "heap_4.o":
        return RTOS_HEAP
    if n in _KERNEL:
        return "其它"
    if n.startswith(("app_", "adc_", "prot_", "phasor", "sample_ring",
                     "sys_", "cfg_", "dbg_", "console_log", "task_watchdog",
                     "sha256", "tpafe", "sd2506", "test_task", "time_sync",
                     "default_task")):
        return "应用"
    return "其它"


def collect(map_path):
    """解析 .map → {'regions': {name: r}, 'subsys': {name: bytes}, 'top': [(obj, B)]}。"""
    data = ram_report.parse_map(str(map_path))
    regions = {r["name"]: r for r in data["regions"]}
    comps = [c for c in data["components"]
             if not ram_report._is_aggregate(c["object"])
             and not c["object"].startswith("(incl.")]
    subsys = {s: 0 for s in SUBSYSTEMS}
    for c in comps:
        subsys[subsystem_of(c["object"])] += c["rw_data"] + c["zi_data"]
    top = sorted(((c["object"], c["rw_data"] + c["zi_data"]) for c in comps),
                 key=lambda t: (-t[1], t[0]))
    top = [t for t in top if t[1] > 0][:TOP_N]
    return {"regions": regions, "region_list": data["regions"],
            "subsys": subsys, "top": top}


def region_pct(r):
    """区占比（%）。**公开名**：check_size_gate.py 复用同一口径，禁第二套算法。"""
    return r["size"] / r["max"] * 100 if r["max"] else 0.0


def _quota_status(name, used):
    if name not in QUOTAS:
        return "参考"
    if name == RTOS_HEAP and used - QUOTAS[name] <= 64:
        # heap_4.o = 堆本体 49,152 B（= 配额）+ 32 B RW 元数据，不算超
        return "堆=配额"
    if used <= QUOTAS[name]:
        return "OK"
    return f"⚠ 超 {(used - QUOTAS[name]) / 1024:.1f} KB"


def render_budget(info):
    """渲染台账受管区（按 .map 区出现顺序动态渲染，新增区自动入表；
    门禁列只约束 GATED_REGIONS，其余区标"参考"）。"""
    L = []
    L.append("| 区域 | 用量 | 容量 | 占比 | 门禁（≤92%） |")
    L.append("|---|---|---|---|---|")
    for r in info["region_list"]:
        gated = r["name"] in GATED_REGIONS
        label = f"{r['name']} (Flash)" if r["name"] == "ER_IROM1" else r["name"]
        u = region_pct(r)
        mark = ("PASS" if u <= THRESHOLD_PCT else "FAIL") if gated else "参考"
        L.append(f"| {label} | {r['size']:,} B | {r['max']:,} B | "
                 f"{u:.1f}% | {mark} |")
    L.append("")
    L.append("| 子系统（静态 RW+ZI；除 RTOS 堆在 CCM 外均在 SRAM1） | 现值 | 配额 | 状态 |")
    L.append("|---|---|---|---|")
    for s in SUBSYSTEMS:
        used = info["subsys"][s]
        quota = f"{QUOTAS[s] // 1024} KB" if s in QUOTAS else "—"
        L.append(f"| {s} | {used:,} B | {quota} | {_quota_status(s, used)} |")
    return "\n".join(L)


# --- 受管区读写（锚点语义同 check_mem_numbers.py） ---------------------------

def extract_managed(text):
    if BEGIN not in text or END not in text:
        return None
    s = text.index(BEGIN) + len(BEGIN)
    e = text.index(END)
    return text[s:e].strip()


def replace_managed(text, block):
    s = text.index(BEGIN) + len(BEGIN)
    e = text.index(END)
    new = "\n" + block + "\n"
    if text[s:e] == new:
        return text, False
    return text[:s] + new + text[e:], True


# --- 子命令 ------------------------------------------------------------------

def _map_or_skip(args):
    map_path = pathlib.Path(args.map)
    if not map_path.exists():
        return None
    return map_path


def cmd_gate(args):
    map_path = _map_or_skip(args)
    if map_path is None:
        print("[SKIP] 无 .map（未做 Keil 构建）→ RAM 预算门禁不适用")
        print("       ⚠ **SKIP 不等于 PASS**：本钩子这一趟什么都没检查。")
        print("       A-5 残余的强制兜底（无需 Keil，改代码后必然拦红）：")
        print("         本地 scripts/ci_local.py 的 L0c 步 / CI 的「体积门禁」步，")
        print("         都跑 scripts/check_size_gate.py（同码凭证 + 92% 红线复核）。")
        print("       改过代码请先 @build，再 python scripts/ram_baseline.py --update。")
        return 0
    info = collect(map_path)
    if not info["regions"]:
        print(f"[FAIL] 未能从 {map_path.name} 解析出执行区", file=sys.stderr)
        return 2

    fails = []
    for name in ("RW_IRAM1", "RW_CCM"):
        r = info["regions"].get(name)
        if not r:
            fails.append(f"缺执行区 {name}（scatter 变更？）")
            continue
        u = region_pct(r)
        if u > THRESHOLD_PCT:
            fails.append(f"{name} 占用 {u:.1f}% > {THRESHOLD_PCT:.0f}%"
                         f"（{r['size']:,}/{r['max']:,} B）")
    if fails:
        print("[FAIL] RAM 预算超阈值：", file=sys.stderr)
        for f in fails:
            print(f"  - {f}", file=sys.stderr)
        print(f"\nTop-{TOP_N} 静态对象（RW+ZI，找增量元凶）：", file=sys.stderr)
        for obj, b in info["top"]:
            print(f"  {b:>7,} B  {obj}", file=sys.stderr)
        print("\n处置：削峰（静态化回退/缓冲复用）或按 docs/内存管理/RAM预算台账.md"
              " §5 走超配额流程；勿盲目抬 scatter。", file=sys.stderr)
        return 1

    # 台账受管区一致性（防漂移；同 mem-numbers 的"事实源唯一"思路）
    if not LEDGER.exists():
        print(f"[FAIL] 台账缺失：{LEDGER.relative_to(REPO_ROOT)}", file=sys.stderr)
        return 1
    text, newline, bom = read_doc(LEDGER)
    expected = render_budget(info)
    got = extract_managed(text)
    if got is None:
        print(f"[FAIL] 台账缺受管锚点：{LEDGER.relative_to(REPO_ROOT)}",
              file=sys.stderr)
        return 1
    if got != expected:
        print("[FAIL] 台账受管区与 .map 不一致（区域水位或子系统配额表漂移）",
              file=sys.stderr)
        print(f"事实源：{map_path.relative_to(REPO_ROOT)}", file=sys.stderr)
        print("修复：python scripts/ram_baseline.py --write", file=sys.stderr)
        return 1

    # 软提醒（不拦）：tsv 落后 / .map 比源码旧
    if TSV_PATH.exists():
        rows = read_tsv_rows()
        ir = info["regions"].get("RW_IRAM1")
        if rows and ir and int(rows[-1]["iram1"]) != ir["size"]:
            print("[WARN] ram_baseline.tsv 末行落后于当前 .map —— "
                  "@build 后请跑 `python scripts/ram_baseline.py --update`")
    else:
        print("[WARN] ram_baseline.tsv 不存在 —— 跑 "
              "`python scripts/ram_baseline.py --update` 建档")
    newest = _newest_source_mtime()
    if newest and newest > map_path.stat().st_mtime:
        print("[WARN] .map 早于最新源码——数字来自旧构建，建议先 @build 再提交")

    ir, cc = info["regions"].get("RW_IRAM1"), info["regions"].get("RW_CCM")
    print(f"RAM 预算门禁 PASS：RW_IRAM1 {region_pct(ir):.1f}% / RW_CCM {region_pct(cc):.1f}%"
          f"（阈值 {THRESHOLD_PCT:.0f}%）")
    return 0


def cmd_write(args):
    map_path = _map_or_skip(args)
    if map_path is None:
        print("[FAIL] 找不到 .map 事实源，先 @build", file=sys.stderr)
        return 1
    info = collect(map_path)
    if not LEDGER.exists():
        print(f"[FAIL] 台账不存在：{LEDGER.relative_to(REPO_ROOT)}", file=sys.stderr)
        return 1
    text, newline, bom = read_doc(LEDGER)
    if BEGIN not in text or END not in text:
        print("[FAIL] 台账缺受管锚点", file=sys.stderr)
        return 1
    new_text, did = replace_managed(text, render_budget(info))
    if did:
        write_doc(LEDGER, new_text, newline, bom)
        print("已同步台账受管区。")
    else:
        print("台账受管区已是最新。")
    return 0


# --- tsv 趋势基线 ------------------------------------------------------------

TSV_HEADER = ("date\tcode\tro\trw\tzi\ter_irom1\ter_pct\tiram1\tiram1_pct"
              "\tccm\tccm_pct\ttop_rwzi")


def read_tsv_rows():
    """tsv 全部行（dict 列表）。**公开名**：check_size_gate.py 用它做凭证互证。"""
    rows = []
    lines = TSV_PATH.read_text(encoding="utf-8").splitlines()
    if len(lines) < 2:
        return rows
    keys = lines[0].split("\t")
    for ln in lines[1:]:
        parts = ln.split("\t")
        if len(parts) == len(keys):
            rows.append(dict(zip(keys, parts)))
    return rows


def _top_field(top):
    return ";".join(f"{obj}={b}" for obj, b in top)


def cmd_update(args):
    map_path = _map_or_skip(args)
    if map_path is None:
        print("[FAIL] 找不到 .map 事实源，先 @build", file=sys.stderr)
        return 1
    info = collect(map_path)
    gt = ram_report.parse_map(str(map_path))["grand_totals"]
    if not gt:
        print("[FAIL] .map 缺 Grand Totals", file=sys.stderr)
        return 2
    today = datetime.now().strftime("%Y-%m-%d")
    reg = info["regions"]
    row = [
        today, str(gt["code"]), str(gt["ro_data"]), str(gt["rw_data"]),
        str(gt["zi_data"]),
        str(reg["ER_IROM1"]["size"]), f"{region_pct(reg['ER_IROM1']):.1f}",
        str(reg["RW_IRAM1"]["size"]), f"{region_pct(reg['RW_IRAM1']):.1f}",
        str(reg["RW_CCM"]["size"]), f"{region_pct(reg['RW_CCM']):.1f}",
        _top_field(info["top"]),
    ]
    lines = (TSV_PATH.read_text(encoding="utf-8").splitlines()
             if TSV_PATH.exists() else [])
    if not lines:
        lines = TSV_HEADER.split("\t")
        lines = [TSV_HEADER]
    body = [ln for ln in lines[1:] if ln.strip()
            and not ln.split("\t")[0] == today]
    body.append("\t".join(row))
    # newline="\n"：Windows 文本模式默认 \n→\r\n，违反 .gitattributes eol=lf
    TSV_PATH.write_text("\n".join([lines[0]] + body) + "\n",
                        encoding="utf-8", newline="\n")
    print(f"已记录 {today}：RW_IRAM1 {reg['RW_IRAM1']['size']:,} B "
          f"({region_pct(reg['RW_IRAM1']):.1f}%) / RW_CCM {reg['RW_CCM']['size']:,} B "
          f"({region_pct(reg['RW_CCM']):.1f}%) → {TSV_PATH.relative_to(REPO_ROOT)}")

    # A-5 残余（2026-10-25）：同一次「真实测量」顺带刷新**体积同码凭证**
    # scripts/size_attestation.json——这样「每次 @build 后跑 --update」这条**既有
    # 纪律**同时喂饱那条无需 Keil 也能判的强制门禁（check_size_gate --ci /
    # --local，即 ci_local 的 L0c 步与 CI 的体积门禁步），不新增人工步骤。
    # 盖戳失败只告警、不改本命令返回值：凭证新鲜度由门禁自己 fail-closed 兜，
    # 失败原因（.map 比源码旧 / 已超线）会在门禁那一侧明确报出来。
    # 为什么在函数内 import：check_size_gate 模块级 import 本模块，反过来即成环。
    try:
        from argparse import Namespace

        import check_size_gate
        check_size_gate.cmd_stamp(Namespace(map=str(map_path)))
    except Exception as exc:  # noqa: BLE001
        print(f"[WARN] 体积同码凭证未刷新：{exc}")
        print("       单独补跑：python scripts/check_size_gate.py --stamp")
    return 0


def cmd_trend(args):
    if not TSV_PATH.exists():
        print("[WARN] tsv 不存在——先跑 --update 建档")
        return 0
    rows = read_tsv_rows()
    if not rows:
        print("[WARN] tsv 为空")
        return 0
    print(f"趋势（最近 {min(args.n, len(rows))} 行，共 {len(rows)} 行）：")
    print(f"{'日期':<12} {'RW_IRAM1':>12} {'RW_CCM':>12} {'Flash':>12}")
    for r in rows[-args.n:]:
        print(f"{r['date']:<12} {int(r['iram1']):>9,} B ({r['iram1_pct']:>4}%)"
              f" {int(r['ccm']):>7,} B ({r['ccm_pct']:>4}%)"
              f" {int(r['er_irom1']):>7,} B ({r['er_pct']:>4}%)")
    if len(rows) >= 2:
        prev = dict()
        for kv in rows[-2]["top_rwzi"].split(";"):
            if "=" in kv:
                obj, b = kv.rsplit("=", 1)
                prev[obj] = int(b)
        grew = []
        for kv in rows[-1]["top_rwzi"].split(";"):
            if "=" not in kv:
                continue
            obj, b = kv.rsplit("=", 1)
            delta = int(b) - prev.get(obj, 0)
            if delta > GROWTH_ALERT_B:
                grew.append((obj, prev.get(obj, 0), int(b), delta))
        if grew:
            print(f"\n⚠ 较上一行 RW+ZI 增长 >{GROWTH_ALERT_B} B 的对象"
                  f"（提交说明须解释，方案 §4.0.3）：")
            for obj, old, new, d in sorted(grew, key=lambda t: -t[3]):
                print(f"  +{d:>6,} B  {obj}  ({old:,} → {new:,})")
        else:
            print("\nTop 对象无 >1 KB 的异常增长。")
    return 0


# --- 工具 --------------------------------------------------------------------

def _newest_source_mtime():
    newest = 0
    for base in ("App", "Drivers/BSP", "Src", "Inc"):
        d = REPO_ROOT / base
        if not d.is_dir():
            continue
        for p in d.rglob("*"):
            if p.suffix in (".c", ".h") and p.is_file():
                newest = max(newest, p.stat().st_mtime)
    lv = REPO_ROOT / "Lib" / "lv_conf.h"
    if lv.is_file():
        newest = max(newest, lv.stat().st_mtime)
    return newest


def main(argv=None):
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8", errors="replace")
        except Exception:
            pass
    ap = argparse.ArgumentParser(
        description="RAM 预算门禁 + 趋势台账（N-27 前半 / 报告 §5.5 三件套）")
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--gate", action="store_true", help="92% 阈值门禁（pre-commit 入口）")
    g.add_argument("--write", action="store_true", help="同步台账受管区")
    g.add_argument("--update", action="store_true", help="追加 tsv 趋势行（@build 后跑）")
    g.add_argument("--trend", action="store_true", help="打印趋势 + 高亮 >1 KB 增长")
    ap.add_argument("-n", type=int, default=10, help="--trend 显示行数")
    ap.add_argument("--map", default=str(MAP_DEFAULT), help=".map 路径")
    args = ap.parse_args(argv)

    if args.write:
        return cmd_write(args)
    if args.update:
        return cmd_update(args)
    if args.trend:
        return cmd_trend(args)
    return cmd_gate(args)  # 默认 = --gate


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