tabs/tab4_detection.py

"""
自定义检测标签页
农村低压配电网电压异常检测项目

本模块实现用户自定义数据的异常检测功能：
- CSV 文件上传
- 数据预览
- 模型选择和阈值调节
- 实时推理 (CPU)
- 检测结果可视化

Author: Rural Voltage Detection Project
Date: 2026
"""

import sys
from pathlib import Path
from typing import Optional, Tuple, Dict, Any

import gradio as gr
import numpy as np
import pandas as pd

# Add parent directories to path for imports
DEMO_DIR = Path(__file__).parent.parent
CODE_DIR = DEMO_DIR
if str(CODE_DIR) not in sys.path:
    sys.path.insert(0, str(CODE_DIR))
if str(DEMO_DIR) not in sys.path:
    sys.path.insert(0, str(DEMO_DIR))

# Import core modules
from core.inference import VoltageAnomalyDetector
from core.data_processor import DataProcessor
from visualization.detection_plots import (
    create_detection_timeline,
    create_score_distribution,
)
from config import (
    MODEL_DIR,
    INFERENCE_CONFIG,
    DEMO_DATA_CONFIG,
)


# ============================================================================
# 全局变量和配置
# ============================================================================

# 可用模型列表
AVAILABLE_MODELS = ["VoltageTimesNet_v2", "TimesNet", "DLinear"]

# 模型描述
MODEL_DESCRIPTIONS = {
    "VoltageTimesNet_v2": "推荐模型: 基于 TimesNet 的改进版本，针对召回率进行优化，适合农村电压异常检测",
    "TimesNet": "基线模型: 使用 FFT 发现周期性，结合 2D 卷积捕获时序模式",
    "DLinear": "轻量级模型: 基于线性分解的简单模型，速度快但精度略低",
}

# 检测器缓存
_detector_cache: Dict[str, VoltageAnomalyDetector] = {}


# ============================================================================
# 辅助函数
# ============================================================================

def get_detector(model_name: str) -> VoltageAnomalyDetector:
    """
    获取或创建检测器实例（带缓存）

    Args:
        model_name: 模型名称

    Returns:
        VoltageAnomalyDetector 实例
    """
    if model_name not in _detector_cache:
        # 查找模型检查点
        checkpoint_path = None
        model_file = MODEL_DIR / f"best_{model_name.lower()}.pth"
        if model_file.exists():
            checkpoint_path = str(model_file)

        # 创建检测器
        detector = VoltageAnomalyDetector(
            model_name=model_name,
            checkpoint_path=checkpoint_path,
            device=INFERENCE_CONFIG.get("device", "cpu"),
        )
        _detector_cache[model_name] = detector

    return _detector_cache[model_name]


def validate_csv_file(file_path: str) -> Tuple[bool, str, Optional[pd.DataFrame]]:
    """
    验证 CSV 文件

    Args:
        file_path: CSV 文件路径

    Returns:
        (是否有效, 消息, DataFrame 或 None)
    """
    try:
        if file_path is None:
            return False, "请先上传 CSV 文件", None

        # 读取文件
        df = pd.read_csv(file_path)

        if df.empty:
            return False, "CSV 文件为空", None

        # 检查数据长度
        min_length = DEMO_DATA_CONFIG.get("window_size", 100) + 10
        if len(df) < min_length:
            return False, f"数据长度不足，至少需要 {min_length} 行数据（当前: {len(df)} 行）", None

        # 检查是否有数值列
        numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
        if len(numeric_cols) == 0:
            return False, "CSV 文件中没有找到数值列", None

        return True, f"文件验证成功: {len(df)} 行, {len(numeric_cols)} 个数值特征", df

    except pd.errors.EmptyDataError:
        return False, "CSV 文件为空或格式错误", None
    except pd.errors.ParserError as e:
        return False, f"CSV 解析错误: {str(e)}", None
    except Exception as e:
        return False, f"读取文件失败: {str(e)}", None


def format_preview_df(df: pd.DataFrame, max_rows: int = 10) -> pd.DataFrame:
    """
    格式化预览 DataFrame

    Args:
        df: 原始 DataFrame
        max_rows: 最大显示行数

    Returns:
        格式化后的 DataFrame
    """
    # 选择数值列
    exclude_cols = ["timestamp", "date", "time", "label", "index", "Unnamed: 0"]
    feature_cols = [c for c in df.columns if c not in exclude_cols]

    # 截取前 N 行
    preview_df = df[feature_cols].head(max_rows).copy()

    # 四舍五入数值
    for col in preview_df.columns:
        if preview_df[col].dtype in [np.float64, np.float32]:
            preview_df[col] = preview_df[col].round(4)

    return preview_df


