guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/subsys/fusion.py

# -*- coding: utf-8 -*-
"""振动↔SCADA 融合面 (2026-08-27 用户令: 振动分析并入 windscada 融合; windcms 独立保留).

分工边界: 谱线/波形/六闸/判据全留 windcms (m5_cms_tcm); 本模块只消费振动线 handoff 接口
(结论/该信多少/盲区), 与我方四源 (温度NBM watch / 润滑柱塞 / 油样索引 / 可靠性工单) 逐台拼证据.

三条融合纪律 (振动线接口 meta 钉死, 本处执行):
  1. 同一失效机制的多观测面 ≠ 多源印证 — 禁按源数加权 (29# 四观测面 = 一条机制链);
  2. 盲区权重: 振动"没报警"在 known_blind_spots 覆盖的模式上不构成阴性证据
     (急性热失效×冲击量 / 早期×谱线层 / 轻度×排名 = 振动瞎区);
  3. claim 窗对齐: 三源数据窗各异 (振动至2026-08 / SCADA至2026-07-07 / 油样至2025-08),
     每条融合行必带各源窗, 窗外无数据处写"未见数"不写"正常".
融合不产判级 (P0/P1 归综合报告线; 判级只转录振动线 verdict 原文)."""
from app_common.app_common_guanlan.api import paths as P
import json, pathlib, re
import pandas as pd
from src.windscada.config import farm

_DATA_ROOT = pathlib.Path(__file__).resolve().parents[3]
HANDOFF = P.m5() / 'handoff_vibration_v2.json'

# 油样部件位目录名变体归一 (38台手工命名不一致; 归一后与融合部件轴对齐)
OIL_COMP_NORM = {
    '主轴前轴承': '主轴前', '主轴后轴承': '主轴后',
    '发电机前轴承': '发电机前', '发电机后轴承': '发电机后',
    '发电机轴承前': '发电机前', '发电机轴承后': '发电机后',
}


GAP_HANDOFF = [None]      # 缺 handoff 时记一句(供上层/页面显示); 见 load_handoff 的 2026-09-17 说明


def load_handoff():
    """振动线接口 v2。

    ★ 2026-09-17 改口径 (用户令"打包不随产物"之后): 原来"缺失就 raise" —— 理由正当
    ("融合面不可静默降级"), 但它把**整条重算链**卡死: 交付包不再随产物 ⇒ 目标机上这件
    handoff 必然缺, 于是 `populate`(⑦) 直接 rc=1, 后面的本体审计全不执行。
    现在改成**响亮降级**: 缺件返回空结构 + 在 `GAP_HANDOFF[0]` 写明缺什么、从哪儿补,
    由上层写进结果与日志(页面/审计能看见), 而不是把链打断。判级本身不造数: 缺件时融合面
    的状态为"不可判", 依据写明缺件。
    """
    if not HANDOFF.exists():
        GAP_HANDOFF[0] = (f'融合面缺件: {HANDOFF} (属"包内无生成端"的件, docs §13.6) —— '
                          f'★用户令 2026-09-17/18: 运行期**不从交付包补齐**, 页面如实显示"不可判"; '
                          f'要出这件必须由研发补生成端(振动六层链 fusion 步), 或放原始件重算')
        return {}
    return json.loads(HANDOFF.read_text(encoding='utf-8'))   # 中文 JSON: 缺 encoding 会按系统 locale(cp936) 读


_HB = None


def huabiao_analysis(t):
    """华标 2026-07 轮 per 台分析短语 (2026-08-28 用户令: 油液填入分析).
    转录检测机构判定(NB/T 10111)+重点抽样值+note, 不产新判级; 29# 缺采响亮."""
    global _HB
    if _HB is None:
        f = (P.ont() / 'oil_2026H2_huabiao.json')
        _HB = json.loads(f.read_text(encoding='utf-8')) if f.exists() else dict(meta={}, 抽样数值=[])
    rows = [r for r in _HB.get('抽样数值', []) if r.get('t') == t]
    if t == 'WTG29':
        return '★本轮缺采 (29#不在17台名单) — TCM-16分支B前置(主轴后Fe/PQ vs 2025-03基线15/12)未验'
    if not rows:
        return None      # 名单内无重点值台由调用方给全批小结
    segs = []
    for r in rows:
        v = f"{r['pos']} Fe{r['Fe']}/PQ{r['PQ']}/疲劳{r.get('疲劳磨粒', '—')}"
        if r.get('note'):
            v += f" ({r['note'][:40]})"
        segs.append(v)
    return '华标判正常(NB/T宽限); 重点值: ' + '; '.join(segs)


def oil_window(cfg=None):
    """油样索引 → per台 最新采样日期 (报告日期口径, 取样日≈报告日−1~2周)."""
    cfg = cfg or farm()
    p = pathlib.Path(cfg['store']) / 'oil_samples_index.parquet'
    if not p.exists():
        raise FileNotFoundError(f'油样索引缺失: {p} (scripts/windscada_watch_channels_build.py 不建它; 见 fusion 模块头)')
    df = pd.read_parquet(p)
    df['comp'] = df['comp'].replace(OIL_COMP_NORM)
    return df


