guanlan-src/scripts/vib_raw_build.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""振动侧一键摄入: data/raw/<场站>/{windcms,m5_cms_tcm} 的原始件 → 窗索引/谱库 → (可选)CMS 报告与知识库.

## 为什么需要它 (2026-09-12 用户令)

用户令: 「数据层里的「CMS 振动评估报告」应遵循 `<安装目录>\\data\\raw\\如东\\windcms`、
「振动线 handoff」应遵循 `<安装目录>\\data\\raw\\如东\\m5_cms_tcm` 存放; 修改系统支持振动数据参与
运行、重算」。此前振动侧只有**产物**(`outputs/<场>/m5_cms_tcm` 90 件 + `outputs/<场>/windcms` 53 件),
源件不在 data/raw, 生成端脚本 (振动线分支的 rudong_tcm_*.py 七件) 也没随包 ⇒ 从零重算时振动侧是断的。

两个事实把这件事限定得很清楚:
  · `src/windcms/pipeline.py` 认的输入就是"含 `*_decode.json` 的目录"(Brande TCM Enterprise 导出),
    这一步**必须能跑**, 否则数据进了 data/raw 也只是躺着;
  · 同分支的六层链 (oem_scan / energy_share / model_run / fusion) 脚本仍未随包 —— 本脚本**不假装**
    能把那几步跑出来: 它只做"索引 + 谱 + (可选)报告/知识库", 其余在报告里如实写"缺失"。