def generate_detection_stats(
    labels: np.ndarray,
    scores: np.ndarray,
    threshold: float,
    total_samples: int,
) -> str:
    """
    生成检测统计信息

    Args:
        labels: 预测标签
        scores: 异常分数
        threshold: 阈值
        total_samples: 原始数据样本数

    Returns:
        Markdown 格式的统计信息
    """
    n_anomaly = int(np.sum(labels))
    n_normal = len(labels) - n_anomaly
    anomaly_ratio = n_anomaly / len(labels) * 100 if len(labels) > 0 else 0

    # 分数统计
    score_mean = float(np.mean(scores))
    score_std = float(np.std(scores))
    score_max = float(np.max(scores))
    score_min = float(np.min(scores))

    stats_md = f"""
### 检测统计

| 指标 | 数值 |
|------|------|
| 原始数据样本数 | {total_samples} |
| 检测窗口数 | {len(labels)} |
| 检测到的异常 | {n_anomaly} ({anomaly_ratio:.2f}%) |
| 正常样本 | {n_normal} ({100-anomaly_ratio:.2f}%) |
| 使用阈值 | {threshold:.4f} |

### 异常分数统计

| 指标 | 数值 |
|------|------|
| 平均分数 | {score_mean:.4f} |
| 标准差 | {score_std:.4f} |
| 最大分数 | {score_max:.4f} |
| 最小分数 | {score_min:.4f} |

> 注: 异常分数表示重构误差，分数越高表示越可能是异常。
"""
    return stats_md


# ============================================================================
# 事件处理函数
# ============================================================================

def handle_file_upload(file) -> Tuple[str, pd.DataFrame, str]:
    """
    处理文件上传事件

    Args:
        file: 上传的文件对象

    Returns:
        (状态消息, 预览 DataFrame, 详细信息)
    """
    if file is None:
        return (
            "等待上传文件...",
            pd.DataFrame(),
            "请上传 CSV 格式的时序数据文件",
        )

    # 验证文件
    is_valid, message, df = validate_csv_file(file.name)

    if not is_valid:
        return (
            f"文件验证失败: {message}",
            pd.DataFrame(),
            message,
        )

    # 生成预览
    preview_df = format_preview_df(df, max_rows=10)

    # 生成详细信息
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    exclude_cols = ["timestamp", "date", "time", "label", "index", "Unnamed: 0"]
    feature_cols = [c for c in numeric_cols if c not in exclude_cols]

    detail_info = f"""
**文件信息:**
- 文件名: {Path(file.name).name}
- 数据行数: {len(df)}
- 特征数量: {len(feature_cols)}
- 特征列表: {', '.join(feature_cols[:8])}{'...' if len(feature_cols) > 8 else ''}
"""

    return (
        f"文件验证成功: {len(df)} 行数据",
        preview_df,
        detail_info,
    )


def handle_model_change(model_name: str) -> str:
    """
    处理模型选择变更

    Args:
        model_name: 选择的模型名称

    Returns:
        模型描述信息
    """
    desc = MODEL_DESCRIPTIONS.get(model_name, "未知模型")
    return f"**{model_name}**: {desc}"


def run_detection(
    file,
    model_name: str,
    threshold: float,
    progress=gr.Progress(),
) -> Tuple[Any, Any, str]:
    """
    执行异常检测

    Args:
        file: 上传的文件对象
        model_name: 模型名称
        threshold: 异常阈值
        progress: Gradio 进度条

    Returns:
        (时间线图, 分布图, 统计信息)
    """
    # 验证输入
    if file is None:
        return None, None, "请先上传 CSV 文件"

    try:
        progress(0, desc="验证文件...")
        is_valid, message, df = validate_csv_file(file.name)

        if not is_valid:
            return None, None, f"文件验证失败: {message}"

        progress(0.1, desc="加载数据...")

        # 提取特征数据
        exclude_cols = ["timestamp", "date", "time", "label", "index", "Unnamed: 0"]
        feature_cols = [c for c in df.columns if c not in exclude_cols
                       and df[c].dtype in [np.float64, np.float32, np.int64, np.int32]]

        if len(feature_cols) == 0:
            return None, None, "未找到有效的数值特征列"

        data = df[feature_cols].values.astype(np.float32)
        data = np.nan_to_num(data, nan=0.0)
        total_samples = len(data)

        progress(0.2, desc="创建数据窗口...")

        # 创建数据处理器
        window_size = DEMO_DATA_CONFIG.get("window_size", 100)
        step_size = DEMO_DATA_CONFIG.get("step_size", 1)

        processor = DataProcessor(
            seq_len=window_size,
            step=step_size,
            normalize=True,
        )

        # 处理数据
        windows = processor.fit_transform(data)
        n_windows = windows.shape[0]

        progress(0.3, desc=f"加载模型 {model_name}...")

        # 获取检测器
        detector = get_detector(model_name)

        # 加载模型（如果尚未加载）
        if not detector.is_loaded:
            detector.load_model()

        progress(0.5, desc="执行推理...")

        # 执行检测
        results = detector.predict(windows, threshold=threshold)

        scores = results["scores"]
        labels = results["labels"]

        progress(0.8, desc="生成可视化...")

        # 准备可视化数据
        # 使用原始数据的前 N 个点（与 scores 长度对齐）
        vis_data = data[:len(scores), :min(3, data.shape[1])]
        vis_feature_names = feature_cols[:min(3, len(feature_cols))]

        # 创建时间线图
        timeline_fig = create_detection_timeline(
            data=vis_data,
            scores=scores,
            labels=labels,
            threshold=threshold,
            feature_names=vis_feature_names,
            max_features=3,
            title="异常检测结果时间线",
        )

        progress(0.9, desc="生成统计信息...")

        # 创建分数分布图（使用预测标签）
        dist_fig = create_score_distribution(
            scores=scores,
            labels=labels,
            threshold=threshold,
            bins=50,
            title="异常分数分布",
        )

        # 生成统计信息
        stats_md = generate_detection_stats(
            labels=labels,
            scores=scores,
            threshold=threshold,
            total_samples=total_samples,
        )

        progress(1.0, desc="完成!")

        return timeline_fig, dist_fig, stats_md

    except ValueError as e:
        return None, None, f"数据处理错误: {str(e)}"
    except RuntimeError as e:
        return None, None, f"模型推理错误: {str(e)}"
    except Exception as e:
        import traceback
        error_detail = traceback.format_exc()
        return None, None, f"检测失败: {str(e)}\n\n详细信息:\n```\n{error_detail}\n```"