def fusion_table(cfg=None, all_turbines=False):
    """逐台融合视图: 振动 verdict (转录) × 温度 watch × 柱塞 × 油样窗.

    行 = 振动线 handoff 里有 verdict 的台; 列 = 各证据轴 + 各源 claim 窗."""
    cfg = cfg or farm()
    h = load_handoff()
    if not h or not (h.get('per_turbine') or {}):
        # 缺 handoff (交付包不随产物 ⇒ 目标机必然缺): 返回**空表但列齐 + 列头齐全**, 上层把
        # GAP_HANDOFF[0] 那句"缺什么/怎么补"写进结果 —— 判级不造数, 页面显示缺件 (2026-09-17)。
        # ★2026-09-18 补: meta 原来给的是**空 dict**, 而 fleet_view 会取 meta['振动窗']/['纪律']/…
        #   ⇒ 上层又变成 KeyError 报给用户(实测 `/api/fleet` 的 fus 回 `{"err": "'振动窗'"}` 那一类)。
        #   缺件就把"缺"写进 meta 的每个键, 页面照常渲染并显示缺件原因。
        gap = GAP_HANDOFF[0] or (f'融合面缺件: {HANDOFF} (包内无生成端, docs §13.6) — '
                                 '运行期不从交付包补齐(用户令 2026-09-17); 要出件须研发补生成端')
        return (pd.DataFrame(columns=['turbine', '部件', '振动结论', '温度', '柱塞', '油样窗', '依据']),
                dict(数据时点='缺件', handoff版本='—', handoff日期='', 证据窗末='',
                     振动窗='缺件 (需 handoff_vibration_v2.json)', SCADA窗='', 油样窗='',
                     纪律=gap, 盲区=[], open_items=[], 色标={}, gap=gap))
    oil = oil_window(cfg)
    from . import temp_nbm
    wp = pathlib.Path(cfg['store']) / 'watch_channels_monthly.parquet'
    watch = pd.read_parquet(wp) if wp.exists() else None

    # 部件 → 温度通道 stem 路由 (温度轴必须与振动结论的部件同对象, 不得拿别的部件温度充数)
    COMP_CH = {'发电机': ('GenBe',), '齿轮箱': ('GeOilTmp', 'IMS', 'HS'), '主轴承': ('MBear',)}
    treg = temp_nbm.registry()
    tdf = treg[0] if isinstance(treg, tuple) else treg

    rows = []
    # v2.1 起 component 为显式字段 (振动线未映射的显式标性质: "引用同族/谱线级/测量链/未映射"
    # — 后三类不当部件级结论消费, 原文透传); 缺字段时回退子块名
    v1 = [(r, r.get('component') or '发电机') for r in h['per_turbine'].get('v1_generator', [])] + \
         [(r, r.get('component') or '齿轮箱') for r in h['per_turbine'].get('v1_gearbox', [])]
    new = [(r, r.get('component', '?')) for r in h['per_turbine'].get('new_2026_08_26', [])]
    for r, comp in v1 + new:
        t = r['turbine']
        stems = next((v for k, v in COMP_CH.items() if k in comp), ())
        sub = tdf[(tdf.turbine == t) & tdf.channel.str.contains('|'.join(stems))] if stems else tdf.iloc[0:0]
        if len(sub):
            top = sub.reindex(sub.dev_K.abs().sort_values(ascending=False).index).iloc[0]
            temp_note = f"{top.channel.replace('tmp_wtc_','').replace('_mean','')} dev{top.dev_K:+.1f}K [{top['判']}]"
        else:
            temp_note = f'未路由到{comp}温度通道'
        # WATCHLIST 台补月度 fleet 位次口径 — NBM 变化型对"稳定高位"失明 (29# 后主轴承 19月位次1-11
        # 而 NBM 判"—" 即此盲区), watchlist 台必须双口径并示
        wch = [c for (wt, c) in temp_nbm.WATCHLIST if wt == t and any(s in c for s in stems)]
        if wch and watch is not None:
            c = wch[0]
            piv = watch.pivot_table(index='month', columns='turbine', values=c)
            last = piv.dropna(how='all').index.max()
            dv = float(piv.loc[last, t] - piv.loc[last].median())
            rk = int(piv.loc[last].rank(ascending=False, method='average')[t])
            temp_note += f"; watch月度({last}): {c.split('_')[2]} dev{dv:+.1f}K 位次{rk}/38"
        # 油样轴: 该台最新样日期 + stale 声明
        to = oil[oil.turbine == t]
        oil_note = '无样'
        if len(to):
            last = to.date.max()
            if last >= '2026-07':
                hb = huabiao_analysis(t)
                oil_note = f"末样{last}: " + (hb or '华标2026-07轮全位判正常/可继续使用 (近共时可作现时判据)')
            else:
                oil_note = f"末样{last} (距振动现窗>12月, stale: 只作回测锚非现时判据)"
                hb29 = huabiao_analysis(t)
                if t == 'WTG29':
                    oil_note += '; ' + hb29
        _vc, _vl = verdict_class(r.get('verdict', ''))
        # 图形化结构字段 (2026-08-28 用户令"能表达清楚的就图形化UI"): 前端据此画偏差条/窗时间轴/时效胶囊,
        # 原字符串列保留作 tooltip 与降级
        _tdev = _tvd = None
        _mt = re.search(r'([A-Za-z0-9_]+) dev([+-][\d.]+)K \[([^\]]*)\]', temp_note)
        if _mt:
            _tch, _tdev, _tvd = _mt.group(1), float(_mt.group(2)), _mt.group(3)
        else:
            _tch = None
        _spans = [[a, b] for a, b in re.findall(r'(\d{4}-\d{2}-\d{2})~(\d{4}-\d{2}-\d{2})', str(r.get('claim_window', '')))]
        _olast = to.date.max() if len(to) else None
        _wl = to_wcms_level
        rows.append(dict(
            turbine=t, 部件=comp,
            振动结论=r.get('verdict', ''), 语义类=_vc, 级=_vl,   # 判级单源: 前端禁自行正则(27#'销案(定论)'曾被染红)
            设备状态=to_owner_state(r.get('verdict', '')),          # 业主口径五级 (windcms report_std 映射)
            证据状态=to_wcms_level(r.get('verdict', '')),           # 可信度九枚举
            振动信号=r.get('signal', r.get('title', ''))[:80],
            温度通道=_tch, 温度dev=_tdev, 温度判=_tvd, 窗区间=_spans,
            油末样=_olast, 油新鲜=bool(_olast and _olast >= '2026-07'),
            claim_window=r.get('claim_window', '未标'),
            温度轴=temp_note, 油样轴=oil_note,
        ))
    _hm = h.get('meta', {})
    _asof = str(_hm.get('date', '?'))
    _hver = str(_hm.get('version', '?'))
    # 证据窗末 = 全表 claim_window 里最大的日期 (转录源的真实末端, 不写死)
    import re as _re2
    _all_d = sorted({d for w in rows for d in _re2.findall(r'20\d\d-\d\d-\d\d', str(w.get('claim_window', '')))})
    _win_end = _all_d[-1] if _all_d else '?'
    # 补齐全场 (2026-08-28 用户令"逐台, 38台"): handoff 只列有结论的台次, 未收录台原来整行不出现
    # → 页面写"28 台次"而全场 38 台, 读者会以为另外 10 台不存在。未收录台按 _unlisted_cell 给出
    # 明确状态 (正常/数据不足·不可判/两口径差异), 不留空行也不假装有结论。
    if all_turbines:
        _have = {r['turbine'] for r in rows}
        for _t in cfg['turbines']:
            if _t in _have:
                continue
            _u = _unlisted_cell(_t)
            _to = oil[oil.turbine == _t]
            _ol = _to.date.max() if len(_to) else None
            rows.append(dict(
                turbine=_t, 部件='(振动线未收录)',
                振动结论=_u['note'], 语义类='unlisted', 级=_u['level'],
                设备状态=_u.get('ostate_hint') or to_owner_state(_u['note']),
                证据状态=to_wcms_level(_u['note']),
                振动信号='—', claim_window='—', 窗区间=[],
                温度通道=None, 温度dev=None, 温度判=None, 温度轴='(未收录台不参与振动侧温度交叉)',
                油末样=_ol, 油新鲜=bool(_ol and _ol >= '2026-07'),
                油样轴=(f'末样{_ol}' if _ol else '无样')))
        rows.sort(key=lambda r: r['turbine'])

    meta = dict(
        色标=windcms_level_color(),
        数据时点=f'振动证据包 · 出具 {_asof} · 证据窗末 {_win_end}',
        handoff版本=_hver, handoff日期=_asof, 证据窗末=_win_end,
        振动窗=f'至{_win_end[:7]} (证据包出件 {_asof})',
        SCADA窗='至2026-07-07', 油样窗='SGS 2024-11~2025-08 (404份/38台) + 华标2026-07 (102份/17台)',
        纪律=h['meta']['reading_discipline'],
        盲区=h['detectability_prior']['known_blind_spots'],
        open_items=h.get('open_items', []),
    )
    return pd.DataFrame(rows), meta


