guanlan-src/scripts/chain_gap_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""六层链"缺料"检查器（2026-09-17 用户令 #2 的收口件）。

作用只有一个：**把"还缺什么"变成一条命令**，而不是靠人记着哪四个脚本没随包。

    python scripts/chain_gap_check.py            # 人看
    python scripts/chain_gap_check.py --check    # 只出结论（缺料 → 退出码 5）

判据（全部是"文件在不在"这种可核事实，不含猜测）：

  A. 四步脚本是否在位 —— 壳 `src/windcms/pipeline.py::analyze` 会调它们；
  B. 两件"标准答案"是否在位 —— `model_run_l6.parquet` / `fusion_38.csv`（逐值对拍的前提）；
  C. 台账里这一族还剩多少"✗ 无生成端"（接料后应显著下降）。

取料单：`docs/向振动线取料单_v0.1.md`；接口规格：`docs/振动六层链_接口规格与缺口_v0.1.md`。
"""
from __future__ import annotations

import argparse
import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / 'scripts'))
from src import paths as P                                              # noqa: E402

# 壳里写死的四步脚本（改壳才会变；这里只是把"缺什么"列清楚）
STEPS = [
    ('oem_scan', 'scripts/rudong_tcm_oem_scan.py', '特征频率扫描（轴承/齿轮/保持架/叶片 1P）'),
    ('energy_share', 'scripts/rudong_line_energy_share.py', '线能量占比'),
    ('model_run', 'scripts/rudong_model_run.py', '六层模型出件（model_run_l6.parquet + 摘要）'),
    ('fusion', 'scripts/rudong_fusion_run.py', '融合面出件（fusion_38.csv 等）'),
]
# 逐值对拍用的"标准答案"（随包件里有，但两件关键件从未随过包）
SAMPLES = [
    ('model_run_l6.parquet', 'model_run 步的标准答案'),
    ('fusion_38.csv', 'fusion 步的标准答案'),
]


def main() -> int:
    ap = argparse.ArgumentParser(description='六层链缺料检查')
    ap.add_argument('--check', action='store_true', help='只出结论（缺料 → rc=5）')
    ap.add_argument('--farm', default=None)
    a = ap.parse_args()
    farm = a.farm or P.farm()
    m5 = P.m5(farm)

    missing_scripts = [s for s in STEPS if not (ROOT / s[1]).is_file()]
    have_fusion_impl = (ROOT / 'scripts/rudong_fusion_run.py').is_file()
    missing_samples = [s for s in SAMPLES if not (m5 / s[0]).is_file()]

    print(f'== 六层链缺料检查 · 场站 {farm} ==')
    print(f'   壳: src/windcms/pipeline.py::analyze（它按固定路径调这四步）')
    print(f'   规格: docs/振动六层链_接口规格与缺口_v0.1.md ｜ 取料单: docs/向振动线取料单_v0.1.md')
    print()
    print('A. 四步脚本:')
    for name, rel, what in STEPS:
        ok = (ROOT / rel).is_file()
        tag = 'OK ' if ok else '缺 '
        extra = ''
        if not ok and name == 'fusion' and have_fusion_impl:
            extra = '（本包已逆向实现其中一件产物: fleet_scalar_z；源码仍缺）'
        print(f'   [{tag}] {name:12s} {rel:44s} {what}{extra}')
    print()
    print('B. 逐值对拍的标准答案:')
    for nm, why in SAMPLES:
        ok = (m5 / nm).is_file()
        print(f'   [{"OK " if ok else "缺 "}] {nm:28s} {why}  ({P.rel(m5 / nm)})')
    # 已完成的那件: 用本器自带 verify 报一句现状（不写盘）
    fr = ROOT / 'scripts/rudong_fusion_run.py'
    if fr.is_file():
        print()
        print('C. 已自研部分（fleet_scalar_z）: 跑 python scripts/rudong_fusion_run.py --verify 看逐值对拍')
    print()
    rc = 5 if missing_scripts else 0
    print(f'   结论: {"四步脚本齐全 ⇒ 六层链可跑, 接着做逐值对拍" if not missing_scripts else str(len(missing_scripts)) + " 个步骤脚本未随包 ⇒ 六层链跑不动（取料单见上）"}')
    if a.check:
        print(f'[{"OK" if rc == 0 else "X"}] chain_gap_check rc={rc}')
    return rc


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())