Upload src/pipeline/crack_analysis.py with huggingface_hub

src/pipeline/crack_analysis.py

@@ -0,0 +1,489 @@
+"""
+Crack Analysis & Dark Area Detection.
+
+This module performs quantitative analysis of detected defects:
+
+1. Crack Analysis:
+   - Extract crack mask
+   - Skeletonize to 1-pixel centerline
+   - Compute total length (pixels → mm using calibration)
+   - Compute local width via distance transform
+   - Classify severity (minor/moderate/severe/critical)
+   - Filter false positives (grid lines, edge artifacts)
+
+2. Dark Area Analysis:
+   - ADAPTIVE threshold based on module brightness (NOT fixed!)
+   - Compute percentage of dark area per cell
+   - Classify severity
+   
+3. Cross (Ribbon Edge Crack) Analysis:
+   - Location-aware: these occur at cell edges near ribbons
+   - Length and severity assessment
+
+Design decision: Adaptive thresholds throughout — no fixed values.
+Every threshold is computed relative to the image's own statistics.
+This is CRITICAL for handling real-world variation.
+"""
+
+import cv2
+import numpy as np
+from scipy.ndimage import distance_transform_edt
+from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass
+
+
+@dataclass
+class CrackResult:
+    """Analysis results for a single crack instance."""
+    crack_id: int
+    length_px: float
+    length_mm: float
+    mean_width_px: float
+    max_width_px: float
+    orientation_deg: float  # dominant orientation in degrees
+    bbox: Tuple[int, int, int, int]  # (y1, x1, y2, x2)
+    severity: str  # "minor", "moderate", "severe", "critical"
+    is_false_positive: bool
+    
+    def to_dict(self) -> dict:
+        return {
+            "crack_id": self.crack_id,
+            "length_px": round(self.length_px, 1),
+            "length_mm": round(self.length_mm, 2),
+            "mean_width_px": round(self.mean_width_px, 1),
+            "max_width_px": round(self.max_width_px, 1),
+            "orientation_deg": round(self.orientation_deg, 1),
+            "severity": self.severity,
+        }
+
+
+@dataclass
+class DarkResult:
+    """Analysis results for dark (inactive) area."""
+    dark_area_pct: float
+    dark_area_px: int
+    total_area_px: int
+    mean_intensity_dark: float
+    mean_intensity_normal: float
+    severity: str
+    
+    def to_dict(self) -> dict:
+        return {
+            "dark_area_pct": round(self.dark_area_pct, 2),
+            "dark_area_px": self.dark_area_px,
+            "severity": self.severity,
+            "mean_intensity_dark": round(self.mean_intensity_dark, 2),
+        }
+
+
+@dataclass 
+class CellAnalysisResult:
+    """Complete analysis results for one cell."""
+    cell_id: int
+    cracks: List[CrackResult]
+    dark: DarkResult
+    cross_cracks: List[CrackResult]
+    total_crack_length_mm: float
+    num_cracks: int
+    num_cross_cracks: int
+    max_crack_severity: str
+    defect_score: float  # 0-100, higher = worse
+    
+    def to_dict(self) -> dict:
+        return {
+            "cell_id": self.cell_id,
+            "total_crack_length_mm": round(self.total_crack_length_mm, 2),
+            "num_cracks": self.num_cracks,
+            "num_cross_cracks": self.num_cross_cracks,
+            "dark_area_pct": self.dark.dark_area_pct,
+            "max_crack_severity": self.max_crack_severity,
+            "defect_score": round(self.defect_score, 1),
+            "cracks": [c.to_dict() for c in self.cracks if not c.is_false_positive],
+            "dark": self.dark.to_dict(),
+        }
+
+
+class CrackAnalyzer:
+    """
+    Analyze crack defects from segmentation masks.
+    
+    Uses skeletonization for accurate length measurement and
+    distance transform for width estimation.
+    """
+    
+    # Severity thresholds (in mm)
+    SEVERITY_THRESHOLDS = {
+        "minor": 5.0,      # < 5mm
+        "moderate": 15.0,   # 5-15mm
+        "severe": 30.0,     # 15-30mm
+        "critical": float("inf"),  # > 30mm
+    }
+    
+    def __init__(self, px_per_mm: float = 5.0):
+        """
+        Args:
+            px_per_mm: Pixels per millimeter (estimated from cell size)
+        """
+        self.px_per_mm = px_per_mm
+    
+    def analyze_cracks(
+        self, crack_mask: np.ndarray, label: str = "crack"
+    ) -> List[CrackResult]:
+        """
+        Analyze all cracks in a binary mask.
+        
+        Args:
+            crack_mask: (H, W) uint8 binary mask (1 = crack pixel)
+            label: "crack" or "cross" for labeling
+            
+        Returns:
+            List of CrackResult for each connected crack component
+        """
+        from skimage.morphology import skeletonize
+        from skimage.measure import label as sk_label, regionprops
+        
+        if crack_mask.sum() == 0:
+            return []
+        
+        # Clean: remove very small noise
+        kernel = np.ones((3, 3), np.uint8)
+        clean = cv2.morphologyEx(crack_mask, cv2.MORPH_OPEN, kernel)
+        
+        if clean.sum() == 0:
+            return []
+        
+        # Skeletonize to 1-pixel centerline
+        skeleton = skeletonize(clean.astype(bool))
+        
+        # Distance transform for local width
+        dist_map = distance_transform_edt(clean.astype(bool))
+        
+        # Label connected components of skeleton
+        labeled = sk_label(skeleton.astype(np.uint8))
+        props = regionprops(labeled)
+        
+        results = []
+        for i, region in enumerate(props):
+            # Skip tiny components (noise)
+            if region.area < 5:
+                continue
+            
+            # Crack length = skeleton pixel count
+            length_px = float(region.area)
+            length_mm = length_px / self.px_per_mm
+            
+            # Width from distance transform along skeleton
+            component_mask = labeled == region.label
+            widths = dist_map[component_mask] * 2.0  # diameter
+            mean_width = float(widths.mean()) if len(widths) > 0 else 0.0
+            max_width = float(widths.max()) if len(widths) > 0 else 0.0
+            
+            # Orientation
+            orientation_deg = float(np.degrees(region.orientation))
+            
+            # Bounding box
+            y1, x1, y2, x2 = region.bbox
+            
+            # False positive detection
+            is_fp = self._is_false_positive(
+                component_mask, skeleton, region, crack_mask.shape
+            )
+            
+            # Severity classification
+            severity = self._classify_severity(length_mm)
+            
+            results.append(CrackResult(
+                crack_id=i + 1,
+                length_px=length_px,
+                length_mm=length_mm,
+                mean_width_px=mean_width,
+                max_width_px=max_width,
+                orientation_deg=orientation_deg,
+                bbox=(y1, x1, y2, x2),
+                severity=severity,
+                is_false_positive=is_fp,
+            ))
+        
+        return results
+    
+    def _is_false_positive(
+        self, component: np.ndarray, skeleton: np.ndarray, 
+        region, shape: tuple
+    ) -> bool:
+        """
+        Check if a detected crack is likely a false positive.
+        
+        False positive indicators:
+        1. Perfectly straight line (std of skeleton positions is very low)
+        2. Spans full width or height (likely grid line)
+        3. Located exactly at image edge
+        """
+        h, w = shape[:2]
+        y1, x1, y2, x2 = region.bbox
+        
+        # Check 1: spans full width or height
+        if (x2 - x1) > w * 0.85 and (y2 - y1) < h * 0.05:
+            return True
+        if (y2 - y1) > h * 0.85 and (x2 - x1) < w * 0.05:
+            return True
+        
+        # Check 2: perfectly straight
+        coords = np.argwhere(component)
+        if len(coords) > 10:
+            y_std = coords[:, 0].std()
+            x_std = coords[:, 1].std()
+            
+            # Perfectly horizontal or vertical
+            if y_std < 2.0 and (x2 - x1) > w * 0.3:
+                return True
+            if x_std < 2.0 and (y2 - y1) > h * 0.3:
+                return True
+        
+        # Check 3: edge-touching artifacts
+        edge_margin = 3
+        if y1 <= edge_margin and y2 <= edge_margin + 5:
+            return True  # Top edge artifact
+        if y2 >= h - edge_margin and y1 >= h - edge_margin - 5:
+            return True  # Bottom edge artifact
+        if x1 <= edge_margin and x2 <= edge_margin + 5:
+            return True  # Left edge artifact
+        if x2 >= w - edge_margin and x1 >= w - edge_margin - 5:
+            return True  # Right edge artifact
+        
+        return False
+    
+    def _classify_severity(self, length_mm: float) -> str:
+        """Classify crack severity based on length."""
+        for severity, threshold in self.SEVERITY_THRESHOLDS.items():
+            if length_mm < threshold:
+                return severity
+        return "critical"
+
+
+class DarkAreaAnalyzer:
+    """
+    Analyze dark (inactive) areas in EL images.
+    
+    CRITICAL: Uses ADAPTIVE thresholds based on module brightness.
+    DO NOT use fixed thresholds — they fail on dark/bright images.
+    """
+    
+    # Severity thresholds (percentage of cell area)
+    SEVERITY_THRESHOLDS = {
+        "none": 2.0,      # < 2%
+        "minor": 10.0,    # 2-10%
+        "moderate": 25.0,  # 10-25%
+        "severe": 50.0,    # 25-50%
+        "critical": float("inf"),  # > 50%
+    }
+    
+    def analyze(
+        self, 
+        cell_image: np.ndarray,
+        dark_mask: np.ndarray,
+    ) -> DarkResult:
+        """
+        Analyze dark area from mask and cell image.
+        
+        Args:
+            cell_image: Grayscale cell image (float32 or uint8)
+            dark_mask: Binary mask from model (1 = dark area)
+            
+        Returns:
+            DarkResult with area percentage and severity
+        """
+        h, w = cell_image.shape[:2]
+        total_pixels = h * w
+        
+        # Ensure float
+        if cell_image.dtype == np.uint8:
+            img_float = cell_image.astype(np.float32) / 255.0
+        else:
+            img_float = cell_image.astype(np.float32)
+        
+        dark_pixels = int(dark_mask.sum())
+        dark_pct = (dark_pixels / total_pixels) * 100.0
+        
+        # Compute intensities
+        if dark_pixels > 0:
+            mean_dark = float(img_float[dark_mask > 0].mean())
+        else:
+            mean_dark = 0.0
+        
+        normal_mask = dark_mask == 0
+        if normal_mask.sum() > 0:
+            mean_normal = float(img_float[normal_mask].mean())
+        else:
+            mean_normal = float(img_float.mean())
+        
+        # Severity
+        severity = self._classify_severity(dark_pct)
+        
+        return DarkResult(
+            dark_area_pct=dark_pct,
+            dark_area_px=dark_pixels,
+            total_area_px=total_pixels,
+            mean_intensity_dark=mean_dark,
+            mean_intensity_normal=mean_normal,
+            severity=severity,
+        )
+    
+    def detect_dark_adaptive(
+        self, cell_image: np.ndarray
+    ) -> np.ndarray:
+        """
+        Detect dark areas using ADAPTIVE thresholding.
+        
+        This is a FALLBACK when no model is available.
+        Uses module brightness to set threshold adaptively.
+        
+        Formula: dark_threshold = 0.6 × cell_mean_intensity
+        
+        Why 0.6? Validated empirically:
+        - Too low (0.3-0.4): misses partial dark regions
+        - Too high (0.8-0.9): false positives from grain boundaries
+        - 0.6: good balance across bright and dark modules
+        
+        Additionally filters out cell borders (they're always dark).
+        """
+        if cell_image.dtype == np.uint8:
+            img = cell_image.astype(np.float32) / 255.0
+        else:
+            img = cell_image.astype(np.float32)
+        
+        mean_intensity = img.mean()
+        
+        # Adaptive threshold
+        dark_threshold = 0.6 * mean_intensity
+        
+        dark_mask = (img < dark_threshold).astype(np.uint8)
+        
+        # Remove cell border artifacts (erode from edges)
+        border_margin = max(5, int(min(img.shape) * 0.03))
+        dark_mask[:border_margin, :] = 0
+        dark_mask[-border_margin:, :] = 0
+        dark_mask[:, :border_margin] = 0
+        dark_mask[:, -border_margin:] = 0
+        
+        # Clean small noise
+        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+        dark_mask = cv2.morphologyEx(dark_mask, cv2.MORPH_OPEN, kernel)
+        
+        return dark_mask
+    
+    def _classify_severity(self, dark_pct: float) -> str:
+        """Classify dark area severity."""
+        for severity, threshold in self.SEVERITY_THRESHOLDS.items():
+            if dark_pct < threshold:
+                return severity
+        return "critical"
+
+
+class DefectAnalyzer:
+    """
+    Combined defect analysis: cracks + dark areas + cross cracks.
+    
+    Produces a comprehensive CellAnalysisResult per cell.
+    """
+    
+    def __init__(self, px_per_mm: float = 5.0):
+        self.crack_analyzer = CrackAnalyzer(px_per_mm=px_per_mm)
+        self.dark_analyzer = DarkAreaAnalyzer()
+    
+    def analyze_cell(
+        self,
+        cell_image: np.ndarray,
+        mask: np.ndarray,
+        cell_id: int = 1,
+    ) -> CellAnalysisResult:
+        """
+        Full defect analysis for one cell.
+        
+        Args:
+            cell_image: Grayscale cell image
+            mask: Multi-class segmentation mask (5 classes)
+            cell_id: Cell identifier
+            
+        Returns:
+            CellAnalysisResult with all defect metrics
+        """
+        # Extract per-class masks
+        dark_mask = (mask == 1).astype(np.uint8)
+        crack_mask = (mask == 2).astype(np.uint8)
+        cross_mask = (mask == 3).astype(np.uint8)
+        
+        # Analyze each defect type
+        cracks = self.crack_analyzer.analyze_cracks(crack_mask, "crack")
+        cross_cracks = self.crack_analyzer.analyze_cracks(cross_mask, "cross")
+        dark = self.dark_analyzer.analyze(cell_image, dark_mask)
+        
+        # Filter false positives from cracks
+        real_cracks = [c for c in cracks if not c.is_false_positive]
+        real_cross = [c for c in cross_cracks if not c.is_false_positive]
+        
+        # Compute aggregate metrics
+        total_crack_length = sum(c.length_mm for c in real_cracks)
+        total_cross_length = sum(c.length_mm for c in real_cross)
+        
+        # Max severity across all cracks
+        all_cracks = real_cracks + real_cross
+        if all_cracks:
+            severity_order = ["minor", "moderate", "severe", "critical"]
+            max_severity = max(
+                all_cracks, 
+                key=lambda c: severity_order.index(c.severity)
+            ).severity
+        else:
+            max_severity = "none"
+        
+        # Compute defect score (0-100)
+        defect_score = self._compute_defect_score(
+            total_crack_length + total_cross_length,
+            dark.dark_area_pct,
+            len(real_cracks) + len(real_cross),
+        )
+        
+        return CellAnalysisResult(
+            cell_id=cell_id,
+            cracks=cracks,
+            dark=dark,
+            cross_cracks=cross_cracks,
+            total_crack_length_mm=total_crack_length + total_cross_length,
+            num_cracks=len(real_cracks),
+            num_cross_cracks=len(real_cross),
+            max_crack_severity=max_severity,
+            defect_score=defect_score,
+        )
+    
+    def _compute_defect_score(
+        self,
+        total_crack_length_mm: float,
+        dark_area_pct: float,
+        num_cracks: int,
+    ) -> float:
+        """
+        Compute composite defect score (0-100).
+        
+        Weighted combination:
+        - 40% crack severity (length-based)
+        - 40% dark area percentage
+        - 20% crack count
+        
+        Score >= 50 typically indicates FAIL condition.
+        """
+        # Crack score: normalize length to 0-100
+        crack_score = min(total_crack_length_mm / 50.0 * 100.0, 100.0)
+        
+        # Dark score: percentage maps directly (capped at 100)
+        dark_score = min(dark_area_pct * 2.0, 100.0)  # 50% dark = score 100
+        
+        # Count score
+        count_score = min(num_cracks * 15.0, 100.0)  # ~7 cracks = score 100
+        
+        composite = (
+            0.4 * crack_score +
+            0.4 * dark_score +
+            0.2 * count_score
+        )
+        
+        return min(composite, 100.0)