# windcms 分级色标 (单源: src/windcms/config.LEVEL_COLOR — 2026-08-28 用户令"色标和分级体系参照 windcms";
# 两套系统对同一批振动结论上色必须一致, 否则同一台在两页颜色不同 = 判级看起来不同)
def windcms_level_color():
    try:
        from src.windcms.config import LEVEL_COLOR
        return dict(LEVEL_COLOR)
    except Exception:
        return {}


# 融合层 verdict 文本 → windcms 九级中的哪一级 (转录映射, 不产新判级)
WCMS_LEVEL = [
    # ★顺序敏感 (同一 substring 坑第 3 次): 关闭类必须最先判 — "销案(定论)" 含"定论"
    (('销案', '正样本', '无新证', '可降', '降观察', '无异常'), '正常'),
    (('撤回',), '撤回'),
    (('不可判', '未评估', '数据不足', 'INSUFFICIENT'), 'INSUFFICIENT'),
    (('定论',), '定论'), (('预警',), '准定论·预警'), (('报警',), '准定论·预警'),
    (('新发',), '候选·新发'), (('记基线',), '候选·记基线'),
    (('候选', '监视', '仲裁', '弱同向', '新证'), '候选'),
    (('观察', '降观察', '保留'), '参考'),
    (('正常',), '正常'),
]


