defect-analysis/app_utils.py

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

import numpy as np
import pandas as pd


DEFECT_SOP_RECOMMENDATIONS = {
    "划痕": ["检查搬运轨道、吸嘴和治具接触面", "复核清洗滚刷与擦拭工位是否有硬质颗粒"],
    "气泡": ["检查贴合压力、真空度、OCA 状态和贴合速度", "复核贴合前清洁与材料开封时长"],
    "漏光": ["检查边缘贴合、背光组装、框胶和压合均匀性", "复核四角/边缘区应力与夹持状态"],
    "色差": ["检查背光、偏光片批次、贴合应力和老化条件", "对比同批材料与相邻工艺参数"],
    "异物": ["检查洁净度、清洗段、静电控制和材料暴露时间", "追溯同批材料与工位环境记录"],
    "亮点": ["复核点灯/AOI 判定、TFT 像素缺陷和异物压伤", "抽查高发区域是否存在压接或污染"],
    "暗点": ["复核点灯/AOI 判定、TFT 像素缺陷和异物压伤", "检查绑定/驱动相关区域异常"],
    "裂纹": ["立即检查切割、搬运、夹持和跌落冲击风险", "对同批面板执行 Hold 与复检"],
}


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 classify_panel_zone(df):
    """按 3C 面板行业常用语义把坐标映射到关键区域。"""
    width = df.get("panel_width_mm", pd.Series(155.0, index=df.index)).replace(0, np.nan)
    height = df.get("panel_height_mm", pd.Series(340.0, index=df.index)).replace(0, np.nan)
    x = df.get("x_mm", width * 0.5)
    y = df.get("y_mm", height * 0.5)
    x_norm = x / width
    y_norm = y / height

    zones = []
    for x, y in zip(x_norm.fillna(0.5), y_norm.fillna(0.5)):
        labels = []
        if x <= 0.1:
            labels.append("左边缘区")
        if x >= 0.9:
            labels.append("右边缘区")
        if y <= 0.1:
            labels.append("下边缘区")
        if y >= 0.9:
            labels.append("上边缘区")
        if (x <= 0.12 or x >= 0.88) and (y <= 0.12 or y >= 0.88):
            labels.append("角落区")
        if 0.68 <= y <= 0.88 and 0.25 <= x <= 0.75:
            labels.append("FPC/绑定区")
        if not labels:
            labels.append("显示中心区")
        zones.append(" / ".join(labels))

    return pd.Series(zones, index=df.index, name="panel_zone")


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()
    zones = classify_panel_zone(df)
    zone_counts = zones.value_counts()
    top_defect_type = type_counts.index[0]
    top_defect_share = float(type_counts.iloc[0] / total_defects)
    top_zone = zone_counts.index[0]
    top_zone_share = float(zone_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,
        "top_zone": top_zone,
        "top_zone_share": top_zone_share,
        "zone_distribution": zone_counts.rename_axis("区域").reset_index(name="缺陷数"),
        "serious_share": serious_share,
        "root_causes": root_causes,
        "daily_trend": daily_trend,
        "pareto": pareto,
        "primary_recommendation": primary_recommendation,
    }


def detect_industry_patterns(df):
    """识别面板行业常见缺陷模式。"""
    if df.empty:
        return []

    patterns = []
    zones = classify_panel_zone(df)
    zone_share = zones.value_counts(normalize=True)
    if any(idx != "显示中心区" and share >= 0.35 for idx, share in zone_share.items()):
        patterns.append(f"区域集中: {zone_share.index[0]} 占比 {zone_share.iloc[0]:.1%}")

    coord_df = df.copy()
    coord_df["x_bin"] = (coord_df["x_mm"] // 5).astype(int)
    coord_df["y_bin"] = (coord_df["y_mm"] // 5).astype(int)
    repeat = coord_df.groupby(["x_bin", "y_bin"])["panel_id"].nunique().max()
    if repeat >= min(3, max(2, df["panel_id"].nunique())):
        patterns.append("跨面板重复坐标: 疑似治具、吸嘴、压头或固定接触点异常")

    if df["x_mm"].nunique() >= 3 and df["y_mm"].nunique() >= 3 and len(df) >= 6:
        corr = abs(pd.Series(df["x_mm"]).corr(pd.Series(df["y_mm"])))
        if pd.notna(corr) and corr >= 0.85:
            patterns.append("线状分布: 疑似搬运划伤、滚轮轨迹或线性压伤")

    batch_share = df["batch_id"].value_counts(normalize=True).iloc[0]
    if batch_share >= 0.5 and df["batch_id"].nunique() > 1:
        patterns.append(f"批次集中: {df['batch_id'].value_counts().index[0]} 占比 {batch_share:.1%}")

    return patterns or ["随机点状分布: 更偏向材料、环境尘埃或偶发检出"]


def generate_industry_diagnosis(df, dashboard):
    """生成 3C 面板行业化诊断结论和排查建议。"""
    if df.empty:
        return {
            "headline": "当前筛选条件下没有可诊断缺陷。",
            "patterns": [],
            "recommendations": ["放宽筛选条件或上传更多检测记录后再诊断。"],
        }

    top_type = dashboard["top_defect_type"]
    top_zone = dashboard.get("top_zone", classify_panel_zone(df).value_counts().index[0])
    top_root = dashboard["root_causes"].iloc[0]["根因候选"] if len(dashboard["root_causes"]) else "当前筛选范围"
    patterns = detect_industry_patterns(df)

    recommendations = []
    if top_type in DEFECT_SOP_RECOMMENDATIONS:
        recommendations.extend(DEFECT_SOP_RECOMMENDATIONS[top_type])
    if "边缘" in top_zone or "角落" in top_zone:
        recommendations.append("优先复核边缘贴合、切割/搬运夹持、吸附接触面和四角应力状态")
    if "FPC" in top_zone or "绑定" in top_zone:
        recommendations.append("重点检查绑定压力、FPC/COF 区域异物、压接参数和 AOI 复判样本")
    if any("跨面板重复" in p for p in patterns):
        recommendations.append("对高发座号对应治具、吸嘴、压头做点检，并抽查同坐标复现样本")
    if dashboard["serious_share"] >= 0.2:
        recommendations.append("严重缺陷占比较高，建议对相关批次执行 Hold、复检或加严抽样")

    deduped = []
    for item in recommendations:
        if item not in deduped:
            deduped.append(item)

    headline = (
        f"{top_zone} 的 {top_type} 最突出，首要候选为 {top_root}。"
        f"建议按工序链路优先排查材料、贴合/搬运接触面和对应治具状态。"
    )

    return {
        "headline": headline,
        "patterns": patterns,
        "recommendations": deduped[:5],
    }