defect-analysis/tests/test_app_utils.py

import math
import unittest

import pandas as pd

from app_utils import (
    apply_defect_filters,
    build_html_report,
    build_ml_factor_insights,
    build_diagnostic_dashboard,
    classify_panel_zone,
    calculate_kpis,
    calculate_spc_metrics,
    generate_industry_diagnosis,
    generate_report_charts,
    normalize_defect_schema,
)


class AppUtilsTest(unittest.TestCase):
    def setUp(self):
        self.df = pd.DataFrame(
            {
                "defect_id": ["D1", "D2", "D3", "D4"],
                "panel_id": ["P1", "P2", "P2", "P3"],
                "batch_id": ["B1", "B1", "B2", "B2"],
                "equipment_id": ["E1", "E1", "E2", "E2"],
                "seat_id": ["S1", "S2", "S1", "S2"],
                "timestamp": pd.to_datetime(
                    [
                        "2026-04-01 00:00:00",
                        "2026-04-01 23:59:59",
                        "2026-04-02 12:00:00",
                        "2026-04-03 00:00:01",
                    ]
                ),
                "defect_type": ["划痕", "亮点", "划痕", "暗点"],
                "severity": ["严重", "轻微", "中等", "严重"],
                "shift": ["白班", "夜班", "白班", "白班"],
                "day": ["2026-04-01", "2026-04-01", "2026-04-02", "2026-04-03"],
            }
        )

    def test_date_filter_includes_full_end_date(self):
        filtered = apply_defect_filters(
            self.df,
            start_date=pd.Timestamp("2026-04-01"),
            end_date=pd.Timestamp("2026-04-01"),
            selected_types=["划痕", "亮点", "暗点"],
            selected_batches=["B1", "B2"],
            selected_equipment=["E1", "E2"],
            selected_seats=["S1", "S2"],
            selected_shift="全部",
            selected_severity="全部",
        )

        self.assertEqual(["D1", "D2"], filtered["defect_id"].tolist())

    def test_kpis_use_same_filter_scope_for_total_panels(self):
        filtered = apply_defect_filters(
            self.df,
            start_date=pd.Timestamp("2026-04-01"),
            end_date=pd.Timestamp("2026-04-02"),
            selected_types=["划痕"],
            selected_batches=["B1", "B2"],
            selected_equipment=["E1", "E2"],
            selected_seats=["S1"],
            selected_shift="全部",
            selected_severity="全部",
        )

        kpis = calculate_kpis(self.df, filtered)

        self.assertEqual(2, kpis["total_panels_inspected"])
        self.assertEqual(2, kpis["defective_panels"])
        self.assertEqual(0.0, kpis["yield_rate"])

    def test_spc_metrics_clamp_estimated_rate_to_valid_probability(self):
        metrics = calculate_spc_metrics(self.df)

        self.assertTrue(math.isfinite(metrics["p_bar"]))
        self.assertTrue(math.isfinite(metrics["ucl"]))
        self.assertTrue(math.isfinite(metrics["lcl"]))
        self.assertLessEqual(metrics["daily"]["defect_rate"].max(), 1.0)

    def test_diagnostic_dashboard_ranks_root_cause_candidates(self):
        dashboard = build_diagnostic_dashboard(self.df)

        self.assertEqual("严重", dashboard["severity_level"])
        self.assertEqual("E1 / S1", dashboard["root_causes"].iloc[0]["根因候选"])
        self.assertEqual("划痕", dashboard["top_defect_type"])
        self.assertIn("优先排查", dashboard["primary_recommendation"])

    def test_diagnostic_dashboard_reports_baseline_lift(self):
        rows = []
        for i in range(10):
            rows.append(
                {
                    "defect_id": f"D{i}",
                    "panel_id": f"P{i}",
                    "batch_id": "B1",
                    "equipment_id": "E1",
                    "seat_id": "S-hot" if i < 8 else "S-cold",
                    "timestamp": pd.Timestamp("2026-04-01"),
                    "defect_type": "气泡",
                    "severity": "严重" if i < 2 else "轻微",
                    "shift": "白班",
                    "day": "2026-04-01",
                }
            )
        df = pd.DataFrame(rows)

        dashboard = build_diagnostic_dashboard(df)
        top = dashboard["root_causes"].iloc[0]

        self.assertEqual("E1 / S-hot", top["根因候选"])
        self.assertGreater(top["异常倍数"], 1.0)

    def test_classify_panel_zone_uses_3c_panel_regions(self):
        zones = classify_panel_zone(
            pd.DataFrame(
                {
                    "x_mm": [2.0, 77.5, 150.0, 80.0],
                    "y_mm": [335.0, 255.0, 170.0, 20.0],
                    "panel_width_mm": [155.0] * 4,
                    "panel_height_mm": [340.0] * 4,
                }
            )
        )

        self.assertIn("角落区", zones.iloc[0])
        self.assertIn("FPC/绑定区", zones.iloc[1])
        self.assertIn("右边缘区", zones.iloc[2])
        self.assertIn("下边缘区", zones.iloc[3])

    def test_industry_diagnosis_generates_panel_sop_recommendation(self):
        rows = []
        for i in range(12):
            rows.append(
                {
                    "defect_id": f"D{i}",
                    "panel_id": f"P{i}",
                    "batch_id": "B1",
                    "equipment_id": "LAM-A01",
                    "seat_id": "R2C3",
                    "timestamp": pd.Timestamp("2026-04-01"),
                    "defect_type": "气泡",
                    "severity": "严重" if i < 4 else "中等",
                    "x_mm": 5.0 + i * 0.3,
                    "y_mm": 250.0,
                    "panel_width_mm": 155.0,
                    "panel_height_mm": 340.0,
                    "shift": "白班",
                    "day": "2026-04-01",
                }
            )
        df = pd.DataFrame(rows)
        dashboard = build_diagnostic_dashboard(df)

        diagnosis = generate_industry_diagnosis(df, dashboard)

        self.assertIn("边缘", diagnosis["headline"])
        self.assertIn("气泡", diagnosis["headline"])
        self.assertTrue(any("贴合" in item for item in diagnosis["recommendations"]))
        self.assertTrue(any("跨面板重复" in pattern for pattern in diagnosis["patterns"]))

    def test_normalize_defect_schema_backfills_industry_fields(self):
        normalized = normalize_defect_schema(self.df)

        self.assertIn("defect_geometry_type", normalized.columns)
        self.assertIn("lam_equipment_id", normalized.columns)
        self.assertIn("clean_equipment_id", normalized.columns)
        self.assertEqual(["point"] * len(normalized), normalized["defect_geometry_type"].tolist())
        self.assertEqual(normalized["equipment_id"].tolist(), normalized["lam_equipment_id"].tolist())
        self.assertEqual(normalized["seat_id"].tolist(), normalized["lam_seat_id"].tolist())
        self.assertTrue((normalized["area_mm2"] >= 0).all())

    def test_diagnostic_dashboard_includes_extended_root_causes(self):
        rows = []
        for i in range(12):
            rows.append(
                {
                    "defect_id": f"D{i}",
                    "panel_id": f"P{i}",
                    "batch_id": "B1",
                    "equipment_id": "LAM-A01",
                    "seat_id": f"R{i % 4 + 1}C1",
                    "inspection_station": "AOI-1",
                    "timestamp": pd.Timestamp("2026-04-01"),
                    "defect_type": "划痕",
                    "severity": "严重" if i < 4 else "轻微",
                    "x_mm": 10 + i,
                    "y_mm": 20 + i,
                    "panel_width_mm": 155.0,
                    "panel_height_mm": 340.0,
                    "hour": 8,
                    "shift": "白班",
                    "day": "2026-04-01",
                    "lam_fixture_id": "FIX-HOT" if i < 9 else "FIX-OK",
                    "lam_nozzle_id": "NZ-01" if i < 9 else "NZ-02",
                    "material_lot_oca": "OCA-HOT" if i < 9 else "OCA-OK",
                }
            )
        df = normalize_defect_schema(pd.DataFrame(rows))

        dashboard = build_diagnostic_dashboard(df)
        extended = dashboard["extended_root_causes"]

        self.assertFalse(extended.empty)
        self.assertEqual("lam_fixture_id", extended.iloc[0]["维度"])
        self.assertEqual("FIX-HOT", extended.iloc[0]["候选值"])
        self.assertGreater(extended.iloc[0]["异常倍数"], 1.0)

    def test_ml_factor_insights_include_model_audit_outputs(self):
        rows = []
        for i in range(40):
            hot = i < 24
            rows.append(
                {
                    "defect_id": f"D{i}",
                    "panel_id": f"P{i}",
                    "batch_id": "B1",
                    "equipment_id": "LAM-A01" if hot else "LAM-B01",
                    "seat_id": "R1C1" if hot else "R2C2",
                    "inspection_station": "AOI-1",
                    "timestamp": pd.Timestamp("2026-04-01 08:00:00"),
                    "defect_type": "气泡" if hot else "划痕",
                    "severity": "严重" if i % 5 == 0 else "轻微",
                    "x_mm": 10.0 + i,
                    "y_mm": 20.0,
                    "panel_width_mm": 155.0,
                    "panel_height_mm": 340.0,
                    "hour": 8,
                    "shift": "白班",
                    "day": "2026-04-01",
                    "lam_fixture_id": "FIX-HOT" if hot else "FIX-OK",
                    "material_lot_oca": "OCA-HOT" if hot else "OCA-OK",
                }
            )
        df = normalize_defect_schema(pd.DataFrame(rows))

        insights = build_ml_factor_insights(df, target_defect_type="气泡", model_name="random_forest", top_n=5)

        self.assertIsNone(insights["error"])
        self.assertEqual("气泡", insights["target_defect_type"])
        self.assertFalse(insights["key_factors"].empty)
        self.assertIn("validation_auc", insights["validation_metrics"])
        self.assertGreater(len(insights["feature_importance"]), 0)

    def test_build_html_report_creates_self_contained_safe_webpage(self):
        html = build_html_report(
            generated_at="2026-05-18 10:00:00",
            date_range_text="2026-04-01 ~ 2026-04-30",
            view_mode="工程师",
            defect_count=12,
            panel_count=8,
            kpis={
                "total_panels_inspected": 8,
                "defective_panels": 3,
                "yield_rate": 62.5,
                "critical_defects": 2,
            },
            type_counts=pd.Series({"气泡<script>": 7, "划痕": 5}),
            equipment_counts=pd.Series({"LAM-A01": 9}),
            seat_top=pd.Series({"R1C1": 4}),
            trend_summary="缺陷数趋势: 上升",
            anomaly_rows=[{"equipment": "LAM-A01", "seat": "R1C1", "count": 4}],
            recommendations=["重点关注气泡"],
            chart_summaries=["TOP1 缺陷类型：气泡，占比 58.3%"],
        )

        self.assertIn("<!doctype html>", html.lower())
        self.assertIn("缺陷集中性分析综合报告", html)
        self.assertIn("气泡&lt;script&gt;", html)
        self.assertNotIn("气泡<script>", html)
        self.assertIn("图表摘要", html)
        self.assertIn("TOP1 缺陷类型：气泡，占比 58.3%", html)

    def test_generate_report_charts_creates_four_offline_images(self):
        df = pd.DataFrame(
            {
                "defect_id": ["D1", "D2", "D3", "D4"],
                "panel_id": ["P1", "P2", "P3", "P4"],
                "batch_id": ["B1"] * 4,
                "equipment_id": ["LAM-A01", "LAM-A01", "LAM-B01", "LAM-B01"],
                "seat_id": ["R1C1", "R1C2", "R2C1", "R2C2"],
                "timestamp": pd.to_datetime(["2026-04-01", "2026-04-02", "2026-04-02", "2026-04-03"]),
                "defect_type": ["气泡", "气泡", "划痕", "异物"],
                "severity": ["轻微", "中等", "严重", "中等"],
                "shift": ["白班"] * 4,
                "day": ["2026-04-01", "2026-04-02", "2026-04-02", "2026-04-03"],
            }
        )
        daily = df.groupby("day").size().rename("缺陷数").reset_index()

        charts = generate_report_charts(df, daily_trend_df=daily)

        self.assertEqual(
            {"type_distribution", "daily_trend", "equipment_distribution", "severity_pie"},
            set(charts),
        )
        self.assertTrue(all(value.startswith("data:image/png;base64,") for value in charts.values()))


if __name__ == "__main__":
    unittest.main()