def to_owner_state(text):
    """振动结论文本 → windcms 业主口径状态等级 (危险/报警/良好/优秀/不可判).

    2026-08-28 用户令"参照 windcms 优秀/良好/报警/危险 级别体系". windcms 有两套并行:
      ① 证据状态 九枚举 (定论/准定论·预警/候选…/INSUFFICIENT) = 可信度轴
      ② 业主口径状态等级 五级 (危险>报警>良好>优秀>不可判) = 设备状态轴
    映射单源 = src/windcms/report_std.line_state (它对未知词 raise, 不静默塌缩成"优秀" —
    该闸是 2026-08-24 钉的, 我们这边照用不复制词表)。
    台级"不可判"另判: 数据不足/未评估的格子直接给"不可判", 不走线级映射
    (线级 INSUFFICIENT 在 windcms 映射成"优秀" = 单条线不足以判 ≠ 整台没数据)。"""
    t = str(text or '')
    if any(k in t for k in ('数据不足', '未评估', '不可判')):
        return '不可判'
    lv = to_wcms_level(t)
    if lv == '—':
        return '不可判'
    try:
        from app_algorithmModel.app_algorithmModel_guanlan.windcms.report_std import line_state
        return line_state(lv)
    except Exception:
        return {'定论': '危险', '准定论·预警': '报警', '候选·新发': '报警', '候选': '良好',
                '候选·记基线': '良好', '参考': '优秀', 'INSUFFICIENT': '优秀',
                '撤回': '优秀', '正常': '优秀'}.get(lv, '不可判')


def to_wcms_level(text):
    """振动结论/格文案 → windcms 级名 (供 UI 取色; 找不到给 '—')."""
    t = str(text or '')
    for kws, lv in WCMS_LEVEL:
        if any(k in t for k in kws):
            return lv
    return '—'


VERDICT_CLASSES = [
    # 顺序敏感: 关闭类先判 — "销案(定论)"含'定论'但语义是关闭 (2026-08-28 用户逮映射粗暴, 27#误红实锤)
    ('closed', ('销案', '正样本', '无新证', '可降', '降观察'), 'ok'),
    ('meta',   ('测量链', '传感器项'), 'meta'),          # 仪器问题, 不入部件状态
    ('alarm',  ('定论', '报警'), 'bad'),
    ('warn',   ('预警',), 'warn'),
    ('hold',   ('保留', '仲裁输入'), 'note'),            # 义务未闭, 非关闭
    ('watch',  ('候选', '监视', '观察', '跟踪', '弱同向', '新证'), 'note'),
]


def verdict_class(verdict):
    """振动线 verdict 文本 → (语义类, 矩阵级). 单源: taxonomy 与本模块矩阵共用.
    未识别词响亮标注而非静默归类."""
    v = str(verdict)
    for cls, kws, lv in VERDICT_CLASSES:
        if any(k in v for k in kws):
            return cls, lv
    return 'unknown', 'note'


def _vib_level(verdict):
    """振动 verdict 文本 → 矩阵格状态级 (转录分级, 不产判级). meta(测量链)降 note 显示."""
    lv = verdict_class(verdict)[1]
    return 'note' if lv == 'meta' else lv


def _ch_cn(ch):
    """通道中文名 (矩阵格用短名; i18n 未收录则回退原缩写, 不静默造名)."""
    from src.windscada import i18n
    nm = i18n.ch_cn(ch, with_tag=False)
    if nm == ch:                       # 未收录 → 去前后缀后原样给
        return ch.replace('tmp_wtc_', '').replace('_mean', '')
    return nm.replace('·均值', '').replace('·最大', '')


_WCMS = None


