anomaly_detection/detect.py

"""
推理入口: 读取前1天 parquet 数据，逐点输出异常检测结果。

数据读取:
  - 每天更新一个 parquet，路径: {PARQUET_ROOT}/{model_name}/{farm_name}/{turbine_name}.parquet
  - 读取时过滤 data_time 在 [昨日 00:00, 今日 00:00) 范围内的数据

检测逻辑:
  1. 打标（运行/限功率/停机/传感器异常）
  2. 停机 / 限功率 → 跳过，不输出
  3. 运行数据 → 模型预测，逐点输出 is_anomaly + score
  4. 传感器异常数据:
       该检测器关心测点有异常 → 逐点输出 result_type='sensor_anomaly', anomaly_label=标签
       该检测器关心测点无异常 → 模型预测，逐点输出

输出表 anomaly_points（逐点）:
  data_time, model_name, farm_name, turbine_name, detector,
  result_type, is_anomaly, anomaly_score, anomaly_label, detect_time

用法:
    python detect.py                    # 检测所有机型所有风机（前1天数据）
    python detect.py --model 机型名称   # 只检测指定机型
    python detect.py --date 2026-02-24  # 指定日期（默认昨天）
"""
import argparse
import sqlite3
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from pathlib import Path

import joblib
import pandas as pd
import numpy as np

from config import MODEL_SAVE_DIR, DB_PATH
from data_loader import list_model_types, list_turbines, load_turbine
from labeler import get_model_statistics, label_dataframe, DETECTOR_SENSOR_COLS
from models.wind_power import PowerCurveDetector, ScatterDetector
from models.yaw import StaticYawDetector, CableTwistDetector
from models.pitch import PitchRegulationDetector, PitchCoordDetector, MinPitchDetector
from models.control_params import PowerQualityDetector, OperationStateDetector


# ── 数据库 ─────────────────────────────────────────────────────────────────────

