"""
独立打标模块：基于预先生成的机型阈值 Excel 对风机数据打标签。

功能：
    apply_labels(df, model_name, threshold_file=None) -> DataFrame
        输入原始数据 DataFrame 和机型名称，返回打标后的 DataFrame，只包含原始列和 label 列、sensor_tag列。

输出新增列：
    - label : 最终业务状态（运行/停机/限功率/传感器异常-xxx）

状态优先级（高→低）：传感器异常 > 停机 > 限功率 > 运行
"""
from typing import Dict, Optional

import numpy as np
import pandas as pd

# ==================== 打标参数 ====================

# 功率越界倍数（相对额定功率）
STATUS_POWER_UPPER_RATIO = 1.25  # 功率上限倍数
STATUS_POWER_LOWER_RATIO = -0.10  # 功率下限倍数
STATUS_SHUTDOWN_RATIO = 0.005  # 停机功率阈值（额定功率的比例），最小取 50kW
STATUS_CURTAIL_LOW_RATIO = 0.02  # 限功率下限（额定功率比例）
STATUS_CURTAIL_HIGH_RATIO = 0.95  # 限功率上限（额定功率比例）
STATUS_CURTAIL_PITCH_OFFSET = 3.0  # 限功率桨距角偏移量（基准桨距角 + 此值）

# 风速越界阈值（m/s）
STATUS_WIND_MAX = 75.0
STATUS_WIND_MIN = -2.0

# 变桨越界阈值（°）
STATUS_PITCH_MAX = 105.0
STATUS_PITCH_MIN = -10.0

# 转速/扭矩越界倍数
STATUS_SPD_UPPER_RATIO = 1.25
STATUS_TORQUE_UPPER_RATIO = 1.25
STATUS_TORQUE_LOWER_ABS = -2000.0

# 逻辑悖论阈值
STATUS_LOGIC_WIND_MIN = 0.1  # 风速低于此值但功率大则为逻辑异常
STATUS_LOGIC_POWER_MIN = 100.0  # 配合上面风速阈值的功率下限

# 机型特殊规则（可扩展）
STATUS_MODEL_SPECIAL_RULES = {
    "EN156-3300": [(2.7, 450)],  # 示例：风速 < 2.7 且功率 > 450 时追加逻辑异常
}

# 传感器异常列与对应可读标签（内部使用，不输出）
_SENSOR_COL_LABEL: Dict[str, str] = {
    "d_val_power": "功率值异常",
    "d_val_wind": "风速值异常",
    "d_val_pitch": "变桨值异常",
    "d_val_spd": "转速值异常",
    "d_val_torque": "扭矩值异常",
    "d_logic_wind_pwr": "风速功率逻辑异常",
    "d_logic_torque_pwr": "转速扭矩逻辑异常",
}

# ==================== 阈值加载（静态数据） ====================