## 这一步跑完, 振动数据就真的参与了运行

    data/raw/<场站>/windcms/.../*_decode.json
        → outputs/<场>/m5_cms_tcm/windows/<窗>/index.parquet     (54 列, 与包内 tcm_index.parquet 同构)
        → outputs/<场>/m5_cms_tcm/windows/<窗>/spectra/*.npz + spectra_meta.parquet
    消费者 (无需改一行代码, 窗是自动发现的):
        src/windcms/data.py::windows()      → 新窗进分析集
        src/windcms/data.py::load_scalars() → 标量进 CMS 报告/页面
        src/windcms/data.py::spectrum()     → 谱图能取到 (此前 spectra 库整个缺失, 谱图是死的)
        scripts/windcms.py report / kb      → 报告与知识库重生成

## ★ 为什么"重生成 CMS 报告"默认**不跑** (2026-09-12 实测)

跑一次 `scripts/windcms.py report` 在本包会**让产物变差**, 不是变好:

| 件 | 随包快照 | 本包重生成后 | 变化 |
|---|---|---|---|
| `windcms/report.md` | 16,857 B (含逐台融合级表 + L4 过闸谱线) | **467 B** (融合级表空、L4 写"无") | **−97.2%** |
| `windcms/overview.html` | 646,218 B | 548,466 B | −15.1% |
| `windcms/index_eng.html` | 443,841 B | 408,112 B | −8.0% |

根因不在数据, 在**缺件**: `src/windcms/data.py::load_model()` 要读
`m5/model_run_l6.parquet` 与 `m5/fusion_38.csv`, 而这两件属六层链的 `model_run` / `fusion` 两步 ——
**那四步脚本没随包**(见 `missing_chain`), 产物也不在。于是重生成 = 用残缺输入覆盖完整快照。

所以: 摄入(索引/谱)默认跑, **报告/知识库默认不跑**; 要跑得显式 `--with-report`, 并且跑之前**先备份**
`outputs/<场>/windcms/`。六层链补齐后这个默认值应当翻过来 (那时重生成才会≥快照)。

## 用法

    python scripts/vib_raw_build.py --farm rudong                     # 摄入(索引+谱), 不动报告
    python scripts/vib_raw_build.py --window w0316                    # 指定窗名
    python scripts/vib_raw_build.py --jobs 12 --meas FFT_             # 并行/测量过滤
    python scripts/vib_raw_build.py --with-report                     # 额外重生成 CMS 报告/知识库(见上, 慎用)
    python scripts/vib_raw_build.py --dry-run                         # 只报会做什么
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import shutil
import subprocess
import sys
import time

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from src.console import soft  # noqa: E402

soft()
PY = sys.executable


def human(n):
    for u in ('B', 'KB', 'MB', 'GB'):
        if n < 1024 or u == 'GB':
            return f'{n:.1f} {u}'
        n /= 1024


def find_roots(station: pathlib.Path):
    """→ 振动侧原始件根列表 (含 *_decode.json 的目录或导出 zip)。"""
    roots = []
    for name in ('windcms', 'm5_cms_tcm'):
        d = station / name
        if not d.is_dir():
            continue
        for z in sorted(d.rglob('*.zip')):
            if z.name.upper().startswith('CMS') or 'decode' in z.name.lower():
                roots.append(z)
        for sub in sorted({p.parent for p in d.rglob('*_decode.json')}):
            # 取**最上层**那个含 decode 文件的目录 (measurement/), 避免每台机组一个 root
            top = sub
            while top.parent != d and any(top.parent.rglob('*_decode.json')):
                top = top.parent
            if top not in roots:
                roots.append(top)
    return roots


def run(step, cmd, log):
    t0 = time.time()
    print(f'[{step}] {" ".join(str(c) for c in cmd)}', flush=True)
    p = subprocess.run([str(c) for c in cmd], cwd=str(ROOT), capture_output=True, text=True,
                       encoding='utf-8', errors='replace')
    tail = (p.stdout or '') + (p.stderr or '')
    rec = dict(step=step, cmd=' '.join(str(c) for c in cmd), rc=p.returncode,
               seconds=round(time.time() - t0, 1), tail=tail[-1200:])
    log.append(rec)
    print(f'    rc={p.returncode}  {rec["seconds"]}s', flush=True)
    if p.returncode != 0:
        print(tail[-2000:], flush=True)
        raise SystemExit(f'步骤 {step} 失败 (rc={p.returncode})')
    return rec


def _rels_of_window(win: pathlib.Path, m5: pathlib.Path) -> dict:
    """某窗的产物 → {相对产物仓的路径: 构建器说明}; 供正常摄入与 --register-only 共用 (单一实现)。"""
    rels = {f'm5_cms_tcm/windows/{win.name}/index.parquet':
            'scripts/rudong_tcm_index.py (54 列, 与包内 tcm_index.parquet 同列名列序)',
            f'm5_cms_tcm/windows/{win.name}/spectra_meta.parquet':
            'scripts/rudong_tcm_spectra.py',
            'm5_cms_tcm/vib_raw_manifest.json': 'scripts/vib_raw_build.py'}
    sp = win / 'spectra'
    if sp.is_dir():
        for p in sp.rglob('*.npz'):
            rels[f'm5_cms_tcm/windows/{win.name}/spectra/{p.relative_to(sp).as_posix()}'] = \
                'scripts/rudong_tcm_spectra.py (npz 分片: DataSets.DataSet.Values 空格串 → float 数组)'
        # 谱库目录里那份 meta 是同一内容的两条读取路径之一, 也登记
        rels[f'm5_cms_tcm/windows/{win.name}/spectra/spectra_meta.parquet'] = \
            'scripts/rudong_tcm_spectra.py (与窗根同名件同内容: 兼顾两种读取约定)'
    return rels


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--farm', default='rudong')
    ap.add_argument('--window', default=None, help='窗名 (默认按数据起始日推 wMMDD)')
    ap.add_argument('--jobs', type=int, default=min(8, (os.cpu_count() or 4)))
    ap.add_argument('--meas', default='FFT_', help="谱摄入的测量过滤 (默认 FFT_; ALL=全转)")
    ap.add_argument('--skip-spectra', action='store_true', help='只做索引 (谱很占盘)')
    ap.add_argument('--with-report', action='store_true',
                    help='额外跑 windcms.py report/kb —— ★默认关: 本包缺六层链的 model_run/fusion 产物, '
                         '重生成会让 report.md/overview.html 掉内容 (见文件头实测表); 跑前先备份 windcms\\')
    ap.add_argument('--register-only', action='store_true',
                    help='不摄入, 只把**已存在**的窗体件补进来源自登记 (幂等修复: 例如先前的摄入跑在自登记'
                         '功能之前, 台账就漏记了那些件)')
    ap.add_argument('--limit', type=int, default=0, help='冒烟: 只摄入前 N 个文件')
    ap.add_argument('--dry-run', action='store_true')
    a = ap.parse_args()

    from src.windscada.config import farm, raw_station_dir
    from src import paths as P
    cfg = farm(a.farm)
    station = pathlib.Path(raw_station_dir(a.farm))
    m5 = P.m5(a.farm)
    roots = find_roots(station)
    print(f'场站原始件目录: {station}')
    print(f'振动侧产物目录: {m5}')

    if a.register_only:
        # 只补登记: 找已存在的窗 (--window 指定, 否则取名字最大的那个), 逐件登记。
        # 放在"源件存在性检查"之前 —— 补登记不需要重新读原始件。
        cands = sorted(p for p in (m5 / 'windows').glob('w[0-9][0-9][0-9][0-9]')
                       if (p / 'index.parquet').exists())
        if a.window:
            cands = [p for p in cands if p.name == a.window]
        if not cands:
            print('没有已存在的窗可登记 (先正常跑一次摄入)')
            return 0
        win = cands[-1]
        rels = _rels_of_window(win, m5)
        from src.derived_manifest import record as _rec
        p = _rec(P.out_root(a.farm), rels, by='scripts/vib_raw_build.py')
        print(f'已登记 {len(rels)} 件 (窗 {win.name}) → {p}')
        return 0

    if not roots:
        print('没找到振动原始件 (*_decode.json 目录或 CMS 导出 zip)。'
              '先用 scripts/place_raw_data.py --scope vib 从现场包落位。')
        return 0
    for r in roots:
        n = len(list(r.rglob('*_decode.json'))) if r.is_dir() else '(zip)'
        sz = sum(p.stat().st_size for p in r.rglob('*_decode.json')) if r.is_dir() else r.stat().st_size
        print(f'  源件根: {r}   {n} 个 decode 文件, {human(sz)}')
    if a.dry_run:
        print('(dry-run, 未摄入)')
        return 0

    windows_dir = m5 / 'windows'
    staging = windows_dir / '_staging_ingest'          # 'test'/'_reimport' 之外的名字, 会被 data.windows() 看见 → 立即改名
    if staging.exists():
        shutil.rmtree(staging)
    staging.mkdir(parents=True, exist_ok=True)

    log = []
    t0 = time.time()
    root = roots[0] if len(roots) == 1 else station / 'windcms'   # 多个根时交给各自目录 (index 支持 rglob)
    idx = staging / 'index.parquet'
    run('index', [PY, str(ROOT / 'scripts/rudong_tcm_index.py'), '--root', str(root), '--out', str(idx),
                  '--jobs', str(a.jobs)] + (['--limit', str(a.limit)] if a.limit else []), log)

    n_spectra = 0
    if not a.skip_spectra:
        run('spectra', [PY, str(ROOT / 'scripts/rudong_tcm_spectra.py'), '--root', str(root),
                        '--out', str(staging / 'spectra'), '--meas', a.meas, '--jobs', str(a.jobs)]
            + (['--limit', str(a.limit)] if a.limit else []), log)
        sm = staging / 'spectra' / 'spectra_meta.parquet'
        if sm.exists():
            import pandas as pd
            n_spectra = len(pd.read_parquet(sm))

    # 窗名: 用数据自身的起始日 (wMMDD), 与既有 w0127/w0707/w0811 同口径
    import pandas as pd
    d = pd.read_parquet(idx)
    tt = pd.to_datetime(d.trigger_time, errors='coerce')
    tmin, tmax = tt.min(), tt.max()
    win = a.window or f'w{tmin:%m%d}'
    final = windows_dir / win
    if final.exists():
        final = windows_dir / f'{win}_reimport_{time.strftime("%m%d%H%M")}'
        print(f'⚠ 窗 {win} 已存在 → 本次摄入落到 {final.name} (data.EXCLUDE_DEFAULT 会排除 _reimport 窗, 不污染生产集)')
    staging.rename(final)
    print(f'\n窗: {final.name}   数据窗 {tmin} → {tmax}   行 {len(d)}   谱 {n_spectra}')

    if a.with_report:
        run('report', [PY, str(ROOT / 'scripts/windcms.py'), 'report', '--farm', a.farm], log)
        run('kb', [PY, str(ROOT / 'scripts/windcms.py'), 'kb', '--farm', a.farm], log)
    else:
        print('（跳过 CMS 报告/知识库重生成 —— 本包缺 model_run/fusion 产物, 重生成会掉内容; '
              '要跑用 --with-report, 且先备份 windcms\\）')

    man = dict(farm=a.farm, window=final.name, window_dir=str(final.relative_to(ROOT)),
               # 源件路径写**相对安装根**的形态: 这文件会随包分发, 绝对路径换机后就是死链
               # (check_transferable.py 会把 outputs 下的绝对路径算作"机器相关路径")
               sources=[str(r.relative_to(ROOT)) if str(r).startswith(str(ROOT)) else str(r) for r in roots],
               sources_note='路径相对<安装目录>', rows=int(len(d)), spectra=int(n_spectra),
               time_min=str(tmin), time_max=str(tmax),
               turbines=int(d.turbine.nunique()), sensors=int(d.sensor_name.nunique()),
               meas_names=int(d.meas_name.nunique()),
               steps=[{k: v for k, v in r.items() if k != 'tail'} for r in log],
               finished=time.strftime('%Y-%m-%d %H:%M:%S'), seconds=round(time.time() - t0, 1),
               missing_chain=['rudong_tcm_oem_scan', 'rudong_line_energy_share', 'rudong_model_run',
                              'rudong_fusion_run'],
               missing_note='六层链的四步脚本未随包 (振动线分支 claude/vibration-data-diagnosis-32b69e); '
                            '本脚本只做索引/谱/报告/知识库, 不冒充跑过那四步')
    (m5 / 'vib_raw_manifest.json').write_text(json.dumps(man, ensure_ascii=False, indent=1), encoding='utf-8')

    # 来源自登记: 让 _provenance.json 把这些件记成 raw-derived (见 src/derived_manifest.py 的说明)
    try:
        from src.derived_manifest import record as _rec
        rels = _rels_of_window(final, m5)
        if a.with_report:
            for p in (P.cms(a.farm)).glob('报告_CMS振动状态评估报告_*.md'):
                # 只登记"今天生成的"这一件 (随包/厂家转录的那些不冒充自算)
                if time.strftime('%Y-%m-%d') in p.name:
                    rels[f'windcms/{p.name}'] = 'scripts/windcms.py report (数据窗: ' + final.name + ')'
        _rec(P.out_root(a.farm), rels, by='scripts/vib_raw_build.py')
        print(f'  自登记 {len(rels)} 件产物 → {P.out_root(a.farm) / "_derived_manifest.json"}')
    except Exception as _e:
        print(f'  ⚠ 来源自登记失败 ({type(_e).__name__}: {_e}) —— 台账会漏记这些件, 但不影响产物本身')

    print(f'\n完成: 用时 {time.time() - t0:.0f}s; 清单 → {m5 / "vib_raw_manifest.json"}')
    print(f'  新窗 {final.name} 已进分析集: windcms 的谱图/标量/报告页会自动收录它 '
          f'(消费者读取时现扫 m5\\windows\\w????\\index.parquet, 无需改代码)')
    print('  windscada 融合面读 handoff (m5_cms_tcm/handoff_vibration_v2.json) —— 那是振动线出件, 本次不动它')
    print('  看效果: python scripts/windcms.py serve --port 8020  → http://127.0.0.1:8020/')
    return 0


if __name__ == '__main__':
    sys.exit(main())