def _unlisted_cell(t):
    """振动线未收录该台时, 说清是哪一种 (2026-08-28 用户令"说清楚问题: 数据不足以判断 还是 正常运行"):
      ① 报告判不可判 → 数据不足, 不可判 (监测盲区/数据缺, 不是正常)
      ② 报告判报警/危险 → 两口径差异! 状态管理判报警而定谳链未收录 (须对拍, 不能显示成"未收录")
      ③ 报告判良好/优秀 → 已评估, 定谳链无需单列 (可作"本轮无异常"读)
      ④ 报告也没有 → 真未评估"""
    g = windcms_grades().get(t)
    if not g:
        return dict(level='unlisted', ostate_hint='不可判',
                    note='未评估 — 不等于正常')
    lv, dt = g['综合'], g['日期']
    if lv in ('不可判',):
        return dict(level='unlisted', ostate_hint='不可判',
                    note='振动线未收录 · 数据不足不可判 (非正常)')
    if lv in ('报警', '危险'):
        return dict(level='warn', ostate_hint=lv,
                    note=f'⚠两口径差异 — 评估报告 {dt} 判{lv}, 振动线未收录')
    # 与其他列同词 (用户: "评估无异常不就是正常吗"); 来源留浮层。
    # ★但等级必须照报告原级 — 原来一律写"正常"会让业主状态统一抬成"优秀",
    #   把报告判"良好"的台降级掉了 (2026-08-28 补全 38 台时实逮)。
    # ★振动列只讲振动 (2026-08-28 用户"还是和之前的振动分析结果不一致"):
    # 原来未收录台在"振动"列写"良好 · 评估报告 2026-08-28" — 那是设备状态评估报告的等级,
    # 不是振动线结论, 摆在振动列里必然被读成振动结论。逐台对拍证实: 26 台有 handoff 结论的
    # 全部一致, 用户看到的"不一致"全部来自这 10 台未收录台的借位显示。
    # 评估报告等级仍经 ostate_hint 驱动右侧设备状态徽章, 信息不丢, 只是不占振动列。
    return dict(level='ok', ostate_hint=lv,
                note='振动线未收录 (设备状态见右侧徽章)')


def windcms_grades():
    """windcms 最新评估报告的 38 台综合等级 (用于把"未收录"说清楚: 数据不足 vs 已评估无异常).
    振动线 handoff 只列有结论的台, 未列不等于正常 — 但也不等于没测; 报告等级能区分这两者。"""
    global _WCMS
    if _WCMS is not None:
        return _WCMS
    _WCMS = {}
    try:
        from src.windcms.config import cms_out     # 与写侧同一解析口 (WINDCMS_OUT; 见 windcms.config.cms_out)
        rp = sorted(cms_out().glob('报告_CMS振动状态评估报告_*.md'))[-1]
        md = rp.read_text(encoding='utf-8')
        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) > 5:
                _WCMS[c[0]] = dict(综合=c[5], 日期=rp.stem.rsplit('_', 1)[-1])
    except Exception:
        pass
    return _WCMS


