wandering1
593c5d0c84
用户裁定（2026-10-28）：L1/L1.5/L2/L2.5/L3 这套层号「谁能看懂」，改用语义名。 映射（同一批定稿，含 L4/L5）： L1 → 语义取数与有效性 L1.5 → 标定折算 L2 → 全周傅氏相量 L2.5 → 测频与遥测快照 L3 → 保护判据 L4 → 出口与记录 L5 → 动作留痕 实测规模：384 站点 / 62 文件（App/algo、App/task、test/host 六个套件、docs/算法、 产品需求、架构总览、启动顺序、ADC 采样节拍、IEC104 上行、采集短板评测、内存管理、 AGENTS.md、README.md、docs/README.md、.pre-commit-config.yaml 与 scripts 里的注释）。 改法（不是全局替换，逐模式有上下文规则）： - 先吃同位语再落裸词：L1 有效性校验 → 语义取数与有效性校验、L2 相量层 → 全周傅氏 相量层、L3 保护 → 保护判据、L2.5 遥测快照 → 测频与遥测快照，避免「保护判据保护判据」 这类重复（第一版就是这么坏掉的，已整体回滚重做）。 - .c/.h 只改注释：先把字符串字面量切出去（引号数为奇数的跨行注释按整行处理）， C 字符串里的 L2-in ring 之类原样保留——AGENTS 铁律「字符串字面量必须纯 ASCII」。 - 层内条目号（L1-3 / L2-3 这类）**本批不动**，属批次 2；另外三族（UI 菜单层级 L0~L3、链路延迟分段 L1~L8、CubeMX 防丢失三层）也不动，属批次 3。 - 顺手修的措辞：范围写法 L1~L3 → 「分层（语义取数与有效性 → 保护判据）」、 CI 段「不含 L1/L2/L3」→「不含采集链分层」、目录地图里同一处并列表述。 验证：@build 0 错 0 警（Code=236456，与改前逐字节同尺寸——纯注释改动）； 内存布局配图已按新脚本重生成（自检通过）；doc-links --all --sections-strict 通过 （195 文档）；mojibake / bom-guard / coding-standard / 新编号门禁自查 全 rc=0； git diff --check 干净。
293 lines
11 KiB
Python
293 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
FreeRTOS 任务表一致性工具。
|
||
|
||
背景
|
||
----
|
||
`README.md` 与 `AGENTS.md` 各有一张 FreeRTOS 任务表（任务名 / 栈大小 / 优先级）。
|
||
历史上这两张表全靠手写，`Src/freertos.c` 改栈后表格忘记同步——2026-09 就出现过
|
||
两张表都比源码落后一大截（写 2048/4096/4096/4096/16384/2048，实际 1024...512）。
|
||
本脚本以 `Src/freertos.c` 为**唯一事实源**，生成 / 校验这两张表。
|
||
|
||
事实源
|
||
------
|
||
解析 `Src/freertos.c` 里形如：
|
||
|
||
const osThreadAttr_t netTask_attributes = {
|
||
.name = "netTask",
|
||
.stack_size = 1024 * 4, /* 注释 */
|
||
.priority = (osPriority_t) osPriorityNormal,
|
||
};
|
||
|
||
的块，抽出 name / stack_words / priority。`stack_size` 一律按 `N * 4`
|
||
（CMSIS-RTOS2 的 stack_size 单位是字节，工程约定写成 `words * 4`）换算为 words。
|
||
|
||
用法
|
||
----
|
||
python scripts/gen_task_table.py # 检查模式（默认）：不一致即退出码 1
|
||
python scripts/gen_task_table.py --check # 同上
|
||
python scripts/gen_task_table.py --write # 就地重写 README/AGENTS 中的任务表
|
||
python scripts/gen_task_table.py --diff # 只打印期望表格，不写文件
|
||
|
||
pre-commit 用法：`python scripts/gen_task_table.py --check`，
|
||
任一被跟踪文档里的任务表与 `Src/freertos.c` 不一致即拦截。
|
||
|
||
退出码：0 通过；1 不一致 / 解析失败。
|
||
"""
|
||
import argparse
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
|
||
FREERTOS_C = REPO_ROOT / "Src" / "freertos.c"
|
||
|
||
# 需要保持同步的文档及其任务表锚点
|
||
# anchor_begin / anchor_end 之间即为「受管表格区」，脚本只替换这一段，
|
||
# 表格外的说明文字（含各文档自己的职责描述/口径注记）原样保留。
|
||
TARGETS = [
|
||
{
|
||
"path": REPO_ROOT / "README.md",
|
||
"begin": "<!-- TASK_TABLE:BEGIN generated by scripts/gen_task_table.py -->",
|
||
"end": "<!-- TASK_TABLE:END -->",
|
||
"header": "| 任务 | 栈大小 | 优先级 | 职责 |\n"
|
||
"|------|--------|--------|------|",
|
||
},
|
||
{
|
||
"path": REPO_ROOT / "AGENTS.md",
|
||
"begin": "<!-- TASK_TABLE:BEGIN generated by scripts/gen_task_table.py -->",
|
||
"end": "<!-- TASK_TABLE:END -->",
|
||
"header": "| Task | 栈 (words) | 优先级 |\n|---|---|---|",
|
||
},
|
||
]
|
||
|
||
# 文档中的展示顺序（与 README/AGENTS 既有一致；不在表内的任务按源码顺序追加）。
|
||
# 不用源码顺序：freertos.c 里 timeSyncTask 定义在最前，但文档按「主任务→辅助任务」排列。
|
||
DISPLAY_ORDER = ["defaultTask", "netTask", "ftpTask", "testTask", "uiTask",
|
||
"timeSyncTask", "adcTask", "iec104Task", "protectTask"]
|
||
|
||
# 各任务在文档里的「职责」描述（README 需要，AGENTS 不需要）。
|
||
# 只影响展示文案，不参与一致性判定；未知任务回落为空。
|
||
ROLE = {
|
||
"defaultTask": "FTL + FatFS 挂载、配置引擎属主、syslog 唯一写者",
|
||
"netTask": "网络初始化 + CH395F 轮询（消费 NET_RECONFIG 按代次重建监听）",
|
||
"ftpTask": "FTP 服务（等待网络+存储就绪）",
|
||
"testTask": "硬件测试入口",
|
||
"uiTask": "LVGL UI 界面",
|
||
"timeSyncTask": "墙钟时间同步",
|
||
"adcTask": "TPAFE5160 采样（TIM3 1kHz 事件驱动 + 原子快照 + STM32 ADC DMA，纯消费者）",
|
||
"iec104Task": "IEC 60870-5-104 上行（:2404 监听，遥信/遥测/定值区）",
|
||
"protectTask": "保护判据 20ms 节拍（保护判据元件 + 采样故障闭锁，B5）",
|
||
}
|
||
|
||
|
||
def sort_tasks(tasks):
|
||
"""按 DISPLAY_ORDER 排序；未列出的任务保持源码相对顺序排在最后。"""
|
||
idx = {n: i for i, n in enumerate(DISPLAY_ORDER)}
|
||
return sorted(tasks, key=lambda t: (idx.get(t[0], len(idx)),))
|
||
|
||
# 任务块：const osThreadAttr_t <id>_attributes = { ... };
|
||
BLOCK_RE = re.compile(
|
||
r"const\s+osThreadAttr_t\s+(?P<id>\w+)_attributes\s*=\s*\{(?P<body>.*?)\};",
|
||
re.S,
|
||
)
|
||
STACK_RE = re.compile(r"\.stack_size\s*=\s*(?P<expr>[^,]+),")
|
||
PRIO_RE = re.compile(r"\.priority\s*=\s*\(osPriority_t\)\s*(?P<prio>\w+)")
|
||
|
||
# 优先级枚举 → 文档里使用的短名（去掉 osPriority 前缀）
|
||
PRIO_SHORT = lambda p: p[len("osPriority"):] if p.startswith("osPriority") else p
|
||
|
||
|
||
def parse_tasks(text: str):
|
||
"""从 freertos.c 文本解析任务列表，按出现顺序返回。
|
||
|
||
stack_size 允许以下写法（工程约定）：
|
||
1024 * 4 -> 1024 words
|
||
(1024 * 4) -> 1024 words
|
||
4096 -> 1024 words（按 bytes/4 折算，并给出提示）
|
||
返回 [(name, words, prio, raw_expr), ...]
|
||
"""
|
||
tasks = []
|
||
for m in BLOCK_RE.finditer(text):
|
||
body = m.group("body")
|
||
sm = STACK_RE.search(body)
|
||
pm = PRIO_RE.search(body)
|
||
if not (sm and pm):
|
||
continue
|
||
expr = sm.group("expr").strip()
|
||
name_lit = re.search(r'\.name\s*=\s*"([^"]+)"', body)
|
||
name = name_lit.group(1) if name_lit else m.group("id")
|
||
|
||
# 去掉注释后求值：`1024 * 4` -> 4096 (bytes)
|
||
clean = re.sub(r"/\*.*?\*/", "", expr, flags=re.S).strip()
|
||
clean = clean.strip("()").strip()
|
||
try:
|
||
num = eval(clean, {"__builtins__": {}}, {}) # noqa: S307 (受控字面量)
|
||
except Exception:
|
||
print(f"[WARN] 无法解析 {name} 的 stack_size 表达式：{expr!r}",
|
||
file=sys.stderr)
|
||
continue
|
||
|
||
# 工程约定 = words*4，故 words = 字节/4
|
||
words = num // 4 if num % 4 == 0 else num
|
||
tasks.append((name, words, PRIO_SHORT(pm.group("prio")), expr))
|
||
return tasks
|
||
|
||
|
||
def render_table(tasks, header: str, with_role: bool) -> str:
|
||
lines = [header]
|
||
for name, words, prio, _expr in tasks:
|
||
if with_role:
|
||
lines.append(f"| {name} | {words} words | {prio} | {ROLE.get(name, '')} |")
|
||
else:
|
||
lines.append(f"| {name} | {words} | {prio} |")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def replace_managed(text: str, target: dict, table: str):
|
||
"""替换 <!-- TASK_TABLE:BEGIN --> ... <!-- TASK_TABLE:END --> 之间的内容。
|
||
|
||
返回 (新文本, 是否变化)。若锚点不存在则原样返回（由调用方决定是否补锚点）。
|
||
"""
|
||
b, e = target["begin"], target["end"]
|
||
if b not in text or e not in text:
|
||
return text, False
|
||
start = text.index(b) + len(b)
|
||
end = text.index(e)
|
||
old = text[start:end]
|
||
new = "\n" + table + "\n"
|
||
if old == new:
|
||
return text, False
|
||
return text[:start] + new + text[end:], True
|
||
|
||
|
||
def extract_managed(text: str, target: dict):
|
||
b, e = target["begin"], target["end"]
|
||
if b not in text or e not in text:
|
||
return None
|
||
start = text.index(b) + len(b)
|
||
end = text.index(e)
|
||
return text[start:end].strip()
|
||
|
||
|
||
def read_doc(path: pathlib.Path):
|
||
"""读文档，返回 (文本, 原换行符, 是否有 BOM)。
|
||
|
||
本仓库 core.autocrlf=true 且无 .gitattributes，工作区换行混杂
|
||
（README=LF、AGENTS=CRLF）；带 BOM 文件（App/cfg/cfg_file.c 等）另有铁律。
|
||
--write 必须**按原样回写**，否则整文件行尾/BOM 漂移 → 巨型无意义 diff。
|
||
"""
|
||
raw = path.read_bytes()
|
||
bom = raw.startswith(b"\xef\xbb\xbf")
|
||
body = raw[3:] if bom else raw
|
||
crlf = body.count(b"\r\n")
|
||
lf_only = body.count(b"\n") - crlf
|
||
newline = "\r\n" if crlf > lf_only else "\n"
|
||
text = body.decode("utf-8", errors="replace")
|
||
# 统一成 \n 供内部处理
|
||
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
return text, newline, bom
|
||
|
||
|
||
def write_doc(path: pathlib.Path, text: str, newline: str, bom: bool):
|
||
"""按原换行符 / BOM 状态回写（text 内部为 \n）。"""
|
||
out = text if newline == "\n" else text.replace("\n", newline)
|
||
data = out.encode("utf-8")
|
||
if bom:
|
||
data = b"\xef\xbb\xbf" + data
|
||
path.write_bytes(data)
|
||
|
||
|
||
def main(argv):
|
||
ap = argparse.ArgumentParser(description="FreeRTOS 任务表生成/校验")
|
||
g = ap.add_mutually_exclusive_group()
|
||
g.add_argument("--check", action="store_true", help="检查模式（默认）")
|
||
g.add_argument("--write", action="store_true", help="就地重写受管表格区")
|
||
g.add_argument("--diff", action="store_true", help="只打印期望表格")
|
||
args = ap.parse_args(argv[1:])
|
||
|
||
if not FREERTOS_C.exists():
|
||
print(f"[FAIL] 找不到事实源：{FREERTOS_C}", file=sys.stderr)
|
||
return 1
|
||
c_text, _nl, _bom = read_doc(FREERTOS_C)
|
||
tasks = parse_tasks(c_text)
|
||
if not tasks:
|
||
print("[FAIL] 未能从 Src/freertos.c 解析出任何任务，请检查正则与源码格式。",
|
||
file=sys.stderr)
|
||
return 1
|
||
tasks = sort_tasks(tasks)
|
||
|
||
if args.diff:
|
||
for t in TARGETS:
|
||
with_role = t["path"].name == "README.md"
|
||
print(f"=== {t['path'].name} ===")
|
||
print(render_table(tasks, t["header"], with_role))
|
||
print()
|
||
return 0
|
||
|
||
changed, missing, problems = [], [], []
|
||
for t in TARGETS:
|
||
if not t["path"].exists():
|
||
problems.append(f"{t['path'].name} 不存在")
|
||
continue
|
||
text, newline, bom = read_doc(t["path"])
|
||
with_role = t["path"].name == "README.md"
|
||
table = render_table(tasks, t["header"], with_role)
|
||
|
||
if args.write:
|
||
new_text, did = replace_managed(text, t, table)
|
||
if did:
|
||
write_doc(t["path"], new_text, newline, bom)
|
||
changed.append(t["path"].name)
|
||
else:
|
||
got = extract_managed(text, t)
|
||
if got is None:
|
||
missing.append(t["path"].name)
|
||
else:
|
||
# 提取区已归一为 \n；与期望表格比较时同样归一
|
||
if got.replace("\r\n", "\n") != table.strip():
|
||
changed.append(t["path"].name)
|
||
|
||
if args.write:
|
||
if missing:
|
||
print("[WARN] 以下文档缺少受管锚点，未写入："
|
||
+ ", ".join(missing))
|
||
print(f" 请在任务表上下各加一行锚点：\n"
|
||
f" {TARGETS[0]['begin']}\n {TARGETS[0]['end']}")
|
||
if changed:
|
||
print("已更新任务表：" + ", ".join(changed))
|
||
if not changed and not missing:
|
||
print("任务表已是最新，无需改动。")
|
||
return 0
|
||
|
||
if problems:
|
||
for p in problems:
|
||
print(f"[FAIL] {p}", file=sys.stderr)
|
||
return 1
|
||
|
||
if missing or changed:
|
||
if missing:
|
||
print(f"[FAIL] 以下文档缺少受管锚点 "
|
||
f"（{TARGETS[0]['begin']} / {TARGETS[0]['end']}）：")
|
||
for n in missing:
|
||
print(f" - {n}")
|
||
if changed:
|
||
print(f"[FAIL] 以下文档的任务表与 Src/freertos.c 不一致：")
|
||
for n in changed:
|
||
print(f" - {n}")
|
||
print("\n事实源（Src/freertos.c）：")
|
||
for name, words, prio, _ in tasks:
|
||
print(f" {name}: {words} words / {prio}")
|
||
print("\n期望表格（README 口径）：")
|
||
print(render_table(tasks, TARGETS[0]["header"], True))
|
||
print("\n修复：运行 `python scripts/gen_task_table.py --write`")
|
||
return 1
|
||
|
||
print(f"任务表一致（{len(tasks)} 个任务，"
|
||
f"{len(TARGETS)} 个文档）。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv))
|