Rename ap.py to app.py

ap.py

@@ -1,1199 +0,0 @@
-# app.py 
-# Layer 1: Color-Metric Logic (LAB Space) → Fail on dark metallic grey
-# Layer 2: ResNet Template Matching → Pass only if matches golden template
-
-import gradio as gr
-import cv2
-import torch
-import torch.nn as nn
-from torchvision import models, transforms
-from PIL import Image
-import numpy as np
-import pickle
-import os
-from dataclasses import dataclass
-from typing import Tuple, List, Optional
-
-# ═════════════════════════════════════════════════════════════════════════════
-# Configuration & Data Classes
-# ═════════════════════════════════════════════════════════════════════════════
-
-@dataclass
-class ColorThresholds:
-    """LAB color space thresholds for saddle surface classification."""
-    # Light Grey (PASS): High L, low saturation
-    light_grey_L_min: float = 110.0      # Lightness threshold
-    light_grey_L_max: float = 200.0
-    light_grey_saturation_max: float = 30.0  # Low color saturation
-    
-    # Dark Metallic Grey (FAIL): Lower L, slightly higher saturation
-    dark_metallic_L_max: float = 105.0   # Darker = defect
-    
-    # Detection confidence
-    defect_pixel_ratio_threshold: float = 0.15  # 15% dark pixels = fail
-
-
-@dataclass
-class DetectionResult:
-    """Result from multi-stage detection."""
-    stage: int  # 1 = color check, 2 = template match
-    status: str  # "PASS" or "FAIL"
-    reason: str
-    confidence: float
-    saddle_colors: List[str]  # Color of each saddle
-    template_match_score: Optional[float] = None
-
-
-# ═════════════════════════════════════════════════════════════════════════════
-# Multi-Stage Detector
-# ═════════════════════════════════════════════════════════════════════════════
-
-class MultiStageEngineDetector:
-    def __init__(
-        self,
-        clahe_clip_limit: float = 2.0,
-        clahe_tile_grid: tuple = (8, 8),
-        color_thresholds: Optional[ColorThresholds] = None,
-    ):
-        # ── ResNet-50 backbone ─────────────────────────────────────────
-        self.model = models.resnet50(weights='IMAGENET1K_V1')
-        self.model = nn.Sequential(*list(self.model.children())[:-1])
-        self.model.eval()
-
-        # ── CLAHE ──────────────────────────────────────────────────────
-        self.clahe = cv2.createCLAHE(
-            clipLimit=clahe_clip_limit,
-            tileGridSize=clahe_tile_grid,
-        )
-
-        # ── ResNet transform ───────────────────────────────────────────
-        self.transform = transforms.Compose([
-            transforms.Resize((224, 224)),
-            transforms.ToTensor(),
-            transforms.Normalize(
-                mean=[0.485, 0.456, 0.406],
-                std=[0.229, 0.224, 0.225],
-            )
-        ])
-
-        # ── Color thresholds ───────────────────────────────────────────
-        self.color_thresholds = color_thresholds or ColorThresholds()
-
-        # ── Templates (golden samples) ─────────────────────────────────
-        self.templates = {}
-        self.load_templates()
-
-    # ═════════════════════════════════════════════════════════════════════════
-    # LAYER 1: Color-Metric Logic (LAB Space)
-    # ═════════════════════════════════════════════════════════════════════════
-
-    def apply_clahe(self, image: np.ndarray) -> np.ndarray:
-        """Apply CLAHE preprocessing."""
-        bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
-        lab = cv2.cvtColor(bgr, cv2.COLOR_BGR2LAB)
-        l, a, b = cv2.split(lab)
-        l_enh = self.clahe.apply(l)
-        lab_enh = cv2.merge([l_enh, a, b])
-        bgr_enh = cv2.cvtColor(lab_enh, cv2.COLOR_LAB2BGR)
-        return cv2.cvtColor(bgr_enh, cv2.COLOR_BGR2RGB)
-
-    def analyze_full_image_color(self, image: np.ndarray) -> Tuple[str, float, dict]:
-        """
-        Analyze entire image for color classification in LAB space.
-        No region detection - checks overall saddle surface color.
-        
-        Returns:
-            color_class: "LIGHT_GREY" or "DARK_METALLIC"
-            confidence: 0-1 confidence score
-            stats: Color statistics
-        """
-        # Convert to LAB
-        bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
-        lab = cv2.cvtColor(bgr, cv2.COLOR_BGR2LAB)
-        l_channel, a_channel, b_channel = cv2.split(lab)
-        
-        # Focus on metallic regions (saddle surfaces)
-        # Create mask for bright metallic areas (saddles are lighter than engine block)
-        _, saddle_mask = cv2.threshold(l_channel, 80, 255, cv2.THRESH_BINARY)
-        
-        # Morphological operations to clean up
-        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (15, 15))
-        saddle_mask = cv2.morphologyEx(saddle_mask, cv2.MORPH_CLOSE, kernel)
-        saddle_mask = cv2.morphologyEx(saddle_mask, cv2.MORPH_OPEN, kernel)
-        
-        # If no saddle regions detected, analyze full image
-        if np.sum(saddle_mask) < 0.05 * saddle_mask.size:
-            saddle_mask = np.ones_like(l_channel) * 255
-        
-        # Calculate mean L (lightness) in saddle regions
-        l_mean = cv2.mean(l_channel, mask=saddle_mask)[0]
-        a_mean = cv2.mean(a_channel, mask=saddle_mask)[0]
-        b_mean = cv2.mean(b_channel, mask=saddle_mask)[0]
-        
-        # Calculate saturation (distance from neutral gray in a-b plane)
-        saturation = np.sqrt((a_mean - 128)**2 + (b_mean - 128)**2)
-        
-        # Calculate percentage of dark pixels in saddle regions
-        dark_pixels = np.sum((l_channel < self.color_thresholds.dark_metallic_L_max) & (saddle_mask > 0))
-        total_pixels = np.sum(saddle_mask > 0)
-        dark_ratio = dark_pixels / total_pixels if total_pixels > 0 else 0
-        
-        # Calculate histogram distribution
-        hist = cv2.calcHist([l_channel], [0], saddle_mask, [256], [0, 256])
-        hist = hist.flatten() / hist.sum()
-        
-        # Check for bimodal distribution (mix of light and dark)
-        # Dark peak (0-105) vs Light peak (110-200)
-        dark_peak_mass = hist[0:105].sum()
-        light_peak_mass = hist[110:200].sum()
-        
-        stats = {
-            'l_mean': l_mean,
-            'a_mean': a_mean,
-            'b_mean': b_mean,
-            'saturation': saturation,
-            'dark_ratio': dark_ratio,
-            'dark_peak_mass': dark_peak_mass,
-            'light_peak_mass': light_peak_mass,
-            'saddle_pixels': total_pixels
-        }
-        
-        # FAIL conditions (any triggers fail)
-        # Condition 1: Mean lightness too low
-        if l_mean < self.color_thresholds.dark_metallic_L_max:
-            confidence = 1.0 - (l_mean / self.color_thresholds.dark_metallic_L_max)
-            return "DARK_METALLIC", confidence, stats
-        
-        # Condition 2: High ratio of dark pixels
-        if dark_ratio > self.color_thresholds.defect_pixel_ratio_threshold:
-            return "DARK_METALLIC", min(dark_ratio * 2, 1.0), stats
-        
-        # Condition 3: Significant dark peak in histogram (corrupted surfaces)
-        if dark_peak_mass > 0.20:  # More than 20% of pixels in dark range
-            return "DARK_METALLIC", dark_peak_mass, stats
-        
-        # PASS condition
-        if (self.color_thresholds.light_grey_L_min <= l_mean <= self.color_thresholds.light_grey_L_max and
-            saturation < self.color_thresholds.light_grey_saturation_max):
-            confidence = min(1.0, light_peak_mass + 0.5)  # Higher if concentrated in light range
-            return "LIGHT_GREY", confidence, stats
-        
-        # Ambiguous - default to LIGHT_GREY with low confidence
-        return "LIGHT_GREY", 0.3, stats
-
-    def layer1_color_check(self, image: np.ndarray) -> DetectionResult:
-        """
-        Layer 1: Full-image color-metric logic.
-        FAIL if image contains dark metallic grey surfaces.
-        PASS if predominantly light grey.
-        """
-        # Normalize input
-        if isinstance(image, Image.Image):
-            image = np.array(image.convert("RGB"))
-        elif image.dtype != np.uint8:
-            image = np.clip(image, 0, 255).astype(np.uint8)
-        
-        # Apply CLAHE preprocessing
-        image_enhanced = self.apply_clahe(image)
-        
-        # Analyze full image color
-        color_class, confidence, stats = self.analyze_full_image_color(image_enhanced)
-        
-        # Result
-        if color_class == "DARK_METALLIC":
-            reason_parts = []
-            if stats['l_mean'] < self.color_thresholds.dark_metallic_L_max:
-                reason_parts.append(f"Low lightness (L={stats['l_mean']:.1f})")
-            if stats['dark_ratio'] > self.color_thresholds.defect_pixel_ratio_threshold:
-                reason_parts.append(f"High dark pixel ratio ({stats['dark_ratio']:.1%})")
-            if stats['dark_peak_mass'] > 0.20:
-                reason_parts.append(f"Dark surface detected ({stats['dark_peak_mass']:.1%})")
-            
-            reason = "Dark metallic grey detected: " + ", ".join(reason_parts)
-            
-            return DetectionResult(
-                stage=1,
-                status="FAIL",
-                reason=reason,
-                confidence=confidence,
-                saddle_colors=[color_class]
-            )
-        else:
-            return DetectionResult(
-                stage=1,
-                status="PASS",
-                reason=f"Light grey surface detected (L={stats['l_mean']:.1f}, light_mass={stats['light_peak_mass']:.1%})",
-                confidence=confidence,
-                saddle_colors=[color_class]
-            )
-
-    # ═════════════════════════════════════════════════════════════════════════
-    # LAYER 2: ResNet Template Matching
-    # ═════════════════════════════════════════════════════════════════════════
-
-    def extract_features(self, image) -> np.ndarray:
-        """Extract ResNet-50 features with CLAHE preprocessing."""
-        # Normalize input
-        if isinstance(image, Image.Image):
-            image = np.array(image.convert("RGB"))
-        elif isinstance(image, np.ndarray):
-            if image.dtype != np.uint8:
-                image = np.clip(image, 0, 255).astype(np.uint8)
-            if image.ndim == 2:
-                image = cv2.cvtColor(image, cv2.COLOR_GRAY2RGB)
-
-        # CLAHE + blur
-        image = self.apply_clahe(image)
-        image = cv2.GaussianBlur(image, (3, 3), 0)
-
-        # ResNet forward pass
-        img_tensor = self.transform(Image.fromarray(image)).unsqueeze(0)
-        with torch.no_grad():
-            features = self.model(img_tensor).squeeze().numpy()
-
-        return features
-
-    @staticmethod
-    def cosine_similarity(f1: np.ndarray, f2: np.ndarray) -> float:
-        return float(np.dot(f1, f2) / (np.linalg.norm(f1) * np.linalg.norm(f2) + 1e-8))
-
-    def layer2_template_match(self, image, threshold: float = 0.70) -> DetectionResult:
-        """
-        Layer 2: ResNet template matching against golden samples.
-        PASS only if similarity >= threshold.
-        FAIL otherwise (even if close).
-        """
-        if not self.templates:
-            return DetectionResult(
-                stage=2,
-                status="FAIL",
-                reason="No golden templates available. Please add templates first.",
-                confidence=1.0,
-                saddle_colors=[],
-                template_match_score=None
-            )
-
-        # Extract features
-        query_feat = self.extract_features(image)
-        
-        # Find best match
-        results = [(name, self.cosine_similarity(query_feat, feat))
-                   for name, feat in self.templates.items()]
-        results.sort(key=lambda x: x[1], reverse=True)
-        
-        best_name, best_score = results[0]
-        
-        # Strict threshold (no false negatives)
-        if best_score >= threshold:
-            return DetectionResult(
-                stage=2,
-                status="PASS",
-                reason=f"Matched golden template '{best_name}' with {best_score:.2%} confidence",
-                confidence=best_score,
-                saddle_colors=[],
-                template_match_score=best_score
-            )
-        else:
-            return DetectionResult(
-                stage=2,
-                status="FAIL",
-                reason=f"Best match '{best_name}' ({best_score:.2%}) below threshold ({threshold:.0%}). Surface defects detected.",
-                confidence=1.0 - best_score,  # High confidence in failure
-                saddle_colors=[],
-                template_match_score=best_score
-            )
-
-    # ═════════════════════════════════════════════════════════════════════════
-    # Multi-Stage Detection Pipeline
-    # ═════════════════════════════════════════════════════════════════════════
-
-    def detect(self, image, template_threshold: float = 0.70) -> Tuple[DetectionResult, str]:
-        """
-        Full two-stage detection pipeline.
-        
-        Returns:
-            result: DetectionResult object
-            summary: Formatted string for UI
-        """
-        # STAGE 1: Color check
-        layer1_result = self.layer1_color_check(image)
-        
-        if layer1_result.status == "FAIL":
-            # Immediate fail on color metric
-            summary = f"""
-## 🔴 DEFECTED PIECE
-
-**Stage 1: Color-Metric Analysis** — FAILED ❌
-
-**Reason**: {layer1_result.reason}
-
-**Analysis Method**: Full-image LAB color space analysis
-
-**Decision**: Part rejected at Stage 1. No template matching performed.
-
----
-**Detection Confidence**: {layer1_result.confidence:.1%}
-"""
-            return layer1_result, summary
-        
-        # STAGE 2: Template matching (only if color check passed)
-        layer2_result = self.layer2_template_match(image, template_threshold)
-        
-        if layer2_result.status == "PASS":
-            summary = f"""
-## ✅ PERFECT PIECE
-
-**Stage 1: Color-Metric Analysis** — PASSED ✓
-- {layer1_result.reason}
-
-**Stage 2: Template Matching** — PASSED ✓
-- {layer2_result.reason}
-
-**Final Decision**: ✅ **APPROVED**
-
----
-**Template Match Score**: {layer2_result.template_match_score:.2%}  
-**Overall Confidence**: {layer2_result.confidence:.1%}
-"""
-        else:
-            summary = f"""
-## 🟡 DEFECTED PIECE
-
-**Stage 1: Color-Metric Analysis** — PASSED ✓
-- {layer1_result.reason}
-
-**Stage 2: Template Matching** — FAILED ❌
-- {layer2_result.reason}
-
-**Final Decision**: ❌ **REJECTED**
-
----
-**Best Template Match**: {layer2_result.template_match_score:.2%}  
-**Threshold**: {template_threshold:.0%}  
-**Confidence in Defect**: {layer2_result.confidence:.1%}
-"""
-        
-        return layer2_result, summary
-
-    # ═════════════════════════════════════════════════════════════════════════
-    # Template Management
-    # ═════════════════════════════════════════════════════════════════════════
-
-    def save_template(self, image, part_name: str) -> str:
-        if image is None or not part_name.strip():
-            return "⚠️ Please provide both an image and a part name."
-        
-        # Verify it's a golden sample (all light grey)
-        color_result = self.layer1_color_check(image)
-        if color_result.status == "FAIL":
-            return f"❌ Cannot save as golden template: {color_result.reason}"
-        
-        # Extract and save features
-        self.templates[part_name.strip()] = self.extract_features(image)
-        with open('templates.pkl', 'wb') as f:
-            pickle.dump(self.templates, f)
-        
-        return f"✅ Golden template '{part_name.strip()}' saved! ({len(self.templates)} total)"
-
-    def load_templates(self):
-        if os.path.exists('templates.pkl'):
-            try:
-                with open('templates.pkl', 'rb') as f:
-                    self.templates = pickle.load(f)
-            except Exception:
-                self.templates = {}
-
-    def delete_template(self, part_name: str) -> str:
-        if part_name in self.templates:
-            del self.templates[part_name]
-            with open('templates.pkl', 'wb') as f:
-                pickle.dump(self.templates, f)
-            return f"🗑️ Template '{part_name}' deleted."
-        return f"⚠️ Template '{part_name}' not found."
-
-
-# ═════════════════════════════════════════════════════════════════════════════
-# Live Edge Preview
-# ═════════════════════════════════════════════════════════════════════════════
-
-def live_edge_preview(frame: np.ndarray) -> np.ndarray:
-    """Real-time CLAHE + Canny edge overlay."""
-    if frame is None:
-        return frame
-    if frame.dtype != np.uint8:
-        frame = np.clip(frame, 0, 255).astype(np.uint8)
-
-    gray = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
-    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
-    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
-    enh = clahe.apply(blurred)
-    edges = cv2.Canny(enh, 30, 120)
-
-    dim = (frame * 0.35).astype(np.uint8)
-    dim[edges > 0] = [0, 220, 220]
-    return dim
-
-
-# ═════════════════════════════════════════════════════════════════════════════
-# Gradio Callbacks
-# ═════════════════════════════════════════════════════════════════════════════
-
-detector = MultiStageEngineDetector()
-
-
-def cb_detect(image, threshold):
-    """Run two-stage detection."""
-    if image is None:
-        return "⚠️ Please provide an image.", None
-    
-    result, summary = detector.detect(image, template_threshold=threshold)
-    label = result.status  # "PASS" or "FAIL"
-    
-    return summary, label
-
-
-def cb_add_template(image, part_name):
-    return detector.save_template(image, part_name)
-
-
-def cb_list_templates():
-    if not detector.templates:
-        return "No golden templates saved yet."
-    lines = [f"- **{n}**" for n in sorted(detector.templates)]
-    return f"**{len(detector.templates)} golden template(s):**\n" + "\n".join(lines)
-
-
-def cb_delete_template(part_name):
-    msg = detector.delete_template(part_name)
-    return msg, cb_list_templates()
-
-
-def cb_capture(frame):
-    return frame
-
-
-def cb_list_templates_raw():
-    """Return comma-separated list of template names for API."""
-    return ",".join(sorted(detector.templates.keys()))
-
-
-# ═════════════════════════════════════════════════════════════════════════════
-# Gradio UI
-# ═════════════════════════════════════════════════════════════════════════════
-
-CSS = """
-@import url('https://fonts.googleapis.com/css2?family=Share+Tech+Mono&family=Barlow:wght@300;400;600;700&display=swap');
-:root {
-    --bg: #0d0f12; --surface: #151820; --border: #2a3045;
-    --accent: #00e5ff; --accent2: #ff6b35; --success: #00e676; --danger: #ff1744;
-    --text: #d0d8e8; --muted: #5a6480;
-    --mono: 'Share Tech Mono', monospace; --sans: 'Barlow', sans-serif;
-}
-body, .gradio-container { background: var(--bg) !important; font-family: var(--sans) !important; color: var(--text) !important; }
-.eng-title { font-family: var(--mono); font-size: 1.6rem; color: var(--accent); letter-spacing: 0.1em; text-transform: uppercase; }
-button.primary { background: var(--accent) !important; color: #000 !important; font-family: var(--mono) !important; font-weight: 700 !important; }
-.gr-markdown { font-family: var(--mono) !important; font-size: 0.8rem !important; background: var(--surface) !important; border: 1px solid var(--border) !important; padding: 14px !important; }
-"""
-
-with gr.Blocks(title="MULTI-STAGE ENGINE DETECTOR") as demo:
-    
-    gr.HTML("""
-    <div class='eng-title' style='text-align:center;padding:20px;'>
-        ⚙ MULTI-STAGE ENGINE SADDLE DETECTOR
-    </div>
-    <div style='text-align:center;color:#5a6480;font-family:Share Tech Mono,monospace;font-size:0.75rem;margin-bottom:20px;'>
-        Layer 1: Color-Metric Logic (LAB) • Layer 2: ResNet Template Matching • Zero False Negatives
-    </div>
-    """)
-
-    with gr.Tabs():
-        # Tab 1: Detect
-        with gr.TabItem("🔍 DETECT"):
-            with gr.Row():
-                with gr.Column():
-                    detect_input = gr.Image(sources=["upload", "webcam"], type="numpy", height=350)
-                    threshold_slider = gr.Slider(0.60, 0.90, 0.70, 0.01, label="Template Match Threshold")
-                    detect_btn = gr.Button("▶ RUN 2-STAGE DETECTION", variant="primary")
-                with gr.Column():
-                    detect_output = gr.Markdown(label="Detection Report")
-                    match_label = gr.Label(label="Final Decision")
-
-            detect_btn.click(
-                fn=cb_detect, 
-                inputs=[detect_input, threshold_slider], 
-                outputs=[detect_output, match_label],
-                api_name="detect_part"
-            )
-
-        # Tab 2: Live Camera
-        with gr.TabItem("📷 LIVE CAMERA"):
-            with gr.Row():
-                with gr.Column():
-                    live_input = gr.Image(sources=["webcam"], streaming=True, type="numpy", height=280)
-                    live_output = gr.Image(label="Edge Preview", height=280)
-                with gr.Column():
-                    captured_frame = gr.Image(type="numpy", height=280, interactive=False)
-                    with gr.Row():
-                        capture_btn = gr.Button("📸 CAPTURE", variant="primary")
-                        detect_live_btn = gr.Button("🔍 DETECT", variant="secondary")
-                    live_threshold = gr.Slider(0.60, 0.90, 0.70, 0.01, label="Threshold")
-                    live_result = gr.Markdown()
-                    live_label = gr.Label()
-
-            live_input.stream(fn=live_edge_preview, inputs=[live_input], outputs=[live_output])
-            capture_btn.click(fn=cb_capture, inputs=[live_input], outputs=[captured_frame])
-            detect_live_btn.click(fn=cb_detect, inputs=[captured_frame, live_threshold], outputs=[live_result, live_label])
-
-        # Tab 3: Add Golden Template
-        with gr.TabItem("➕ GOLDEN TEMPLATE"):
-            with gr.Row():
-                with gr.Column():
-                    template_input = gr.Image(sources=["upload", "webcam"], type="numpy", height=300)
-                    part_name_input = gr.Textbox(label="Template Name", placeholder="e.g. perfect_sample_1")
-                    add_btn = gr.Button("💾 SAVE AS GOLDEN TEMPLATE", variant="primary")
-                with gr.Column():
-                    add_output = gr.Markdown()
-                    gr.HTML("<div style='color:#5a6480;padding:10px;'>⚠️ Only save perfect pieces as golden templates. System verifies all saddles are light grey before saving.</div>")
-
-            add_btn.click(
-                fn=cb_add_template, 
-                inputs=[template_input, part_name_input], 
-                outputs=[add_output],
-                api_name="add_template"
-            )
-
-        # Tab 4: Manage Templates
-        with gr.TabItem("📋 TEMPLATES"):
-            with gr.Row():
-                with gr.Column():
-                    template_list = gr.Markdown()
-                    refresh_btn = gr.Button("↻ REFRESH", variant="secondary")
-                with gr.Column():
-                    delete_name = gr.Textbox(label="Delete Template", placeholder="exact name")
-                    delete_btn = gr.Button("🗑 DELETE", variant="secondary")
-                    delete_status = gr.Markdown()
-
-            refresh_btn.click(fn=cb_list_templates, outputs=[template_list], api_name="list_templates")
-            delete_btn.click(fn=cb_delete_template, inputs=[delete_name], outputs=[delete_status, template_list])
-            
-            # Hidden raw list endpoint for API access
-            raw_list_btn = gr.Button("RAW LIST", visible=False)
-            raw_list_btn.click(fn=cb_list_templates_raw, outputs=[], api_name="list_templates_raw")
-            
-            demo.load(fn=cb_list_templates, outputs=[template_list])
-
-
-if __name__ == "__main__":
-    demo.launch(css=CSS, server_name="0.0.0.0", server_port=7860)
-    
-
-
-
-
-# # app.py 
-# import gradio as gr
-# import torch
-# import torch.nn as nn
-# from torchvision import models, transforms
-# from PIL import Image
-# import numpy as np
-# import pickle
-# import os
-# import cv2
-# from typing import List, Dict, Tuple, Optional
-# from dataclasses import dataclass
-# from pathlib import Path
-
-# @dataclass
-# class TemplateData:
-#     """Store multiple feature vectors for robust matching"""
-#     features_list: List[np.ndarray]
-#     part_name: str
-#     num_images: int
-#     mean_features: Optional[np.ndarray] = None
-    
-#     def __post_init__(self):
-#         """Calculate mean features after initialization"""
-#         if self.features_list and self.mean_features is None:
-#             self.mean_features = np.mean(self.features_list, axis=0)
-
-
-# class EnginePartDetector:
-    
-#     def __init__(
-#         self,
-#         clahe_clip_limit: float = 9.5,
-#         clahe_tile_grid: Tuple[int, int] = (8, 8),
-#         device: str = 'cuda' if torch.cuda.is_available() else 'cpu',
-#         templates_path: str = 'templates.pkl'
-#     ):
-#         self.device = device
-#         self.templates_path = templates_path
-        
-#         # ── ResNet-50 backbone (feature extractor) ───────────────────────
-#         print(f"Loading ResNet50 on {self.device}...")
-#         self.model = models.resnet50(weights='IMAGENET1K_V1')
-#         self.model = nn.Sequential(*list(self.model.children())[:-1])
-#         self.model.to(self.device)
-#         self.model.eval()
-        
-#         # ── CLAHE for contrast enhancement ────────────────────────────────
-#         self.clahe = cv2.createCLAHE(
-#             clipLimit=clahe_clip_limit,
-#             tileGridSize=clahe_tile_grid,
-#         )
-        
-#         # ── Image augmentation for training robustness ────────────────────
-#         self.base_transform = transforms.Compose([
-#             transforms.Resize((224, 224)),
-#             transforms.ToTensor(),
-#             transforms.Normalize(
-#                 mean=[0.485, 0.456, 0.406],
-#                 std=[0.229, 0.224, 0.225],
-#             )
-#         ])
-        
-#         # Augmentation for multi-image templates (simulate lighting variations)
-#         self.augment_transforms = [
-#             # Original
-#             self.base_transform,
-#             # Brightness variations
-#             transforms.Compose([
-#                 transforms.Resize((224, 224)),
-#                 transforms.ColorJitter(brightness=0.3),
-#                 transforms.ToTensor(),
-#                 transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-#             ]),
-#             transforms.Compose([
-#                 transforms.Resize((224, 224)),
-#                 transforms.ColorJitter(brightness=0.5),
-#                 transforms.ToTensor(),
-#                 transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-#             ]),
-#             # Contrast variations
-#             transforms.Compose([
-#                 transforms.Resize((224, 224)),
-#                 transforms.ColorJitter(contrast=0.3),
-#                 transforms.ToTensor(),
-#                 transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-#             ]),
-#         ]
-        
-#         self.templates: Dict[str, TemplateData] = {}
-#         self.load_templates()
-#         print(f"Loaded {len(self.templates)} templates from disk")
-    
-#     # ── Preprocessing Pipeline ────────────────────────────────────────────
-    
-#     def apply_clahe(self, image: np.ndarray) -> np.ndarray:
-#         """Apply CLAHE to enhance contrast and recover shadow details"""
-#         bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
-#         lab = cv2.cvtColor(bgr, cv2.COLOR_BGR2LAB)
-        
-#         l_channel, a_channel, b_channel = cv2.split(lab)
-#         l_enhanced = self.clahe.apply(l_channel)
-        
-#         lab_enhanced = cv2.merge([l_enhanced, a_channel, b_channel])
-#         bgr_enhanced = cv2.cvtColor(lab_enhanced, cv2.COLOR_LAB2BGR)
-#         rgb_enhanced = cv2.cvtColor(bgr_enhanced, cv2.COLOR_BGR2RGB)
-        
-#         return rgb_enhanced
-    
-#     def preprocess_image(self, image: np.ndarray) -> np.ndarray:
-#         """Apply CLAHE and mild denoising"""
-#         # Ensure uint8 RGB
-#         if image.dtype != np.uint8:
-#             image = image.astype(np.uint8)
-        
-#         # CLAHE for contrast enhancement
-#         image = self.apply_clahe(image)
-        
-#         # Mild Gaussian blur to reduce metallic sheen noise
-#         image = cv2.GaussianBlur(image, (3, 3), 0)
-        
-#         return image
-    
-#     # ── Feature Extraction ────────────────────────────────────────────────
-    
-#     @torch.no_grad()
-#     def extract_features(self, image, use_augmentation: bool = False) -> np.ndarray:
-       
-#         # Normalize input to numpy uint8 RGB
-#         if isinstance(image, Image.Image):
-#             image = np.array(image.convert("RGB"))
-#         elif isinstance(image, np.ndarray) and image.dtype != np.uint8:
-#             image = image.astype(np.uint8)
-        
-#         # Apply preprocessing
-#         image = self.preprocess_image(image)
-#         image_pil = Image.fromarray(image)
-        
-#         if use_augmentation:
-#             # Extract features with multiple augmentations
-#             features_list = []
-#             for transform in self.augment_transforms:
-#                 img_tensor = transform(image_pil).unsqueeze(0).to(self.device)
-#                 features = self.model(img_tensor).squeeze().cpu().numpy()
-#                 features_list.append(features)
-#             return np.array(features_list)
-#         else:
-#             # Single feature extraction
-#             img_tensor = self.base_transform(image_pil).unsqueeze(0).to(self.device)
-#             features = self.model(img_tensor).squeeze().cpu().numpy()
-#             return features
-    
-#     @torch.no_grad()
-#     def extract_batch_features(self, images: List[np.ndarray]) -> List[np.ndarray]:
-#         features_list = []
-        
-#         # Preprocess all images
-#         preprocessed = []
-#         for img in images:
-#             if isinstance(img, Image.Image):
-#                 img = np.array(img.convert("RGB"))
-#             img = self.preprocess_image(img)
-#             preprocessed.append(Image.fromarray(img))
-        
-#         # Batch processing
-#         batch_size = 8
-#         for i in range(0, len(preprocessed), batch_size):
-#             batch = preprocessed[i:i+batch_size]
-#             batch_tensors = torch.stack([
-#                 self.base_transform(img) for img in batch
-#             ]).to(self.device)
-            
-#             batch_features = self.model(batch_tensors).squeeze().cpu().numpy()
-#             if len(batch) == 1:
-#                 batch_features = batch_features.reshape(1, -1)
-            
-#             features_list.extend(batch_features)
-        
-#         return features_list
-    
-#     # ── Similarity Matching ───────────────────────────────────────────────
-    
-#     @staticmethod
-#     def cosine_similarity(feat1: np.ndarray, feat2: np.ndarray) -> float:
-#         return np.dot(feat1, feat2) / (np.linalg.norm(feat1) * np.linalg.norm(feat2) + 1e-8)
-    
-#     def compute_multi_similarity(
-#         self, 
-#         query_features: np.ndarray, 
-#         template_data: TemplateData,
-#         method: str = 'mean'
-#     ) -> float:
-       
-#         similarities = [
-#             self.cosine_similarity(query_features, template_feat)
-#             for template_feat in template_data.features_list
-#         ]
-        
-#         if method == 'mean':
-#             return np.mean(similarities)
-#         elif method == 'max':
-#             return np.max(similarities)
-#         elif method == 'median':
-#             return np.median(similarities)
-#         else:
-#             return np.mean(similarities)
-    
-#     # ── Template Management ───────────────────────────────────────────────
-    
-#     def save_template(
-#         self, 
-#         images: List[np.ndarray], 
-#         part_name: str,
-#         use_augmentation: bool = True
-#     ) -> str:
-        
-#         if not images or not part_name:
-#             return "❌ Please provide both images and part name"
-        
-#         if not part_name.strip():
-#             return "❌ Part name cannot be empty"
-        
-#         print(f"Processing {len(images)} images for '{part_name}'...")
-        
-#         # Extract features from all images
-#         all_features = []
-        
-#         for idx, image in enumerate(images):
-#             if use_augmentation:
-#                 # Extract features with augmentation (multiple versions per image)
-#                 aug_features = self.extract_features(image, use_augmentation=True)
-#                 all_features.extend(aug_features)
-#                 print(f"  Image {idx+1}: Extracted {len(aug_features)} augmented features")
-#             else:
-#                 # Single feature per image
-#                 features = self.extract_features(image, use_augmentation=False)
-#                 all_features.append(features)
-#                 print(f"  Image {idx+1}: Extracted 1 feature")
-        
-#         # Create template
-#         template_data = TemplateData(
-#             features_list=all_features,
-#             part_name=part_name,
-#             num_images=len(images)
-#         )
-        
-#         self.templates[part_name] = template_data
-        
-#         # Save to disk
-#         try:
-#             with open(self.templates_path, 'wb') as f:
-#                 pickle.dump(self.templates, f)
-            
-#             total_features = len(all_features)
-#             return (f"✅ Template '{part_name}' saved successfully!\n"
-#                    f"📊 {len(images)} images → {total_features} feature vectors\n"
-#                    f"💪 Strong reference built for robust matching")
-#         except Exception as e:
-#             return f"❌ Error saving template: {str(e)}"
-    
-#     def load_templates(self):
-#         """Load templates from disk"""
-#         if os.path.exists(self.templates_path):
-#             try:
-#                 with open(self.templates_path, 'rb') as f:
-#                     self.templates = pickle.load(f)
-#                 print(f"Loaded {len(self.templates)} templates")
-#             except Exception as e:
-#                 print(f"Error loading templates: {e}")
-#                 self.templates = {}
-#         else:
-#             self.templates = {}
-    
-#     def delete_template(self, part_name: str) -> str:
-#         """Delete a template"""
-#         if part_name in self.templates:
-#             del self.templates[part_name]
-#             with open(self.templates_path, 'wb') as f:
-#                 pickle.dump(self.templates, f)
-#             return f"✅ Template '{part_name}' deleted"
-#         return f"❌ Template '{part_name}' not found"
-    
-#     # ── Part Detection ────────────────────────────────────────────────────
-    
-#     def match_part(
-#         self, 
-#         image: np.ndarray, 
-#         threshold: float = 0.7,
-#         similarity_method: str = 'mean'
-#     ) -> Tuple[str, Optional[str]]:
-        
-#         if image is None:
-#             return "❌ Please provide an image", None
-        
-#         if not self.templates:
-#             return "⚠️ No templates available. Please add templates first.", None
-        
-#         # Extract query features
-#         query_features = self.extract_features(image, use_augmentation=False)
-        
-#         # Compute similarities for all templates
-#         results = []
-#         for part_name, template_data in self.templates.items():
-#             similarity = self.compute_multi_similarity(
-#                 query_features, 
-#                 template_data, 
-#                 method=similarity_method
-#             )
-#             results.append((part_name, similarity, template_data.num_images))
-        
-#         # Sort by similarity
-#         results.sort(key=lambda x: x[1], reverse=True)
-        
-#         # Format output
-#         best_match = results[0]
-#         output_text = f"🔍 **Best Match**: {best_match[0]}\n"
-#         output_text += f"📊 **Confidence**: {best_match[1]:.2%}\n"
-#         output_text += f"📸 **Template Size**: {best_match[2]} reference images\n\n"
-        
-#         if best_match[1] >= threshold:
-#             output_text += "✅ **Status**: MATCHED\n\n"
-#             matched_label = best_match[0]
-#         else:
-#             output_text += "❌ **Status**: NO MATCH (below threshold)\n\n"
-#             matched_label = None
-        
-#         output_text += "**All Results:**\n"
-#         for part, sim, num_imgs in results:
-#             output_text += f"- {part}: {sim:.2%} ({num_imgs} ref images)\n"
-        
-#         return output_text, matched_label
-
-
-# # ── Global Detector Instance ──────────────────────────────────────────────
-
-# detector = EnginePartDetector()
-
-
-# # ── Gradio Interface Functions ────────────────────────────────────────────
-
-# def add_template(images, part_name, use_augmentation):
-#     """Add template from multiple images"""
-#     if images is None or len(images) == 0:
-#         return "❌ Please upload at least one image"
-    
-#     # Convert images to list if single image
-#     if not isinstance(images, list):
-#         images = [images]
-    
-#     return detector.save_template(images, part_name.strip(), use_augmentation)
-
-
-# def detect_part(image, threshold, method):
-#     """Detect part in image"""
-#     return detector.match_part(image, threshold, method)
-
-
-# def list_templates():
-#     """List all saved templates"""
-#     if not detector.templates:
-#         return "📭 No templates saved yet"
-    
-#     output = "📋 **Saved Templates:**\n\n"
-#     for name, template_data in detector.templates.items():
-#         total_features = len(template_data.features_list)
-#         output += f"- **{name}**\n"
-#         output += f"  └─ {template_data.num_images} images → {total_features} feature vectors\n\n"
-    
-#     return output
-
-
-# def delete_template_ui(part_name):
-#     """Delete a template"""
-#     return detector.delete_template(part_name)
-
-
-# # ── Gradio UI ─────────────────────────────────────────────────────────────
-
-# with gr.Blocks(title="Engine Part Detection System", theme=gr.themes.Soft()) as demo:
-#     gr.Markdown("""
-#     # 🔧 Engine Part Detection System
-#     ### Multi-Image Template Matching with ResNet50
-    
-#     **Key Features:**
-#     - 🖼️ **Multi-Image Templates**: Build robust references from multiple angles/lighting
-#     - 💡 **Lighting Resistant**: CLAHE preprocessing + augmentation handles shadows
-#     - 🎯 **High Accuracy**: Multiple feature vectors per template ensure reliable matching
-    
-#     **Workflow:**
-#     1. **Add Templates**: Upload 3-10 images per part (different angles/lighting)
-#     2. **Detect Parts**: Real-time matching with confidence scores
-#     """)
-    
-#     with gr.Tab("🔍 Detect Part"):
-#         gr.Markdown("### Upload or capture an image to identify the engine part")
-        
-#         with gr.Row():
-#             with gr.Column():
-#                 detect_input = gr.Image(
-#                     sources=["upload", "webcam"], 
-#                     type="numpy",
-#                     label="Input Image"
-#                 )
-                
-#                 with gr.Row():
-#                     threshold_slider = gr.Slider(
-#                         0.5, 0.95, 
-#                         value=0.7, 
-#                         step=0.05,
-#                         label="Similarity Threshold"
-#                     )
-#                     method_dropdown = gr.Dropdown(
-#                         choices=['mean', 'max', 'median'],
-#                         value='mean',
-#                         label="Matching Method"
-#                     )
-                
-#                 detect_btn = gr.Button("🔍 Detect Part", variant="primary", size="lg")
-                
-#             with gr.Column():
-#                 detect_output = gr.Textbox(
-#                     label="Detection Results", 
-#                     lines=12,
-#                     max_lines=15
-#                 )
-#                 match_label = gr.Label(label="Matched Part", num_top_classes=3)
-        
-#         detect_btn.click(
-#             fn=detect_part,
-#             inputs=[detect_input, threshold_slider, method_dropdown],
-#             outputs=[detect_output, match_label],
-#             api_name="detect"
-#         )
-    
-#     with gr.Tab("➕ Add Template"):
-#         gr.Markdown("""
-#         ### Build a Golden Template
-#         Upload **3-10 images** of the same part with:
-#         - Different lighting conditions (bright, dim, shadows)
-#         - Multiple angles (front, side, top)
-#         - Various backgrounds
-#         """)
-        
-#         with gr.Row():
-#             with gr.Column():
-#                 template_input = gr.File(
-#                     file_count="multiple",
-#                     file_types=["image"],
-#                     label="📸 Upload Multiple Images (3-10 recommended)"
-#                 )
-                
-#                 part_name_input = gr.Textbox(
-#                     label="Part Name",
-#                     placeholder="e.g., spark_plug, piston, cylinder_head",
-#                     info="Use descriptive names without spaces"
-#                 )
-                
-#                 augmentation_checkbox = gr.Checkbox(
-#                     value=True,
-#                     label="🔄 Enable Augmentation (Recommended)",
-#                     info="Automatically generates variations for robustness"
-#                 )
-                
-#                 add_btn = gr.Button("💾 Save Template", variant="primary", size="lg")
-                
-#             with gr.Column():
-#                 add_output = gr.Textbox(
-#                     label="Status", 
-#                     lines=8,
-#                     max_lines=10
-#                 )
-                
-#                 gr.Markdown("""
-#                 **Tips:**
-#                 - More images = better accuracy
-#                 - Vary lighting: normal, bright, shadowed
-#                 - Include different angles
-#                 - Augmentation adds 4x variations per image
-#                 """)
-        
-#         # Handle file upload and conversion
-#         def process_files(files):
-#             if files is None:
-#                 return []
-#             images = []
-#             for file in files:
-#                 img = Image.open(file.name)
-#                 images.append(np.array(img))
-#             return images
-        
-#         template_images_state = gr.State([])
-        
-#         template_input.change(
-#             fn=process_files,
-#             inputs=template_input,
-#             outputs=template_images_state
-#         )
-        
-#         add_btn.click(
-#             fn=add_template,
-#             inputs=[template_images_state, part_name_input, augmentation_checkbox],
-#             outputs=add_output,
-#             api_name="add_template"
-#         )
-    
-#     with gr.Tab("📋 Manage Templates"):
-#         gr.Markdown("### View and manage saved templates")
-        
-#         with gr.Row():
-#             with gr.Column():
-#                 template_list = gr.Textbox(
-#                     label="Saved Templates", 
-#                     lines=15,
-#                     max_lines=20
-#                 )
-#                 refresh_btn = gr.Button("🔄 Refresh List", size="sm")
-                
-#             with gr.Column():
-#                 delete_part_name = gr.Textbox(
-#                     label="Template Name to Delete",
-#                     placeholder="Enter exact part name"
-#                 )
-#                 delete_btn = gr.Button("🗑️ Delete Template", variant="stop")
-#                 delete_output = gr.Textbox(label="Status", lines=3)
-        
-#         refresh_btn.click(
-#             fn=list_templates,
-#             outputs=template_list,
-#             api_name="list_templates"
-#         )
-        
-#         delete_btn.click(
-#             fn=delete_template_ui,
-#             inputs=delete_part_name,
-#             outputs=delete_output
-#         ).then(
-#             fn=list_templates,
-#             outputs=template_list
-#         )
-        
-#         # Auto-load templates on tab open
-#         demo.load(fn=list_templates, outputs=template_list)
-    
-#     with gr.Tab("ℹ️ Help"):
-#         gr.Markdown("""
-#         ## How It Works
-        
-#         ### 🎯 Multi-Image Template Matching
-#         Each template consists of multiple feature vectors extracted from:
-#         - Your uploaded reference images (3-10 recommended)
-#         - Automatic augmentations (brightness, contrast variations)
-        
-#         This creates a **"golden template"** that's robust to:
-#         - ✅ Lighting changes (shadows, bright spots)
-#         - ✅ Different angles and perspectives
-#         - ✅ Background variations
-#         - ✅ Camera/phone variations
-        
-#         ### 🔬 Technical Details
-#         - **Backbone**: ResNet50 pre-trained on ImageNet
-#         - **Preprocessing**: CLAHE (contrast enhancement) + Gaussian blur
-#         - **Matching**: Cosine similarity across multiple feature vectors
-#         - **Method Options**:
-#           - `mean`: Average similarity (balanced, recommended)
-#           - `max`: Best single match (more lenient)
-#           - `median`: Middle similarity (outlier-resistant)
-        
-#         ### 💡 Best Practices
-        
-#         1. **Template Creation**:
-#            - Upload 5-7 images per part minimum
-#            - Include normal, bright, and shadowed lighting
-#            - Capture multiple angles
-#            - Keep augmentation enabled
-        
-#         2. **Detection**:
-#            - Start with 0.70 threshold
-#            - Adjust based on results (lower = more lenient)
-#            - Use `mean` method for general use
-        
-#         3. **Troubleshooting**:
-#            - Low confidence? Add more template images
-#            - False positives? Increase threshold
-#            - Missing detections? Lower threshold or use `max` method
-        
-#         ### 📊 Example Workflow
-#         ```
-#         Template Building:
-#         ├─ Part: "spark_plug"
-#         ├─ Images: 5 photos
-#         ├─ Augmentation: 4x per image
-#         └─ Total: 20 feature vectors
-        
-#         Detection:
-#         ├─ Query: New spark plug photo
-#         ├─ Similarity: 87% (mean across 20 vectors)
-#         └─ Result: ✅ MATCHED
-#         ```
-#         """)
-
-
-# if __name__ == "__main__":
-#     demo.launch(
-#         server_name="0.0.0.0",
-#         server_port=7860,
-#         share=False
-#     )

