#!/usr/bin/env python3 # -*- coding: utf-8 -*- r"""融合面 handoff 的**观澜自算**生成端（用户令 2026-09-19「所有的计算均要形成观澜的源代码」）。 ## 为什么需要它 `/detail/v2` 的「需要关注」与「全场状态」两块，取数走 `/api/fleet` 的 `fus`，而 `fus` 的 链盘行(需要关注)与逐台矩阵(全场状态)都来自 `src/windscada/subsys/fusion.py` 读的 `outputs/<场>/m5_cms_tcm/handoff_vibration_v2.json`。那件此前是**振动线人工交证件**： 全库 0 处写入方 ⇒ 清过产物、只从 data/raw 重算的机器上必然不在位 ⇒ 两块页面**空白** （不是报错，也不是没数据 —— 是那件没有生成端）。 用户令把口径定死了：**所有计算都要落到观澜自己的源代码里**，别的电脑装完（放上原始件、重算） 系统就要正常跑。所以本器把这份 handoff **按同一份结构自己算出来**： 输入（全部是观澜从 data/raw 重算出来的产物） · m5_cms_tcm/model_run_l6.parquet 六层链 L6 过闸谱线（scripts/rudong_model_run.py） · m5_cms_tcm/fusion_38.csv 逐台融合裁决（scripts/rudong_fusion_run.py） · windcms/报告_CMS振动状态评估报告_*.md 逐台逐部件状态等级 + 证据状态（scripts/windcms.py report） · m5_cms_tcm/windows/<窗>/index.parquet 窗内真实日期区间（claim_window 用实算值） 输出 · m5_cms_tcm/handoff_vibration_v2.json 与消费者同结构（meta / detectability_prior / per_turbine / open_items） ## 与"正本"的关系（必须说清，不许冒充） 原 handoff 是**人对人的交证件**（含人工裁决、校准更新、现场动作），本器产出的是**同结构的观澜自算件**： · `meta.from` = 观澜自算（不是振动线），`meta.version` = `guanlan-selfcomputed-v1`； · 每条 verdict 由观澜自己的判据给出（L6 过闸线 + 融合裁决 + 报告状态等级），**不含**任何人工裁决； · `open_items` 首条如实声明"本件非振动线正本，正本到场应以正本为准"； · 若盘上已有**正本**（`meta.from` 里不含"观澜自算"），本器**不覆盖**（除非显式 `--force`）。 用法: python scripts/rudong_fusion_handoff.py # 有正本则跳过; 没有则生成 python scripts/rudong_fusion_handoff.py --status # 只报现状 python scripts/rudong_fusion_handoff.py --force # 覆盖（自算件之间可覆盖） """ from __future__ import annotations import argparse import json import pathlib import sys import time import pandas as pd 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 HANDOFF = None # 运行时按场站解析 (P.m5()) VERSION = 'guanlan-selfcomputed-v1' # 状态等级 → verdict 文本（词表必须与 fusion.VERDICT_CLASSES 对得上, 否则整列落成 unknown）: # 危险 → '定论' (bad) · 报警 → '准定论·预警' (warn) · 良好/优秀 → '无新证' (ok) · 不可判 → '测量链不可判' (meta) VERDICT_OF = {'危险': '定论', '报警': '准定论·预警', '良好': '无新证(本轮未见异常)', '优秀': '无新证(本轮未见异常)', '不可判': '测量链不可判'} COMP_OF = {'发电机': '发电机', '齿轮箱': '齿轮箱', '主轴承前': '主轴承', '主轴承后': '主轴承'} def status(farm: str | None = None) -> int: p = P.m5(farm) / 'handoff_vibration_v2.json' if not p.is_file(): print(json.dumps(dict(path=P.rel(p), exists=False, kind=None, note='不在位 —— /detail/v2 的「需要关注/全场状态」会如实为空; ' '跑本器可从观澜自算产物生成'), ensure_ascii=False)) return 0 try: h = json.loads(p.read_text(encoding='utf-8')) except Exception as e: print(json.dumps(dict(path=P.rel(p), exists=True, kind='unreadable', err=str(e)[:120]))) return 1 meta = h.get('meta') or {} kind = 'self' if '观澜自算' in str(meta.get('from', '')) else 'primary' n = sum(len(v or []) for v in (h.get('per_turbine') or {}).values()) print(json.dumps(dict(path=P.rel(p), exists=True, kind=kind, version=meta.get('version'), date=meta.get('date'), from_=meta.get('from'), per_turbine=n, note='正本(振动线交证件)' if kind == 'primary' else '观澜自算件(同结构)'), ensure_ascii=False)) return 0 def _report_levels(farm: str | None) -> tuple: """逐台逐部件的状态等级/证据状态 + 报告日期 + 数据窗日期 —— 直接取重算出的 CMS 报告（不重复判级）。""" import re cms = P.cms(farm) reps = sorted(cms.glob('报告_CMS振动状态评估报告_*.md')) if not reps: return {}, None, None rp = reps[-1] md = rp.read_text(encoding='utf-8') date = rp.stem.rsplit('_', 1)[-1] # 表头: | 机组号 | 主轴承前 | 主轴承后 | 齿轮箱 | 发电机 | 综合 | 融合级(模型) | CMS 红/黄 | 行动 | out = {} if '## 附录 A' in md: for l in md.split('## 附录 A')[1].splitlines(): if not l.startswith('| WTG'): continue c = [x.strip() for x in l.strip('|').split('|')] if len(c) < 6: continue out[c[0]] = dict(主轴承前=c[1], 主轴承后=c[2], 齿轮箱=c[3], 发电机=c[4], 综合=c[5]) win = None m = re.search(r'数据窗\s+(\S+)\s+(\d{4}-\d{2}-\d{2})\s*→\s*(\d{4}-\d{2}-\d{2})', md) if m: win = (m.group(2), m.group(3)) return out, date, win def build(farm: str | None = None, force: bool = False) -> int: m5 = P.m5(farm) dest = m5 / 'handoff_vibration_v2.json' if dest.is_file() and not force: try: cur = json.loads(dest.read_text(encoding='utf-8')) if '观澜自算' not in str((cur.get('meta') or {}).get('from', '')): print(f'[跳过] {P.rel(dest)} 是**振动线正本**(非观澜自算) —— 不覆盖。' f'要强制覆盖用 --force（不建议：正本含人工裁决）') return 0 except Exception: pass levels, rep_date, win = _report_levels(farm) if not levels: print('[X] 没有重算出的 CMS 报告（windcms/报告_CMS振动状态评估报告_*.md）—— ' '先跑 scripts/windcms.py report（或在重算链 ④b 里跑）') return 2 f38 = m5 / 'fusion_38.csv' fus = pd.read_csv(f38, dtype=str) if f38.is_file() else pd.DataFrame() ts = pd.Timestamp.now() w_lo, w_hi = (win or ('', '')) if not w_lo: from src.windcms import config as _wc, data as _wd sp = _wd.window_spans(_wc.farm(str(farm or 'rudong'))) rows = [v for v in sp.values() if v.get('rows')] if rows: w_lo, w_hi = str(rows[0]['t_min'])[:10], str(rows[0]['t_max'])[:10] window_txt = f'{w_lo}~{w_hi}' if w_lo else '未标' per = {'v1_generator': [], 'v1_gearbox': [], 'new_2026_08_26': []} by_comp = {'发电机': 'v1_generator', '齿轮箱': 'v1_gearbox', '主轴承': 'new_2026_08_26'} n_bad = n_warn = 0 for tid, lv in sorted(levels.items()): fu = fus[fus['台'] == tid].iloc[0].to_dict() if len(fus) and (fus['台'] == tid).any() else {} for comp_cn, state in (('主轴承前', lv.get('主轴承前')), ('主轴承后', lv.get('主轴承后')), ('齿轮箱', lv.get('齿轮箱')), ('发电机', lv.get('发电机'))): comp = COMP_OF.get(comp_cn, comp_cn) verdict = VERDICT_OF.get(str(state), '测量链不可判') if not str(state) or str(state) == '优秀': continue # 优秀不必逐条登记（矩阵按"未列≠正常"渲染, 但全列 38×4 太吵） if verdict == '定论': n_bad += 1 elif verdict.startswith('准定论'): n_warn += 1 per[by_comp.get(comp, 'new_2026_08_26')].append(dict( id=f'{tid}-{comp_cn}', turbine=tid, component=comp, title=f'{comp_cn} 状态等级 {state}', verdict=verdict, claim_window=f'{window_txt}（观澜自算, 窗内实算区间）', source='观澜自算: m5_cms_tcm/model_run_l6.parquet + fusion_38.csv + windcms 报告', fusion_grade=fu.get('融合'), cms=f"{fu.get('CMS', '')}", report_date=rep_date)) from src.windcms.data import span_text as _span # 单一实现: 窗区间口径 from src.windcms import config as _wc2 try: span = _span(_wc2.farm(str(farm or 'rudong'))) except Exception: span = window_txt doc = dict( meta=dict(version=VERSION, date=ts.strftime('%Y-%m-%d'), from_='观澜自算 (scripts/rudong_fusion_handoff.py)', to='windscada 融合面 (/api/fleet 的 fus: 需要关注 / 全场状态 / 振动融合)', supersedes='（无）', principle='判级全部来自观澜自己的确定性判据, 不含人工裁决; 与振动线正本同结构', reading_discipline='列内状态=判级转录; 「未列」≠正常; 相对判据必配同域绝对量; ' '缺件时如实为空, 不从交付包补齐(用户令 2026-09-17)', derived_from=['m5_cms_tcm/model_run_l6.parquet', 'm5_cms_tcm/fusion_38.csv', 'windcms/报告_CMS振动状态评估报告_*.md', 'm5_cms_tcm/windows/*/index.parquet'], data_window=span, n_verdicts=sum(len(v) for v in per.values()), n_bad=n_bad, n_warn=n_warn), detectability_prior=dict( purpose='振动面能力边界与盲区（观澜自算口径, 与振动线正本的可能不同）', classes=dict( spectral_line=dict(note='特征频率族: 六道前提闸(G1–G8) + 同域绝对锚 + 证据族; ' '不过闸的 x_fleet/选择性/趋势一律不进下游'), broadband_level=dict(note='宽带层: CMS rms_200/Indicator; 未越限不能作正常依据'), impact_time_domain=dict(note='冲击轴(时域六窗): 与谱线层互补, 盲区不同'), iso_velocity=dict(note='ISO 速度当量: 唯一已校准的绝对锚(黄 4.5 / 红 11.0 mm/s)'), data_quality=dict(note='L0 数据质量: 不过门则该测点任何结论不可采信')), meas_crossing=dict(note='同一物理量的不同测量(原始谱/包络谱)不可跨域比绝对量'), **{'known_blind_spots': [ '包络域无 ISO 绝对锚 ⇒ 该域结论封顶「候选」', '主轴承前后 8–13 kHz 存在同路电气调幅成分, 该带不用于机械源定位', '无正样本锚(实物/内窥)时不得出「定论」; 本件为**观澜自算**, 不具备跨线裁决权', '单窗数据时"逐窗趋势"信息量有限(按日聚合后可见)']}, 端到端闭环=dict(note='判级→行动→复测闭环在综合报告线, 本面只出证据与判级')), per_turbine=per, field_actions_v1=[], calibration_updates_v1=[], open_items=['★本件是**观澜自算**的融合面（非振动线 handoff 正本）: 由 scripts/rudong_fusion_handoff.py ' '从重算产物生成; 振动线正本到场时应以正本为准（把正本放回该路径即可, 本器不再覆盖正本）。', '峰值拾取仍为本器口径(目标频率 ±2 bin 取最大), 见 docs/振动六层链_接口规格与缺口_v0.1.md §3.1', f'数据窗 {span}'] if True else []) dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(json.dumps(doc, ensure_ascii=False, indent=1), encoding='utf-8') print(f'已写 {P.rel(dest)}: 逐台条目 {doc["meta"]["n_verdicts"]} 条 ' f'(危险/报警 {n_bad} · 预警 {n_warn}) · 数据窗 {span}') try: from src import derived_manifest as DM DM.record(P.out_root(farm), {dest.relative_to(P.out_root(farm)).as_posix(): 'scripts/rudong_fusion_handoff.py (观澜自算融合面: L6 过闸线+融合裁决+报告状态等级; 同结构于振动线手交件)'}, by='rudong_fusion_handoff') print(' 已自登记 → _derived_manifest.json') except Exception as e: print(f' [i] 自登记跳过: {type(e).__name__}: {e}') return 0 def main() -> int: ap = argparse.ArgumentParser(description='融合面 handoff 的观澜自算生成端（同结构于振动线手交件）') ap.add_argument('--farm', default=None) ap.add_argument('--status', action='store_true', help='只报现状（正本/自算件/不在位）') ap.add_argument('--force', action='store_true', help='覆盖已有自算件（正本仍不覆盖）') a = ap.parse_args() if a.status: return status(a.farm) return build(a.farm, a.force) if __name__ == '__main__': for _s in (sys.stdout, sys.stderr): try: _s.reconfigure(errors='replace') except Exception: pass sys.exit(main())