guanlan-src/app_ontology/app_ontology_guanlan/sop/pitch_validation.py

# -*- coding: utf-8 -*-
"""pitch_validation.py — 变桨校核: 三叶片桨距不平衡 (单叶静态零位偏 / frozen) 检出 + 定位.

裁决对象 = per 机组 **三叶片桨距一致性** — 真三叶 IPC (独立变桨) 场的冗余三元组奇偶
(同转子三叶 below-rated 应同 fine pitch; 单叶零位偏/卡死破对称)。区别 yaw (对风) / sensor (仪表)。

来源: gym P 赛道 three_blade champion (EXP-P1-01 立 / P1-03 跨场 n=2 海勒斯+敖都稳健, 远景 EN182-6250
1min IPC 场)。gym 分数≠交付, 本件 = 走设计 §7 四闸 (冻结suite/3原则审/RV-1独立审/labeled DP test) 的生产实现。

★物理 (冗余三元组奇偶, 同齿比/三相电流族): 健康三叶 below-rated → 三桨中位近相等 (fine pitch β≈0);
  单叶零位偏 δ → 该叶中位偏 δ → 三桨极差 spread=max−min 抬 → flag + 定位偏离中位最大的叶 (独有定位轴)。
  **intra-turbine 相对互比 (三桨自比) → 天生无 fleet 批次盲** (对照臂 fleet_z 跨机 z 过敏 INV fp 20/64;
  region2_abs 绝对锚自指盲 fp 13/64 = Cp 自参照陷阱族); 迁移 IPC 场阈稳。

★产化硬化 (送审前真盘 probe 逮):
  - **gen 下界 (filter gap 修)**: gym _br 仅 P<0.7·rated **无下界** → 停机/停运台 (P≡0, 三叶顺桨 89-95°) 混入
    → spread 94° 假阳 (海勒斯 9ZlypWTP 整窗 P≡0 实证)。修: below-rated = **gen_power_min < P < br·rated**
    (真发电 fine-pitch 段); 健康台加下界 spread 逐位不变 (4wpykJt9 0.765→0.765), 停机台 → INSUFFICIENT。
  - **第五角: 三叶独立性守卫**: 集控广播机型 / 早期单叶 = 三桨为同一测量写三列 (非真 IPC 面) → spread≡0 =
    伪"平衡" (EXP-P1-00/06 王家湾 negative: '清点≠可用'); pair-diff < derived_tol → INSUFFICIENT note
    (承传感器 _phase_derived_guard, sensor 五角教训第五角)。
  - **顺桨/非fine-pitch 守卫**: 任一叶中位 > feather_deg → 疑停机/顺桨混入 (gen 下界应已滤, belt-and-suspenders) → note。
  - **fine-pitch gate (RV-1 逮, 角C)**: br_ws 上沿 (如 9 m/s) 够到**变桨调节区** (海勒斯 ws8-9 三桨中位 5.08° 非 β≈0) →
    collective-pitch 瞬态污染静态极差 → per样本**三桨中位 < fine_pitch_max(3°)** 过闸, 只留 Region-2 静态段
    (单叶偏 row-median 仍≈0° 不受影响, 调节区三叶同偏被剔); 海勒斯健康台 spread 略降 (JXYvbOK2 0.957→0.941, margin↑),
    windy 场调节区占比大时 confound 更大 → farm-robust。

★诚实界 (gym known_limitations + 3原则审):
  - **共模盲 (headline)**: 单台 collective 集距同偏 (三叶同 δ) → 极差不变 → 本臂**天然盲** (EXP-P1-02 自咬);
    集距零位偏须绝对轴 (region2_abs, 但自指盲) 或现场独立量 → 本臂只判"叶间不平衡"非"整体零位对不对"。
  - **定位歧义 (RV-1 逮, 角A)**: loc_blade = odd-one-out (偏离另两叶最大者); **单叶坏 与 双叶同偏不可辨**
    (后者 loc 反指健康叶, 同三相电气双相同偏反指) → 两非loc叶彼此近时 note 提示, 需现场三叶对照/绝对量定坏叶。
  - **razor-thin margin (RV-1 逮, 角B)**: 海勒斯健康 max spread 0.957 vs 阈 1.0 = 4% margin (JXYvbOK2 子窗越阈)
    → spread∈[0.7×阈,阈] 报 borderline=True (监视非定论, 亚阈值轻度不平衡); 阈 1° 是 gym a-priori 非 tune, 窗选敏感。
  - **半广播守卫 (RV-1 逮, 角D)**: 一对桨距近同 (n_copy==1) = 仅 2 独立通道 → note "定位不可靠" (spread 仍可检出);
    ≥2 对近同 = 全广播 → INSUFFICIENT。
  - **相对≠绝对**: 三桨互比给不出绝对桨距零位; 绝对零位须现场独立量 (pitch-error skill [A])。
  - **数据面**: 限**真三叶 IPC 场** (远景 EN182 类, 三叶独立可观测); 集控广播/单叶机型三叶面不存在 → 守卫弃权。
  - **检出下限 ~1-1.5°** (gym 海勒斯 δ=0.5/1.0/1.5 → recall 0/0.67/0.83); 定位命中 ≥0.9。
  - **spread 幅值语境**: spread ≫10° ≠ 零位偏 (那是卡叶/停机/顺桨) → 调用方按幅值分诊 (零位偏 O(1°) vs 卡叶 O(10°+))。
  - **INSUFFICIENT 归因二分 (RV-1 逮, 角C-b)**: fine-pitch gate 会把"**发电但全坐高集距** (~10°, 降额/集距offset)"的台
    推过 min_n → INSUFFICIENT; 此类**非停机** (海勒斯 j1eytYUu/zdR8V4b1 实证: P>50 数百~数千样本坐 10°) → note 须区分
    "发电但高集距(走绝对轴)" vs "真停机/无数据", **别把活着的高集距异常台静默归入停机桶** (承 sensor 五角"信号别静默埋")。

DP 验收 = tests/sop/test_pitch_three_blade_dp.py。
"""
from __future__ import annotations

