defect-analysis/tests/test_app_utils.py

import math
import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import pandas as pd

from app_utils import (
    apply_defect_filters,
    build_diagnostic_dashboard,
    calculate_kpis,
    calculate_spc_metrics,
)


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)


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