# 静态阈值数据（从机型参数阈值表.xlsx转换）
_THRESHOLD_CACHE: Dict[str, dict] = {
    "CCWE1500-82.DF": {
        "p_max_observed": 1554.0,
        "torque_limit": 17924.0,
        "spd_limit": 1861.59,
        "baseline_pitch": 0.04
    },
    "CCWE3000-122.HD": {
        "p_max_observed": 3047.0,
        "torque_limit": None,
        "spd_limit": 362.3,
        "baseline_pitch": 0.19
    },
    "DEW-D10000-185": {
        "p_max_observed": 10135.3,
        "torque_limit": 10841040.0,
        "spd_limit": 10.45,
        "baseline_pitch": -0.5
    },
    "DEW-D3800-148": {
        "p_max_observed": 3508.8,
        "torque_limit": 3320535.25,
        "spd_limit": 11.42,
        "baseline_pitch": 0.0
    },
    "DEW-G3600-165": {
        "p_max_observed": 3730.5,
        "torque_limit": 21010.0,
        "spd_limit": 1838.41,
        "baseline_pitch": 0.15
    },
    "DEW-G5500-183": {
        "p_max_observed": 5548.6,
        "torque_limit": 32636.0,
        "spd_limit": 1746.57,
        "baseline_pitch": 2.85
    },
    "DEW-G6250-186": {
        "p_max_observed": 6314.0,
        "torque_limit": 92110.56,
        "spd_limit": 710.82,
        "baseline_pitch": 0.1
    },
    "DEW-H8350-A-212": {
        "p_max_observed": 7085.1,
        "torque_limit": 125065.52,
        "spd_limit": 8.82,
        "baseline_pitch": 1.89
    },
    "DF103-2000": {
        "p_max_observed": 2024.2,
        "torque_limit": 1264956.0,
        "spd_limit": 16.37,
        "baseline_pitch": -0.5
    },
    "DF131-2500": {
        "p_max_observed": 2527.9,
        "torque_limit": 2058254.75,
        "spd_limit": 13.25,
        "baseline_pitch": 0.0
    },
    "EN141-2650": {
        "p_max_observed": 2744.6,
        "torque_limit": 15705.3,
        "spd_limit": 1778.21,
        "baseline_pitch": -0.79
    },
    "EN156-3300": {
        "p_max_observed": 3359.5,
        "torque_limit": 18575.35,
        "spd_limit": 1835.56,
        "baseline_pitch": -1.45
    },
    "EN171-4500": {
        "p_max_observed": 4559.4,
        "torque_limit": 25390.95,
        "spd_limit": 1795.83,
        "baseline_pitch": -0.85
    },
    "EN182-6250": {
        "p_max_observed": 6326.0,
        "torque_limit": 34601.38,
        "spd_limit": 1802.05,
        "baseline_pitch": -0.25
    },
    "FD77-1500": {
        "p_max_observed": 1534.2,
        "torque_limit": 8058.38,
        "spd_limit": 1846.84,
        "baseline_pitch": 0.0
    },
    "G52-850": {
        "p_max_observed": 851.5,
        "torque_limit": None,
        "spd_limit": 1653.3,
        "baseline_pitch": -2.2
    },
    "G58-850": {
        "p_max_observed": 813.5,
        "torque_limit": None,
        "spd_limit": 1670.4,
        "baseline_pitch": 0.48
    },
    "GW140-2500": {
        "p_max_observed": 2575.0,
        "torque_limit": None,
        "spd_limit": 12.31,
        "baseline_pitch": 3.63
    },
    "GW140-3300": {
        "p_max_observed": 3486.0,
        "torque_limit": None,
        "spd_limit": None,
        "baseline_pitch": 4.25
    },
    "GW191-5000": {
        "p_max_observed": 5004.0,
        "torque_limit": None,
        "spd_limit": 705.6,
        "baseline_pitch": 0.0
    },
    "GW191-6250": {
        "p_max_observed": 6258.6,
        "torque_limit": None,
        "spd_limit": 686.2,
        "baseline_pitch": 2.2
    },
    "GW82-1500": {
        "p_max_observed": 1570.0,
        "torque_limit": None,
        "spd_limit": 18.12,
        "baseline_pitch": 0.2
    },
    "GWH191-5270": {
        "p_max_observed": 4916.6,
        "torque_limit": None,
        "spd_limit": 709.99,
        "baseline_pitch": 0.29
    },
    "GWH191-6700": {
        "p_max_observed": 5905.7,
        "torque_limit": None,
        "spd_limit": 685.9,
        "baseline_pitch": 0.08
    },
    "MY2.0se-121": {
        "p_max_observed": 2079.0,
        "torque_limit": None,
        "spd_limit": 1798.0,
        "baseline_pitch": 0.03
    },
    "MY3.0se-145": {
        "p_max_observed": 3212.0,
        "torque_limit": None,
        "spd_limit": 316.0,
        "baseline_pitch": 0.07
    },
    "MY4.0se-166": {
        "p_max_observed": 4160.0,
        "torque_limit": None,
        "spd_limit": 1095.0,
        "baseline_pitch": -0.45
    },
    "MY5.0se-155": {
        "p_max_observed": 5551.0,
        "torque_limit": None,
        "spd_limit": 284.6,
        "baseline_pitch": 0.0
    },
    "MY5.0se-166": {
        "p_max_observed": 5162.0,
        "torque_limit": 102.4,
        "spd_limit": 1031.1,
        "baseline_pitch": -0.49
    },
    "MY5.2se-166": {
        "p_max_observed": 5242.0,
        "torque_limit": None,
        "spd_limit": None,
        "baseline_pitch": -0.37
    },
    "MY6.25se-172": {
        "p_max_observed": 6346.0,
        "torque_limit": None,
        "spd_limit": 228.6,
        "baseline_pitch": -0.49
    },
    "S48-750": {
        "p_max_observed": 780.0,
        "torque_limit": None,
        "spd_limit": 1524.0,
        "baseline_pitch": 0.0
    },
    "SL1500-82": {
        "p_max_observed": 1544.3,
        "torque_limit": None,
        "spd_limit": 1841.5,
        "baseline_pitch": 0.01
    },
    "SL3000-113": {
        "p_max_observed": 3055.7,
        "torque_limit": None,
        "spd_limit": 1229.26,
        "baseline_pitch": 0.02
    },
    "SWT7000-154": {
        "p_max_observed": 7163.4,
        "torque_limit": None,
        "spd_limit": 10.8,
        "baseline_pitch": -1.69
    },
    "UP1500-82": {
        "p_max_observed": 1502.4,
        "torque_limit": 103.87,
        "spd_limit": 1777.1,
        "baseline_pitch": 0.03
    },
    "UP1500-86": {
        "p_max_observed": 1558.4,
        "torque_limit": 103.53,
        "spd_limit": 1773.6,
        "baseline_pitch": 0.03
    },
    "V100-2000": {
        "p_max_observed": 1999.8,
        "torque_limit": None,
        "spd_limit": 1734.83,
        "baseline_pitch": -2.7
    },
    "V110-2200": {
        "p_max_observed": 2204.1,
        "torque_limit": None,
        "spd_limit": 1757.7,
        "baseline_pitch": 0.9
    },
    "V52-850": {
        "p_max_observed": 854.8,
        "torque_limit": None,
        "spd_limit": 1693.8,
        "baseline_pitch": -1.97
    },
    "V60-850": {
        "p_max_observed": 853.2,
        "torque_limit": None,
        "spd_limit": 1673.9,
        "baseline_pitch": -3.06
    },
    "V80-2000": {
        "p_max_observed": 2043.9,
        "torque_limit": None,
        "spd_limit": 1748.7,
        "baseline_pitch": -1.74
    },
    "V90-2000": {
        "p_max_observed": 2005.7,
        "torque_limit": None,
        "spd_limit": 1730.4,
        "baseline_pitch": -2.16
    },
    "W2000-111": {
        "p_max_observed": 2111.0,
        "torque_limit": 100.04,
        "spd_limit": 1872.0,
        "baseline_pitch": 0.18
    },
    "W2500-135": {
        "p_max_observed": 2566.0,
        "torque_limit": None,
        "spd_limit": 1504.6,
        "baseline_pitch": 0.0
    },
    "W3450-146": {
        "p_max_observed": 3533.0,
        "torque_limit": None,
        "spd_limit": 1604.8,
        "baseline_pitch": 0.2
    },
    "XE122-2500": {
        "p_max_observed": 2568.0,
        "torque_limit": None,
        "spd_limit": 14.32,
        "baseline_pitch": 0.5
    },
    "XE72-2000": {
        "p_max_observed": 2084.0,
        "torque_limit": None,
        "spd_limit": 23.58,
        "baseline_pitch": -0.27
    }
}


