|
|
@@ -1,275 +1,318 @@
|
|
|
-#!/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 → 报告与知识库重生成
|
|
|
-
|
|
|
-## ★ 报告默认**跑**了 (2026-09-18 用户令后口径翻转)
|
|
|
-
|
|
|
-2026-09-12 的旧口径是"默认不跑": 那时拿**随包快照**当基准, 重生成会让 `windcms/report.md`
|
|
|
-从 16,857 B 掉到 467 B (−97%), 因为 `data.load_model()` 要读的 `m5/model_run_l6.parquet` 与
|
|
|
-`m5/fusion_38.csv` 属六层链 `model_run`/`fusion` 两步, 那两步脚本没随包。
|
|
|
-
|
|
|
-用户令 2026-09-18「`/detail` 各页面不能用旧版产出补, 必须基于输入数据重算」把这条基准**换掉了**:
|
|
|
-交付包里根本没有旧产物可比, 缺的那个"完整快照"也不该再出现在页面上。于是:
|
|
|
-
|
|
|
- · 报告里少的只有**融合级**那一列(如实写 `—`, 因为 model_run/fusion 两步确实没跑),
|
|
|
- 其余判级(状态等级/证据状态/行动等级/逐台页/总览)全部由 `data/raw/<场站>/windcms` 的
|
|
|
- `*_decode.json` 现算 —— 实测 2026-09-18: 38 台判级出来(报警 3 / 优秀 35), `/api/vibcms` 立即可用;
|
|
|
- · 故本脚本**默认**带报告与知识库 (`--no-report` 可关), `rebuild_all.py` 的 ④b 同步。
|
|
|
-
|
|
|
-## 用法
|
|
|
-
|
|
|
- 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 --no-report # 只摄入, 不动报告/知识库
|
|
|
- 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('--no-report', dest='with_report', action='store_false',
|
|
|
- help='只摄入(索引+谱), 不跑 windcms.py report/kb')
|
|
|
- ap.set_defaults(with_report=True) # ★2026-09-18 口径翻转: 见文件头"报告默认跑了"
|
|
|
- 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:
|
|
|
- # ★2026-09-19 补链序: 六层链的业务顺序是 oem_scan → energy_share → **model_run → fusion** → report,
|
|
|
- # 报告要读 model_run_l6.parquet 与 fusion_38.csv 才有"融合级"那一列。原先这里直接跑 report,
|
|
|
- # 于是清过产物的机器上报告永远缺融合级(整列 '—')。现在按序补齐(两步都是**按口径重建**, 无标准答案
|
|
|
- # 对拍 —— 口径与来路写在各自产物与台账里, 见 docs/振动六层链_接口规格与缺口_v0.1.md §3)。
|
|
|
- run('model_run', [PY, str(ROOT / 'scripts/rudong_model_run.py'), '--window', final.name], log)
|
|
|
- run('fusion', [PY, str(ROOT / 'scripts/rudong_fusion_run.py'), '--window', final.name], log)
|
|
|
- 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)
|
|
|
- # 六层链 fusion 步里**唯一已被标准答案验证过**的那一件 (fleet_scalar_z): 窗索引 → 同工况族中位
|
|
|
- # → 稳健 z。上面那次 fusion 已经一并出它 (同一条命令: --window final.name)。
|
|
|
- else:
|
|
|
- print('（按 --no-report 跳过 CMS 报告/知识库重生成 —— 页面上"振动评估"一栏会显示为无产物）')
|
|
|
-
|
|
|
- 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())
|
|
|
+#!/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 → 报告与知识库重生成
|
|
|
+
|
|
|
+## ★ 报告默认**跑**了 (2026-09-18 用户令后口径翻转)
|
|
|
+
|
|
|
+2026-09-12 的旧口径是"默认不跑": 那时拿**随包快照**当基准, 重生成会让 `windcms/report.md`
|
|
|
+从 16,857 B 掉到 467 B (−97%), 因为 `data.load_model()` 要读的 `m5/model_run_l6.parquet` 与
|
|
|
+`m5/fusion_38.csv` 属六层链 `model_run`/`fusion` 两步, 那两步脚本没随包。
|
|
|
+
|
|
|
+用户令 2026-09-18「`/detail` 各页面不能用旧版产出补, 必须基于输入数据重算」把这条基准**换掉了**:
|
|
|
+交付包里根本没有旧产物可比, 缺的那个"完整快照"也不该再出现在页面上。于是:
|
|
|
+
|
|
|
+ · 报告里少的只有**融合级**那一列(如实写 `—`, 因为 model_run/fusion 两步确实没跑),
|
|
|
+ 其余判级(状态等级/证据状态/行动等级/逐台页/总览)全部由 `data/raw/<场站>/windcms` 的
|
|
|
+ `*_decode.json` 现算 —— 实测 2026-09-18: 38 台判级出来(报警 3 / 优秀 35), `/api/vibcms` 立即可用;
|
|
|
+ · 故本脚本**默认**带报告与知识库 (`--no-report` 可关), `rebuild_all.py` 的 ④b 同步。
|
|
|
+
|
|
|
+## 用法
|
|
|
+
|
|
|
+ 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 --no-report # 只摄入, 不动报告/知识库
|
|
|
+ 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('--no-report', dest='with_report', action='store_false',
|
|
|
+ help='只摄入(索引+谱), 不跑 windcms.py report/kb')
|
|
|
+ ap.set_defaults(with_report=True) # ★2026-09-18 口径翻转: 见文件头"报告默认跑了"
|
|
|
+ ap.add_argument('--manifest-only', action='store_true',
|
|
|
+ help='只补写 vib_raw_manifest.json（依已存在的窗; 不重跑摄入）')
|
|
|
+ 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 a.manifest_only:
|
|
|
+ # 2026-09-19: 摄入已跑完、清单却没写成的情形（实测: ④b 在 report 步崩掉 → 本脚本在写清单**之前**
|
|
|
+ # 就退出 ⇒ `m5_cms_tcm/vib_raw_manifest.json` 缺失, 页面归口审计按"溯源缺失"报 rc=6、整条链报失败）。
|
|
|
+ # 本模式**只补写清单**, 不重跑摄入（重跑会再花 ~22 分钟并另立一个 _reimport 窗）。
|
|
|
+ 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]
|
|
|
+ import pandas as _pd
|
|
|
+ ix = _pd.read_parquet(win / 'index.parquet',
|
|
|
+ columns=[c for c in ('turbine', 'sensor_name', 'meas_name', 'trigger_time')
|
|
|
+ if c in _pd.read_parquet(win / 'index.parquet').columns])
|
|
|
+ sm = win / 'spectra_meta.parquet'
|
|
|
+ n_spec = int(len(_pd.read_parquet(sm, columns=['shard']))) if sm.is_file() else 0
|
|
|
+ ts = _pd.to_datetime(ix['trigger_time'], errors='coerce') if 'trigger_time' in ix else None
|
|
|
+ man = dict(farm=a.farm, window=win.name, window_dir=str(win.relative_to(ROOT)),
|
|
|
+ sources=[str(r.relative_to(ROOT)) if str(r).startswith(str(ROOT)) else str(r) for r in roots],
|
|
|
+ sources_note='路径相对<安装目录>; ★本清单由 --manifest-only 依**已存在的窗**补写 '
|
|
|
+ '(2026-09-19), 不是摄入当次的原始记录 —— steps 留空, 时间取窗首末',
|
|
|
+ rows=int(len(ix)), spectra=n_spec,
|
|
|
+ time_min=(str(ts.min()) if ts is not None and len(ts) else ''), time_max=(str(ts.max()) if ts is not None and len(ts) else ''),
|
|
|
+ turbines=int(ix.turbine.nunique()) if 'turbine' in ix else 0,
|
|
|
+ sensors=int(ix.sensor_name.nunique()) if 'sensor_name' in ix else 0,
|
|
|
+ meas_names=int(ix.meas_name.nunique()) if 'meas_name' in ix else 0,
|
|
|
+ steps=[], finished=time.strftime('%Y-%m-%d %H:%M:%S'), seconds=None,
|
|
|
+ missing_chain=[], missing_note='')
|
|
|
+ (m5 / 'vib_raw_manifest.json').write_text(json.dumps(man, ensure_ascii=False, indent=1), encoding='utf-8')
|
|
|
+ print(f'已补写 {P.rel(m5 / "vib_raw_manifest.json")}: 窗 {win.name} · {len(ix)} 行 · {n_spec} 谱')
|
|
|
+ try:
|
|
|
+ from src.derived_manifest import record as _rec
|
|
|
+ rels = _rels_of_window(win, m5)
|
|
|
+ _rec(P.out_root(a.farm), rels, by='scripts/vib_raw_build.py (manifest-only)')
|
|
|
+ print(f' 已自登记 {len(rels)} 件')
|
|
|
+ except Exception as _e:
|
|
|
+ print(f' [i] 自登记跳过: {type(_e).__name__}: {_e}')
|
|
|
+ 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:
|
|
|
+ # ★2026-09-19 补链序: 六层链的业务顺序是 oem_scan → energy_share → **model_run → fusion** → report,
|
|
|
+ # 报告要读 model_run_l6.parquet 与 fusion_38.csv 才有"融合级"那一列。原先这里直接跑 report,
|
|
|
+ # 于是清过产物的机器上报告永远缺融合级(整列 '—')。现在按序补齐(两步都是**按口径重建**, 无标准答案
|
|
|
+ # 对拍 —— 口径与来路写在各自产物与台账里, 见 docs/振动六层链_接口规格与缺口_v0.1.md §3)。
|
|
|
+ run('model_run', [PY, str(ROOT / 'scripts/rudong_model_run.py'), '--window', final.name], log)
|
|
|
+ run('fusion', [PY, str(ROOT / 'scripts/rudong_fusion_run.py'), '--window', final.name], log)
|
|
|
+ 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)
|
|
|
+ # 六层链 fusion 步里**唯一已被标准答案验证过**的那一件 (fleet_scalar_z): 窗索引 → 同工况族中位
|
|
|
+ # → 稳健 z。上面那次 fusion 已经一并出它 (同一条命令: --window final.name)。
|
|
|
+ else:
|
|
|
+ print('（按 --no-report 跳过 CMS 报告/知识库重生成 —— 页面上"振动评估"一栏会显示为无产物）')
|
|
|
+
|
|
|
+ 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())
|
zhouyang.xie