defect-analysis/app_utils.py

"""缺陷分析页面的可测试业务逻辑。"""

import numpy as np
import pandas as pd


def normalize_date_bounds(start_date, end_date):
    """把日期范围转换成左闭右开的时间边界，确保结束日期整天被包含。"""
    start_ts = pd.Timestamp(start_date).normalize()
    end_exclusive = pd.Timestamp(end_date).normalize() + pd.Timedelta(days=1)
    return start_ts, end_exclusive


def apply_defect_filters(
    df,
    *,
    start_date,
    end_date,
    selected_types,
    selected_batches,
    selected_equipment,
    selected_seats,
    selected_shift="全部",
    selected_severity="全部",
):
    """应用页面筛选条件。"""
    start_ts, end_exclusive = normalize_date_bounds(start_date, end_date)
    mask = (
        (df["timestamp"] >= start_ts)
        & (df["timestamp"] < end_exclusive)
        & (df["defect_type"].isin(selected_types))
        & (df["batch_id"].isin(selected_batches))
        & (df["equipment_id"].isin(selected_equipment))
    )
    if selected_shift != "全部":
        mask &= df["shift"] == selected_shift
    if selected_severity != "全部":
        mask &= df["severity"] == selected_severity
    if selected_seats:
        mask &= df["seat_id"].isin(selected_seats)

    return df[mask].copy()


def calculate_kpis(source_df, filtered_df):
    """基于当前筛选结果计算页面 KPI。"""
    total_panels_inspected = filtered_df["panel_id"].nunique()
    defective_panels = filtered_df["panel_id"].nunique()
    total_defects = len(filtered_df)
    critical_defects = int((filtered_df["severity"] == "严重").sum()) if total_defects else 0
    top_defect_type = filtered_df["defect_type"].mode().iloc[0] if total_defects else "-"
    yield_rate = (1 - defective_panels / max(total_panels_inspected, 1)) * 100

    return {
        "total_panels_inspected": int(total_panels_inspected),
        "defective_panels": int(defective_panels),
        "yield_rate": float(yield_rate),
        "total_defects": int(total_defects),
        "critical_defects": int(critical_defects),
        "top_defect_type": top_defect_type,
    }


