guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/taxonomy.py

# -*- coding: utf-8 -*-
"""B翼 系统划分 (2026-08-25 用户令: 按交付分册的系统学重组, 温度NBM是方法不是系统).
九系统 = 交付汇编 01_分系统 蓝本: 变桨/偏航/主轴承/齿轮箱/发电机/变流器/主控与传感网 (+叶片/塔架=SCADA无轴, 边界注记).
每系统状态 = worst(该系统全部轴), 依据原文保留; 对账基线 = reference/rudong/<系统>_delivered_grades.json."""
from app_common.app_common_guanlan.api import paths as P
import json, pathlib
import numpy as np, pandas as pd
from src.windscada.config import farm
from app_ETL.app_ETL_guanlan.api import load_10min

SYSTEMS = ['变桨', '偏航', '主轴承', '齿轮箱', '发电机', '变流器', '主控与传感网', '叶片与叶根', '塔架与基础']
# 9 系统 = 交付 9 分册一一对应 (2026-08-26 补齐: 先前只做 7 个, 缺叶片与叶根/塔架与基础)
NO_SCADA = {'叶片与叶根': '无SCADA轴(分册蓝本=CMS叶频/巡检)', '塔架与基础': '无SCADA轴(分册蓝本=CMS塔频/巡检)'}

# 温度通道 → 系统 (通道身份按契约命名指纹; GFilB*=定值80℃加热器带 [恒值哨兵], Brk=高速轴刹车归齿轮箱侧)
TEMP2SYS = {
    'MBearGTm': '主轴承', 'MBearHTm': '主轴承',
    'GeOilTmp': '齿轮箱', 'IMSGenTm': '齿轮箱', 'IMSRotTm': '齿轮箱', 'HSGenTmp': '齿轮箱', 'HSRotTmp': '齿轮箱',
    'GFilB1Tm': '齿轮箱', 'GFilB2Tm': '齿轮箱', 'GFilB3Tm': '齿轮箱', 'BrkTmpGn': '齿轮箱', 'BrkTmpGr': '齿轮箱',
    'GenBeGTm': '发电机', 'GenBeRTm': '发电机', 'Gen1U1Tm': '发电机', 'Gen1V1Tm': '发电机', 'Gen1W1Tm': '发电机',
    'ConvWTmp': '变流器', 'TrafoTMa': '变流器', 'TraOilTF': '变流器', 'TraRooTF': '变流器',
    'A1ExtTmp': '主控与传感网', 'A21IntTm': '主控与传感网', 'A3LefTmp': '主控与传感网', 'A3RigTmp': '主控与传感网',
    'NacelTmp': '主控与传感网', 'HubTemp': '主控与传感网',
    'HydOilTm': '变桨',
    'AmbieTmp': None,   # 环温=热汇参考, 不归系统
}
LUB2SYS = {'din_wtc_MBLubPis_timeon': '主轴承', 'din_wtc_GBLubPis_timeon': '齿轮箱', 'dot_wtc_GenLubPu_timeon': '发电机'}

def build_aux_store(cfg=None, since='2026-01-01'):
    """轻量补仓: 三系统润滑柱塞 duty (s/日) + ScComErr 通讯计数增量."""
    cfg = cfg or farm()
    rows = []
    for t in cfg['turbines']:
        rec = dict(turbine=t)
        try:
            d = load_10min(t, cfg, groups=['A.状态', 'B.润滑液压'])
            cur = d[d.ts >= since]
            byday = cur.groupby(cur.ts.dt.date)
            for col, sysname in LUB2SYS.items():
                rec[f'lub_{sysname}'] = float(byday[col].sum().mean())
            v = cur['cnt_wtc_ScComErr_accum'].dropna()
            rec['scomerr'] = float(v.iloc[-1] - v.iloc[0]) if len(v) > 2 else np.nan
        except Exception as e:
            rec['err'] = str(e)[:60]
        rows.append(rec); print(t, flush=True)
    df = pd.DataFrame(rows)
    df.to_parquet(pathlib.Path(cfg['store']) / 'system_aux.parquet')
    print('→ system_aux.parquet', df.shape)