import numpy as np
import pandas as pd


def pitch_three_blade_verdict(
    df: pd.DataFrame, *, blade_pitch_cols: list, power_col: str, ws_col: str, turbine_col: str,
    rated_power: float, gen_power_min: float = 50.0, br_power_frac: float = 0.7,
    br_ws: tuple = (4.0, 9.0), imb_thresh: float = 1.0, min_n: int = 200,
    derived_tol: float = 0.02, feather_deg: float = 20.0, fine_pitch_max: float = 3.0,
) -> dict:
    """per-机组 **三叶片桨距不平衡** 裁决 (真三叶 IPC 场).

    返回 {tid: {"spread": 三桨中位极差(°), "flag": bool, "blade_meds": [3], "loc_blade": 1-3 (坏叶),
                "n": below-rated 发电样本, "note"?: 弃权/低置信}}。
    机制: below-rated **发电 fine-pitch** 段 (gen_power_min < P < br_power_frac·rated ∧ ws∈br_ws ∧
      三桨中位<fine_pitch_max=纯Region-2静态, 剔调节区) per台三桨中位;
      spread = max−min; flag = spread > imb_thresh (1° a-priori, 海勒斯健康 p90=0.68/max=0.95 → margin);
      定位 = 偏离三桨中位最大的叶 (高或低侧)。
    """
    # NaN-safe: Arrow str backend 下 astype(str) 留 float NaN → sorted 混型崩; dropna + 显式 str()
    tids = sorted({str(x) for x in df[turbine_col].dropna().unique()})
    NULL = {"spread": 0.0, "flag": False, "blade_meds": None, "loc_blade": None, "n": 0}
    bc = [c for c in blade_pitch_cols if c in df.columns]
    if len(bc) != 3 or power_col not in df.columns or ws_col not in df.columns:
        return {t: {**NULL, "note": "需恰 3 桨距列 + power + ws; abstain"} for t in tids}
    P = pd.to_numeric(df[power_col], errors="coerce").to_numpy(float)
    W = pd.to_numeric(df[ws_col], errors="coerce").to_numpy(float)
    B = np.column_stack([pd.to_numeric(df[c], errors="coerce").to_numpy(float) for c in bc])
    finite = np.isfinite(P) & np.isfinite(W) & np.isfinite(B).all(axis=1)
    # fine-pitch gate (RV-1 逮 角C: br_ws 上沿够到变桨调节区 → 桨距非 β≈0 静态): per样本三桨中位 < fine_pitch_max
    #   → 只留 Region-2 fine-pitch 静态段, 剔除调节区 collective-pitch 瞬态污染 (海勒斯 ws8-9 三桨中位 5.08° 实证)。
    #   单叶偏不受影响 (一叶+8°其余0° → row median 仍≈0° 过闸); collective 调节 (三叶同~5°) → 被剔。
    row_med = np.median(B, axis=1)
    br_gen = finite & (P > gen_power_min) & (P < br_power_frac * rated_power) & (W >= br_ws[0]) & (W <= br_ws[1])
    br = br_gen & (row_med < fine_pitch_max)                   # + fine-pitch gate
    tv = df[turbine_col].astype(str).to_numpy()
    out = {}
    for t in tids:
        sel = br & (tv == t)
        if sel.sum() < min_n:
            # INSUFFICIENT 归因二分 (RV-1 逮 角C-b): 分"真停机/无发电"vs"发电但全在调节区/高集距" — 后者别静默埋成停机
            gen_pre = br_gen & (tv == t)                       # gate 前的 below-rated 发电样本
            if gen_pre.sum() >= min_n:                         # 有足量发电样本, 却被 fine-pitch gate 剔光
                hi_med = float(np.median(row_med[gen_pre]))
                out[t] = {**NULL, "n": int(sel.sum()),
                          "note": f"有 below-rated 发电样本 (n={int(gen_pre.sum())}) 但全在变桨调节区/高集距 "
                                  f"(三桨中位≈{hi_med:.1f}°≥fine_pitch_max) → 非 Region-2 静态, 本臂弃权; "
                                  f"旁注: 集距偏高疑降额/集距offset, 走绝对轴 (region2_abs / 现场量) — 非停机"}
            else:
                out[t] = {**NULL, "n": int(sel.sum()),
                          "note": "below-rated 发电样本不足 (疑停机/停运/无本工况数据); INSUFFICIENT"}
            continue
        Bt = B[sel]
        # 第五角: 三叶独立性 — 计数近同桨距对 (RV-1 逮全∧半广播)
        d = [float(np.median(np.abs(Bt[:, i] - Bt[:, j]))) for i, j in ((0, 1), (0, 2), (1, 2))]
        n_copy = sum(1 for x in d if x < derived_tol)
        if n_copy >= 2:                                       # ≥2 对近同 = ≤1 独立通道 → 全广播不可测
            out[t] = {**NULL, "n": int(sel.sum()),
                      "note": f"三叶桨距近同 (pair-diff<{derived_tol}°): 疑集控广播/单叶非真IPC面 → INSUFFICIENT"}
            continue
        meds = [float(np.median(Bt[:, b])) for b in range(3)]
        spread = max(meds) - min(meds)
        med_m = float(np.median(meds))
        loc = (int(np.argmax(meds)) + 1 if (max(meds) - med_m) >= (med_m - min(meds))
               else int(np.argmin(meds)) + 1)               # 偏离另两叶最大的叶 (odd-one-out; 高/低侧)
        borderline = bool(imb_thresh * 0.7 < spread <= imb_thresh)
        rec = {"spread": round(spread, 3), "flag": bool(spread > imb_thresh), "borderline": borderline,
               "blade_meds": [round(m, 3) for m in meds], "loc_blade": loc, "n": int(sel.sum())}
        notes = []
        if n_copy == 1:                                       # 角D: 半广播 (一对复制) → 定位不可靠
            notes.append("疑半广播/两叶复制 (一对 pair 近同): 仅 2 独立通道, 定位不可靠, spread 仍可检出")
        others = sorted(m for k, m in enumerate(meds) if k != loc - 1)   # 角A: 两非loc叶
        if spread > imb_thresh and len(others) == 2 and abs(others[1] - others[0]) < 0.3 * spread:
            notes.append("两非loc叶彼此近: 单叶坏∨**双叶同偏**不可辨 (后者 loc 反指健康叶) → 现场三叶对照/绝对量")
        if borderline:                                        # 角B: 亚阈值 razor-thin
            notes.append(f"borderline (spread {spread:.2f}∈[{imb_thresh*0.7:.2f},{imb_thresh:.2f}]): 疑轻度不平衡, 监视非定论")
        if any(abs(m) > feather_deg for m in meds):           # 顺桨/非fine-pitch 混入 (gen下界应已滤)
            notes.append("某叶中位非fine-pitch (>feather_deg): 疑停机/顺桨混入, spread非零位偏语境")
        elif spread > 10.0:                                   # 幅值语境: 零位偏 O(1°) vs 卡叶 O(10°+)
            notes.append("spread≫零位偏量级: 疑卡叶/机械故障非静态零位偏, 按幅值分诊")
        if notes:
            rec["note"] = "; ".join(notes)
        out[t] = rec
    return out