def calculate_spc_metrics(df):
    """计算 SPC 所需数据，防止模拟分母造成非法概率。"""
    daily = df.groupby("day").agg(
        total_defects=("defect_id", "count"),
        panels_with_defects=("panel_id", "nunique"),
    ).reset_index()
    daily["day"] = pd.to_datetime(daily["day"])
    daily = daily.sort_values("day").reset_index(drop=True)

    if len(daily) < 2:
        return {
            "daily": daily,
            "p_bar": 0.0,
            "ucl": 0.0,
            "lcl": 0.0,
            "uwl": 0.0,
            "lwl": 0.0,
            "sigma_p": 0.0,
        }

    total_days = (df["timestamp"].max() - df["timestamp"].min()).days + 1
    total_unique_panels = df["panel_id"].nunique()
    estimated = max(total_unique_panels // max(total_days // 7, 1), 1)
    daily["estimated_inspected"] = np.maximum(estimated, daily["panels_with_defects"])
    daily["defect_rate"] = (
        daily["panels_with_defects"] / daily["estimated_inspected"]
    ).clip(lower=0, upper=1)

    p_bar = float(np.clip(daily["defect_rate"].mean(), 0, 1))
    n_avg = float(daily["estimated_inspected"].mean())
    sigma_p = float(np.sqrt(max(p_bar * (1 - p_bar), 0) / n_avg)) if n_avg > 0 else 0.0

    return {
        "daily": daily,
        "p_bar": p_bar,
        "ucl": min(1.0, p_bar + 3 * sigma_p),
        "lcl": max(0.0, p_bar - 3 * sigma_p),
        "uwl": min(1.0, p_bar + 2 * sigma_p),
        "lwl": max(0.0, p_bar - 2 * sigma_p),
        "sigma_p": sigma_p,
    }


def build_diagnostic_dashboard(df):
    """生成诊断驾驶舱需要的摘要、根因候选和趋势数据。"""
    total_defects = len(df)
    if total_defects == 0:
        return {
            "severity_level": "正常",
            "top_defect_type": "-",
            "top_defect_share": 0.0,
            "serious_share": 0.0,
            "root_causes": pd.DataFrame(),
            "daily_trend": pd.DataFrame(),
            "pareto": pd.DataFrame(),
            "primary_recommendation": "当前筛选条件下没有缺陷记录。",
        }

    type_counts = df["defect_type"].value_counts()
    top_defect_type = type_counts.index[0]
    top_defect_share = float(type_counts.iloc[0] / total_defects)
    serious_share = float((df["severity"] == "严重").sum() / total_defects)

    root_causes = (
        df.groupby(["equipment_id", "seat_id"])
        .agg(
            缺陷数=("defect_id", "count"),
            涉及面板=("panel_id", "nunique"),
            主要缺陷=("defect_type", lambda s: s.mode().iloc[0]),
            严重数=("severity", lambda s: int((s == "严重").sum())),
        )
        .reset_index()
    )
    root_causes["根因候选"] = root_causes["equipment_id"] + " / " + root_causes["seat_id"]
    root_causes["占比"] = root_causes["缺陷数"] / total_defects
    root_causes["严重占比"] = root_causes["严重数"] / root_causes["缺陷数"].clip(lower=1)
    equipment_totals = df.groupby("equipment_id")["defect_id"].count()
    equipment_seat_counts = df.groupby("equipment_id")["seat_id"].nunique().clip(lower=1)
    root_causes["期望缺陷数"] = root_causes["equipment_id"].map(
        equipment_totals / equipment_seat_counts
    ).clip(lower=0.001)
    root_causes["异常倍数"] = (root_causes["缺陷数"] / root_causes["期望缺陷数"]).round(2)
    count_score = root_causes["缺陷数"] / root_causes["缺陷数"].max()
    panel_score = root_causes["涉及面板"] / df["panel_id"].nunique()
    lift_score = (root_causes["异常倍数"] / 3).clip(upper=1)
    root_causes["风险分"] = (
        count_score * 55 + lift_score * 25 + root_causes["严重占比"] * 15 + panel_score * 5
    ).round(1)
    root_causes = root_causes.sort_values(["风险分", "缺陷数"], ascending=False).head(8)
    root_causes = root_causes[
        ["根因候选", "缺陷数", "占比", "异常倍数", "涉及面板", "主要缺陷", "严重占比", "风险分"]
    ].reset_index(drop=True)

    pareto = type_counts.rename_axis("缺陷类型").reset_index(name="缺陷数")
    pareto["占比"] = pareto["缺陷数"] / total_defects
    pareto["累计占比"] = pareto["占比"].cumsum()

    daily_trend = df.groupby("day").size().rename("缺陷数").reset_index()
    daily_trend["day"] = pd.to_datetime(daily_trend["day"])
    daily_trend = daily_trend.sort_values("day")

    if serious_share >= 0.2 or (len(root_causes) > 0 and root_causes.iloc[0]["占比"] >= 0.15):
        severity_level = "严重"
    elif serious_share >= 0.1 or top_defect_share >= 0.35:
        severity_level = "关注"
    else:
        severity_level = "正常"

    if len(root_causes) > 0:
        top_root = root_causes.iloc[0]
        primary_recommendation = (
            f"优先排查 {top_root['根因候选']}，该组合贡献 {top_root['占比']:.1%} "
            f"缺陷，异常倍数 {top_root['异常倍数']:.2f}x，主要类型为 {top_root['主要缺陷']}。"
        )
    else:
        primary_recommendation = f"优先排查 {top_defect_type} 相关工艺参数。"

    return {
        "severity_level": severity_level,
        "top_defect_type": top_defect_type,
        "top_defect_share": top_defect_share,
        "serious_share": serious_share,
        "root_causes": root_causes,
        "daily_trend": daily_trend,
        "pareto": pareto,
        "primary_recommendation": primary_recommendation,
    }