ORDER = {'报警': 0, '良好': 1, '不可判': 2, '优秀': 3, '—': 3}

def _worst(states):
    js = [s for s in states if s in ORDER]
    return sorted(js, key=lambda x: ORDER[x])[0] if js else '不可判'

def _load_tc(cfg):
    import pandas as _pd, pathlib as _pl
    f = _pl.Path(cfg['store']) / 'thermal_chain.parquet'
    return _pd.read_parquet(f).set_index('turbine') if f.exists() else None


def system_matrix(cfg=None, span=None):
    """38台 × 7系统 状态 + 依据. 轴: 分系统模块(变桨/偏航/蓄能) + 温度通道归系统 + 润滑柱塞×fleet + 测风/通讯.

    `span=(起, 止)` 给了就**按所选时间窗重算判级**（用户令 2026-09-21，含两端）：
    四个判级面各自把它当判级窗（变桨=日粒度件切片、偏航/蓄能/温度=走 slim10min 窄仓重算；
    温度的 ref/ref_sm 对照窗仍固定）。不给 = 各面原口径（近 60/90 日）。
    """
    cfg = cfg or farm()
    TC = _load_tc(cfg)

    # ★ 2026-09-17 缺件容错: 交付包不随产物 ⇒ 这批"无生成端"的件(pitch_daily/yaw_daily/system_aux/
    #   thermal_chain…)在目标机上必然缺。这里逐个兜住: 缺件 = 该面无数据 (记 GAPS, 页面显示缺件),
    #   绝不让 KeyError/FileNotFoundError 把本体层整步打断 (实测清产物后重算就死在这里)。
    GAPS: list[str] = []

    def _missing(what: str, why: str = ''):
        tag = f'{what} 缺件 ({why})' if why else f'{what} 缺件'
        if tag not in GAPS:
            GAPS.append(tag)

    def _load_reg(fn, what, idx_col):
        try:
            reg = fn()
            reg = reg[0] if isinstance(reg, tuple) else reg
            return reg.set_index(idx_col)
        except FileNotFoundError as e:
            _missing(what, str(e).split("'")[-2] if "'" in str(e) else str(e)[:60])
        except Exception as e:
            _missing(what, f'{type(e).__name__}')
        return pd.DataFrame()

    def _load_parquet(name):
        f = pathlib.Path(cfg['store']) / name
        if not f.exists():
            _missing(name)
            return pd.DataFrame()
        try:
            return pd.read_parquet(f).set_index('turbine')
        except Exception as e:
            _missing(name, f'{type(e).__name__}')
            return pd.DataFrame()

    def _cell(i, t_, col, default='不可判'):
        """按台取一格 —— 表空/行缺/列缺都返回 default (缺件时该轴不可判, 不是崩溃)。"""
        try:
            v = i.loc[t_, col]
        except Exception:
            return default
        try:
            if v != v:          # NaN
                return default
        except Exception:
            pass
        return v

    try:
        from .subsys import structure as stxmod
        _stx, STXD = stxmod.registry(cfg)
        STX = _stx.set_index('机组')
    except Exception:
        STX, STXD = None, {}
    from .subsys import pitch as pitchmod, yaw as yawmod, hydraulic, temp_nbm
    # ★2026-09-21 用户令「判级也按所选时间窗重算」: span 给了就逐面按窗重算（各面自己决定怎么按窗取数）。
    #   四个面里只有"变桨"的输入是日粒度序列；偏航/蓄能/温度走 slim10min 窄仓重算（同一份 load_10min 的列子集）。
    p_i = _load_reg(lambda: pitchmod.registry(span=span), '变桨面 pitch_daily.parquet', '机组')
    y_i = _load_reg(lambda: yawmod.registry(span=span), '偏航面 yaw_daily.parquet', '机组')
    h_i = _load_reg(lambda: hydraulic.registry(span=span), '蓄能面 (液压深挖件)', 'turbine')
    treg = temp_nbm.registry(win='cur', span=span)
    tdf = treg[0] if isinstance(treg, tuple) else treg
    if tdf is None or not len(tdf):
        tdf = pd.DataFrame(columns=['turbine', 'channel', 'dev_K', '判'])
    aux = _load_parquet('system_aux.parquet')
    ydf = _load_parquet('yaw_daily.parquet')
    pairs = temp_nbm.pairs_table()
    try:
        pcd = pd.read_parquet(pathlib.Path(cfg['store']) / 'powercurve_dev.parquet')
        pc_a = {r.turbine: f"dev{r.dev_w:+.1%} ws偏置{r.ws_bias:+.2f}" for _, r in pcd.iterrows() if '风速计' in str(r['判别'])}
    except Exception:
        pc_a = {}
    lub_med = {s: aux[f'lub_{s}'].median() for s in ('主轴承', '齿轮箱', '发电机')}
    sc_med = aux['scomerr'].median()
    tmap = {'温度候选(定向查)': '报警', '传感器不可信(A类)': '报警', '记基线观察': '良好', '不可判(返服嫌疑)': '不可判'}
    result = {}
    for t in cfg['turbines']:
        sysd = {}
        # 每系统: (states, 依据 list)
        acc = {s: ([], []) for s in SYSTEMS}
        # 变桨 = 分系统四维 + 蓄能深挖
        acc['变桨'][0].append(_cell(p_i, t, '系统级'))
        _pd_ = [_cell(p_i, t, d, '') for d in pitchmod.DIMS]
        if any(_pd_):
            acc['变桨'][1].append('四维:' + '/'.join(f'{d}{v}' for d, v in zip(pitchmod.DIMS, _pd_) if v))
        else:
            acc['变桨'][1].append('四维: 缺件, 不可判')
        hv = _cell(h_i, t, '判', '—')
        acc['变桨'][0].append('报警' if '候选' in hv else ('良好' if '记基线' in hv else '优秀'))
        if hv != '—': acc['变桨'][1].append(f"蓄能: {h_i.loc[t, '依据']}")
        # 偏航
        acc['偏航'][0].append(_cell(y_i, t, '系统级'))
        _yd_ = [_cell(y_i, t, d, '') for d in yawmod.DIMS]
        if any(_yd_):
            acc['偏航'][1].append('四维:' + '/'.join(f'{d[:2]}{v}' for d, v in zip(yawmod.DIMS, _yd_) if v))
        else:
            acc['偏航'][1].append('四维: 缺件, 不可判')
        # 温度通道归系统
        for _, r in tdf[tdf['turbine'] == t].iterrows():
            ch = r['channel'].replace('tmp_wtc_', '').replace('_mean', '').replace('_max', '')
            sysname = TEMP2SYS.get(ch)
            if not sysname: continue
            st = tmap.get(str(r['判']), None)
            if st:
                acc[sysname][0].append(st)
                acc[sysname][1].append(f"{ch} {r['dev_K']:+.1f}K({r['判'][:4]})")
        # 配对差 (主轴承双温差 / 发电机DE-NDE)
        for _, pr in pairs[pairs['turbine'] == t].iterrows():
            sysname = '主轴承' if '主轴承' in str(pr['pair']) or 'MBear' in str(pr['pair']) else '发电机'
            if abs(pr['pair_dev']) >= 4:
                acc[sysname][0].append('良好')
                acc[sysname][1].append(f"配对差{pr['pair']} {pr['pair_diff']:+.1f}K(偏{pr['pair_dev']:+.1f})")
        # 润滑柱塞 ×fleet (筛查级; 单窗相对门弱, 只报不定级除非极端)
        for s in ('主轴承', '齿轮箱', '发电机'):
            v = _cell(aux, t, f'lub_{s}', float('nan'))
            x = v / max(lub_med[s], 1e-9) if v == v else np.nan
            if x == x and (x >= 2.5 or x <= 0.25):
                acc[s][0].append('良好')
                acc[s][1].append(f"润滑duty {v:.0f}s/日 ({x:.2f}×fleet)")
        # 热链轴 (M13): 节点−热汇ΔT / 冷却执行器 duty / 三相极差 → 归所属系统
        if TC is not None and t in TC.index:
            tr = TC.loc[t]
            for col, sysn, thr, lab in (('齿轮箱高速轴−油', '齿轮箱', 4.0, '高速轴−油ΔT'),
                                        ('齿轮箱中速轴−油', '齿轮箱', 4.0, '中速轴−油ΔT'),
                                        ('齿箱油−冷却水', '齿轮箱', 4.0, '油−冷却水ΔT'),
                                        ('发电机绕组−冷却水', '发电机', 6.0, '绕组−冷却水ΔT')):
                v = tr.get(col)
                if v == v and (v - TC[col].median()) >= thr:
                    acc[sysn][0].append('良好')
                    acc[sysn][1].append(f'{lab} {v:.1f}K (超全场中位{v - TC[col].median():+.1f}K)')
            ph = tr.get('三相极差')
            if ph == ph and ph >= max(3.0, 2.5 * TC['三相极差'].median()):
                acc['发电机'][0].append('良好')
                acc['发电机'][1].append(f'绕组三相极差 {ph:.1f}K ({ph / TC["三相极差"].median():.1f}×fleet, 仪表/接线候选)')
            cd = tr.get('冷却器高速_高油温档'); cm = TC['冷却器高速_高油温档'].median()
            if cd == cd and cm > 0.02 and cd >= max(1.5 * cm, cm + 0.1):
                acc['齿轮箱'][0].append('良好')
                acc['齿轮箱'][1].append(f'冷却器高速档 duty {cd:.2f} ({cd / cm:.2f}×fleet, 同油温档) 换热能力低候选')
        # 叶片与叶根 / 塔架与基础 (M16 结构件)
        if STX is not None and t in STX.index:
            sr = STX.loc[t]
            for dim in ('叶片A', '叶片B', '叶片C', '叶根螺栓'):
                if sr.get(dim) in ('报警', '良好'):
                    acc['叶片与叶根'][0].append(sr[dim]); acc['叶片与叶根'][1].append(f"{dim}: {STXD[t]['叶片'][dim][:60]}")
            for dim in ('塔筒频率', '塔筒振动', '塔筒湿度'):
                v = sr.get(dim)
                if v in ('报警', '良好'):
                    acc['塔架与基础'][0].append(v); acc['塔架与基础'][1].append(f"{dim}: {STXD[t]['塔架'][dim][:70]}")
                elif v == '监测不可用':
                    acc['塔架与基础'][0].append('不可判'); acc['塔架与基础'][1].append(f"{dim}: {STXD[t]['塔架'][dim][:70]}")
        # 主控与传感网: 测风(含曲线A类风速计路由) + 通讯
        if pc_a.get(t):
            acc['主控与传感网'][0].append('报警')
            acc['主控与传感网'][1].append(f"曲线判风速计偏置A类 ({pc_a[t]})")
        wf = float((_cell(ydf, t, 'windfau_s', 0) or 0) + (_cell(ydf, t, 'windfau_t', 0) or 0))
        yfa = int(_cell(ydf, t, 'D4风传感', 0) or 0)
        if wf >= 100 or yfa >= 30: acc['主控与传感网'][0].append('报警'); acc['主控与传感网'][1].append(f'测风故障计数{wf:.0f}+报警{yfa}')
        elif wf >= 30 or yfa >= 5: acc['主控与传感网'][0].append('良好'); acc['主控与传感网'][1].append(f'测风计数{wf:.0f}/报警{yfa}')
        sc = _cell(aux, t, 'scomerr', float('nan'))
        scx = sc / max(sc_med, 1e-9) if sc == sc else np.nan
        if scx == scx and scx >= 3 and sc >= 50:
            acc['主控与传感网'][0].append('良好'); acc['主控与传感网'][1].append(f'通讯错误增量{sc:.0f} ({scx:.1f}×fleet)')
        for s in SYSTEMS:
            states, why = acc[s]
            base = _worst(states) if states else '优秀'
            sysd[s] = dict(状态=base, 依据='; '.join(why) if why else '各轴无非常态')
        result[t] = sysd

    if GAPS:
        # 整表挂一条缺件说明 (页面/维护页据此显示"哪些面因缺件不可判"), 不静默造数
        result['_缺件说明'] = dict(GAPS=list(GAPS),
                                注='这些是"包内没有生成端"的产物 (docs §7); 交付包按用户令不随产物, '
                                   '运行期**不由交付包补齐** (用户令 2026-09-17): 放原始件后重算, '
                                   '或由研发补生成端 (逐族可逆性见 docs §13.6)。')

    # 振动映射进分系统状态 (2026-08-28 用户令): handoff 判级按 component 归系统, 取严合并+依据可溯.
    # 定论/报警/预警级 → 该系统升'报警'; 候选/监视/观察级 → 状态不动只追加依据 (筛查级展示态,
    # 振动判级本身已过双模审, 非本层产判级 — 转录取严)
    try:
        from .subsys import fusion as _fus
        _ftab, _ = _fus.fusion_table(cfg)
        _SYSMAP = [('发电机', '发电机'), ('齿轮箱', '齿轮箱'), ('行星', '齿轮箱'), ('中速', '齿轮箱'),
                   ('高速', '齿轮箱'), ('主轴承', '主轴承'), ('主轴', '主轴承'), ('润滑泵', '主轴承')]
        import re as _re
        for _, r in _ftab.iterrows():
            t2, comp, lv = r.turbine, str(r.部件), str(r.振动结论)
            tgt = next((sysn for kw, sysn in _SYSMAP if kw in comp), None)
            if not tgt or t2 not in result:
                continue
            _cls, _mlv = _fus.verdict_class(lv)
            alarm = _mlv in ('bad', 'warn') and _cls != 'meta'   # 测量链问题不入部件状态
            _dates = _re.findall(r'20\d{2}-\d{2}(?:-\d{2})?', str(r.claim_window))
            _cw = f"窗至{max(_dates)}" if _dates else '窗未标'
            note = f"振动: {comp} {lv} ({_cw}; 振动线转录)"
            d2 = result[t2][tgt]
            if alarm:
                d2['状态'] = '报警'
            if d2['依据'] == '各轴无非常态':
                d2['依据'] = note
            elif '振动:' not in d2['依据']:
                d2['依据'] += '; ' + note
        # ── windcms 评估报告转录 (2026-08-28 用户令: 与 windcms 核对设备状态评级) ──
        # 三方对拍逮出两类真问题, 都是"我方界面比 CMS 乐观":
        #   A 漏报 6 处: CMS 判报警而本层显示优秀/良好 (09#齿轮箱 / 16#·20#·23#·38#主轴承 / 17#发电机)
        #   C 10 处: CMS 判不可判而本层显示"优秀" — 把"监测面没覆盖"显示成"设备好", 是纪律问题
        # 合并规则 (取严, 不覆盖本层更强的证据):
        #   CMS 报警/危险 → 抬到该级 (本层已≥该级则不动)
        #   CMS 不可判 且 本层无自有异常证据 → 标不可判 (没测到 ≠ 正常)
        #   CMS 良好/优秀 → 不动 (本层可能有 SCADA 侧证据, 那是另一条轴)
        try:
            _wg = _fus.windcms_grades()
            _WCOL = {'主轴承': ('主轴承前', '主轴承后'), '齿轮箱': ('齿轮箱',), '发电机': ('发电机',)}
            _ORD = ['危险', '报警', '良好', '优秀', '不可判']
            import pathlib as _pl, re as _re3
            from src.windcms.config import cms_out as _cms_out     # 读写同一口 (WINDCMS_OUT 不能被读侧忽略)
            _wmd = sorted(_cms_out().glob('报告_CMS振动状态评估报告_*.md'))
            _wdate = _wmd[-1].stem.rsplit('_', 1)[-1] if _wmd else '?'
            _wpart = {}
            if _wmd:
                for _l in _wmd[-1].read_text(encoding='utf-8').split('## 附录 A')[1].splitlines():
                    if _l.startswith('| WTG'):
                        _c = [x.strip() for x in _l.strip('|').split('|')]
                        _wpart[_c[0]] = dict(主轴承前=_c[1], 主轴承后=_c[2], 齿轮箱=_c[3], 发电机=_c[4])
            for _t3, _row3 in _wpart.items():
                if _t3 not in result:
                    continue
                for _sysn3, _keys in _WCOL.items():
                    _lvs = [_row3[k] for k in _keys if _row3.get(k) in _ORD]
                    if not _lvs:
                        continue
                    _w = min(_lvs, key=_ORD.index)
                    _d3 = result[_t3][_sysn3]
                    _cur = _d3['状态']
                    _note3 = f"CMS评估报告{_wdate}判{_w}"
                    if _w in ('危险', '报警') and _cur not in ('危险', '报警'):
                        _d3['状态'] = _w
                        _d3['依据'] = (_note3 + ('; ' + _d3['依据'] if _d3['依据'] != '各轴无非常态' else '')).strip()
                    elif _w == '不可判' and _cur in ('优秀', '良好'):
                        _d3['状态'] = '不可判'
                        _d3['依据'] = f"{_note3} (监测面未覆盖 — 没测到≠正常)" + (
                            '; 本层: ' + _d3['依据'] if _d3['依据'] != '各轴无非常态' else '')
                    elif _note3 not in _d3['依据'] and _w != _cur:
                        _d3['依据'] = (_d3['依据'] + f"; {_note3}").lstrip('; ') if _d3['依据'] != '各轴无非常态' else _note3
        except Exception as _e3:
            for _t3 in result:
                for _s3 in ('主轴承', '齿轮箱', '发电机'):
                    result[_t3][_s3]['依据'] += f'; ⚠CMS报告转录失败({_e3})'

    except Exception as e:
        # 振动映射失败必须可见 (依据里响亮标注, 不静默降级)
        for t2 in result:
            for sysn in ('主轴承', '齿轮箱', '发电机'):
                result[t2][sysn]['依据'] += f'; ⚠振动映射失败({str(e)[:40]})'
    return result

