Files
STM32F4-Base/scripts/ram_baseline.py
T
wandering1 d3107f374f docs: 待处理内容.md 解耦为 docs/任务规划/ 文件夹(README 排序权威 + 6 个职责单一文件)
- rename_ref.py 搬迁:git mv + 全库引用改写 33 文件 82 处
- README.md 保留 §0 编号空间 / §1 未闭环一览 / §2 推进顺序权威 / §3 维护约定
- 拆出 6 文件:W1-采集保护收尾(任务卡五字段)、排期详述、暂缓-RS485、
  登记簿、测试与架构专项、基线与待确认
- 外部引用重定位 ~50 处:裸「登记簿 §X」、§5.1→基线与待确认 §2.1、
  §2.4→暂缓-RS485、§6→基线与待确认 §3、脚本注释 T-x/N-x→登记簿
- 新增并行写纪律:任务分支不改本文件夹,状态列由 Lead 主工作树更新
- 验证:doc-links 141 文档全绿;mem-numbers 一致;refgraph 重生成;
  @build 0 错误 0 警告;新文件无 BOM / LF / 零乱码
2026-10-12 13:42:47 +08:00

408 lines
16 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 的对象——方案 §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 # 方案 §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 _util(r):
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 = _util(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(" (CI 盲区已登记 docs/任务规划/测试与架构专项.md A-5 残余;"
"本地改过代码请先 @build)")
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 = _util(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()
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 {_util(ir):.1f}% / RW_CCM {_util(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 = []
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"{_util(reg['ER_IROM1']):.1f}",
str(reg["RW_IRAM1"]["size"]), f"{_util(reg['RW_IRAM1']):.1f}",
str(reg["RW_CCM"]["size"]), f"{_util(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"({_util(reg['RW_IRAM1']):.1f}%) / RW_CCM {reg['RW_CCM']['size']:,} B "
f"({_util(reg['RW_CCM']):.1f}%) → {TSV_PATH.relative_to(REPO_ROOT)}")
return 0
def cmd_trend(args):
if not TSV_PATH.exists():
print("[WARN] tsv 不存在——先跑 --update 建档")
return 0
rows = _read_tsv()
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"(提交说明须解释,方案 §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())