# -*- coding: utf-8 -*-
"""B翼 液压蓄能深挖轴 (windscada v1.1). 变桨液压 HydPress 的蓄能器健康三轴:
  A1 停机保压衰减率: 停机连续段 (≥1h, P<10kW) 内 HydPress_mean 斜率 bar/h — 泄漏/氮气预充失效直测;
     油温变化门 |ΔHydOilTm|≤2K (PV=nRT: 降温本身降压, 不设门=假衰减);
  A2 同油温档带宽横比: band=max-min per行, 按 HydOilTm 5K 档 ×fleet 同档中位 (油路压力必须油温匹配, 粘度反混);
  A3 缸级蓄能报警: 3411/3412/3413 蓄能器A/B/C检查超时 + 7106 变桨泵时间过长 + 7100 变桨液压过热.
边界: 1min 层无液压压力列 (实测) → 全部 10min 面; 判级=筛查级候选, 非定谳.
互证面: 温度NBM 液压油温簇 {20,19,08,03} (蓄能失效→频繁充压→发热)."""
import numpy as np, pandas as pd, pathlib
from src.windscada.config import farm
from app_ETL.app_ETL_guanlan.api import load_10min

ACC_CODES = {'3411': 'A缸超时', '3412': 'B缸超时', '3413': 'C缸超时', '7106': '泵超时', '7100': '液压过热'}

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

    span 模式走 `slim10min` 窄仓（同一份 `load_10min` 的列子集）且 **不写盘** —— 按窗重算是
    服务时算给页面看的，不能覆盖正式产物 `hydraulic_accum.parquet`（原口径 = `since` 起）。
    """
    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
                d = slim.load(t, cfg, span=span,
                              columns=['grd_wtc_ActPower_mean', 'prs_wtc_HydPress_mean',
                                       'prs_wtc_HydPress_max', 'prs_wtc_HydPress_min',
                                       'tmp_wtc_HydOilTm_mean']).sort_values('ts').reset_index(drop=True)
            else:
                d = load_10min(t, cfg, groups=['A.功率', 'B.变桨', 'B.温度NBM'])
                d = d[d.ts >= since].sort_values('ts').reset_index(drop=True)
            # A2 同油温档带宽 (发电态)
            op = d[d['grd_wtc_ActPower_mean'] > 100].copy()
            op['band'] = op['prs_wtc_HydPress_max'] - op['prs_wtc_HydPress_min']
            op['tbin'] = (op['tmp_wtc_HydOilTm_mean'] // 5 * 5)
            gb = op.groupby('tbin')['band'].agg(['median', 'size'])
            import json as _json
            rec['band_by_tbin'] = _json.dumps({str(float(k)): [float(v['median']), int(v['size'])] for k, v in gb.iterrows() if v['size'] >= 24})
            rec['hydoil_med'] = float(op['tmp_wtc_HydOilTm_mean'].median())
            # A1 停机保压段: 连续 P<10kW ≥6行(1h); 段内压力斜率 + 油温变化
            stop = (d['grd_wtc_ActPower_mean'] < 10).astype(int)
            seg_id = (stop.diff() != 0).cumsum()
            decays = []
            for _, g in d[stop == 1].groupby(seg_id[stop == 1]):
                if len(g) < 6: continue
                dt_h = (g.ts.iloc[-1] - g.ts.iloc[0]).total_seconds() / 3600
                if dt_h <= 0: continue
                dT = g['tmp_wtc_HydOilTm_mean'].iloc[-1] - g['tmp_wtc_HydOilTm_mean'].iloc[0]
                if abs(dT) > 2: continue        # 油温变化门 (排 PV=nRT 假衰减)
                p = g['prs_wtc_HydPress_mean'].dropna()
                if len(p) < 6 or p.iloc[0] == 0: continue
                decays.append((p.iloc[-1] - p.iloc[0]) / dt_h)
            rec['decay_med'] = float(np.median(decays)) if decays else np.nan
            rec['n_stop_seg'] = len(decays)
            rec['minp_p5'] = float(op['prs_wtc_HydPress_min'].quantile(0.05))
        except Exception as e:
            rec['err'] = 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:                       # ★按窗：报警计数也只看窗内（原口径 = since 起）
        a = al[(al['t_on'] >= str(span[0])) & (al['t_on'] <= str(span[1]) + ' 23:59:59')]
    else:
        a = al[(al['t_on'] >= since)]
    for code, name in ACC_CODES.items():
        cnt = a[a.code == code].groupby('turbine').size()
        df[name] = df['turbine'].map(cnt).fillna(0).astype(int)
    if write:
        df.to_parquet(pathlib.Path(cfg['store']) / 'hydraulic_accum.parquet')
        print('→ hydraulic_accum.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']) / 'hydraulic_accum.parquet')
    # fleet 同档带宽基线
    import json as _json
    fleet_bins = {}
    parsed = {}
    for _, r in df.iterrows():
        bb = _json.loads(r['band_by_tbin']) if isinstance(r['band_by_tbin'], str) and r['band_by_tbin'] else {}
        parsed[r['turbine']] = bb
        for tb, (m, n) in bb.items():
            fleet_bins.setdefault(tb, []).append(m)
    fleet_med = {tb: float(np.median(v)) for tb, v in fleet_bins.items() if len(v) >= 10}
    dec = df['decay_med'].dropna()
    dec_med, dec_mad = float(dec.median()), float((dec - dec.median()).abs().median()) or 0.05
    out, detail = [], {}
    for _, r in df.iterrows():
        t = r['turbine']; axes = []
        # A2: 台内各油温档 ×fleet 同档, 取加权中位
        ratios = [(m / fleet_med[tb], n) for tb, (m, n) in parsed.get(r['turbine'], {}).items() if tb in fleet_med and fleet_med[tb] > 0]
        bandx = float(np.median([x for x, _ in ratios])) if ratios else np.nan
        if bandx == bandx and bandx >= 1.5: axes.append(f'同油温档带宽 {bandx:.2f}×fleet')
        # A1: 泄压率显著低于 fleet (更负 = 掉得快)
        dz = (r['decay_med'] - dec_med) / (1.4826 * dec_mad) if r['decay_med'] == r['decay_med'] else np.nan
        if dz == dz and dz <= -3 and r['decay_med'] < -0.5: axes.append(f'停机泄压 {r["decay_med"]:.2f}bar/h (z{dz:.1f}, n段{int(r["n_stop_seg"])})')
        # A3: 缸级/泵报警
        acc_n = int(r['A缸超时'] + r['B缸超时'] + r['C缸超时'])
        pump_n = int(r['泵超时'] + r['液压过热'])
        if acc_n: axes.append('蓄能缸检超时 ' + '/'.join(f'{c}{int(r[c])}' for c in ('A缸超时', 'B缸超时', 'C缸超时') if r[c]))
        if pump_n: axes.append(f'泵超时/过热 {pump_n}')
        verdict = '蓄能失效候选(定向查)' if len(axes) >= 2 else ('记基线观察' if axes else '—')
        why = '; '.join(axes) if axes else f"带宽 {bandx:.2f}×同档fleet, 泄压 {r['decay_med'] if r['decay_med']==r['decay_med'] else float('nan'):.2f}bar/h"
        out.append(dict(turbine=t, bandx=round(bandx, 2) if bandx == bandx else None,
                        decay=round(float(r['decay_med']), 2) if r['decay_med'] == r['decay_med'] else None,
                        hydoil=round(float(r['hydoil_med']), 1), acc_alarms=acc_n, pump_alarms=pump_n,
                        判=verdict, 依据=why))
        detail[t] = axes
    return pd.DataFrame(out).sort_values('bandx', ascending=False).reset_index(drop=True), detail