|
|
@@ -0,0 +1,193 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+r"""六层链 `oem_scan` 步 —— **逆向工程实现（分母已就位/分子待口径）**，2026-09-17。
|
|
|
+
|
|
|
+## 现状：一半已经落地并逐值验过，另一半明确挡住
|
|
|
+
|
|
|
+壳 `src/windcms/pipeline.py::analyze` 会调本脚本（无参数，读 `M5_WINDOWS` / `SCAN_BANDS` 环境变量），
|
|
|
+产出特征频率扫描件（`bearing_freq_scan` / `gear_freq_scan` / `cage_slip_scan` / `oem_frequency_scan`）。
|
|
|
+
|
|
|
+**已落地（分母 + 配置，逐值对拍通过）**：
|
|
|
+
|
|
|
+1. **扫描配置**：每个部件用哪条谱、什么分辨率 —— `reference/rudong/oem_scan_plan.json`
|
|
|
+ （11 个部件：`shard` 取 `Env_6000_4000_850_Tr` / `FFT_6000_Tr` / `FFT_2000_fast_Tr` / `FFT_62_Tr`，
|
|
|
+ `dx_hz` 1.0625 / 0.9375 / 0.625 / 0.03875）。
|
|
|
+2. **理论频率（分母）**：`option` 串 → `reference/rudong/oem_bearing_freqs.json` 条目，规则
|
|
|
+ = 位置 token（`Pos.150R`→`150r`、`139.1/1060 f-576853.PRL`→`1391/1060`、`6338 M/C3`→`6338m/c3`）
|
|
|
+ + 同位置多候选时按型号 token 定夺（NSK 284 vs FAG 273；NSK 137 vs FAG 151）。
|
|
|
+ **实测：11/11 个部件的 `BPFI_hz`/`BPFO_hz`/`BSF_hz` 与随包样件逐值一致**（`--verify-plan`）。
|
|
|
+
|
|
|
+**未落地（分子）**：`*_fleet_ratio` / `*_local_ratio` 要的是**观测频率 ÷ 理论频率**，而"在谱里怎么取观测峰"
|
|
|
+（带宽 / 取什么量最大 / 是否插值 / 是否按 rpm 阶次跟踪 / 多记录怎么合并）是缺的口径
|
|
|
+（见 `docs/向振动线取料单_v0.1.md` 第 1 项、`docs/振动六层链_接口规格与缺口_v0.1.md` §7）。
|
|
|
+
|
|
|
+**因此本脚本默认行为是"响亮失败"**（rc=3），绝不写半成品：
|
|
|
+在口径到位前，`windcms.py analyze --steps oem_scan` 会明确告诉你缺什么、去哪儿要，而不是产出一堆
|
|
|
+只有一半是真的扫描件让人误以为重算成功了。
|
|
|
+
|
|
|
+## 用法
|
|
|
+
|
|
|
+ python scripts/rudong_tcm_oem_scan.py --verify-plan # 分母+配置 与随包样件逐值对拍（不写盘）
|
|
|
+ python scripts/rudong_tcm_oem_scan.py --plan # 打印扫描计划（部件/测点/谱/分辨率/理论频率）
|
|
|
+ python scripts/rudong_tcm_oem_scan.py --status # 机器可读状态（供 chain_gap_check 用）
|
|
|
+ python scripts/rudong_tcm_oem_scan.py # 真跑（口径未到位 ⇒ rc=3, 不写盘）
|
|
|
+
|
|
|
+退出码: 0 成功/对拍通过 · 2 缺参考件 · 3 **峰值拾取口径未到位**（不写盘） · 5 对拍不一致
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import argparse
|
|
|
+import json
|
|
|
+import pathlib
|
|
|
+import re
|
|
|
+import sys
|
|
|
+
|
|
|
+import pandas as pd
|
|
|
+
|
|
|
+ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
|
+sys.path.insert(0, str(ROOT))
|
|
|
+from src import paths as P # noqa: E402
|
|
|
+
|
|
|
+PLAN = ROOT / 'reference' / 'rudong' / 'oem_scan_plan.json'
|
|
|
+THEORY = ROOT / 'reference' / 'rudong' / 'oem_bearing_freqs.json'
|
|
|
+# 口径闸: 峰值拾取口径到位后把它翻成 True, 本脚本才会真的写产物
|
|
|
+PEAK_PICKING_READY = False
|
|
|
+MISSING = ('峰值拾取口径: 理论频率附近**怎么取观测峰**(带宽 / 取什么量最大 / 是否插值 / '
|
|
|
+ '是否按 rpm 阶次跟踪 / 多记录怎么合并)',) # ★ 末尾逗号: 这是"缺失项清单", 不是一整句话
|
|
|
+
|
|
|
+
|
|
|
+def _norm(s: str) -> str:
|
|
|
+ return re.sub(r'[^0-9a-z/]', '', str(s).lower())
|
|
|
+
|
|
|
+
|
|
|
+def pos_token(option: str) -> str:
|
|
|
+ """option 串 → 位置 token（见模块头第 2 条的实测规则）。"""
|
|
|
+ s = str(option)
|
|
|
+ m = re.search(r'pos\.?\s*([0-9]+[LR]?|[0-9.]+/[0-9]+)', s, re.I)
|
|
|
+ if m:
|
|
|
+ return _norm(m.group(1))
|
|
|
+ m = re.search(r'([0-9.]+/[0-9]+)', s)
|
|
|
+ if m:
|
|
|
+ return _norm(m.group(1))
|
|
|
+ m = re.search(r'([0-9]{3,4}\s*[A-Z]?\s*/\s*[A-Z0-9]+)', s)
|
|
|
+ if m:
|
|
|
+ return _norm(m.group(1))
|
|
|
+ return _norm(s)
|
|
|
+
|
|
|
+
|
|
|
+def theory_of(option: str, bearings: dict):
|
|
|
+ tok = pos_token(option)
|
|
|
+ cands = [k for k, v in bearings.items() if pos_token(v.get('pos', '')) == tok and tok]
|
|
|
+ if not cands:
|
|
|
+ return None, tok, []
|
|
|
+ if len(cands) == 1:
|
|
|
+ return cands[0], tok, cands
|
|
|
+ no = _norm(option)
|
|
|
+ for k in cands:
|
|
|
+ mo = _norm(bearings[k].get('model', ''))
|
|
|
+ if mo and (mo[:10] in no or mo[:8] in no):
|
|
|
+ return k, tok, cands
|
|
|
+ for k in cands:
|
|
|
+ mk = str(bearings[k].get('model', '')).lower()
|
|
|
+ for v in ('nsk', 'fag', 'ina'):
|
|
|
+ if v in mk and v in no:
|
|
|
+ return k, tok, cands
|
|
|
+ return cands[0], tok, cands
|
|
|
+
|
|
|
+
|
|
|
+def load_plan():
|
|
|
+ if not PLAN.is_file():
|
|
|
+ raise SystemExit(f'[X] 缺扫描配置 {P.rel(PLAN)}')
|
|
|
+ if not THEORY.is_file():
|
|
|
+ raise SystemExit(f'[X] 缺理论频率表 {P.rel(THEORY)}')
|
|
|
+ plan = json.loads(PLAN.read_text(encoding='utf-8'))
|
|
|
+ th = json.loads(THEORY.read_text(encoding='utf-8'))['bearings']
|
|
|
+ out = []
|
|
|
+ for c in plan['components']:
|
|
|
+ k, tok, cands = theory_of(c['option'], th)
|
|
|
+ e = th.get(k) or {}
|
|
|
+ out.append(dict(**c, theory_key=k, pos_token=tok, n_cands=len(cands),
|
|
|
+ BPFI_hz=e.get('BPFI_Hz'), BPFO_hz=e.get('BPFO_Hz'), BSF_hz=e.get('BSF_Hz'),
|
|
|
+ shaft_Hz=e.get('shaft_Hz'), model=e.get('model', '')))
|
|
|
+ return plan, out
|
|
|
+
|
|
|
+
|
|
|
+def verify_plan(farm: str | None = None) -> int:
|
|
|
+ """分母 + 配置 与随包样件逐值对拍（样件 = outputs/<场>/m5_cms_tcm/oem_frequency_scan.parquet）。"""
|
|
|
+ farm = farm or P.farm()
|
|
|
+ sample = P.m5(farm) / 'oem_frequency_scan.parquet'
|
|
|
+ if not sample.is_file():
|
|
|
+ print(f'[X] 没有样件可对拍: {P.rel(sample)}')
|
|
|
+ return 2
|
|
|
+ _plan, rows = load_plan()
|
|
|
+ s = pd.read_parquet(sample)
|
|
|
+ got = pd.DataFrame(rows)
|
|
|
+ m = s[['component', 'sensor', 'shard', 'option', 'dx_hz', 'BPFI_hz', 'BPFO_hz', 'BSF_hz']].drop_duplicates()
|
|
|
+ j = m.merge(got, on=['component', 'sensor', 'shard', 'option'], how='outer', suffixes=('_样', '_本'),
|
|
|
+ indicator=True)
|
|
|
+ both = j[j['_merge'] == 'both']
|
|
|
+ ok = 0
|
|
|
+ bad = []
|
|
|
+ for _, r in both.iterrows():
|
|
|
+ good = all(abs(float(r[f'{c}_样']) - float(r[f'{c}_本'])) < 1e-9 for c in ('BPFI_hz', 'BPFO_hz', 'BSF_hz')) \
|
|
|
+ and abs(float(r['dx_hz_样']) - float(r['dx_hz_本'])) < 1e-9
|
|
|
+ ok += good
|
|
|
+ if not good:
|
|
|
+ bad.append(r['component'])
|
|
|
+ print(f'== oem_scan 分母+配置 逐值对拍 · 场站 {farm} ==')
|
|
|
+ print(f' 样件组件 {len(m)} 个 · 本器 {len(got)} 个 · 同键 {len(both)} 个'
|
|
|
+ f' · 仅样件 {int((j["_merge"] == "left_only").sum())} · 仅本器 {int((j["_merge"] == "right_only").sum())}')
|
|
|
+ print(f' BPFI_hz/BPFO_hz/BSF_hz + dx_hz 逐值一致: {ok}/{len(both)}' + (f' 不一致: {bad}' if bad else ''))
|
|
|
+ for _, r in got.iterrows():
|
|
|
+ print(f' {r["component"]:24s} {r["sensor"]:22s} {r["shard"]:22s} dx={r["dx_hz"]:<8} '
|
|
|
+ f'BPFI={r["BPFI_hz"]:<6} BPFO={r["BPFO_hz"]:<6} BSF={r["BSF_hz"]:<6} ← {str(r["theory_key"])[:34]}')
|
|
|
+ rc = 0 if (ok == len(both) and len(both) == len(m) == len(got)) else 5
|
|
|
+ print(f' 结论: {"分母映射成立（理论频率 + 扫描配置逐值一致）" if rc == 0 else "有不一致, 见上"} rc={rc}')
|
|
|
+ return rc
|
|
|
+
|
|
|
+
|
|
|
+def status(farm: str | None = None) -> int:
|
|
|
+ """机器可读状态（JSON）—— 供 scripts/chain_gap_check.py 汇总。"""
|
|
|
+ _plan, rows = load_plan()
|
|
|
+ st = dict(step='oem_scan', complete=bool(PEAK_PICKING_READY), components=len(rows),
|
|
|
+ denominator='ok（理论频率 + 扫描配置已落地, --verify-plan 逐值对拍）',
|
|
|
+ missing=list(MISSING) if not PEAK_PICKING_READY else [],
|
|
|
+ request_sheet='docs/向振动线取料单_v0.1.md')
|
|
|
+ print(json.dumps(st, ensure_ascii=False))
|
|
|
+ return 0 if PEAK_PICKING_READY else 3
|
|
|
+
|
|
|
+
|
|
|
+def main() -> int:
|
|
|
+ ap = argparse.ArgumentParser(description='六层链 oem_scan 步（逆向实现: 分母已就位/分子待口径）')
|
|
|
+ ap.add_argument('--farm', default=None)
|
|
|
+ ap.add_argument('--verify-plan', action='store_true', help='分母+配置 与随包样件逐值对拍（不写盘）')
|
|
|
+ ap.add_argument('--plan', action='store_true', help='打印扫描计划')
|
|
|
+ ap.add_argument('--status', action='store_true', help='机器可读状态（JSON）')
|
|
|
+ a = ap.parse_args()
|
|
|
+ if a.status:
|
|
|
+ return status(a.farm)
|
|
|
+ if a.verify_plan:
|
|
|
+ return verify_plan(a.farm)
|
|
|
+ _plan, rows = load_plan()
|
|
|
+ if a.plan:
|
|
|
+ print(f'== 扫描计划（{len(rows)} 个部件）==')
|
|
|
+ for r in rows:
|
|
|
+ print(f' {r["component"]:24s} {r["sensor"]:22s} {r["shard"]:22s} dx={r["dx_hz"]:<8} '
|
|
|
+ f'BPFI={r["BPFI_hz"]:<6} BPFO={r["BPFO_hz"]:<6} BSF={r["BSF_hz"]:<6} shaft={r["shaft_Hz"]}')
|
|
|
+ return 0
|
|
|
+ if not PEAK_PICKING_READY:
|
|
|
+ print('[X] 峰值拾取口径未到位 —— 本步**拒绝**产出半成品（只写分母会很危险: 那些件看着像扫描结果, 实则没有分子）')
|
|
|
+ print(f' 缺的是: {MISSING[0]}')
|
|
|
+ print(' 要料: docs/向振动线取料单_v0.1.md 第 1 项（源码最佳, 一句话亦可）')
|
|
|
+ print(' 已就位: 扫描配置 + 理论频率 —— 用 --verify-plan 可逐值对拍（11/11 通过）')
|
|
|
+ return 3
|
|
|
+ raise SystemExit('[!] PEAK_PICKING_READY=True 但实现未完成 —— 请先补上分子侧实现再翻这个开关')
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ for _s in (sys.stdout, sys.stderr):
|
|
|
+ try:
|
|
|
+ _s.reconfigure(errors='replace')
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ sys.exit(main())
|
zhouyang.xie