# ==================== 变桨电机三元组 parity 校核 (与三叶桨距角正交) ====================

def pitch_motor_parity_verdict(
    df, *, motor_cur_cols: list, turbine_col: str, motor_temp_cols: list | None = None,
    cur_active_min: float = 3.0, imb_thresh: float | None = None,
    k_mad: float = 3.5, uniform_cv_floor: float = 0.05, derived_tol: float = 0.1,
    min_active_n: int = 200, temp_confirm_deg: float = 5.0, low_duty_frac: float = 0.05,
) -> dict:
    """per-机组 **变桨电机三元组不平衡** 校核 (电机/轴承健康) — 与 pitch_three_blade_verdict
    (叶间**桨距角**不平衡, 标定轴) **正交**: 本轴 = 三**电机电流** imbalance (单电机摩擦/轴承退化)。

    返回 {tid: {"m": active段中位 imbalance, "flag": bool, "z": fleet稳健z, "loc_motor": 1-3 (偏离电机),
                "n_active": active样本, "note"?: 弃权/低置信}}。
    机制 (no black box, parity 族同三相电流/齿比): m = max_i|I_i/Ibar−1| (Ibar=三电机均值); **标度不变**
      → 三电机共缩 (真整体变桨负载) → m 逐位不变 (王牌, 实测 Δm 2e-16); 单电机 gain/摩擦 → 该电机偏 → m 抬。
      **active 条件化** (max(I_i)>cur_active_min = 主动变桨; 消 idle 电流≈0 的比值噪声) + **per台中位** (over
      多方位角 = 消 azimuth 1P 载荷调制)。flag = fleet 稳健守卫 (hi_thresh 地板 scale, 承 yaw §48 F2) 高离群。
    ★confound 五角 checklist 前置 (§48):
      - **派生守卫**: 任两电机 pair-diff 中位 < derived_tol → 疑复制列 (三电机同一测量) → INSUFFICIENT。
      - **id 列守卫**: turbine_col 全空 → INSUFFICIENT (承 yaw turbine_hash 坏列)。
      - **active 条件化**: idle 电流≈0 → 比值噪声; 必 active (max>cur_active_min); active<min_active_n → 弃权。
      - **azimuth 1P**: 电机电流随方位角 (重力/气动相位) 调制 → per台中位平均掉 1P。
    ★诚实界:
      - **候选/筛查级非确诊**: 相对非绝对 (电机效率标定未知); 单电机高电流 = 摩擦/轴承退化**候选**, 现场核电机/减速/轴承。
      - **温度交叉判别 (C1, RV-1 逮 + 真盘证)**: 传入 motor_temp_cols → loc 电机温度共升 (loc_temp_dev>temp_confirm_deg)
        = I²R/摩擦发热 → **升置信排电流传感器 gain artifact** (传感器 gain 错抬读数**不产热**); 温度未共升 → 疑测量侧
        (采集/传感器) 非发热退化。真盘: 台51 电机2 电流35A+温度54.7°C(+28°C双坐实真退化)/台63 电机1(+19.5°C)/台58(+6°C弱)。
      - **电流瞬态** (active 才有信号) vs 温度 sustained; 温度亦可作主轴 (motor_temp_cols 换 motor_cur_cols 位跑)。
      - **窗界 (C3)**: 短窗 (真盘 10 天) 下"恒偏 (传感器/电机个体差) vs 渐变 (退化)"不可解, flag=窗内持续偏; 须更长窗看 onset 趋势定退化。
      - **t0 起单电机恒偏** (无 onset) 本臂靠绝对 m (非差分) 可检, 但"恒偏是真退化还是电机个体差"无铭牌不可辨 (同齿比静态恒偏)。
      - **双电机同偏反指健康电机** (loc odd-one-out 结构性反指, 承桨距角A/电气双相) → 两非loc电机近时 note, 需现场三电机对照。
      - **fleet-relative → 批次盲** (全场一致高不平衡 → 无flag); 批次嫌疑场须绝对基准。
    来源: gym EXP-PM1-01 (TRUE: H2检出76-86 / H3可分性Δm 2e-16 / 真盘候选台58·63) → §7四闸并回 src/ 2026-07-16。
    DP = tests/sop/test_pitch_motor_parity_dp.py。
    """
    # NaN-safe: Arrow str backend 下 astype(str) 留 float NaN → sorted 混型崩; dropna + 显式 str()
    tids = sorted({str(x) for x in df[turbine_col].dropna().unique()})
    NULL = {"m": None, "flag": False, "z": None, "loc_motor": None, "n_active": 0}
    mc = [c for c in motor_cur_cols if c in df.columns]
    if len(mc) != 3:
        return {t: {**NULL, "note": "需恰 3 电机电流列; abstain"} for t in tids}
    if df[turbine_col].notna().sum() == 0 or (len(tids) == 1 and tids[0] in ("nan", "None", "")):
        return {t: {**NULL, "note": "turbine id 列全空/坏列; INSUFFICIENT"} for t in tids}
    I = np.column_stack([pd.to_numeric(df[c], errors="coerce").to_numpy(float) for c in mc])
    tv = df[turbine_col].astype(str).to_numpy()
    import warnings
    with warnings.catch_warnings(), np.errstate(invalid="ignore", divide="ignore"):
        warnings.simplefilter("ignore", RuntimeWarning)       # 全NaN行由 active/finite mask 滤, 抑良性空切片告警
        active = (np.nanmax(I, axis=1) > cur_active_min) & np.isfinite(I).all(axis=1)
        Ibar = np.nanmean(I, axis=1)
        dev = I / Ibar[:, None] - 1.0
        m = np.nanmax(np.abs(dev), axis=1)
    tc = [c for c in (motor_temp_cols or []) if c in df.columns]        # C1: 温度交叉判别 (可选, RV-1 逮)
    T = (np.column_stack([pd.to_numeric(df[c], errors="coerce").to_numpy(float) for c in tc])
         if len(tc) == 3 else None)
    out = {}
    per_m = {}
    n_total = {t: int((tv == t).sum()) for t in tids}
    for t in tids:
        s = (tv == t) & active & np.isfinite(m)
        if s.sum() < min_active_n:
            out[t] = {**NULL, "n_active": int(s.sum()), "note": "active(主动变桨)样本不足; INSUFFICIENT"}
            continue
        # 派生守卫: 三电机是否复制列 (第五角)
        d = [float(np.nanmedian(np.abs(I[s, i] - I[s, j]))) for i, j in ((0, 1), (0, 2), (1, 2))]
        if min(d) < derived_tol:
            out[t] = {**NULL, "n_active": int(s.sum()),
                      "note": f"疑派生/复制电机电流 (pair-diff<{derived_tol}A): parity 不可证; INSUFFICIENT"}
            continue
        per_m[t] = float(np.median(m[s]))
        out[t] = {"m": round(per_m[t], 4), "flag": False, "z": None, "loc_motor": None,
                  "n_active": int(s.sum())}
        md = np.median(dev[s], axis=0)                        # 定位: 偏离最大电机 (odd-one-out)
        j = int(np.nanargmax(np.abs(md)))
        out[t]["loc_motor"] = j + 1
        notes = []
        # C1 温度交叉判别 (RV-1 逮): loc 电机温度共升 → I²R/摩擦发热 = 真电机退化, 排电流传感器 gain artifact (只抬读不产热)
        if T is not None:
            with np.errstate(invalid="ignore"):
                tmed = [float(np.nanmedian(T[s, i])) for i in range(3)]
            if all(np.isfinite(tmed)):
                loc_tdev = tmed[j] - float(np.median([tmed[k] for k in range(3) if k != j]))
                out[t]["loc_temp_dev_deg"] = round(loc_tdev, 1)
                out[t]["temp_confirms"] = bool(loc_tdev > temp_confirm_deg)
                notes.append("loc电机温度共升 → 电机侧真退化 (排电流传感器gain artifact)" if loc_tdev > temp_confirm_deg
                             else "loc电机温度未共升 → 电流偏疑传感器/采集侧 (非发热退化), 现场核采集链")
        others = sorted(md[k] for k in range(3) if k != j)
        if len(others) == 2 and abs(others[1] - others[0]) < 0.3 * abs(md[j]):
            notes.append("两非loc电机近: 单电机坏∨双电机同偏不可辨 (后者loc反指健康电机) → 现场三电机对照")
        if n_total[t] and s.sum() / n_total[t] < low_duty_frac:   # C2: 低 active 占空 = 疑停机/降额
            notes.append(f"active 占空低 ({s.sum()/n_total[t]:.1%}<{low_duty_frac:.0%}): 疑停机/降额少变桨, m 基于少数机动")
        if notes:
            out[t]["note"] = "; ".join(notes)
    if len(per_m) < 5:
        return out                                            # 不足 fleet z (已报 per台 m)
    vals = np.array(list(per_m.values())); med = float(np.median(vals))
    mad = float(np.median(np.abs(vals - med)))
    scale = max(1.4826 * mad, uniform_cv_floor * abs(med)) if med != 0 else (1.4826 * mad or float(np.std(vals)))
    thr = (imb_thresh if imb_thresh is not None else med + k_mad * scale)
    for t in per_m:
        z = (per_m[t] - med) / scale if scale > 0 else 0.0
        out[t]["z"] = round(float(z), 2)
        out[t]["flag"] = bool(per_m[t] > thr)
    return out