# -*- coding: utf-8 -*-
"""L1 工况门: 限电识别与运行状态分类 (统一技术栈前置硬门; 口径=零位专项 20260821 已审: 未被卡=PowerRef>1.15×实发).
自校验锚: 专项实测全场限电时段占比 2025=19.7% / 2026(1-7月)=76.9% — 复现不出即分类器错.
边界: 停机型限电须状态码/调度令 (报警轴已有 远程停机-业主 1007×359条, 接入待 M4c)。
锁页互证 (dsh_learn D运行知识): ActRegSt 恒 NotActive 非限电码; 场级受限同步(中位31/38)=park调度; 受限时间 2025≈0.20/2026≈0.79 与本门命令面锚一致; 七态逐台锁表=后训练_创新_如东工况归属.csv."""
import pandas as pd, numpy as np
from src.windscada.config import farm
from app_ETL.app_ETL_guanlan.api import load_10min

CUTIN_MS = 3.0     # 切入风速 (SWT-4.0-130); 低风待机门
STATES = ['正常发电', '限电命令·未绑定', '限电·绑定', '停机', '低风待机']
# 双口径 (2026-08-24 数据裁决, WTG13 单台: 命令面/全行 0.160/0.726 ≈ 专项锚 0.197/0.769; 绑定面/发电行 0.104/0.511):
#   命令面 = PowerRef<0.95×额定 (被调度限的时段, 专项统计口径); 绑定面 = 命令面 ∧ ref≤1.15×实发 (损失真发生, 专项剥离口径的补集)

def classify(d, rated_kw=4000):
    p = d['grd_wtc_ActPower_mean']; ref = d['tur_wtc_PowerRef_endvalue']; rpm = d['tur_wtc_GenRpm_mean']
    # ★2026-08-26 修 (云端外审逮→根因): 低风门原用 tur_wtc_PriAnemo_mean = **全场恒0.01死哨兵**,
    #   且该列因恒值未进契约 → d.get() 返回 None → 低风门**静默失效**, "低风待机"从未产生,
    #   切入风速以下的待机(实测占停机行 32%)全被算成"停机" ⇒ 可用率/MTBF/停机帕累托系统性偏。
    #   修法: 用活风源 SecAnemo; 风通道缺失必须**响亮失败**, 不许静默跳过判据。
    ws = d.get('tur_wtc_SecAnemo_mean')
    if ws is None:
        raise KeyError('classify: 缺 tur_wtc_SecAnemo_mean — 低风门不可跳过 (守护失败必响亮; '
                       'PriAnemo 是恒0.01死哨兵不可作风源)')
    st = pd.Series('正常发电', index=d.index)
    cmd = ref < 0.95 * rated_kw
    bind = cmd & (ref <= 1.15 * p.clip(lower=1)) & (p > 100)
    st[cmd & (p > 100) & ~bind] = '限电命令·未绑定'
    st[bind] = '限电·绑定'
    stop = (p <= 0) & (rpm < 100)
    lowwind = stop & ws.notna() & (ws < CUTIN_MS)      # 切入风速以下 = 待机, 非停机
    st[stop] = '停机'
    st[lowwind] = '低风待机'
    return st

def farm_daily(cfg=None, turbines=None):
    cfg = cfg or farm()
    rows = []
    for t in (turbines or cfg['turbines']):
        try:
            d = load_10min(t, cfg, groups=['A.功率', 'A.风况', 'A.转速'])
        except FileNotFoundError:
            continue
        d['state'] = classify(d, cfg['rated_kw'])
        d['date'] = d['ts'].dt.date
        g = d.groupby('date')['state'].value_counts().unstack(fill_value=0)
        g['turbine'] = t
        rows.append(g.reset_index())
    return pd.concat(rows, ignore_index=True).fillna(0)

def self_check(cfg=None, sample=6):
    """自校验: 抽样台复现专项限电占比 (2025≈19.7%, 2026≈76.9%)."""
    cfg = cfg or farm()
    ts_ = cfg['turbines'][::max(1, len(cfg['turbines']) // sample)][:sample]
    out = {}
    for t in ts_:
        d = load_10min(t, cfg, groups=['A.功率', 'A.风况', 'A.转速'])
        d['state'] = classify(d, cfg['rated_kw'])
        d['year'] = d['ts'].dt.year
        cmd_share = d.groupby('year').apply(lambda x: float((x['tur_wtc_PowerRef_endvalue'] < 0.95 * cfg['rated_kw']).mean()), include_groups=False)
        op = d[d['grd_wtc_ActPower_mean'] > 100]
        bind_share = op.groupby('year')['state'].apply(lambda s: (s == '限电·绑定').mean())
        out[t] = {int(y): dict(命令面=round(float(cmd_share.get(y, float('nan'))), 3), 绑定面=round(float(bind_share.get(y, float('nan'))), 3)) for y in cmd_share.index}
    return out


DISPATCH_CODES = {'1007': '远程停机-业主', '1008': '远程停机-西门子'}   # M4a 自动发现, 与 dsh_learn 记忆互证

def dispatch_intervals(cfg=None):
    """调度令停机区间 (报警事件层). 用于把'停机'细分出'停机(调度令)' — 停机型限电的证据轴."""
    import pandas as pd
    cfg = cfg or farm()
    al = pd.read_parquet(cfg['store'] / 'alarms.parquet')
    d = al[al.code.isin(DISPATCH_CODES)].dropna(subset=['t_on', 't_off'])
    return d[['turbine', 'code', 't_on', 't_off']].rename(columns={'t_on': 'time_on', 't_off': 'time_off'})

def refine_stop(d, turbine, intervals):
    """把 classify() 的'停机'行细分: 落在调度令区间内 → '停机(调度令)'."""
    iv = intervals[intervals.turbine == turbine]
    if not len(iv):
        return d
    stop = d['state'] == '停机'
    if not stop.any():
        return d
    ts = d.loc[stop, 'ts'].values
    hit = np.zeros(stop.sum(), dtype=bool)
    for _, r in iv.iterrows():
        hit |= (ts >= np.datetime64(r.time_on)) & (ts <= np.datetime64(r.time_off))
    idx = d.index[stop][hit]
    d.loc[idx, 'state'] = '停机(调度令)'
    return d