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

# -*- coding: utf-8 -*-
"""B翼 热链与冷却执行器件 (windscada M13; SOP §4.6 局部摩擦判据 + §4.6c①a 参考通道存活守卫).
★补上分册对账缺口"水冷/冷却行为轴我无" — 原生层实有整组冷却执行器 (先前未契约化).
轴:
  T1 节点−即时热汇 ΔT (局部摩擦判据, 每测点先声明热汇通道):
      发电机绕组(U/V/W max) − 变流器冷却水 ConvWTmp
      齿轮箱 HS/IMS 轴承 − 齿轮箱油 GeOilTmp   (超 fleet 中位 ≥4℃ = 该轴承真摩擦候选)
      齿轮箱油 GeOilTmp − 冷却水 ConvWTmp      (大 = 油冷器/油路, 非轴承)
      配对 DE−NDE 张开符号 + 绕组三相极差
  T2 冷却执行器 duty (counts+timeon 三件套, 动作次数必配时间占比):
      同油温档 冷却器高/低速 duty ×fleet   (同档 duty 高 = 换热能力低/换热器脏)
      油冷器旁通阀 OilCoByp duty            (常开 = 不走油冷器)
      齿轮箱加热 GearHeat duty × 环温        (反-环温缺失 = 加热器卡ON 家族)
      油泵 GearPuHS/LS duty
  T3 判别: 温控调节卡死家族 (加热器卡ON=反-环温 / 冷却卡全流=过冷 / 冷却不投入=贴环温)
守卫: 热汇通道先过 liveness (死/冻结/卡值) → 死则该 ΔT 判 INSUFFICIENT 不排名 (§4.6c①a);
      ΔT 判据用 fleet 中位相对门, 不用固定绝对阈 (防设计签名假阳)."""
import numpy as np, pandas as pd, pathlib
from src.windscada.config import farm
from app_ETL.app_ETL_guanlan.api import load_10min

PAIRS = [('发电机绕组−冷却水', ['tmp_wtc_Gen1U1Tm_mean', 'tmp_wtc_Gen1V1Tm_mean', 'tmp_wtc_Gen1W1Tm_mean'], 'tmp_wtc_ConvWTmp_mean'),
         ('齿轮箱高速轴−油', ['tmp_wtc_HSGenTmp_mean', 'tmp_wtc_HSRotTmp_mean'], 'tmp_wtc_GeOilTmp_mean'),
         ('齿轮箱中速轴−油', ['tmp_wtc_IMSGenTm_mean', 'tmp_wtc_IMSRotTm_mean'], 'tmp_wtc_GeOilTmp_mean'),
         ('齿箱油−冷却水', ['tmp_wtc_GeOilTmp_mean'], 'tmp_wtc_ConvWTmp_mean')]
ACT = {'冷却器高速': 'dot_wtc_GearCoHS', '冷却器低速': 'dot_wtc_GearCoLS', '油冷器旁通': 'dot_wtc_OilCoByp',
       '齿箱加热': 'dot_wtc_GearHeat', '油泵高速': 'dot_wtc_GearPuHS', '油泵低速': 'dot_wtc_GearPuLS'}
PBIN = 500      # 载荷匹配档 kW
OBIN = 5        # 油温档 ℃

def _live(sr):
    s = sr.dropna()
    if len(s) < 200: return False
    if s.nunique() <= 2: return False
    if float(s.value_counts(normalize=True).iloc[0]) > 0.5: return False   # 卡值门 (§4.6c①a stuck_frac)
    return True

