guanlan-src/scripts/rudong_fusion_handoff.py

#!/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())