STM32F4-Base/scripts/gen_task_table.py

#!/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 节拍（L3 元件 + 采样故障闭锁，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))