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

# -*- coding: utf-8 -*-
"""B翼 偏航系统 (windscada M6c v3). 判据蓝本=偏航分册四维度 + 报警族按码归维 + 真动态轴.

★2026-08-25 边界订正 (旧版错误声明"对风差无模拟通道/泵压无数据轴"): 那是只看 1min 转发层与
  契约内 134 列得出的, **未扫全 10min 原生层 598 列**。原生层实有且全场活:
    tur_wtc_NacelPos_*   机舱位置 (0-360°)      → 对风误差可重构 err=wrap(风向−机舱位置)
    tur_wtc_ScYawPos_mean 偏航累计位置 (-380~754) → 扭缆角轴
    flg_wtc_YawStat_endvalue 命令位 ('|'静止/'>'顺/'<'逆)
    flg_wtc_ScYawUnw_counts  解缆动作
    prs_wtc_YawPmPrs_*   偏航泵压 (小站泵压)
  1min 转发层的 cabin_position/twisted_cable_angle/yaw_error1 确为全场 38/38 死零 (转发层死列≠信号死)。

维度:
  D1 偏航轴承与润滑: 润滑泵时长×fleet + 柱塞出脂反馈 + 10201
  D2 驱动装置: 偏航动作次数/行程 + 10100/10125/10105
  D3 液压制动: 卡钳压水平/带宽 + 泵压 + 10113/7122
  D4 对风控制: 静态偏置(符号均值) + 动态散布σ(清洗后) + 报警轴 (10400/10401 + 风传感族)
  A类 数据质量: 机舱位置通道归零/拖偏 (判据=corr(err,机舱位置) 负相关 ∧ |err|>45°行占比≥2%)
边界 (skill yaw-dynamic 铁律): 10min 层测不到**死区/偏航速率/滞环** (需≤30s), 该三项 INSUFFICIENT;
  err 为 vane 口径 → 相对/上限, 非可回收 MWh; 行程=活动量代理非真磨损量。"""
import numpy as np, pandas as pd, pathlib
from src.windscada.config import farm
from app_ETL.app_ETL_guanlan.api import load_10min

DIMS = ['偏航轴承与润滑', '驱动装置', '液压制动', '对风控制']
FAM = {
    'D1润滑不可能': ['10201'],
    'D2过热': ['10100'], 'D2变流器警告': ['10125'], 'D2解缆': ['10105'],
    'D3制动液压错误': ['10113'], 'D3油位低': ['7122'],
    'D4偏航失败': ['10400', '10401'],
    'D4风传感': ['8108', '8110', '8184', '8185', '8186', '8187', '8171'],
}