def matrix(cfg=None, months=None):
    """§1 全场融合矩阵: 38台 × 4源 (振动/温度/润滑/油液), 每格状态+时效.

    months = 全局时间窗的月列表 (None=全程) — 2026-08-27 用户令"整体时间窗": 温度/柱塞格的
    现窗=所选窗末月, 记忆轨迹裁剪到窗内 (趋势参考=窗内趋势); 油液格=窗内样本;
    振动格 verdict 是振动线 claim 窗的转录, **不随本窗变** (证据窗固定, 列头注明).
    振动格纪律: handoff 只列有结论的台 → 未列台显示"未列", **未列≠正常** (盲区见§3);
    油液全场 stale (末样距今>12月) → 该列只作时效展示不作现时状态."""
    cfg = cfg or farm()
    h = load_handoff()
    oil = oil_window(cfg)
    if months:
        oil = oil[oil.date.str[:7].isin(months)]
    st = pathlib.Path(cfg['store'])
    from . import temp_nbm
    treg = temp_nbm.registry()
    tdf = treg[0] if isinstance(treg, tuple) else treg

    # 振动: 逐台聚合 handoff verdicts (一台多部件取最重级)
    # ★2026-09-18 修: 这里原来直接取 `h['per_turbine']` —— 而 `load_handoff()` 在缺件时返回的
    #   正是**空 dict**(2026-09-17 改成响亮降级, 不再 raise), 于是清过产物、重算后没有
    #   handoff_vibration_v2.json 的机器上, `/api/fleet` 的融合面回的是 `{"err": "'per_turbine'"}`
    #   —— 一个 KeyError 被当成"这一面坏了"显示给用户, 而不是"缺哪件"。缺件就走空表 + gap 说明。
    vib = {}
    if not h or not (h.get('per_turbine') or {}):
        # 缺件就走**空表 + 列头说明**(与 fusion_table 同一口径): 页面显示"缺哪件", 不显示 KeyError。
        gap = GAP_HANDOFF[0] or (f'融合面缺件: {HANDOFF} (包内无生成端, docs §13.6) —— '
                                 '运行期不从交付包补齐(用户令 2026-09-17); 要出件须研发补生成端')
        return [], dict(振动=f'缺件: {HANDOFF.name}', 温度='', 润滑='', 油液='', months=[], gap=gap)
    v1 = [(r, '发电机') for r in h['per_turbine'].get('v1_generator', [])] + \
         [(r, '齿轮箱') for r in h['per_turbine'].get('v1_gearbox', [])]
    new = [(r, r.get('component', '?')) for r in h['per_turbine'].get('new_2026_08_26', [])]
    SEV = {'bad': 3, 'warn': 2, 'note': 1, 'ok': 0}
    for r, comp in v1 + new:
        t = r['turbine']; lv = _vib_level(r.get('verdict', ''))
        cur = vib.get(t)
        if cur is None or SEV[lv] > SEV[cur['level']]:
            vib[t] = dict(level=lv, note=f"{comp}: {r.get('verdict','')}")
        elif SEV[lv] == SEV[cur['level']] and comp not in cur['note']:
            cur['note'] += f"; {comp}: {r.get('verdict','')}"

    # 润滑 (主轴承柱塞): **信息列非判据列** — 跨台单次时长天然异质 (正常月 p10~p90=22~961s,
    # 各台润滑设定不同, 横比不成立; 2026-08-27 实测), 且现窗6天窗全场共模拖长 (大风连转窗效应).
    # 单格判级只随机制链定案 (29# = 19个月纵向形态+温度/振动时序锁定, 非单月横比), 由 handoff 驱动.
    lubp = st / 'mblub_monthly.parquet'
    lub = pd.read_parquet(lubp) if lubp.exists() else None
    if lub is not None and months:
        lub = lub[lub.month.isin(months)]
        if not len(lub): lub = None          # 窗内无柱塞数据 (如纯2024窗) → 该列"窗内无数据"
    lub_cell = {}
    if lub is not None:
        last_m = lub.month.max()
        cur = lub[lub.month == last_m].set_index('turbine')
        per = (cur.pis / cur.cnt.replace(0, pd.NA)).dropna()
        rank = per.rank(ascending=False, method='min')
        chain_bad = {r['turbine'] for r, _ in new if '润滑' in str(r.get('component', '')) + str(r.get('title', ''))}
        for t in per.index:
            v = float(per[t]); rk = int(rank[t])
            lv = 'bad' if t in chain_bad else 'note'
            # tag 窗无关措辞: "泵打不满"特指 2026-07 形态, 单月历史窗下挂具体形态词会时间错位
            tag = ' [润滑断供机制链台(v5定案), 各期形态见轨迹]' if t in chain_bad else ''
            lub_cell[t] = dict(level=lv, note=f'{v:.0f}s/次 第{rk}/38{tag}', hist_label='单次循环时长逐月(s)')

    # 温度格双口径 (振动线⑥采纳, 全台不只 WATCHLIST 台): NBM 变化口径 + watch 月度水平位次口径 —
    # 变化型对"稳定高位"失明 (29# 后主轴承 19月位次1-11 而 NBM 判"—"), 两口径并示互补盲区
    wp = pathlib.Path(cfg['store']) / 'watch_channels_monthly.parquet'
    watch = pd.read_parquet(wp) if wp.exists() else None
    if watch is not None and months:
        watch = watch[watch.month.isin(months)]
        if not len(watch): watch = None

    def _level_rank(ch, t):
        # ★窗对齐 (2026-08-28 用户逮"振动融合没核对时间窗"): 原取全局末月, 而列头声明的是所选窗末月
        # ⇒ 选 2025H2 时列头写"窗末月2025-12"而格内是 2026-07 的数, 声明与实际不符。
        # NBM 变化口径本就是固定近90日 (列头已标"不随本窗"), 但 watch 月度水平位次必须落在所选窗内。
        if watch is None or ch not in watch.columns: return ''
        piv = watch.pivot_table(index='month', columns='turbine', values=ch)
        piv = piv.dropna(how='all')
        if months:
            piv = piv[piv.index.isin([str(m) for m in months])]
        if not len(piv): return ''
        last = piv.index.max()
        row = piv.loc[last]
        if pd.isna(row.get(t)): return ''
        dv = float(row[t] - row.median())
        rk = int(row.rank(ascending=False, method='average')[t])
        return f" · 现水平 {dv:+.1f}K 第{rk}/38"

    # 历史记忆轨迹 (2026-08-27 用户令: 显示最新窗, 旧窗变成记忆看趋势 — 现窗快照会丢形态,
    # 如 29# 柱塞两段式塌→回升→翻转拖长; 每格带逐月序列, 前端画迷你趋势线现窗点高亮)
    def _temp_hist(ch, t):
        if watch is None or ch not in watch.columns: return None
        piv = watch.pivot_table(index='month', columns='turbine', values=ch).dropna(how='all')
        if months:
            piv = piv[piv.index.isin([str(m) for m in months])]
        if t not in piv.columns: return None
        dev = (piv[t] - piv.median(axis=1)).round(1)
        return [None if pd.isna(x) else float(x) for x in dev]

    lub_hist = {}
    if lub is not None:
        lper = lub.assign(per=(lub.pis / lub.cnt.replace(0, pd.NA))).pivot_table(index='month', columns='turbine', values='per')
        for t in lper.columns:
            lub_hist[t] = [None if pd.isna(x) else round(float(x), 1) for x in lper[t]]
    months_axis = sorted(watch.month.unique()) if watch is not None else []

    # 温度列只收传动链三部件的通道 (2026-08-28 用户令"各列只列与问题相关的部件: 主轴/发电机/齿轮箱"):
    # 本矩阵的行是"该台传动链状态", 拿变压器/液压油温的最大偏来填温度列 = 跨部件充数,
    # 与 fusion_table 早已执行的 COMP_CH 路由纪律("不得拿别的部件温度充数")一致。
    # 非传动链通道 (变压器/机舱/环温/液压) 在分系统页各自的轴上看, 不进本列。
    _DRIVE_STEM = ('MBear', 'MBearH', 'MBearG',            # 主轴承前/后
                   'GeOilTmp', 'IMS', 'HS', 'GBox',        # 齿轮箱油/中间轴/高速轴
                   'GenBe', 'GenIV', 'GenSt')              # 发电机轴承/绕组
    rows = []
    for t in cfg['turbines']:
        v = vib.get(t)
        sub = tdf[tdf.turbine == t]
        sub = sub[sub.channel.str.contains('|'.join(_DRIVE_STEM), na=False)]
        hit = sub[sub['判'] != '—']
        if len(hit):
            top = hit.reindex(hit.dev_K.abs().sort_values(ascending=False).index).iloc[0]
            nm = top.channel.replace('tmp_wtc_', '').replace('_mean', '')
            tcell = dict(level='warn', note=f"{_ch_cn(top.channel)} {top.dev_K:+.1f}K {top['判']}" + _level_rank(top.channel, t),
                         hist=_temp_hist(top.channel, t), hist_label=f'{nm} 月度dev轨迹')
        elif len(sub):
            top = sub.reindex(sub.dev_K.abs().sort_values(ascending=False).index).iloc[0]
            nm = top.channel.replace('tmp_wtc_', '').replace('_mean', '')
            tcell = dict(level='ok', note=f"{_ch_cn(top.channel)} {top.dev_K:+.1f}K" + _level_rank(top.channel, t),
                         hist=_temp_hist(top.channel, t), hist_label=f'{nm} 月度dev轨迹')
        else:
            tcell = dict(level='ok', note='传动链温度通道无越档')
        to = oil[oil.turbine == t]
        if len(to):
            last = to.date.max()
            fresh = last >= '2026-07'          # 华标2026-07轮台 = 有近共时样, 不再挂 stale
            _hbn = huabiao_analysis(t) if fresh else (huabiao_analysis(t) if t == 'WTG29' else None)
            ocell = dict(level=('ok' if fresh else 'stale'),
                         note=f"{len(to)}样 末{last[:7]}" + ('' if fresh else ' [stale>12月]')
                              + (f" · {_hbn}" if _hbn else (' · 华标2026-07全位判正常' if fresh else '')),
                         hist_label='采样时间线')
        else:
            ocell = dict(level='stale', note='窗内无样' if months else '无样')
        _vc = v or _unlisted_cell(t)
        for _cell in (_vc, tcell, lub_cell.get(t) or {}, ocell):
            if isinstance(_cell, dict):
                _cell['wlevel'] = to_wcms_level(_cell.get('note', ''))     # 证据状态轴 (可信度)
                _cell['ostate'] = _cell.get('ostate_hint') or to_owner_state(_cell.get('note', ''))   # 业主口径状态轴
        rows.append(dict(
            turbine=t,
            振动=_vc,
            温度=tcell, 润滑={**lub_cell.get(t, dict(level='note', note='无仓')), 'hist': lub_hist.get(t)}, 油液=ocell,
        ))
    heads = dict(振动='证据窗固定, 不随本窗 (至2026-08)',
                 温度=f"窗末月{watch.month.max() if watch is not None else '?'} (变化口径=同型机群比对·近90日固定)",
                 润滑=f"窗末月{lub.month.max() if lub is not None else '无数据'}",
                 油液='窗内样本 (华标2026-07轮17台=近共时; 其余台末样2025-08 stale)')
    heads['months'] = months_axis
    return rows, heads


