defect-analysis/tests/test_production_modules.py

import unittest

import pandas as pd

from defect_analysis.data_quality import build_data_quality_report
from defect_analysis.root_cause import build_extended_root_causes
from defect_analysis.schemas import (
    CORE_REQUIRED_COLUMNS,
    TEMPLATE_COLUMNS,
    get_missing_required_columns,
    normalize_defect_schema,
)


class ProductionModulesTest(unittest.TestCase):
    def test_schema_exposes_core_and_template_columns(self):
        self.assertIn("defect_id", CORE_REQUIRED_COLUMNS)
        self.assertIn("lam_fixture_id", TEMPLATE_COLUMNS)
        self.assertIn("material_lot_oca", TEMPLATE_COLUMNS)

    def test_schema_normalization_keeps_legacy_csv_compatible(self):
        legacy = pd.DataFrame(
            {
                "defect_id": ["D1"],
                "panel_id": ["P1"],
                "batch_id": ["B1"],
                "equipment_id": ["LAM-A01"],
                "seat_id": ["R1C1"],
                "inspection_station": ["AOI-1"],
                "timestamp": [pd.Timestamp("2026-04-01 08:00:00")],
                "defect_type": ["划痕"],
                "severity": ["严重"],
                "x_mm": [10.0],
                "y_mm": [20.0],
                "panel_width_mm": [155.0],
                "panel_height_mm": [340.0],
                "hour": [8],
                "shift": ["白班"],
                "day": ["2026-04-01"],
            }
        )

        normalized = normalize_defect_schema(legacy)

        self.assertEqual([], get_missing_required_columns(normalized))
        self.assertEqual("point", normalized.loc[0, "defect_geometry_type"])
        self.assertEqual("LAM-A01", normalized.loc[0, "lam_equipment_id"])

    def test_data_quality_report_flags_invalid_coordinates_and_traceability(self):
        df = normalize_defect_schema(
            pd.DataFrame(
                {
                    "defect_id": ["D1", "D2"],
                    "panel_id": ["P1", ""],
                    "batch_id": ["B1", "B1"],
                    "equipment_id": ["LAM-A01", "LAM-A01"],
                    "seat_id": ["R1C1", "R1C2"],
                    "inspection_station": ["AOI-1", "AOI-1"],
                    "timestamp": [pd.Timestamp("2026-04-01"), pd.Timestamp("2026-04-01")],
                    "defect_type": ["划痕", "未知缺陷"],
                    "severity": ["严重", "轻微"],
                    "x_mm": [10.0, 999.0],
                    "y_mm": [20.0, 20.0],
                    "panel_width_mm": [155.0, 155.0],
                    "panel_height_mm": [340.0, 340.0],
                    "hour": [8, 8],
                    "shift": ["白班", "白班"],
                    "day": ["2026-04-01", "2026-04-01"],
                    "lam_fixture_id": ["FIX-1", ""],
                }
            )
        )

        report = build_data_quality_report(df)

        self.assertLess(report["score"], 100)
        self.assertLess(report["coordinate_valid_rate"], 1.0)
        self.assertLess(report["traceability_rate"], 1.0)
        self.assertTrue(any("坐标" in issue for issue in report["issues"]))

    def test_data_quality_report_handles_missing_columns(self):
        report = build_data_quality_report(pd.DataFrame({"defect_id": ["D1"]}))

        self.assertLess(report["score"], 100)
        self.assertEqual(0.0, report["coordinate_valid_rate"])
        self.assertTrue(report["issues"])

    def test_root_cause_module_returns_extended_candidates(self):
        rows = []
        for i in range(10):
            rows.append(
                {
                    "defect_id": f"D{i}",
                    "panel_id": f"P{i}",
                    "batch_id": "B1",
                    "equipment_id": "LAM-A01",
                    "seat_id": "R1C1",
                    "inspection_station": "AOI-1",
                    "timestamp": pd.Timestamp("2026-04-01"),
                    "defect_type": "划痕",
                    "severity": "严重" if i < 3 else "轻微",
                    "x_mm": 10.0,
                    "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 i < 8 else "FIX-OK",
                }
            )
        df = normalize_defect_schema(pd.DataFrame(rows))

        candidates = build_extended_root_causes(df, dimensions=["lam_fixture_id"])

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

    def test_root_cause_empty_dimensions_do_not_fallback_to_defaults(self):
        df = normalize_defect_schema(
            pd.DataFrame(
                {
                    "defect_id": ["D1"],
                    "panel_id": ["P1"],
                    "batch_id": ["B1"],
                    "equipment_id": ["LAM-A01"],
                    "seat_id": ["R1C1"],
                    "inspection_station": ["AOI-1"],
                    "timestamp": [pd.Timestamp("2026-04-01")],
                    "defect_type": ["划痕"],
                    "severity": ["严重"],
                    "x_mm": [10.0],
                    "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-1"],
                }
            )
        )

        candidates = build_extended_root_causes(df, dimensions=[])

        self.assertTrue(candidates.empty)


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