def build_store(cfg=None, span=None, write=True):
    """偏航面登记件。`span=(起, 止)` 给了就按**所选时间窗**（含两端）重算（用户令 2026-09-21）。

    span 模式下数据源换成 `slim10min` 窄仓（同一份 `load_10min` 的列子集 ⇒ 同列同值），
    并 **不写盘**（`write=False`）：按窗重算是"服务时算给页面看"，不能覆盖正式产物
    `yaw_daily.parquet`（那是"数据末端往前 90 天"的口径，缺了它其它消费者会变味）。
    """
    cfg = cfg or farm()
    rows = []
    for t in cfg['turbines']:
        rec = dict(turbine=t)
        try:
            if span:
                from app_ETL.app_ETL_guanlan.api import slim
                cur = slim.load(t, cfg, span=span,
                                columns=['grd_wtc_ActPower_mean', 'prs_wtc_YawPress_mean',
                                         'prs_wtc_YawPress_max', 'prs_wtc_YawPress_min',
                                         'dot_wtc_YawLubPu_timeon', 'din_wtc_Y1LubPis_timeon',
                                         'flg_wtc_ScYawOpe_counts', 'cnt_wtc_WindFauS_endvalue',
                                         'cnt_wtc_WindFauT_endvalue'])
            else:
                d = load_10min(t, cfg, groups=['A.功率', 'B.偏航', 'B.润滑液压'])
                cur = d[d.ts >= str(d.ts.max() - pd.Timedelta(days=90))[:10]]
            op = cur[cur['grd_wtc_ActPower_mean'] > 100]
            byday = cur.groupby(cur.ts.dt.date)
            rec['yawpress_med'] = float(op['prs_wtc_YawPress_mean'].median())
            rec['yawpress_band'] = float((op['prs_wtc_YawPress_max'] - op['prs_wtc_YawPress_min']).median())
            rec['yawlub_s_day'] = float(byday['dot_wtc_YawLubPu_timeon'].sum().mean())
            rec['y1pis_s_day'] = float(byday['din_wtc_Y1LubPis_timeon'].sum().mean())
            rec['scyaw_cnt_day'] = float(byday['flg_wtc_ScYawOpe_counts'].sum().mean())
            for c, k in (('cnt_wtc_WindFauS_endvalue', 'windfau_s'), ('cnt_wtc_WindFauT_endvalue', 'windfau_t')):
                v = cur[c].dropna()
                rec[k] = float(v.iloc[-1] - v.iloc[0]) if len(v) > 2 else np.nan   # 计数器窗内增量
            rec['n_op'] = int(len(op))
        except Exception as e:
            rec['err10'] = str(e)[:60]
        rows.append(rec)
        if write:
            print(t, flush=True)
    df = pd.DataFrame(rows)
    al = pd.read_parquet(pathlib.Path(cfg['store']) / 'alarms.parquet')
    if span:                       # ★按窗：报警族计数也只看窗内（原口径 = 2026 年全年）
        a_w = al[(al['t_on'] >= str(span[0])) & (al['t_on'] <= str(span[1]) + ' 23:59:59')]
    else:
        a_w = al[al['t_on'].dt.year == 2026]
    for name, codes in FAM.items():
        cnt = a_w[a_w.code.isin(codes)].groupby('turbine').size()
        df[name] = df['turbine'].map(cnt).fillna(0).astype(int)
    if write:
        st = pathlib.Path(cfg['store']); df.to_parquet(st / 'yaw_daily.parquet')
        print('→ yaw_daily.parquet', df.shape)
    return df