# ============================================================================
# 创建标签页
# ============================================================================

def create_detection_tab():
    """
    创建自定义检测标签页

    Returns:
        Gradio Tab 组件
    """
    with gr.Tab("自定义检测") as tab:
        gr.Markdown("""
        ## 自定义数据异常检测

        上传您自己的时序数据文件，使用训练好的模型进行异常检测。

        **使用说明:**
        1. 上传 CSV 格式的时序数据文件
        2. 选择检测模型
        3. 调整异常阈值
        4. 点击「开始检测」按钮

        > 提示: 数据文件应包含数值型特征列，至少需要 100+ 行数据。
        """)

        with gr.Row():
            # 左侧: 输入区域
            with gr.Column(scale=1):
                gr.Markdown("### 数据上传")

                # 文件上传
                file_input = gr.File(
                    label="上传 CSV 文件",
                    file_types=[".csv"],
                    file_count="single",
                )

                # 文件状态
                file_status = gr.Textbox(
                    label="文件状态",
                    value="等待上传文件...",
                    interactive=False,
                    lines=1,
                )

                # 文件详情
                file_detail = gr.Markdown(
                    value="请上传 CSV 格式的时序数据文件"
                )

                gr.Markdown("### 模型配置")

                # 模型选择
                model_selector = gr.Radio(
                    choices=AVAILABLE_MODELS,
                    value="VoltageTimesNet_v2",
                    label="选择模型",
                )

                # 模型描述
                model_desc = gr.Markdown(
                    value=f"**VoltageTimesNet_v2**: {MODEL_DESCRIPTIONS['VoltageTimesNet_v2']}"
                )

                # 阈值滑块
                threshold_slider = gr.Slider(
                    minimum=0.1,
                    maximum=1.0,
                    value=0.5,
                    step=0.01,
                    label="异常阈值",
                    info="阈值越低，检测越敏感（检出更多异常）；阈值越高，检测越保守",
                )

                # 检测按钮
                detect_btn = gr.Button(
                    "开始检测",
                    variant="primary",
                    size="lg",
                )

            # 右侧: 数据预览
            with gr.Column(scale=1):
                gr.Markdown("### 数据预览 (前 10 行)")

                preview_table = gr.Dataframe(
                    label="数据预览",
                    headers=None,
                    interactive=False,
                    wrap=True,
                )

        gr.Markdown("---")
        gr.Markdown("### 检测结果")

        # 结果区域
        with gr.Row():
            # 时间线图
            with gr.Column(scale=2):
                timeline_plot = gr.Plot(
                    label="异常检测时间线",
                )

        with gr.Row():
            # 分布图
            with gr.Column(scale=1):
                dist_plot = gr.Plot(
                    label="异常分数分布",
                )

            # 统计信息
            with gr.Column(scale=1):
                stats_output = gr.Markdown(
                    value="等待检测...",
                    label="检测统计",
                )

        # ====================================================================
        # 事件绑定
        # ====================================================================

        # 文件上传事件
        file_input.change(
            fn=handle_file_upload,
            inputs=[file_input],
            outputs=[file_status, preview_table, file_detail],
        )

        # 模型选择事件
        model_selector.change(
            fn=handle_model_change,
            inputs=[model_selector],
            outputs=[model_desc],
        )

        # 检测按钮事件
        detect_btn.click(
            fn=run_detection,
            inputs=[file_input, model_selector, threshold_slider],
            outputs=[timeline_plot, dist_plot, stats_output],
        )

    return tab


# ============================================================================
# 测试代码
# ============================================================================

if __name__ == "__main__":
    # 创建测试应用
    with gr.Blocks(title="自定义检测测试", theme="soft") as demo:
        create_detection_tab()

    demo.launch(
        server_name="0.0.0.0",
        server_port=7862,
        share=False,
    )