def reconcile(cfg=None):
    """逐系统对账交付分册 (仅比 SCADA 可覆盖维; 分册振动维=CMS证据, SCADA 不可见, 不比)."""
    cfg = cfg or farm()
    mx = system_matrix(cfg)
    ref = pathlib.Path(cfg['contract']).parent
    out = {}
    books = {'主轴承': '主轴承_delivered_grades.json', '齿轮箱': '齿轮箱_delivered_grades.json',
             '发电机': '发电机_delivered_grades.json', '变流器': '变流器_delivered_grades.json',
             '主控与传感网': '主控传感网_delivered_grades.json'}
    # SCADA 可覆盖维 (白名单; 其余维=CMS/巡检证据不比)
    cover = {'主轴承': ['温度'], '齿轮箱': ['中速轴', '高速轴'],
             '发电机': ['温升'], '变流器': ['水冷系统', '变压器油温'],
             '主控与传感网': ['测风', '通讯与控制']}   # 只列我真有轴的维; 润滑排油/散热/防凝露/电源电池等未建轴不入对账
    for s, fn in books.items():
        g = json.loads((ref / fn).read_text(encoding='utf-8'))
        book_alarm = {t for t, d in g.items() if any('报警' in str(d.get(c, '')) for c in cover[s])}
        mine_alarm = {t for t, d in mx.items() if d[s]['状态'] == '报警'}
        mine_near = {t for t, d in mx.items() if d[s]['状态'] in ('报警', '良好')}
        out[s] = dict(分册可比维报警=sorted(book_alarm), 我方报警=sorted(mine_alarm),
                      交集=sorted(book_alarm & mine_alarm), 分册独有=sorted(book_alarm - mine_alarm),
                      我独有=sorted(mine_alarm - book_alarm),
                      近失覆盖=f"{len(book_alarm & mine_near)}/{len(book_alarm)} (分册报警台落我报警∪良好)",
                      近失漏=sorted(book_alarm - mine_near))
    return out