app.py

@@ -0,0 +1,753 @@
+
+import os
+import sys
+import cv2
+import time
+import asyncio
+import logging
+import argparse
+import threading
+import numpy as np
+from enum import Enum
+from typing import Optional, Dict, Any, Callable, Tuple, List
+from dataclasses import dataclass, field
+from datetime import datetime
+from PIL import Image
+
+# ─────────────────────────────────────────────────────────────────
+# Logging
+# ─────────────────────────────────────────────────────────────────
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s │ %(name)-12s │ %(levelname)-7s │ %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("detection")
+
+
+# ─────────────────────────────────────────────────────────────────
+# Data Models
+# ─────────────────────────────────────────────────────────────────
+
+class Verdict(str, Enum):
+    PASS = "PASS"
+    FAIL = "FAIL"
+    UNKNOWN = "UNKNOWN"
+    ERROR = "ERROR"
+
+
+@dataclass
+class QualityMetrics:
+    """Image quality measurements."""
+    brightness: float = 0.0
+    contrast: float = 0.0
+    sharpness: float = 0.0
+    is_blurred: bool = False
+    resolution: Tuple[int, int] = (0, 0)
+
+    @property
+    def quality_score(self) -> float:
+        return min(100.0, self.sharpness / 2.0)
+
+
+@dataclass
+class SegmentedROI:
+    """A detected region of interest from segmentation."""
+    bbox: Tuple[int, int, int, int]  # x, y, w, h
+    contour: Any = None
+    cropped_image: Optional[Image.Image] = None
+    mask: Optional[np.ndarray] = None
+    area: float = 0.0
+    circularity: float = 0.0
+    label: str = "part"
+
+
+@dataclass
+class DetectionResult:
+    """Complete result of a single detection pass."""
+    verdict: Verdict = Verdict.UNKNOWN
+    confidence: float = 0.0
+    matched_class: str = ""
+    quality: QualityMetrics = field(default_factory=QualityMetrics)
+    visualization_b64: Optional[str] = None
+    all_scores: Dict[str, float] = field(default_factory=dict)
+    segments_found: int = 0
+    status_detail: str = ""
+    timestamp: str = ""
+    elapsed_ms: float = 0.0
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "verdict": self.verdict.value,
+            "confidence": round(self.confidence, 4),
+            "matched_class": self.matched_class,
+            "quality": {
+                "brightness": round(self.quality.brightness, 2),
+                "contrast": round(self.quality.contrast, 2),
+                "sharpness": round(self.quality.sharpness, 2),
+                "is_blurred": self.quality.is_blurred,
+                "quality_score": round(self.quality.quality_score, 2),
+                "resolution": list(self.quality.resolution),
+            },
+            "visualization": self.visualization_b64,
+            "all_scores": self.all_scores,
+            "segments_found": self.segments_found,
+            "status_detail": self.status_detail,
+            "timestamp": self.timestamp,
+            "elapsed_ms": round(self.elapsed_ms, 1),
+        }
+
+
+@dataclass
+class SessionStats:
+    """Running totals for an auto-inspection session."""
+    total: int = 0
+    passed: int = 0
+    failed: int = 0
+    unknown: int = 0
+    errors: int = 0
+    start_time: Optional[float] = None
+
+    @property
+    def elapsed_seconds(self) -> float:
+        if self.start_time is None:
+            return 0.0
+        return time.time() - self.start_time
+
+    def record(self, verdict: Verdict):
+        self.total += 1
+        if verdict == Verdict.PASS:
+            self.passed += 1
+        elif verdict == Verdict.FAIL:
+            self.failed += 1
+        elif verdict == Verdict.UNKNOWN:
+            self.unknown += 1
+        else:
+            self.errors += 1
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "total": self.total,
+            "passed": self.passed,
+            "failed": self.failed,
+            "unknown": self.unknown,
+            "errors": self.errors,
+            "elapsed_seconds": round(self.elapsed_seconds, 1),
+        }
+
+
+# ─────────────────────────────────────────────────────────────────
+# Image Analyzer — Validation, Quality, and Segmentation
+# ─────────────────────────────────────────────────────────────────
+
+class ImageAnalyzer:
+    """
+    Handles all pre-AI image analysis:
+      - Quality validation (brightness, contrast, sharpness)
+      - Part segmentation via contour + morphological analysis
+      - ROI extraction for focused detection
+    """
+
+    # Thresholds
+    MIN_RESOLUTION = (320, 240)
+    MAX_INPUT_DIM = 1024
+    BRIGHTNESS_FLOOR = 15
+    BRIGHTNESS_CEIL = 245
+    CONTRAST_FLOOR = 5
+    BLUR_THRESHOLD = 100.0  # Laplacian variance below this = blurry
+
+    # Segmentation tunables
+    MORPHO_KERNEL = 5
+    MIN_CONTOUR_AREA_RATIO = 0.005   # Minimum area relative to image area
+    MAX_CONTOUR_AREA_RATIO = 0.85    # Maximum area relative to image area
+    CIRCULARITY_THRESHOLD = 0.15     # Minimum circularity for a valid part contour
+
+    def measure_quality(self, img: Image.Image) -> QualityMetrics:
+        """Compute image quality metrics without modifying the image."""
+        arr = np.array(img.convert("RGB"))
+        gray = cv2.cvtColor(arr, cv2.COLOR_RGB2GRAY)
+        laplacian_var = float(cv2.Laplacian(gray, cv2.CV_64F).var())
+
+        return QualityMetrics(
+            brightness=float(np.mean(arr)),
+            contrast=float(np.std(arr)),
+            sharpness=laplacian_var,
+            is_blurred=laplacian_var < self.BLUR_THRESHOLD,
+            resolution=(img.width, img.height),
+        )
+
+    def validate(self, img: Image.Image) -> Tuple[bool, str]:
+        
+        w, h = img.size
+        if w < self.MIN_RESOLUTION[0] or h < self.MIN_RESOLUTION[1]:
+            return False, f"Resolution too low: {w}×{h} (need {self.MIN_RESOLUTION[0]}×{self.MIN_RESOLUTION[1]})"
+
+        aspect = w / h
+        if aspect < 0.2 or aspect > 5.0:
+            return False, f"Unusual aspect ratio: {aspect:.2f}"
+
+        metrics = self.measure_quality(img)
+        if metrics.brightness < self.BRIGHTNESS_FLOOR:
+            return False, "Image too dark"
+        if metrics.brightness > self.BRIGHTNESS_CEIL:
+            return False, "Image too bright / overexposed"
+        if metrics.contrast < self.CONTRAST_FLOOR:
+            return False, "Insufficient contrast — blank or uniform image"
+
+        return True, "OK"
+
+    def prepare(self, img: Image.Image) -> Image.Image:
+        
+        if img.mode != "RGB":
+            img = img.convert("RGB")
+        img.thumbnail((self.MAX_INPUT_DIM, self.MAX_INPUT_DIM), Image.Resampling.LANCZOS)
+        return img
+
+    # ── Part Segmentation ────────────────────────────────────────
+
+    def segment_parts(self, img: Image.Image) -> List[SegmentedROI]:
+       
+        arr = np.array(img.convert("RGB"))
+        gray = cv2.cvtColor(arr, cv2.COLOR_RGB2GRAY)
+        img_area = gray.shape[0] * gray.shape[1]
+
+        # Adaptive threshold deals better with shadows than global Otsu
+        binary = cv2.adaptiveThreshold(
+            gray, 255,
+            cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
+            cv2.THRESH_BINARY_INV,
+            blockSize=31,
+            C=10,
+        )
+
+        # Morphological closing fills holes inside parts
+        kernel = cv2.getStructuringElement(
+            cv2.MORPH_ELLIPSE,
+            (self.MORPHO_KERNEL, self.MORPHO_KERNEL),
+        )
+        closed = cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel, iterations=3)
+
+        # Optional: small opening to remove noise specks
+        opened = cv2.morphologyEx(closed, cv2.MORPH_OPEN, kernel, iterations=1)
+
+        contours, _ = cv2.findContours(opened, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+        rois: List[SegmentedROI] = []
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            ratio = area / img_area
+
+            # Filter by relative area
+            if ratio < self.MIN_CONTOUR_AREA_RATIO or ratio > self.MAX_CONTOUR_AREA_RATIO:
+                continue
+
+            # Circularity = 4π × area / perimeter²   (1.0 for perfect circle)
+            perimeter = cv2.arcLength(cnt, True)
+            circularity = (4 * np.pi * area / (perimeter ** 2)) if perimeter > 0 else 0
+
+            if circularity < self.CIRCULARITY_THRESHOLD:
+                continue
+
+            x, y, w, h = cv2.boundingRect(cnt)
+
+            # Create a mask for this contour and crop
+            mask = np.zeros(gray.shape, dtype=np.uint8)
+            cv2.drawContours(mask, [cnt], -1, 255, thickness=cv2.FILLED)
+
+            # Crop the bounding box region
+            crop_arr = arr[y:y + h, x:x + w].copy()
+            crop_mask = mask[y:y + h, x:x + w]
+            # Apply mask — set background to black
+            crop_arr[crop_mask == 0] = 0
+            cropped_pil = Image.fromarray(crop_arr)
+
+            rois.append(SegmentedROI(
+                bbox=(x, y, w, h),
+                contour=cnt,
+                cropped_image=cropped_pil,
+                mask=crop_mask,
+                area=area,
+                circularity=circularity,
+                label=f"part_{len(rois)}",
+            ))
+
+        # Sort by area descending — largest part first
+        rois.sort(key=lambda r: r.area, reverse=True)
+        log.info(f"Segmentation: found {len(rois)} part region(s) from {len(contours)} contours")
+        return rois
+
+    def draw_segmentation_overlay(
+        self, img: Image.Image, rois: List[SegmentedROI], verdict: Optional[Verdict] = None
+    ) -> Image.Image:
+        
+        arr = np.array(img.convert("RGB")).copy()
+
+        color_map = {
+            Verdict.PASS: (0, 200, 100),
+            Verdict.FAIL: (220, 60, 60),
+            Verdict.UNKNOWN: (220, 180, 0),
+            Verdict.ERROR: (128, 128, 128),
+            None: (100, 180, 255),
+        }
+        color = color_map.get(verdict, (100, 180, 255))
+
+        for roi in rois:
+            x, y, w, h = roi.bbox
+            cv2.rectangle(arr, (x, y), (x + w, y + h), color, 2)
+
+            # Label with area info
+            label = f"{roi.label} ({roi.circularity:.2f})"
+            font_scale = max(0.4, min(1.0, w / 300))
+            cv2.putText(arr, label, (x, max(y - 8, 15)),
+                        cv2.FONT_HERSHEY_SIMPLEX, font_scale, color, 1, cv2.LINE_AA)
+
+        # Verdict stamp in top-right
+        if verdict is not None:
+            stamp = verdict.value
+            (tw, th), _ = cv2.getTextSize(stamp, cv2.FONT_HERSHEY_SIMPLEX, 1.2, 3)
+            sx = arr.shape[1] - tw - 20
+            sy = th + 20
+            cv2.rectangle(arr, (sx - 10, sy - th - 10), (sx + tw + 10, sy + 10), color, cv2.FILLED)
+            cv2.putText(arr, stamp, (sx, sy),
+                        cv2.FONT_HERSHEY_SIMPLEX, 1.2, (255, 255, 255), 3, cv2.LINE_AA)
+
+        return Image.fromarray(arr)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Detection Engine — Orchestrates the full pipeline
+# ─────────────────────────────────────────────────────────────────
+
+class DetectionEngine:
+    
+
+    def __init__(self, hf_client=None, threshold: float = 0.70):
+        self.analyzer = ImageAnalyzer()
+        self.threshold = threshold
+
+        # Lazy-init the HF client so index.py stays importable
+        # without triggering network calls at import time.
+        self._hf_client = hf_client
+        self._hf_initialized = hf_client is not None
+
+    @property
+    def hf(self):
+        if not self._hf_initialized:
+            from hf_client import HuggingFaceClient
+            self._hf_client = HuggingFaceClient()
+            self._hf_initialized = True
+        return self._hf_client
+
+    async def run(self, img: Image.Image, threshold: Optional[float] = None) -> DetectionResult:
+       
+        t0 = time.time()
+        result = DetectionResult(timestamp=datetime.utcnow().isoformat())
+        thr = threshold if threshold is not None else self.threshold
+
+        # ── Step 1: Quality Gate ─────────────────────────────────
+        valid, reason = self.analyzer.validate(img)
+        result.quality = self.analyzer.measure_quality(img)
+
+        if not valid:
+            result.verdict = Verdict.ERROR
+            result.status_detail = f"Quality rejected: {reason}"
+            result.elapsed_ms = (time.time() - t0) * 1000
+            log.warning(f"Quality gate failed: {reason}")
+            return result
+
+        # ── Step 2: Segment Parts ────────────────────────────────
+        rois = self.analyzer.segment_parts(img)
+        result.segments_found = len(rois)
+
+        if len(rois) == 0:
+            log.info("No part segments found — sending full image to AI")
+
+        # ── Step 3: Prepare for AI ───────────────────────────────
+        # Send the full image (the backend has its own ROI logic).
+        # The segmentation here is for local overlay + future use.
+        prepared = self.analyzer.prepare(img)
+
+        # ── Step 4: AI Classification ────────────────────────────
+        try:
+            ai_result = await self.hf.detect_part(prepared, thr)
+        except Exception as exc:
+            result.verdict = Verdict.ERROR
+            result.status_detail = f"AI backend error: {exc}"
+            result.elapsed_ms = (time.time() - t0) * 1000
+            log.error(f"AI call failed: {exc}")
+            return result
+
+        if not ai_result.get("success"):
+            result.verdict = Verdict.ERROR
+            result.status_detail = f"AI returned failure: {ai_result.get('error', 'unknown')}"
+            result.elapsed_ms = (time.time() - t0) * 1000
+            return result
+
+        # ── Step 5: Interpret Verdict ────────────────────────────
+        best_match = str(ai_result.get("best_match", "")).strip()
+        confidence = float(ai_result.get("confidence", 0.0))
+        all_scores = ai_result.get("all_scores", {})
+        status_text = str(ai_result.get("status_text", ""))
+
+        result.confidence = confidence
+        result.matched_class = best_match
+        result.all_scores = all_scores
+        result.status_detail = status_text
+
+        result.verdict = self._interpret_verdict(best_match, status_text)
+
+        # ── Step 6: Visualization ────────────────────────────────
+        # Build a composite overlay: segmentation boxes + verdict stamp
+        vis_img = self.analyzer.draw_segmentation_overlay(prepared, rois, result.verdict)
+
+        # Convert overlay to base64 for transport
+        import io, base64
+        buf = io.BytesIO()
+        vis_img.save(buf, format="JPEG", quality=85)
+        result.visualization_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
+
+        # Also attach the AI backend's visualization if available
+        vis_path = ai_result.get("visualization")
+        if vis_path and os.path.exists(vis_path):
+            try:
+                with open(vis_path, "rb") as f:
+                    result.visualization_b64 = base64.b64encode(f.read()).decode("utf-8")
+            finally:
+                try:
+                    os.remove(vis_path)
+                except OSError:
+                    pass
+
+        # Cleanup any temp files from the HF client
+        for tmp in ai_result.get("_temp_paths", []):
+            if tmp and tmp != vis_path:
+                try:
+                    os.remove(tmp)
+                except OSError:
+                    pass
+
+        result.elapsed_ms = (time.time() - t0) * 1000
+        log.info(
+            f"Detection: {result.verdict.value} │ "
+            f"class={best_match} │ conf={confidence:.3f} │ "
+            f"segments={len(rois)} │ {result.elapsed_ms:.0f}ms"
+        )
+        return result
+
+    @staticmethod
+    def _interpret_verdict(best_match: str, status_text: str) -> Verdict:
+        
+        match_upper = best_match.upper()
+        status_lower = status_text.lower()
+
+        # Localization failures (bolt holes not found, etc.)
+        failure_markers = ["no bolt holes", "localization failed", "insufficient hole"]
+        if any(marker in status_lower for marker in failure_markers):
+            return Verdict.UNKNOWN
+
+        # Empty / none match
+        if not match_upper or match_upper == "NONE" or match_upper == "UNKNOWN":
+            return Verdict.UNKNOWN
+
+        # Explicit verdict from backend status text
+        status_upper = status_text.upper()
+        if "PASS" in status_upper:
+            return Verdict.PASS
+        if "FAIL" in status_upper:
+            return Verdict.FAIL
+
+        # Fallback: class-name heuristic
+        if "PERFECT" in match_upper:
+            return Verdict.PASS
+
+        # Everything else (Defect, Damaged, etc.) is a FAIL
+        return Verdict.FAIL
+
+
+# ─────────────────────────────────────────────────────────────────
+# Camera Source — Manages OpenCV camera lifecycle
+# ─────────────────────────────────────────────────────────────────
+
+class CameraSource:
+    
+
+    WARMUP_FRAMES = 5  # Discard first N frames (often garbled)
+
+    def __init__(self, camera_id: int = 0):
+        self.camera_id = camera_id
+        self._cap: Optional[cv2.VideoCapture] = None
+
+    @staticmethod
+    def detect_available(max_check: int = 5) -> List[int]:
+        """Probe for available camera indices."""
+        available = []
+        for idx in range(max_check):
+            cap = cv2.VideoCapture(idx, cv2.CAP_DSHOW if os.name == "nt" else cv2.CAP_ANY)
+            if cap.isOpened():
+                ret, _ = cap.read()
+                if ret:
+                    available.append(idx)
+                cap.release()
+        return available
+
+    def open(self) -> bool:
+        """Open the camera and discard warm-up frames."""
+        backend = cv2.CAP_DSHOW if os.name == "nt" else cv2.CAP_ANY
+        self._cap = cv2.VideoCapture(self.camera_id, backend)
+
+        if not self._cap.isOpened():
+            log.error(f"Cannot open camera {self.camera_id}")
+            return False
+
+        # Set resolution hints
+        self._cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
+        self._cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
+
+        # Discard warm-up frames
+        for _ in range(self.WARMUP_FRAMES):
+            self._cap.read()
+
+        log.info(f"Camera {self.camera_id} opened — "
+                 f"{int(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH))}×"
+                 f"{int(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT))}")
+        return True
+
+    def grab(self) -> Optional[Image.Image]:
+        """Capture a single frame as a PIL Image (RGB)."""
+        if self._cap is None or not self._cap.isOpened():
+            return None
+        ret, frame = self._cap.read()
+        if not ret or frame is None:
+            return None
+        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        return Image.fromarray(rgb)
+
+    def release(self):
+        """Release the camera resource."""
+        if self._cap is not None:
+            self._cap.release()
+            self._cap = None
+            log.info(f"Camera {self.camera_id} released")
+
+    @property
+    def is_open(self) -> bool:
+        return self._cap is not None and self._cap.isOpened()
+
+
+# ─────────────────────────────────────────────────────────────────
+# Auto Inspector — Continuous detection loop
+# ─────────────────────────────────────────────────────────────────
+
+class AutoInspector:
+
+    def __init__(
+        self,
+        engine: DetectionEngine,
+        camera_id: int = 0,
+        interval: float = 3.0,
+    ):
+        self.engine = engine
+        self.camera = CameraSource(camera_id)
+        self.interval = max(1.0, interval)  # Floor at 1 second
+
+        self._stop_event = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self.stats = SessionStats()
+
+    @property
+    def is_running(self) -> bool:
+        return self._thread is not None and self._thread.is_alive()
+
+    def start(self, on_result: Optional[Callable] = None):
+       
+        if self.is_running:
+            log.warning("Auto-inspection is already running")
+            return
+
+        self._stop_event.clear()
+        self.stats = SessionStats(start_time=time.time())
+
+        self._thread = threading.Thread(
+            target=self._run_loop,
+            args=(on_result,),
+            daemon=True,
+            name="auto-inspector",
+        )
+        self._thread.start()
+        log.info(f"Auto-inspection started — camera={self.camera.camera_id}, interval={self.interval}s")
+
+    def stop(self):
+        """Signal the loop to stop and wait for cleanup."""
+        if not self.is_running:
+            return
+        log.info("Stopping auto-inspection...")
+        self._stop_event.set()
+        if self._thread:
+            self._thread.join(timeout=10)
+        self.camera.release()
+        log.info(f"Auto-inspection stopped — {self.stats.to_dict()}")
+
+    def _run_loop(self, on_result: Optional[Callable]):
+        """Internal loop that runs in a background thread."""
+        # Create a new event loop for this thread
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        self._loop = loop
+
+        try:
+            if not self.camera.open():
+                log.error("Failed to open camera — aborting auto-inspection")
+                return
+
+            while not self._stop_event.is_set():
+                frame = self.camera.grab()
+                if frame is None:
+                    log.warning("Frame grab failed — retrying in 1s")
+                    self._stop_event.wait(1.0)
+                    continue
+
+                # Run detection synchronously within this thread's event loop
+                result = loop.run_until_complete(self.engine.run(frame))
+                self.stats.record(result.verdict)
+
+                if on_result:
+                    try:
+                        on_result(result, self.stats)
+                    except Exception as cb_err:
+                        log.error(f"Callback error: {cb_err}")
+
+                # Wait for the interval (interruptible)
+                self._stop_event.wait(self.interval)
+
+        except Exception as exc:
+            log.error(f"Auto-inspection loop crashed: {exc}", exc_info=True)
+        finally:
+            self.camera.release()
+            loop.close()
+
+
+# ─────────────────────────────────────────────────────────────────
+# Single-Image Detection (convenience function)
+# ─────────────────────────────────────────────────────────────────
+
+async def detect_image(
+    image_path: str,
+    threshold: float = 0.70,
+    engine: Optional[DetectionEngine] = None,
+) -> DetectionResult:
+    
+    if not os.path.isfile(image_path):
+        raise FileNotFoundError(f"Image not found: {image_path}")
+
+    img = Image.open(image_path)
+    eng = engine or DetectionEngine(threshold=threshold)
+    return await eng.run(img, threshold)
+
+
+# ───���─────────────────────────────────────────────────────────────
+# CLI — Run as standalone script
+# ─────────────────────────────────────────────────────────────────
+
+def _print_result(result: DetectionResult, stats: Optional[SessionStats] = None):
+  
+    v = result.verdict.value
+    color = {"PASS": "\033[92m", "FAIL": "\033[91m", "UNKNOWN": "\033[93m", "ERROR": "\033[90m"}
+    reset = "\033[0m"
+    c = color.get(v, "")
+
+    print(f"\n{'─' * 50}")
+    print(f"  {c}█ {v}{reset}  │  class: {result.matched_class or '—'}  │  conf: {result.confidence:.1%}")
+    print(f"  segments: {result.segments_found}  │  quality: {result.quality.quality_score:.0f}  │  {result.elapsed_ms:.0f}ms")
+
+    if result.all_scores:
+        scores_str = "  ".join(f"{k}: {v:.1%}" for k, v in result.all_scores.items())
+        print(f"  scores: {scores_str}")
+
+    if result.status_detail:
+        detail = result.status_detail[:120].replace("\n", " ")
+        print(f"  detail: {detail}")
+
+    if stats:
+        s = stats
+        print(f"  session: {s.total} total  │  ✓{s.passed}  ✗{s.failed}  ?{s.unknown}  │  {s.elapsed_seconds:.0f}s")
+    print(f"{'─' * 50}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Engine Part Detection — Standalone Detection Pipeline",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Examples:
+  python index.py                            # Auto-detect camera, run continuous
+  python index.py --camera 0 --interval 5    # Camera 0, every 5 seconds
+  python index.py --image part.jpg           # Single image detection
+  python index.py --list-cameras             # Show available cameras
+        """,
+    )
+
+    grp = parser.add_mutually_exclusive_group()
+    grp.add_argument("--image", "-i", type=str, help="Path to a single image file for detection")
+    grp.add_argument("--camera", "-c", type=int, default=None, help="Camera index (default: auto-detect)")
+    grp.add_argument("--list-cameras", action="store_true", help="List available cameras and exit")
+
+    parser.add_argument("--threshold", "-t", type=float, default=0.70, help="Detection threshold (default: 0.70)")
+    parser.add_argument("--interval", type=float, default=3.0, help="Seconds between captures in auto mode (default: 3.0)")
+    parser.add_argument("--quiet", "-q", action="store_true", help="Suppress verbose output")
+
+    args = parser.parse_args()
+
+    if args.quiet:
+        logging.getLogger("detection").setLevel(logging.WARNING)
+
+    # ── List cameras ─────────────────────────────────────────────
+    if args.list_cameras:
+        print("Scanning for cameras...")
+        cams = CameraSource.detect_available()
+        if cams:
+            print(f"Found {len(cams)} camera(s): {cams}")
+        else:
+            print("No cameras detected.")
+        sys.exit(0)
+
+    # ── Single image mode ────────────────────────────────────────
+    if args.image:
+        print(f"Analyzing: {args.image}")
+        result = asyncio.run(detect_image(args.image, args.threshold))
+        _print_result(result)
+        sys.exit(0 if result.verdict != Verdict.ERROR else 1)
+
+    # ── Auto inspection mode (camera) ────────────────────────────
+    camera_id = args.camera
+    if camera_id is None:
+        print("Auto-detecting cameras...")
+        available = CameraSource.detect_available()
+        if not available:
+            print("No cameras found. Use --image for file-based detection.")
+            sys.exit(1)
+        camera_id = available[0]
+        print(f"Using camera {camera_id}")
+
+    engine = DetectionEngine(threshold=args.threshold)
+    inspector = AutoInspector(engine, camera_id=camera_id, interval=args.interval)
+
+    print(f"\n  Auto Inspection Mode")
+    print(f"  Camera: {camera_id}  │  Interval: {args.interval}s  │  Threshold: {args.threshold}")
+    print(f"  Press Ctrl+C to stop\n")
+
+    inspector.start(on_result=_print_result)
+
+    try:
+        while inspector.is_running:
+            time.sleep(0.5)
+    except KeyboardInterrupt:
+        print("\n\nStopping...")
+    finally:
+        inspector.stop()
+        print(f"\nSession summary: {inspector.stats.to_dict()}")
+
+
+if __name__ == "__main__":
+    main()