def build_store(cfg=None, days=180):
    cfg = cfg or farm()
    rows = []
    for t in cfg['turbines']:
        d = load_10min(t, cfg, groups=['A.功率', 'B.温度NBM', 'B.冷却执行器'])
        d = d[d.ts >= str(d.ts.max() - pd.Timedelta(days=days))[:10]]
        op = d[d['grd_wtc_ActPower_mean'] > 500].copy()
        if not len(op): continue
        op['pb'] = (op['grd_wtc_ActPower_mean'] // PBIN * PBIN)
        op['ob'] = (op['tmp_wtc_GeOilTmp_mean'] // OBIN * OBIN)
        rec = dict(turbine=t, n=len(op))
        # T1 ΔT (热汇 liveness 守卫)
        for name, nodes, sink in PAIRS:
            if sink not in op.columns or not _live(op[sink]):
                rec[name] = np.nan; rec[name + '_守卫'] = '热汇通道不可信'; continue
            nod = [c for c in nodes if c in op.columns and _live(op[c])]
            if not nod:
                rec[name] = np.nan; rec[name + '_守卫'] = '节点通道不可信'; continue
            hot = op[nod].max(axis=1)
            dt = (hot - op[sink])
            per_pb = dt.groupby(op['pb']).median()
            n_pb = dt.groupby(op['pb']).size()
            keep = n_pb[n_pb >= 30].index
            rec[name] = float(per_pb.loc[keep].median()) if len(keep) else np.nan
            rec[name + '_守卫'] = ''
        rec['DE_NDE'] = float((op['tmp_wtc_GenBeGTm_mean'] - op['tmp_wtc_GenBeRTm_mean']).median())
        ph = op[['tmp_wtc_Gen1U1Tm_mean', 'tmp_wtc_Gen1V1Tm_mean', 'tmp_wtc_Gen1W1Tm_mean']]
        rec['三相极差'] = float((ph.max(axis=1) - ph.min(axis=1)).median())
        # T2 执行器 duty (秒/行 → 占比) 同油温档
        for nm, base in ACT.items():
            c_on, c_ct = base + '_timeon', base + '_counts'
            if c_on not in op.columns: rec[nm] = np.nan; continue
            duty = op[c_on] / 600.0
            # ★稀发执行器必用 mean 聚合: 中位对 counts~1/日 的位必然塌成 0 (自逮)
            per_ob = duty.groupby(op['ob']).mean()
            n_ob = duty.groupby(op['ob']).size()
            keep = n_ob[n_ob >= 30].index
            rec[nm] = float(per_ob.loc[keep].mean()) if len(keep) else np.nan
            # 高油温档 duty (换热需求最大处) — 冷却能力的真检验面
            hot_ob = [b for b in keep if b >= 45]
            rec[nm + '_高油温档'] = float(per_ob.loc[hot_ob].mean()) if hot_ob else np.nan
            rec[nm + '_次日'] = float(op[c_ct].sum() / max(len(op) / 144, 1)) if c_ct in op.columns else np.nan
        # T3 反-环温 (加热器): duty 与环温的相关; 正常加热器应负相关
        if 'dot_wtc_GearHeat_timeon' in op.columns and _live(op['tmp_wtc_AmbieTmp_mean']):
            hd = op['dot_wtc_GearHeat_timeon'] / 600.0
            rec['加热反环温corr'] = float(hd.corr(op['tmp_wtc_AmbieTmp_mean'])) if hd.std() > 0 else np.nan
        rec['油温中位'] = float(op['tmp_wtc_GeOilTmp_mean'].median())
        rec['冷却水中位'] = float(op['tmp_wtc_ConvWTmp_mean'].median()) if 'tmp_wtc_ConvWTmp_mean' in op.columns else np.nan
        rec['环温中位'] = float(op['tmp_wtc_AmbieTmp_mean'].median())
        rows.append(rec); print(t, flush=True)
    df = pd.DataFrame(rows)
    df.to_parquet(pathlib.Path(cfg['store']) / 'thermal_chain.parquet')
    print('→ thermal_chain.parquet', df.shape)

def typology(cfg=None, cols=None):
    """劣化分型 (纪律: 劣化判据必须落时序现窗非全窗).
    持续上行 = 末6月均 − 前6月均 ≥ 1.0K (病情在演进) / 稳定高位 = 末6月高但Δ小 (长期固有, 变化型判据看不见)
    / 已回落 = 前高末低 (历史尖峰, 勿当现况)。"""
    cfg = cfg or farm()
    f = pathlib.Path(cfg['store']) / 'thermal_monthly.parquet'
    if not f.exists(): return {}
    d = pd.read_parquet(f)
    cols = cols or [n for n, _, _ in PAIRS]
    out = {}
    for col in cols:
        if col not in d.columns: continue
        piv = d.pivot_table(index='month', columns='turbine', values=col)
        dev = piv.sub(piv.median(axis=1), axis=0)
        for t in dev.columns:
            s_ = dev[t].dropna()
            if len(s_) < 12: continue
            d0, d1 = float(s_.head(6).mean()), float(s_.tail(6).mean())
            out.setdefault(t, {})[col] = dict(前6=round(d0, 2), 末6=round(d1, 2), Δ=round(d1 - d0, 2),
                                              末月=round(float(s_.iloc[-1]), 2),
                                              型=('持续上行' if d1 - d0 >= 1.0 else
                                                 '已回落' if d0 - d1 >= 1.0 else '稳定'))
    return out


def registry(cfg=None):
    cfg = cfg or farm()
    df = pd.read_parquet(pathlib.Path(cfg['store']) / 'thermal_chain.parquet')
    out, detail = [], {}
    TY = typology(cfg)
    med = df.median(numeric_only=True)
    for _, r in df.iterrows():
        t = r['turbine']; axes = []
        for name, _, _ in PAIRS:
            g = r.get(name + '_守卫', '')
            if g: axes.append(f'{name}: {g} → INSUFFICIENT'); continue
            v = r.get(name)
            if v != v: continue
            dev = v - med[name]
            if '−油' in name and dev >= 4:
                ty = TY.get(t, {}).get(name, {})
                tag = ''
                if ty:
                    tag = (f" [时序: 前6月{ty['前6']:+.1f}→末6月{ty['末6']:+.1f} (Δ{ty['Δ']:+.1f}) = {ty['型']}]"
                           + ('  ★病情在演进, 优先' if ty['型'] == '持续上行' else
                              '  (长期固有非新发展)' if ty['型'] == '稳定' else '  (历史峰已回落, 勿当现况)'))
                axes.append(f'{name} {v:.1f}K (超全场中位 {dev:+.1f}K ≥4K) → 该轴承摩擦候选{tag}')
            elif '油−冷却水' in name and dev >= 4:
                axes.append(f'{name} {v:.1f}K ({dev:+.1f}K) → 油冷器/油路候选 (非轴承)')
            elif '绕组−冷却水' in name and dev >= 6:
                axes.append(f'{name} {v:.1f}K ({dev:+.1f}K) → 绕组散热候选')
        for nm in ACT:
            v = r.get(nm)
            if v != v or med.get(nm, np.nan) != med.get(nm, np.nan): continue
            m = med[nm]
            if nm == '油冷器旁通' and v >= 0.5 and v >= m + 0.2:
                axes.append(f'{nm} duty {v:.2f} (全场{m:.2f}) → 旁通常开=不走油冷器')
            elif nm.startswith('冷却器') and m > 0.02 and v >= max(1.5 * m, m + 0.1):
                axes.append(f'{nm} 同油温档 duty {v:.2f} ({v/m:.2f}×全场) → 换热能力低候选(换热器/风扇)')
            elif nm == '齿箱加热' and v >= max(1.5 * m, m + 0.1) and m > 0.01:
                axes.append(f'{nm} duty {v:.2f} ({v/m:.2f}×全场)')
        ph = r.get('三相极差')
        if ph == ph and ph >= max(3.0, 2.5 * med['三相极差']):
            axes.append(f'绕组三相极差 {ph:.1f}K ({ph/med["三相极差"]:.1f}×全场) → 仪表/接线候选 (SOP §4.6b: 破平衡=测量非部件)')
        c = r.get('加热反环温corr')
        if c == c and c > -0.05 and r.get('齿箱加热', 0) > 0.05:
            axes.append(f'加热器 duty 与环温相关 {c:+.2f} (正常应显著负) → 加热器卡ON候选')
        for nm in ACT:
            if nm in med and abs(float(df[nm].std())) < 1e-6:
                pass    # 全场恒定 → 无分辨力, 在 note 里统一声明, 不逐台刷屏
        real = [a for a in axes if 'INSUFFICIENT' not in a]
        rising = any('持续上行' in a for a in axes)
        verdict = '热链候选(定向查)' if (len(real) >= 2 or rising) else ('记基线观察' if real else '—')
        out.append(dict(turbine=t, 判=verdict, 依据='; '.join(axes) if axes else
                        f"绕组−水 {r.get('发电机绕组−冷却水', float('nan')):.1f}K, HS−油 {r.get('齿轮箱高速轴−油', float('nan')):.1f}K, "
                        f"油−水 {r.get('齿箱油−冷却水', float('nan')):.1f}K 均在全场带内",
                        油温=round(float(r['油温中位']), 1), 冷却水=round(float(r['冷却水中位']), 1) if r['冷却水中位'] == r['冷却水中位'] else None))
        detail[t] = axes
    return pd.DataFrame(out), detail