RENDERED_DP_KEYS = {'purpose', 'classes', 'meas_crossing', 'known_blind_spots', '端到端闭环'}


def capability():
    """§3 振动能力边界: **全量透传 + 白名单渲染** (2026-08-27 双线约定, 替代白名单透传 —
    v2.2 "端到端闭环"新键曾静默漏消费, 教训='读同路径自动吃到'只对既有字段成立).
    生产方新增顶层键 → unrendered 自动列出提示接线, 不再依赖交接消息记得说.

    ★2026-09-18 修 (同 matrix 的教训): 这里原来直接取 `h['detectability_prior']` —— 缺 handoff 时
    `load_handoff()` 返回空 dict, 于是 `KeyError: 'detectability_prior'` 被 fleet_view 包成
    `fus={"err": "'per_turbine'"}` 之类的**程序错**显示给人。缺件就该说"缺件"。
    """
    h = load_handoff()
    dp = h.get('detectability_prior') or {}
    if not dp:
        return dict(classes={}, blind=[], meas={}, purpose='', loop={}, unrendered=[],
                    gap=GAP_HANDOFF[0] or (f'振动能力边界需 {HANDOFF.name} (包内无生成端, docs §13.6); '
                                           '本机该件不在位 ⇒ 如实为空'))
    return dict(classes=dp.get('classes', {}), blind=dp.get('known_blind_spots', []),
                meas=dp.get('meas_crossing', {}), purpose=dp.get('purpose', ''),
                loop=dp.get('端到端闭环', {}),
                unrendered=sorted(k for k in dp if k not in RENDERED_DP_KEYS))