def registry(cfg=None, span=None):
    """偏航面判级。`span=(起, 止)` 给了就按**所选时间窗**重算（判据/阈值不动，只换数据切片）。"""
    cfg = cfg or farm()
    if span:
        df = build_store(cfg, span=span, write=False)
    else:
        df = pd.read_parquet(pathlib.Path(cfg['store']) / 'yaw_daily.parquet')
    ep = pathlib.Path(cfg['store']) / 'yaw_err_clean.parquet'
    ERR = pd.read_parquet(ep) if ep.exists() else None
    med = df.median(numeric_only=True)
    rows, detail = [], {}
    for _, r in df.iterrows():
        t = r['turbine']; d = dict(维度={}, 依据={})
        # D1 轴承润滑: 泵时长×fleet + 柱塞出脂反馈 + 10201
        lubx = r['yawlub_s_day'] / max(med['yawlub_s_day'], 1e-9) if r['yawlub_s_day'] == r['yawlub_s_day'] else np.nan
        pisx = r['y1pis_s_day'] / max(med['y1pis_s_day'], 1e-9) if r['y1pis_s_day'] == r['y1pis_s_day'] else np.nan
        st1 = '不可判'
        if lubx == lubx:
            st1 = '报警' if (lubx >= 1.8 or r['D1润滑不可能'] >= 5 or (pisx == pisx and pisx <= 0.3 and lubx >= 1.0)) else \
                  '良好' if (lubx >= 1.4 or r['D1润滑不可能'] >= 1) else '优秀'
        d['依据'][DIMS[0]] = (f"润滑泵 {r['yawlub_s_day']:.0f}s/日 ({lubx:.2f}×fleet), 柱塞反馈 {r['y1pis_s_day']:.0f}s/日 ({pisx:.2f}×), "
                              f"润滑不可能报警{int(r['D1润滑不可能'])}")
        # D2 驱动: 动作次数×fleet + 过热/变流器/解缆报警
        mvx = r['scyaw_cnt_day'] / max(med['scyaw_cnt_day'], 1e-9) if r['scyaw_cnt_day'] == r['scyaw_cnt_day'] else np.nan
        st2 = '不可判'
        if mvx == mvx:
            st2 = '报警' if (r['D2过热'] >= 5 or r['D2变流器警告'] >= 10 or mvx >= 1.8) else \
                  '良好' if (r['D2过热'] >= 1 or r['D2变流器警告'] >= 3 or mvx >= 1.4 or r['D2解缆'] >= med['D2解缆'] * 2 + 5) else '优秀'
        d['依据'][DIMS[1]] = (f"偏航动作 {r['scyaw_cnt_day']:.0f}次/日 ({mvx:.2f}×fleet), 解缆{int(r['D2解缆'])}次, "
                              f"过热{int(r['D2过热'])}, 变流器警告{int(r['D2变流器警告'])}")
        # D3 液压制动: 卡钳压水平/带宽 + 制动液压错误/油位低
        px = r['yawpress_med']; bandx = r['yawpress_band'] / max(med['yawpress_band'], 1e-9)
        st3 = '报警' if (r['D3制动液压错误'] >= 3 or r['D3油位低'] >= 3 or bandx >= 2.0) else \
              '良好' if (r['D3制动液压错误'] >= 1 or r['D3油位低'] >= 1 or bandx >= 1.4 or abs(px - med['yawpress_med']) > 15) else '优秀'
        d['依据'][DIMS[2]] = (f"卡钳压中位 {px:.0f} ({px - med['yawpress_med']:+.0f} vs fleet), 带宽 {r['yawpress_band']:.0f} ({bandx:.2f}×), "
                              f"制动液压错误{int(r['D3制动液压错误'])}, 油位低{int(r['D3油位低'])}")
        # D4 对风: 真对风误差轴 (机舱位置重构, 清洗后) + 报警轴
        wf = (r['windfau_s'] if r['windfau_s'] == r['windfau_s'] else 0) + (r['windfau_t'] if r['windfau_t'] == r['windfau_t'] else 0)
        e = ERR.loc[t] if ERR is not None and t in ERR.index else None
        st4 = '报警' if (r['D4偏航失败'] >= 10 or r['D4风传感'] >= 30 or wf >= 100) else \
              '良好' if (r['D4偏航失败'] >= 2 or r['D4风传感'] >= 5 or wf >= 30) else '优秀'
        etxt = '对风误差轴不可用'
        if e is not None:
            if abs(e['bias_clean']) >= 2.0:
                st4 = '报警'; etxt = f"静态偏置 {e['bias_clean']:+.2f}° ≥2° → 零位候选(转静态偏航专项)"
            else:
                etxt = (f"静态偏置 {e['bias_clean']:+.2f}° (全场|中位|{ERR['bias_clean'].abs().median():.2f}), "
                        f"动态散布σ {e['sd_clean']:.2f}° (全场{ERR['sd_clean'].median():.2f}), |误差|>10°占时 {e['gt10_clean']:.1%}")
        d['依据'][DIMS[3]] = (f"{etxt}; 偏航失败{int(r['D4偏航失败'])}, 风传感报警{int(r['D4风传感'])}, 计数增量{wf:.0f} "
                              f"[死区/速率/滞环: 10min层不可测=INSUFFICIENT]")
        # A类 数据质量 (不入四维评级, 单列: 它污染一切依赖机舱位置的分析)
        if e is not None and e['drop'] >= 0.02 and e['corr_pos'] <= -0.15:
            d['数据质量'] = (f"A类·机舱位置通道归零/拖偏: |误差|>45°行占 {e['drop']:.1%}, "
                            f"corr(误差,机舱位置)={e['corr_pos']:+.2f} (健康台≈0) → 偏航编码器/PLC读回现场核; "
                            f"清洗前σ {e['sd_raw']:.1f}° 清洗后 {e['sd_clean']:.1f}°")
        for dim, st in zip(DIMS, (st1, st2, st3, st4)):
            d['维度'][dim] = st
        judged = [v for v in d['维度'].values() if v != '不可判']
        detail_extra = d.get('数据质量')
        order = {'报警': 0, '良好': 1, '优秀': 2}
        rows.append(dict(机组=t, 系统级=sorted(judged, key=lambda x: order[x])[0] if judged else '不可判',
                         **d['维度'], 数据质量=detail_extra or ''))
        detail[t] = d
    return pd.DataFrame(rows).sort_values('机组').reset_index(drop=True), detail