# -*- 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