# 传动链三部件的报警/工单族过滤 (2026-08-27 用户令: 齿轮箱/发电机/主轴的报警与工单进融合板块)
# 报警按 text 关键词 (码表对译后中文); 发电机族排除逆变器/通讯类 (13121 属变流器侧非发电机本体);
# 主轴工单含"传动链"二级但排除滑环 (集团模板"(双馈)传动链"错标, 本场鼠笼异步无滑环属发电机励磁问题域)
DRIVETRAIN = {
    '齿轮箱': dict(alarm_pat=r'齿轮', wo_pat=r'齿轮'),
    '发电机': dict(alarm_pat=r'发电机', alarm_excl=r'逆变器|通讯', wo_pat=r'发电机'),
    '主轴': dict(alarm_pat=r'主轴', wo_pat=r'主轴', wo_extra='传动链', wo_excl=r'滑环'),
}


def drivetrain_events(months=None, cfg=None):
    """传动链三部件 报警+工单, 按月窗过滤 (months=None 为全程).

    ⚠两源时间几乎不重叠: 报警仓 2025-01 起 / 工单 2020~2024-11 止 (2025-26 月表为集团模板无本场记录)
    — 选 2026 类窗工单必空, 空=数据边界非"无工单", 边界声明随节常显."""
    cfg = cfg or farm()
    st = pathlib.Path(cfg['store'])
    a = pd.read_parquet(st / 'alarms.parquet')
    w = pd.read_parquet(st / 'workorders.parquet')
    w = w[w.t_report > '2019']                     # 1970 epoch 脏日期剔除
    a['month'] = a.t_on.dt.to_period('M').astype(str)
    w['month'] = w.t_report.dt.to_period('M').astype(str)
    if months:
        a = a[a.month.isin(months)]
        w = w[w.month.isin(months)]
    out = {}
    for comp, cf in DRIVETRAIN.items():
        aa = a[a.text.str.contains(cf['alarm_pat'], na=False)]
        if cf.get('alarm_excl'):
            aa = aa[~aa.text.str.contains(cf['alarm_excl'], na=False)]
        code_stats = (aa.groupby(['code', 'text']).agg(条数=('t_on', 'size'), 台数=('turbine', 'nunique'),
                                                       时长h=('dur_true_s', lambda x: round(float(x.sum()) / 3600, 1)))
                      .reset_index().sort_values('条数', ascending=False))
        per_t = aa.groupby('turbine').size().sort_values(ascending=False)
        recent = aa.sort_values('t_on', ascending=False).head(12)
        alarm_rows = [dict(t=r.turbine, code=r.code, text=r.text, on=str(r.t_on)[:16],
                           dur_min=round(float(r.dur_true_s) / 60, 1) if r.dur_true_s == r.dur_true_s else None)
                      for _, r in recent.iterrows()]
        cols = ['故障位置二级', '故障位置三级', '维修对象', '故障名称']
        hit = w[cols].astype(str).apply(lambda c: c.str.contains(cf['wo_pat'], na=False)).any(axis=1)
        if cf.get('wo_extra'):
            hit = hit | w['故障位置二级'].astype(str).str.contains(cf['wo_extra'], na=False)
        if cf.get('wo_excl'):
            hit = hit & ~w[cols].astype(str).apply(lambda c: c.str.contains(cf['wo_excl'], na=False)).any(axis=1)
        ww = w[hit].sort_values('t_report', ascending=False)
        wo_rows = [dict(t=r.turbine, date=str(r.t_report)[:10], name=str(r['故障名称'])[:40],
                        act=f"{r['维修动作']}·{r['维修对象']}" if pd.notna(r['维修动作']) else str(r['排查项目'])[:30],
                        cause=str(r['故障原因'])[:40] if pd.notna(r['故障原因']) else '')
                   for _, r in ww.head(15).iterrows()]
        cs = [dict(code=str(r['code']), text=str(r['text']), 条数=int(r['条数']), 台数=int(r['台数']),
                   时长h=float(r['时长h'])) for r in code_stats.head(8).to_dict('records')]
        out[comp] = dict(
            报警=dict(总条数=int(len(aa)), 台数=int(aa.turbine.nunique()), 码统计=cs,
                    top台={t: int(n) for t, n in per_t.head(6).items()}, 近况=alarm_rows),
            工单=dict(总条数=int(len(ww)), 台数=int(ww.turbine.nunique()), 明细=wo_rows),
        )
    bounds = ('报警仓 2025-01-01~2026-07-15 / 工单 2020-01~2024-11-21 (2025-26 月表为集团模板无本场记录) '
              '— 两源时间几乎不重叠: 2026类窗工单空=数据边界非"无工单"; 全程窗才两源齐')
    return out, bounds