def init_db(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS anomaly_points (
            id            INTEGER PRIMARY KEY AUTOINCREMENT,
            data_time     TEXT,     -- 数据时间戳（来自 parquet 的 data_time 列）
            model_name    TEXT,     -- 机型
            farm_name     TEXT,     -- 风场
            turbine_name  TEXT,     -- 风机
            detector      TEXT,     -- 检测器名称
            result_type   TEXT,     -- 'anomaly_detection' | 'sensor_anomaly'
            is_anomaly    INTEGER,  -- 1=异常, 0=正常（sensor_anomaly 时恒为 1）
            anomaly_score REAL,     -- 异常分数（越低越异常；sensor_anomaly 时为 NULL）
            anomaly_label TEXT,     -- 异常标签（sensor_anomaly 时填传感器异常类型，anomaly_detection 时为 NULL）
            detect_time   TEXT      -- 本次检测运行时间
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_ap_turbine_time
        ON anomaly_points (turbine_name, data_time)
    """)
    conn.commit()
    return conn


def _bulk_insert(conn, rows: list):
    """批量写入，rows 为 tuple 列表。"""
    if not rows:
        return
    conn.executemany(
        "INSERT INTO anomaly_points VALUES (NULL,?,?,?,?,?,?,?,?,?,?)",
        rows
    )
    conn.commit()


def _delete_existing(conn, turbine_name: str, farm_name: str, target_date: str):
    """写入前删除该风机该日期的旧记录，防止重复运行产生重复数据。"""
    conn.execute(
        "DELETE FROM anomaly_points WHERE turbine_name=? AND farm_name=? AND data_time LIKE ?",
        (turbine_name, farm_name, f"{target_date}%"),
    )
    conn.commit()


# ── 数据时间过滤 ───────────────────────────────────────────────────────────────

def _filter_date(df: pd.DataFrame, target_date: datetime.date) -> pd.DataFrame:
    """
    过滤 data_time 列在 target_date 当天的数据。
    data_time 列不存在时返回原始数据（不过滤）。
    """
    if "data_time" not in df.columns:
        return df
    dt = pd.to_datetime(df["data_time"], errors="coerce")
    mask = dt.dt.date == target_date
    return df[mask & dt.notna()].copy()


# ── 核心检测函数（逐点输出）────────────────────────────────────────────────────

def _run_detector(
    model_name: str, farm_name: str, turbine_name: str,
    model, detector_name: str,
    df_running: pd.DataFrame,
    df_sensor: pd.DataFrame,
    detect_time: str,
) -> list:
    """
    对单个检测器执行检测，返回 rows 列表（由调用方统一写入数据库）。
    model: 已加载的检测器对象（机型级预加载，避免每台风机重复 joblib.load）。
    """
    if model is None:
        return []

    rows = []
    sensor_cols = DETECTOR_SENSOR_COLS.get(detector_name, [])

    # ── 处理传感器异常数据 ────────────────────────────────────────────────────
    if not df_sensor.empty:
        existing_sc = [c for c in sensor_cols if c in df_sensor.columns]
        if existing_sc:
            has_anom = df_sensor[existing_sc].any(axis=1)
        else:
            has_anom = pd.Series(False, index=df_sensor.index)

        # 有传感器异常 → 直接标记
        df_direct = df_sensor[has_anom]
        if not df_direct.empty:
            dt_col = df_direct["data_time"].astype(str) if "data_time" in df_direct.columns \
                     else pd.Series("", index=df_direct.index)
            st_col = df_direct["status"].astype(str) if "status" in df_direct.columns \
                     else pd.Series("传感器异常", index=df_direct.index)
            for t, st in zip(dt_col.values, st_col.values):
                rows.append((
                    t, model_name, farm_name, turbine_name, detector_name,
                    "sensor_anomaly", 1, None, st, detect_time,
                ))

        # 无传感器异常 → 加入模型检测队列
        df_sensor_clean = df_sensor[~has_anom]
    else:
        df_sensor_clean = pd.DataFrame()

    # ── 合并需要模型预测的数据 ────────────────────────────────────────────────
    frames = [f for f in [df_running, df_sensor_clean] if not f.empty]
    if frames:
        df_detect = pd.concat(frames, ignore_index=False)
        try:
            result = model.predict(df_detect)
            # 按 index 对齐时间戳，避免 predict 内部 dropna 后行数不一致导致错位
            time_map = df_detect["data_time"].astype(str).to_dict() if "data_time" in df_detect.columns \
                       else {i: "" for i in df_detect.index}
            is_anom_col = result["anomaly"].astype(int) if "anomaly" in result.columns \
                          else pd.Series(0, index=result.index)
            score_col = result["score"] if "score" in result.columns \
                        else pd.Series(np.nan, index=result.index)
            for idx in result.index:
                sc_val = score_col.loc[idx]
                sc = float(sc_val) if pd.notna(sc_val) else None
                rows.append((
                    time_map.get(idx, ""),
                    model_name, farm_name, turbine_name, detector_name,
                    "anomaly_detection",
                    int(is_anom_col.loc[idx]),
                    sc,
                    None,
                    detect_time,
                ))
            n_anom = int(is_anom_col.sum())
            print(f"    [{detector_name}] 检测 {len(result)} 点，异常 {n_anom} 点")
        except Exception as e:
            print(f"    [{detector_name}] 检测失败: {e}")

    return rows


# ── 单台风机检测 ───────────────────────────────────────────────────────────────

def detect_turbine(rec: dict, models: dict, conn: sqlite3.Connection,
                   model_stats: dict, target_date, detect_time: str):
    """models: {detector_name: loaded_model_object}，机型级预加载。"""
    mn, fn, tn, path = rec["model_name"], rec["farm_name"], rec["turbine_name"], rec["path"]

    all_cols = [
        "data_time",
        "wind_spd", "p_active", "gen_spd", "actual_torque",
        "yaw_ang", "twist_ang",
        "pitch_ang_set_1", "pitch_ang_set_2", "pitch_ang_set_3",
        "pitch_ang_act_1", "pitch_ang_act_2", "pitch_ang_act_3",
        "pitch_spd_1", "pitch_spd_2", "pitch_spd_3",
        "rotor_spd", "theory_p_active", "p_reactive", "grid_freq",
        "grid_ia", "grid_ib", "grid_ic",
        "grid_ua", "grid_ub", "grid_uc",
        "ambient_temp",
    ]
    df_raw = load_turbine(path, required_cols=["p_active"], optional_cols=all_cols)
    if df_raw is None:
        return

    df_raw = _filter_date(df_raw, target_date)
    if df_raw.empty:
        print(f"  {fn}/{tn}: 无 {target_date} 数据，跳过")
        return

    # 写入前去重
    _delete_existing(conn, tn, fn, str(target_date))

    labeled   = label_dataframe(df_raw, model_stats, mn)
    df_run    = labeled[labeled["status"] == "运行"].copy()
    df_sensor = labeled[labeled["status"].str.startswith("传感器异常")].copy()

    total = len(labeled)
    print(f"  {fn}/{tn}: 总{total} | 运行{len(df_run)} | "
          f"传感器异常{len(df_sensor)} | 停机/限功率{total-len(df_run)-len(df_sensor)}")

    # 并行运行9个检测器（模型对象只读，线程安全），主线程统一写入避免 SQLite 锁冲突
    DETECTOR_NAMES = [
        "wind_power_curve", "wind_power_scatter",
        "yaw_static", "yaw_twist",
        "pitch_regulation", "pitch_coord", "pitch_min",
        "ctrl_power_quality", "ctrl_op_state",
    ]
    all_rows = []
    with ThreadPoolExecutor(max_workers=min(9, len(DETECTOR_NAMES))) as executor:
        futures = {
            executor.submit(
                _run_detector, mn, fn, tn,
                models.get(det_name), det_name,
                df_run, df_sensor, detect_time
            ): det_name
            for det_name in DETECTOR_NAMES
        }
        for future in as_completed(futures):
            det_name = futures[future]
            try:
                all_rows.extend(future.result())
            except Exception as e:
                print(f"    [{det_name}] 并行检测异常: {e}")
    _bulk_insert(conn, all_rows)


# ── 机型级检测 ─────────────────────────────────────────────────────────────────

def detect_model_type(model_name: str, conn: sqlite3.Connection,
                      target_date, detect_time: str):
    model_dir = MODEL_SAVE_DIR / model_name
    if not model_dir.exists():
        print(f"[SKIP] 机型 {model_name} 无已训练模型，请先运行 train.py")
        return

    turbines = list_turbines(model_name)
    print(f"\n机型 {model_name}: 共 {len(turbines)} 台风机，检测日期 {target_date}")

    # ── 加载 model_stats（优先从 pkl，回退到全量数据） ──
    stats_path = model_dir / "model_stats.pkl"
    if stats_path.exists():
        model_stats = joblib.load(stats_path)
        print(f"  [stats] 从 model_stats.pkl 加载")
    else:
        from data_loader import load_model_type
        df_all = load_model_type(model_name, required_cols=["p_active"],
                                 optional_cols=["gen_spd", "actual_torque", "pitch_ang_act_1"])
        if df_all.empty:
            print(f"[SKIP] 机型 {model_name} 无数据")
            return
        model_stats = get_model_statistics(df_all)
        del df_all
        print(f"  [stats] 从全量数据计算（建议重新训练以生成 model_stats.pkl）")

    # ── 机型级预加载所有模型（一次 joblib.load，所有风机复用） ──
    PKL_MAP = {
        "wind_power_curve":   ("wind_power_curve.pkl",   PowerCurveDetector.load),
        "wind_power_scatter": ("wind_power_scatter.pkl", ScatterDetector.load),
        "yaw_static":         ("yaw_static.pkl",         StaticYawDetector.load),
        "yaw_twist":          ("yaw_twist.pkl",          CableTwistDetector.load),
        "pitch_regulation":   ("pitch_regulation.pkl",   PitchRegulationDetector.load),
        "pitch_coord":        ("pitch_coord.pkl",        PitchCoordDetector.load),
        "pitch_min":          ("pitch_min.pkl",           MinPitchDetector.load),
        "ctrl_power_quality": ("ctrl_power_quality.pkl", PowerQualityDetector.load),
        "ctrl_op_state":      ("ctrl_op_state.pkl",      OperationStateDetector.load),
    }
    models = {}
    for det_name, (pkl_file, load_fn) in PKL_MAP.items():
        pkl_path = model_dir / pkl_file
        if pkl_path.exists():
            try:
                models[det_name] = load_fn(pkl_path)
            except Exception as e:
                print(f"  [WARN] 加载 {pkl_file} 失败: {e}")
    print(f"  [模型] 预加载 {len(models)}/{len(PKL_MAP)} 个检测器")

    for rec in turbines:
        print(f"  检测: {rec['farm_name']} / {rec['turbine_name']}")
        detect_turbine(rec, models, conn, model_stats, target_date, detect_time)


# ── 主流程 ─────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="风机异常检测推理（逐点输出）")
    parser.add_argument("--model", type=str, default=None, help="指定机型名称")
    parser.add_argument("--date",  type=str, default=None,
                        help="指定检测日期 YYYY-MM-DD，默认为昨天")
    args = parser.parse_args()

    if args.date:
        target_date = datetime.strptime(args.date, "%Y-%m-%d").date()
    else:
        target_date = (datetime.now() - timedelta(days=1)).date()

    detect_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    conn = init_db(DB_PATH)
    print(f"结果数据库: {DB_PATH}")
    print(f"检测日期:   {target_date}")

    if args.model:
        detect_model_type(args.model, conn, target_date, detect_time)
    else:
        for mt in list_model_types():
            detect_model_type(mt, conn, target_date, detect_time)

    conn.close()
    print("\n检测完成。")


if __name__ == "__main__":
    main()