wandering1
90db528032
W1-0 台账三件套：scripts/ram_baseline.py（四模式）+ ram_baseline.tsv 趋势 + docs/内存管理/RAM预算台账.md + pre-commit mem-budget 钩子（RW_IRAM1/RW_CCM >92% 即 FAIL + Top-10 附表）；审查报告 §5.5 三条闭环。 B4 IEC 60870-5-104 上行（被控站，TCP 2404）：App/protocol/iec104/ 三模块 （asdu 编解码 / pts 点表+数据源 vtable / link APCI+FSM，纯逻辑零 I/O）+ App/task/iec104_serv.c（iec104Task，NET_READY 屏障 100ms 分片喂位）。 82 断言 L0.5 全绿 + iec104-unit-test 钩子（第 31 钩子）； 设计说明 docs/子系统设计/IEC104上行设计说明.md 全量回填。 B5 protectTask（L3 板端接线）：App/task/protect_task.c——20ms 判据节拍 prot_elm_run + L5-4 采样故障闭锁（adc_health 电平位→元件复位）+ L5-1 定值 校验和巡检；新增 prot_elm_request_reset() 收口跨上下文复归（104 YK11 落请求、 eval 头部消费）；定值/元件初始化移入 app_main_init（内核前）。TC-PE-010 （prot_elm 28 用例）。 门禁与内存：TASK_WD_PROTECT=8 入 required_mask（产品 0x77→0x1F7、HW 层 0x1F5，看门狗单测 39 断言同步）；FreeRTOS 堆 48→56KB（.ioc L1 参数 + FreeRTOSConfig.h 同步，RAM预算台账 §6 登记配额 48→56KB）。 ZI 144,008→152,204（+8,196 B = 堆 +8KB，B5 扩容）；Code=204,348。 stack_report：protectTask 512 words 静态链 872 B/余 57%、iec104Task 640 words 1456 B/余 43% 均 PASS；RW_IRAM1 73.1% / RW_CCM 87.9%（<92%）。 验证：@build 0 错 0 警；ci_local 全绿（31 钩子 + 13 主机侧单测）； doc-links 135 文档 PASS。登记簿 B4/B5 销项、W1 波次完成；保护规划/看门狗 说明/AGENTS（9 任务）/RAM 三文档全部同步。 遗留（验收窗口，非代码空洞）：台架第三方 104 主站往返（送检窗口）、 标准源注入动作判据、@test PROT 锚点 + protect 栈水位 + 新堆 min_ever 复测。
408 lines
16 KiB
Python
408 lines
16 KiB
Python
#!/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())
|