def get_model_stats(model_name: str) -> dict:
    """根据机型名称获取对应的阈值字典"""
    thresh = _THRESHOLD_CACHE
    if model_name not in thresh:
        raise ValueError(f"未找到机型 {model_name} 的阈值配置")
    return thresh[model_name]


# ==================== 核心打标函数 ====================

def _label_dataframe(
        df_input: pd.DataFrame,
        stats: Optional[dict] = None,
        model_name: Optional[str] = None
) -> pd.DataFrame:
    """
    内部打标函数,返回包含所有中间列(d_val_* 等)的完整 DataFrame。
    参数含义同 apply_labels。
    """
    if stats is None:
        if model_name is None:
            raise ValueError("必须提供 stats 或 model_name")
        stats = get_model_stats(model_name)

    df = df_input.copy()
    P_MAX = stats["p_max_observed"]
    PITCH_BASE = stats["baseline_pitch"]

    # ── 1. 传感器异常列 ────────────────────────────────────────────────────────
    df["d_val_power"] = (
            (df["p_active"] > P_MAX * STATUS_POWER_UPPER_RATIO) |
            (df["p_active"] < P_MAX * STATUS_POWER_LOWER_RATIO)
    )

    if "wind_spd" in df.columns:
        df["d_val_wind"] = (df["wind_spd"] > STATUS_WIND_MAX) | (df["wind_spd"] < STATUS_WIND_MIN)
    else:
        df["d_val_wind"] = False

    pitch_cols = [c for c in df.columns if "pitch_ang_act" in c]
    df["d_val_pitch"] = False
    for col in pitch_cols:
        df["d_val_pitch"] |= (df[col] > STATUS_PITCH_MAX) | (df[col] < STATUS_PITCH_MIN)

    df["d_val_spd"] = False
    if stats["spd_limit"] and "gen_spd" in df.columns:
        df["d_val_spd"] = (
                (df["gen_spd"] > stats["spd_limit"] * STATUS_SPD_UPPER_RATIO) |
                (df["gen_spd"] < -200)
        )

    df["d_val_torque"] = False
    if stats["torque_limit"] and "actual_torque" in df.columns:
        df["d_val_torque"] = (
                (df["actual_torque"] > stats["torque_limit"] * STATUS_TORQUE_UPPER_RATIO) |
                (df["actual_torque"] < STATUS_TORQUE_LOWER_ABS)
        )

    # ── 2. 逻辑悖论列 ──────────────────────────────────────────────────────────
    if "wind_spd" in df.columns:
        df["d_logic_wind_pwr"] = (
                (df["wind_spd"] < STATUS_LOGIC_WIND_MIN) &
                (df["p_active"] > STATUS_LOGIC_POWER_MIN)
        )
        # 风速冻结检测：滑动窗口内风速几乎不变但功率明显不为0（传感器卡死）
        wind_std = df["wind_spd"].rolling(window=60, min_periods=2).std()
        df["d_logic_wind_pwr"] |= (
                (wind_std < 0.05) &
                (df["p_active"] > STATUS_LOGIC_POWER_MIN)
        )
        # 机型特殊规则
        if model_name is not None:
            for wind_thresh, pwr_thresh in STATUS_MODEL_SPECIAL_RULES.get(model_name, []):
                df["d_logic_wind_pwr"] |= (
                        (df["wind_spd"] < wind_thresh) & (df["p_active"] > pwr_thresh)
                )
    else:
        df["d_logic_wind_pwr"] = False

    df["d_logic_torque_pwr"] = False
    if stats["torque_limit"] and "actual_torque" in df.columns:
        df["d_logic_torque_pwr"] = (
                (df["p_active"] > P_MAX * 0.1) &
                (df["actual_torque"].abs() < stats["torque_limit"] * 0.01)
        )

    # ── 3. 汇总传感器异常标签 ───────────────────────────
    all_sensor_cols = list(_SENSOR_COL_LABEL.keys())
    existing = [c for c in all_sensor_cols if c in df.columns]

    if existing:
        tag_matrix = np.where(
            df[existing].values,
            np.array([_SENSOR_COL_LABEL[c] for c in existing]),
            "",
        )
        df["sensor_anomaly_tags"] = pd.Series(
            [",".join(t for t in row if t) for row in tag_matrix],
            index=df.index,
        )
    else:
        df["sensor_anomaly_tags"] = ""

    # ── 4. 业务状态打标 ────────────────────────────────────────────────────────
    any_sensor_anomaly = df[[c for c in existing if c in df.columns]].any(axis=1)

    df["label"] = "运行"

    shutdown_thresh = max(50.0, P_MAX * STATUS_SHUTDOWN_RATIO)
    mask_shutdown = (~df["d_val_power"]) & (df["p_active"] <= shutdown_thresh)
    df.loc[mask_shutdown, "label"] = "停机"

    if "pitch_ang_act_1" in df.columns:
        mask_curtail = (
                (df["label"] != "停机") &
                (~df["d_val_power"]) &
                (~df["d_val_pitch"]) &
                (df["p_active"] > P_MAX * STATUS_CURTAIL_LOW_RATIO) &
                (df["p_active"] < P_MAX * STATUS_CURTAIL_HIGH_RATIO) &
                (df["pitch_ang_act_1"] > (PITCH_BASE + STATUS_CURTAIL_PITCH_OFFSET))
        )
        df.loc[mask_curtail, "label"] = "限功率"

    # 传感器异常覆盖
    mask_sensor = any_sensor_anomaly
    df.loc[mask_sensor, "label"] = "传感器异常"

    # 新增 sensor_tag 列：7位整数，每位1=正常，2=异常 
    # 7位整数从左至右 分别为功率值异常（最高位）、风速值异常、变桨值异常、转速值异常、扭矩值异常、风速功率逻辑异常、转速扭矩逻辑异常（最低位）
    sensor_order = [
        'd_val_power', 'd_val_wind', 'd_val_pitch', 'd_val_spd', 'd_val_torque', 'd_logic_wind_pwr',
        'd_logic_torque_pwr'
    ]
    # 确保这些列都存在
    tag_str_series = df[sensor_order].apply(
        lambda row: ''.join('2' if x else '1' for x in row), axis=1
    )
    df['sensor_tag'] = tag_str_series.astype(int)

    return df


def apply_labels(
        df: pd.DataFrame,
        model_name: str
) -> pd.DataFrame:
    """
    便捷打标函数：传入原始 DataFrame 和机型名称，返回打标后的 DataFrame。

    返回的 DataFrame 包含原始数据的所有列，以及新增的 `label` 列（最终业务状态）。

    参数
    ----------
    df : pd.DataFrame
        原始数据，必须包含 p_active 列，可选 wind_spd, pitch_ang_act_1, gen_spd, actual_torque 等。
    model_name : str
        机型名称（需与静态数据中的机型名称一致）。

    返回
    -------
    pd.DataFrame
        包含原始数据列 + label 列的 DataFrame。
    """
    # 调用内部打标函数，获取完整结果（包含中间列）
    full_labeled = _label_dataframe(df, model_name=model_name)
    # 只保留原始列和 label 列
    result_cols = list(df.columns) + ['label', 'sensor_tag']
    return full_labeled[result_cols]