""" Quality Inspection Service Defect detection for manufacturing and quality control """ import logging from typing import List, Dict, Any, Optional, Tuple from io import BytesIO import numpy as np from PIL import Image import cv2 logger = logging.getLogger(__name__) class QualityService: """Quality inspection and defect detection""" def __init__(self): self.defect_types = [ "scratch", "dent", "crack", "stain", "misalignment", "missing_component", "color_variation", "surface_defect" ] async def inspect_surface( self, image_bytes: bytes, sensitivity: str = "medium", reference_image: Optional[bytes] = None ) -> Dict[str, Any]: """ Inspect surface for defects Args: image_bytes: Image to inspect sensitivity: low, medium, or high reference_image: Optional reference/good image for comparison Returns: Inspection results """ # Load image image = Image.open(BytesIO(image_bytes)) image_np = np.array(image) # Convert to grayscale if len(image_np.shape) == 3: gray = cv2.cvtColor(image_np, cv2.COLOR_RGB2GRAY) else: gray = image_np # Set sensitivity parameters sensitivity_params = { "low": {"blur": 5, "threshold": 30, "min_area": 500}, "medium": {"blur": 3, "threshold": 20, "min_area": 200}, "high": {"blur": 1, "threshold": 10, "min_area": 50} } params = sensitivity_params.get(sensitivity, sensitivity_params["medium"]) defects = [] # Edge detection for scratches and cracks edges = cv2.Canny(gray, 50, 150) scratch_defects = await self._detect_linear_defects(edges, params) defects.extend(scratch_defects) # Blob detection for dents and stains blob_defects = await self._detect_blob_defects(gray, params) defects.extend(blob_defects) # If reference image provided, do comparison if reference_image: comparison_defects = await self._compare_with_reference( image_np, reference_image, params ) defects.extend(comparison_defects) # Calculate quality score quality_score = self._calculate_quality_score(defects, image.size) return { "quality_score": round(quality_score, 1), "pass": quality_score >= 80, "defects_found": len(defects), "defects": defects, "sensitivity": sensitivity, "image_size": {"width": image.width, "height": image.height} } async def _detect_linear_defects( self, edges: np.ndarray, params: Dict ) -> List[Dict]: """Detect linear defects like scratches and cracks""" defects = [] # Use Hough Line Transform lines = cv2.HoughLinesP( edges, rho=1, theta=np.pi / 180, threshold=50, minLineLength=30, maxLineGap=10 ) if lines is not None: for line in lines: x1, y1, x2, y2 = line[0] length = np.sqrt((x2 - x1)**2 + (y2 - y1)**2) if length > params["min_area"] / 5: defects.append({ "type": "scratch", "severity": "high" if length > 100 else "medium", "location": { "x1": int(x1), "y1": int(y1), "x2": int(x2), "y2": int(y2) }, "size": round(length, 1) }) return defects[:10] # Limit to top 10 async def _detect_blob_defects( self, gray: np.ndarray, params: Dict ) -> List[Dict]: """Detect blob-like defects (dents, stains)""" defects = [] # Apply Gaussian blur blurred = cv2.GaussianBlur(gray, (params["blur"]*2+1, params["blur"]*2+1), 0) # Adaptive thresholding thresh = cv2.adaptiveThreshold( blurred, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2 ) # Find contours contours, _ = cv2.findContours( thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) for contour in contours: area = cv2.contourArea(contour) if area > params["min_area"]: x, y, w, h = cv2.boundingRect(contour) # Classify based on shape perimeter = cv2.arcLength(contour, True) circularity = 4 * np.pi * area / (perimeter * perimeter) if perimeter > 0 else 0 if circularity > 0.7: defect_type = "dent" else: defect_type = "stain" severity = "high" if area > 1000 else "medium" if area > 500 else "low" defects.append({ "type": defect_type, "severity": severity, "location": {"x": int(x), "y": int(y), "width": int(w), "height": int(h)}, "area": round(area, 1), "circularity": round(circularity, 2) }) return defects[:20] # Limit to top 20 async def _compare_with_reference( self, image: np.ndarray, reference_bytes: bytes, params: Dict ) -> List[Dict]: """Compare with reference image to find differences""" defects = [] # Load reference ref_image = Image.open(BytesIO(reference_bytes)) ref_np = np.array(ref_image) # Resize if needed if image.shape[:2] != ref_np.shape[:2]: ref_np = cv2.resize(ref_np, (image.shape[1], image.shape[0])) # Convert to grayscale if len(image.shape) == 3: gray1 = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY) else: gray1 = image if len(ref_np.shape) == 3: gray2 = cv2.cvtColor(ref_np, cv2.COLOR_RGB2GRAY) else: gray2 = ref_np # Calculate difference diff = cv2.absdiff(gray1, gray2) _, thresh = cv2.threshold(diff, params["threshold"], 255, cv2.THRESH_BINARY) # Find contours of differences contours, _ = cv2.findContours( thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) for contour in contours: area = cv2.contourArea(contour) if area > params["min_area"]: x, y, w, h = cv2.boundingRect(contour) defects.append({ "type": "deviation_from_reference", "severity": "high" if area > 1000 else "medium", "location": {"x": int(x), "y": int(y), "width": int(w), "height": int(h)}, "area": round(area, 1) }) return defects def _calculate_quality_score( self, defects: List[Dict], image_size: Tuple[int, int] ) -> float: """Calculate overall quality score (0-100)""" if not defects: return 100.0 total_area = image_size[0] * image_size[1] defect_area = sum(d.get("area", 100) for d in defects) # Penalty based on defect coverage coverage_penalty = (defect_area / total_area) * 100 # Penalty based on severity severity_penalties = {"high": 10, "medium": 5, "low": 2} severity_penalty = sum( severity_penalties.get(d.get("severity", "low"), 2) for d in defects ) score = 100 - min(coverage_penalty + severity_penalty, 100) return max(score, 0) async def inspect_dimensions( self, image_bytes: bytes, expected_dimensions: Dict[str, float], pixels_per_unit: float = 1.0 ) -> Dict[str, Any]: """ Check if object dimensions match expected values Args: image_bytes: Image data expected_dimensions: Expected width, height, etc. pixels_per_unit: Calibration factor Returns: Dimension analysis results """ image = Image.open(BytesIO(image_bytes)) image_np = np.array(image) # Convert and find contours if len(image_np.shape) == 3: gray = cv2.cvtColor(image_np, cv2.COLOR_RGB2GRAY) else: gray = image_np _, thresh = cv2.threshold(gray, 127, 255, cv2.THRESH_BINARY) contours, _ = cv2.findContours( thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) if not contours: return {"error": "No object detected"} # Find largest contour (main object) largest = max(contours, key=cv2.contourArea) x, y, w, h = cv2.boundingRect(largest) # Convert to units measured = { "width": w / pixels_per_unit, "height": h / pixels_per_unit } # Compare with expected deviations = {} for key in expected_dimensions: if key in measured: expected = expected_dimensions[key] actual = measured[key] deviation = ((actual - expected) / expected) * 100 deviations[key] = { "expected": expected, "measured": round(actual, 2), "deviation_percent": round(deviation, 2), "within_tolerance": abs(deviation) < 5 # 5% tolerance } all_within_tolerance = all(d["within_tolerance"] for d in deviations.values()) return { "pass": all_within_tolerance, "measurements": measured, "deviations": deviations, "bounding_box": {"x": x, "y": y, "width": w, "height": h} } # Singleton instance quality_service = QualityService()