""" AI Image Classification Pipeline Maps images → IOM objects + defect codes + reality signals """ from typing import Dict, List, Optional, Tuple from dataclasses import dataclass import json @dataclass class ImageAnalysis: """Result of AI image analysis""" detected_objects: List[Dict] defect_codes: List[Dict] reality_signals: List[Dict] confidence: float scene_type: str urban_context: Dict def to_dict(self) -> Dict: return { "detected_objects": self.detected_objects, "defect_codes": self.defect_codes, "reality_signals": self.reality_signals, "confidence": round(self.confidence, 2), "scene_type": self.scene_type, "urban_context": self.urban_context } class ImageClassifier: """Classifies images into IOM objects and signals""" def __init__(self): self.object_patterns = self._load_object_patterns() self.defect_patterns = self._load_defect_patterns() self.scene_classifiers = self._load_scene_classifiers() def _load_object_patterns(self) -> Dict: """Load object detection patterns""" return { "window": { "goid_prefix": "BYG-FAC-WIN", "keywords": ["window", "glass", "frame", "glazing"], "confidence_threshold": 0.7 }, "road_surface": { "goid_prefix": "TRN-ROD-SUR", "keywords": ["road", "pavement", "asphalt", "concrete", "surface"], "confidence_threshold": 0.8 }, "street_light": { "goid_prefix": "BEL-STR-LIG", "keywords": ["light", "lamp", "pole", "illumination"], "confidence_threshold": 0.75 }, "sign": { "goid_prefix": "COM-DIS-SGN", "keywords": ["sign", "billboard", "poster", "display"], "confidence_threshold": 0.7 }, "ev_charger": { "goid_prefix": "ENE-EVC-CHA", "keywords": ["charger", "ev", "electric", "charging_station"], "confidence_threshold": 0.8 }, "building_facade": { "goid_prefix": "BYG-FAC-WAL", "keywords": ["wall", "facade", "building", "exterior"], "confidence_threshold": 0.8 }, "vehicle": { "goid_prefix": "TRN-VEH", "keywords": ["car", "truck", "bus", "vehicle", "pickup"], "confidence_threshold": 0.85 }, "pedestrian": { "goid_prefix": "HUM-PED", "keywords": ["person", "pedestrian", "people", "crowd"], "confidence_threshold": 0.9 } } def _load_defect_patterns(self) -> Dict: """Load defect detection patterns""" return { "1100": { # Surface rust "keywords": ["rust", "corrosion", "oxidation", "brown_stain"], "severity": 2 }, "2100": { # Dirt accumulation "keywords": ["dirt", "grime", "stain", "pollution", "dust"], "severity": 2 }, "2300": { # Surface damage "keywords": ["crack", "chip", "scratch", "damage", "wear"], "severity": 3 }, "2400": { # Graffiti "keywords": ["graffiti", "tag", "paint", "vandalism"], "severity": 2 }, "4100": { # Missing parts "keywords": ["missing", "absent", "broken_off", "detached"], "severity": 4 }, "4200": { # Broken parts "keywords": ["broken", "damaged", "fractured", "destroyed"], "severity": 4 }, "5100": { # Physical blockage "keywords": ["blocked", "obstructed", "barrier", "closure"], "severity": 3 }, "6200": { # Water damage "keywords": ["water", "flood", "leak", "moisture", "stain"], "severity": 4 } } def _load_scene_classifiers(self) -> Dict: """Load scene classification patterns""" return { "urban_street": { "keywords": ["road", "sidewalk", "building", "street"], "signals": ["walkability", "noise_level", "heat_stress"] }, "construction_site": { "keywords": ["construction", "crane", "scaffold", "building"], "signals": ["construction_completion", "temporary_structure_density"] }, "commercial_area": { "keywords": ["shop", "restaurant", "sign", "crowd"], "signals": ["public_life", "informal_economy", "street_food_density"] }, "residential": { "keywords": ["house", "apartment", "residential", "home"], "signals": ["family_presence", "rent_burden", "community_activity"] }, "industrial": { "keywords": ["factory", "warehouse", "industrial", "smoke"], "signals": ["odor_index", "noise_level", "air_quality"] }, "park_green": { "keywords": ["park", "tree", "grass", "green"], "signals": ["shade_index", "heat_stress", "family_presence"] } } def analyze_image( self, image_path: str, location: Optional[Dict] = None, metadata: Optional[Dict] = None ) -> ImageAnalysis: """ Analyze an image and extract IOM objects, defects, and signals In production, this would use actual AI models (YOLO, CLIP, etc.) For now, simulate with pattern matching """ # Simulate AI detection detected_objects = self._simulate_object_detection(image_path) defect_codes = self._simulate_defect_detection(image_path, detected_objects) scene_type = self._classify_scene(image_path) reality_signals = self._extract_signals(scene_type, detected_objects, defect_codes) # Calculate overall confidence confidence = sum(obj.get("confidence", 0) for obj in detected_objects) / max(len(detected_objects), 1) # Extract urban context urban_context = self._extract_urban_context(detected_objects, scene_type) return ImageAnalysis( detected_objects=detected_objects, defect_codes=defect_codes, reality_signals=reality_signals, confidence=confidence, scene_type=scene_type, urban_context=urban_context ) def _simulate_object_detection(self, image_path: str) -> List[Dict]: """Simulate object detection (replace with actual AI in production)""" # In production: YOLO, Detectron2, etc. # For now, return example based on filename objects = [] # Example: image contains a window if "window" in image_path.lower() or "facade" in image_path.lower(): objects.append({ "type": "window", "goid_prefix": "BYG-FAC-WIN", "confidence": 0.85, "bbox": [100, 100, 300, 400] }) # Example: image contains a road if "road" in image_path.lower() or "street" in image_path.lower(): objects.append({ "type": "road_surface", "goid_prefix": "TRN-ROD-SUR", "confidence": 0.92, "bbox": [0, 300, 640, 480] }) # Example: image contains a vehicle if "vehicle" in image_path.lower() or "car" in image_path.lower() or "truck" in image_path.lower(): objects.append({ "type": "vehicle", "goid_prefix": "TRN-VEH", "confidence": 0.88, "bbox": [200, 200, 500, 400] }) # Default: building facade if not objects: objects.append({ "type": "building_facade", "goid_prefix": "BYG-FAC-WAL", "confidence": 0.75, "bbox": [0, 0, 640, 480] }) return objects def _simulate_defect_detection( self, image_path: str, detected_objects: List[Dict] ) -> List[Dict]: """Simulate defect detection""" defects = [] # Check for defects based on object type and image name for obj in detected_objects: obj_type = obj.get("type", "") # Window defects if obj_type == "window": if "dirty" in image_path.lower() or "dirt" in image_path.lower(): defects.append({ "code": "2100", "type": "dirt_accumulation", "confidence": 0.82, "severity": 2 }) if "broken" in image_path.lower() or "crack" in image_path.lower(): defects.append({ "code": "2300", "type": "surface_damage", "confidence": 0.78, "severity": 3 }) # Road defects if obj_type == "road_surface": if "pothole" in image_path.lower() or "damage" in image_path.lower(): defects.append({ "code": "2300", "type": "surface_damage", "confidence": 0.90, "severity": 4 }) if "graffiti" in image_path.lower(): defects.append({ "code": "2400", "type": "graffiti", "confidence": 0.85, "severity": 2 }) return defects def _classify_scene(self, image_path: str) -> str: """Classify scene type""" # In production: scene classification model # For now, simple keyword matching if any(kw in image_path.lower() for kw in ["street", "road", "sidewalk"]): return "urban_street" elif any(kw in image_path.lower() for kw in ["construction", "building"]): return "construction_site" elif any(kw in image_path.lower() for kw in ["shop", "store", "commercial"]): return "commercial_area" elif any(kw in image_path.lower() for kw in ["park", "green", "tree"]): return "park_green" else: return "urban_street" def _extract_signals( self, scene_type: str, detected_objects: List[Dict], defect_codes: List[Dict] ) -> List[Dict]: """Extract reality signals from analysis""" signals = [] # Scene-based signals scene_signals = self.scene_classifiers.get(scene_type, {}).get("signals", []) for signal_type in scene_signals: signals.append({ "signal_type": signal_type, "value": 50.0, # Baseline "confidence": 0.6, "source": "scene_classification" }) # Object-based signals for obj in detected_objects: obj_type = obj.get("type", "") if obj_type == "vehicle": signals.append({ "signal_type": "noise_level", "value": 65.0, # dB estimate "confidence": 0.5, "source": "object_detection" }) if obj_type == "road_surface": signals.append({ "signal_type": "walkability", "value": 70.0, "confidence": 0.6, "source": "object_detection" }) # Defect-based signals for defect in defect_codes: code = defect.get("code", "") if code in ["2100", "2200"]: signals.append({ "signal_type": "visual_maintenance", "value": 30.0, "confidence": 0.75, "source": "defect_detection" }) if code in ["2300", "4100", "4200"]: signals.append({ "signal_type": "infrastructure_reliability", "value": 40.0, "confidence": 0.7, "source": "defect_detection" }) return signals def _extract_urban_context( self, detected_objects: List[Dict], scene_type: str ) -> Dict: """Extract urban context from analysis""" return { "scene_type": scene_type, "object_count": len(detected_objects), "object_types": list(set(obj.get("type") for obj in detected_objects)), "density_estimate": len(detected_objects) * 10, # Simple heuristic "activity_level": "high" if len(detected_objects) > 5 else "medium" if len(detected_objects) > 2 else "low" } def batch_process( self, image_paths: List[str], locations: Optional[List[Dict]] = None ) -> List[ImageAnalysis]: """Process multiple images""" results = [] for i, image_path in enumerate(image_paths): location = locations[i] if locations and i < len(locations) else None result = self.analyze_image(image_path, location) results.append(result) return results # Example usage def example_image_analysis(): """Example: Analyze an image""" classifier = ImageClassifier() # Analyze image result = classifier.analyze_image( image_path="bangkok_street_road_damage.jpg", location={"lat": 13.7563, "lng": 100.5018} ) print("=== Image Analysis ===") print(f"Scene Type: {result.scene_type}") print(f"Confidence: {result.confidence:.2f}") print("\nDetected Objects:") for obj in result.detected_objects: print(f" {obj['type']} ({obj['goid_prefix']}): {obj['confidence']:.2f}") print("\nDefect Codes:") for defect in result.defect_codes: print(f" {defect['code']} ({defect['type']}): severity {defect['severity']}") print("\nReality Signals:") for signal in result.reality_signals: print(f" {signal['signal_type']}: {signal['value']:.1f} (confidence: {signal['confidence']})") print("\nUrban Context:") print(f" Activity Level: {result.urban_context['activity_level']}") print(f" Density Estimate: {result.urban_context['density_estimate']}") return result if __name__ == '__main__': example_image_analysis()