Spaces:

Askhedi
/

Car_Damage_Insurance_Fraud_Detector

Running

App Files Files Community

astrosbd commited on Jun 19, 2025

Commit

d09366d

verified ·

1 Parent(s): cd8cefd

V1.2

Browse files

Files changed (1) hide show

app.py +533 -2

app.py CHANGED Viewed

@@ -1,3 +1,530 @@
 def process_image(input_image, damage_threshold, deepfake_threshold, skip_damage, device_str, usage_count, recipient_email):
     """Process an image through the detection pipeline and send results by email"""
     # Handle empty or None usage_count
@@ -212,7 +739,6 @@ def process_image(input_image, damage_threshold, deepfake_threshold, skip_damage
     return analysis_text + f"\n\n📧 {email_message}", usage_count, email_message
 def create_gradio_interface():
     # Define a theme
     theme = gr.themes.Soft(
@@ -428,4 +954,9 @@ def create_gradio_interface():
             outputs=[usage_display]
         )
-    return app

+#!/usr/bin/env python3
+import importlib.util
+import os
+import sys
+import time
+import cv2
+import torch
+import numpy as np
+import gradio as gr
+from PIL import Image
+from torchvision import transforms
+import torch.nn as nn
+import traceback
+from huggingface_hub import hf_hub_download
+from torchvision.models import vit_b_16
+# Email functionality imports
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.mime.image import MIMEImage
+from email.mime.base import MIMEBase
+from email import encoders
+import base64
+import io
+from datetime import datetime
+# Add current directory to path
+if not os.getcwd() in sys.path:
+    sys.path.append(os.getcwd())
+# Check if detectron2 is installed
+if importlib.util.find_spec("detectron2") is None:
+    print("Installing PyTorch and Detectron2...")
+    os.system("pip install torch torchvision --extra-index-url https://download.pytorch.org/whl/cpu")
+    os.system("pip install git+https://github.com/facebookresearch/detectron2.git")
+    print("Installation complete!")
+# Check for detectron2
+try:
+    from detectron2.engine import DefaultPredictor
+    from detectron2.config import get_cfg
+    from detectron2.utils.visualizer import Visualizer, ColorMode
+    from detectron2 import model_zoo
+    DETECTRON2_AVAILABLE = True
+except ImportError:
+    print("Warning: Detectron2 is not installed. Damage detection will not be available.")
+    DETECTRON2_AVAILABLE = False
+# Define model paths
+DEFAULT_DAMAGE_MODEL_PATH = "./output/model_final.pth"
+DEFAULT_DEEPFAKE_MODEL_PATH = "./output/vit_deepfake_final.pth"
+# Sample images for demo (add your own paths)
+SAMPLE_IMAGES = [
+    "./test3.png",
+    "./test5.png",
+]
+# Maximum number of tries allowed
+MAX_TRIES = 5
+# Email configuration using environment variables
+EMAIL_CONFIG = {
+    'SMTP_SERVER': 'smtp.mail.ovh.net',
+    'EMAIL': os.getenv('login_email', 'sales@askhedi.fr'),
+    'PASSWORD': os.getenv('password_email'),
+    'SMTP_PORT': 465
+}
+# Télécharger le modèle deepfake depuis Hugging Face
+try:
+    huggingface_model_path = hf_hub_download(
+        repo_id="Askhedi/Car_damage_fraud_detector",
+        filename="vit_deepfake_final.pth",
+        token=os.getenv('key')
+    )
+    print(f"Modèle téléchargé depuis Hugging Face: {huggingface_model_path}")
+except Exception as e:
+    print(f"Erreur lors du téléchargement du modèle depuis Hugging Face: {e}")
+    huggingface_model_path = None
+def send_results_by_email(recipient_email, analysis_text, result_image, original_filename="car_image"):
+    """Send analysis results by email"""
+    if not EMAIL_CONFIG['PASSWORD']:
+        return False, "❌ Email configuration not available. Please contact support."
+    if not recipient_email or "@" not in recipient_email:
+        return False, "❌ Please provide a valid email address"
+    try:
+        # Create message
+        msg = MIMEMultipart('related')
+        msg['From'] = EMAIL_CONFIG['EMAIL']
+        msg['To'] = recipient_email
+        msg['Subject'] = f"🚗 Car Damage Analysis Results - {original_filename}"
+        # Create HTML body
+        html_body = f"""
+        <html>
+        <head>
+            <title>Car Damage Analysis Results</title>
+            <style>
+                body {{ font-family: Arial, sans-serif; margin: 20px; }}
+                .header {{ background-color: #f0f0f0; padding: 15px; border-radius: 5px; }}
+                .results {{ margin: 20px 0; white-space: pre-wrap; }}
+                .footer {{ color: #666; font-size: 12px; margin-top: 30px; }}
+            </style>
+        </head>
+        <body>
+            <div class="header">
+                <h1>🚗 Car Damage Fraud Detection Results</h1>
+                <p>Analysis performed on: {datetime.now().strftime('%d/%m/%Y at %H:%M:%S')}</p>
+            </div>
+            <div class="results">
+                <h3>📋 Analysis Details:</h3>
+                <pre>{analysis_text}</pre>
+            </div>
+            <div class="footer">
+                <p><em>This analysis was generated by the Car Damage Fraud Detector AI system.</em></p>
+                <p>Powered by Askhedi - Advanced AI Detection Services</p>
+            </div>
+        </body>
+        </html>
+        """
+        msg.attach(MIMEText(html_body, 'html'))
+        # Attach result image if available
+        if result_image is not None:
+            # Convert numpy array to image bytes
+            if isinstance(result_image, np.ndarray):
+                # Convert from RGB to PIL Image
+                pil_image = Image.fromarray(result_image.astype('uint8'))
+                img_buffer = io.BytesIO()
+                pil_image.save(img_buffer, format='PNG')
+                img_data = img_buffer.getvalue()
+                # Attach image
+                img_part = MIMEBase('application', 'octet-stream')
+                img_part.set_payload(img_data)
+                encoders.encode_base64(img_part)
+                img_part.add_header(
+                    'Content-Disposition',
+                    f'attachment; filename="analysis_result_{original_filename}.png"'
+                )
+                msg.attach(img_part)
+        # Send email
+        server = smtplib.SMTP_SSL(EMAIL_CONFIG['SMTP_SERVER'], EMAIL_CONFIG['SMTP_PORT'])
+        server.login(EMAIL_CONFIG['EMAIL'], EMAIL_CONFIG['PASSWORD'])
+        server.send_message(msg)
+        server.quit()
+        return True, f"✅ Results sent successfully to {recipient_email}"
+    except Exception as e:
+        return False, f"❌ Error sending email: {str(e)}"
+def setup_device(device_str):
+    """Set up the computation device"""
+    if device_str == 'auto':
+        if torch.cuda.is_available():
+            return torch.device('cuda:0')
+        elif hasattr(torch, 'backends') and hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
+            return torch.device('mps')
+        else:
+            return torch.device('cpu')
+    elif device_str == 'cuda' and torch.cuda.is_available():
+        return torch.device('cuda:0')
+    elif device_str == 'mps' and hasattr(torch, 'backends') and hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
+        return torch.device('mps')
+    else:
+        print(f"Warning: Device {device_str} not available, using CPU instead.")
+        return torch.device('cpu')
+def setup_damage_detector(model_path, threshold=0.7):
+    """Set up the damage detection model"""
+    if not DETECTRON2_AVAILABLE:
+        print("Detectron2 is not installed. Cannot set up damage detector.")
+        return None, None
+    try:
+        print(f"Checking model path: {model_path}")
+        print(f"Model exists: {os.path.exists(model_path)}")
+        if model_path is None or not os.path.exists(model_path):
+            print(f"Error: Damage model file not found at {model_path}")
+            return None, None
+        cfg = get_cfg()
+        cfg.merge_from_file(model_zoo.get_config_file("COCO-InstanceSegmentation/mask_rcnn_R_50_FPN_3x.yaml"))
+        cfg.MODEL.WEIGHTS = model_path
+        cfg.MODEL.ROI_HEADS.NUM_CLASSES = 1  # Only one class (damage)
+        cfg.MODEL.ROI_HEADS.SCORE_THRESH_TEST = threshold
+        # Use CPU if on Mac (MPS)
+        cfg.MODEL.DEVICE = "cpu"
+        print("Forcing Detectron2 to use CPU")
+        predictor = DefaultPredictor(cfg)
+        return predictor, cfg
+    except Exception as e:
+        print(f"Detailed error: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        return None, None
+def load_vit_deepfake_model(model_path, device):
+    """Load the Vision Transformer (ViT) model for deepfake detection"""
+    if model_path is None:
+        print("No deepfake model specified. Skipping deepfake detection.")
+        return None
+    try:
+        # Create ViT model with binary classification head
+        model = vit_b_16(weights=None)
+        # Modify the classifier head for binary classification (real vs fake)
+        in_features = model.heads.head.in_features
+        model.heads.head = nn.Linear(in_features, 2)
+        # Load weights
+        print(f"Loading ViT deepfake model from: {model_path}")
+        checkpoint = torch.load(model_path, map_location='cpu')
+        if isinstance(checkpoint, dict) and 'model_state_dict' in checkpoint:
+            model.load_state_dict(checkpoint['model_state_dict'])
+        elif isinstance(checkpoint, dict) and 'state_dict' in checkpoint:
+            model.load_state_dict(checkpoint['state_dict'])
+        else:
+            model.load_state_dict(checkpoint)
+        # Move model to device and set to evaluation mode
+        model = model.to(device)
+        model.eval()
+        return model
+    except Exception as e:
+        print(f"Error loading ViT deepfake model: {e}")
+        import traceback
+        traceback.print_exc()
+        return None
+def preprocess_for_vit(image, device):
+    """Prétraite une image pour le modèle ViT de torchvision"""
+    try:
+        print("Début du prétraitement de l'image...")
+        import torch
+        from torchvision import transforms
+        from PIL import Image
+        import numpy as np
+        import cv2
+        # Transformation standard pour ViT (similaire à celle utilisée lors de l'entraînement)
+        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])
+        ])
+        # Convertir l'image en RGB si nécessaire
+        if len(image.shape) == 3 and image.shape[2] == 3:
+            if image.dtype != np.uint8:
+                image = (image * 255).astype(np.uint8)
+            rgb_img = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+        else:
+            rgb_img = image
+        print("Image convertie en RGB")
+        # Convertir en PIL Image et appliquer les transformations
+        pil_img = Image.fromarray(rgb_img)
+        img_tensor = transform(pil_img).unsqueeze(0).to(device)  # Ajouter la dimension batch et envoyer au device
+        print("Image prétraitée avec succès")
+        return img_tensor
+    except Exception as e:
+        print(f"ERREUR lors du prétraitement de l'image: {e}")
+        traceback.print_exc()
+        return None
+def detect_damage(img, damage_detector):
+    """Detect damage in an image"""
+    try:
+        if img is None:
+            raise ValueError("Invalid image")
+        # If no detector, use whole image
+        if damage_detector is None:
+            h, w = img.shape[:2]
+            damage_regions = [{
+                "box": (0, 0, w, h),
+                "score": 1.0,
+                "mask": None
+            }]
+            return img, None, damage_regions
+        # Run inference
+        outputs = damage_detector(img)
+        # Get regions
+        instances = outputs["instances"].to("cpu")
+        boxes = instances.pred_boxes.tensor.numpy() if instances.has("pred_boxes") else []
+        scores = instances.scores.numpy() if instances.has("scores") else []
+        masks = instances.pred_masks.numpy() if instances.has("pred_masks") else []
+        damage_regions = []
+        for i in range(len(boxes)):
+            x1, y1, x2, y2 = map(int, boxes[i])
+            damage_regions.append({
+                "box": (x1, y1, x2, y2),
+                "score": float(scores[i]),
+                "mask": masks[i] if len(masks) > i else None
+            })
+        # If no regions found, use whole image
+        if not damage_regions:
+            h, w = img.shape[:2]
+            damage_regions = [{
+                "box": (0, 0, w, h),
+                "score": 1.0,
+                "mask": None
+            }]
+        return img, outputs, damage_regions
+    except Exception as e:
+        print(f"Error detecting damage: {e}")
+        traceback.print_exc()
+        # Return whole image if error
+        if 'img' in locals() and img is not None:
+            h, w = img.shape[:2]
+            damage_regions = [{
+                "box": (0, 0, w, h),
+                "score": 1.0,
+                "mask": None
+            }]
+            return img, None, damage_regions
+        return None, None, []
+def check_deepfake_vit(image, damage_regions, deepfake_model, device, threshold=0.5):
+    """Vérifie si les régions endommagées sont des deepfakes en utilisant un modèle ViT de torchvision"""
+    results = []
+    if deepfake_model is None:
+        print("Le modèle deepfake est None, ignorant la détection")
+        return []
+    print(f"Début de la détection de deepfake avec {len(damage_regions)} régions")
+    detailed_info = []
+    try:
+        # Si pas de régions endommagées, vérifier l'image entière
+        if not damage_regions:
+            print("Pas de régions endommagées, vérification de l'image entière")
+            img_tensor = preprocess_for_vit(image, device)
+            if img_tensor is None:
+                print("Échec du prétraitement de l'image")
+                return []
+            # Exécuter l'inférence - Passer directement le tensor
+            print("Exécution de l'inférence sur l'image entière")
+            with torch.no_grad():
+                outputs = deepfake_model(img_tensor)  # Modèle torchvision attend directement le tensor
+                # Obtenir les prédictions
+                logits = outputs
+                probabilities = torch.nn.functional.softmax(logits, dim=1)
+                # Obtenir les probabilités de classe
+                for i in range(probabilities.shape[1]):
+                    prob = probabilities[0, i].item()
+                    print(f"Probabilité classe {i}: {prob*100:.2f}%")
+                fake_prob = probabilities[0, 1].item()  # Probabilité d'être faux (classe 1)
+                real_prob = probabilities[0, 0].item()  # Probabilité d'être réel (classe 0)
+                is_fake = fake_prob > threshold
+                detailed_info.append(f"Image entière: RÉEL={real_prob*100:.2f}%, FAUX={fake_prob*100:.2f}%")
+                results.append({
+                    "region": "full_image",
+                    "deepfake_prob": float(fake_prob),
+                    "real_prob": float(real_prob),
+                    "is_fake": bool(is_fake),
+                    "detailed_info": detailed_info[-1]
+                })
+            return results
+        # Traiter chaque région endommagée
+        for i, region in enumerate(damage_regions):
+            print(f"Traitement de la région {i}...")
+            x1, y1, x2, y2 = region["box"]
+            x1, y1 = max(0, x1), max(0, y1)
+            x2, y2 = min(image.shape[1], x2), min(image.shape[0], y2)
+            # Ne traiter que les régions valides
+            if x2 > x1 and y2 > y1:
+                # Extraire la région
+                roi = image[y1:y2, x1:x2]
+                print(f"Taille de la région {i}: {roi.shape}")
+                # Prétraiter
+                img_tensor = preprocess_for_vit(roi, device)
+                if img_tensor is None:
+                    print(f"Échec du prétraitement de la région {i}")
+                    continue
+                # Inférence - Passer directement le tensor
+                print(f"Exécution de l'inférence sur la région {i}")
+                with torch.no_grad():
+                    outputs = deepfake_model(img_tensor)  # Modèle torchvision attend directement le tensor
+                    # Obtenir les prédictions
+                    logits = outputs
+                    probabilities = torch.nn.functional.softmax(logits, dim=1)
+                    # Obtenir les probabilités de classe
+                    for j in range(probabilities.shape[1]):
+                        prob = probabilities[0, j].item()
+                        print(f"Région {i} - Probabilité classe {j}: {prob*100:.2f}%")
+                    fake_prob = probabilities[0, 1].item()  # Probabilité d'être faux (classe 1)
+                    real_prob = probabilities[0, 0].item()  # Probabilité d'être réel (classe 0)
+                    is_fake = fake_prob > threshold
+                    region_info = f"Région {i}: RÉEL={real_prob*100:.2f}%, FAUX={fake_prob*100:.2f}%"
+                    detailed_info.append(region_info)
+                    results.append({
+                        "region_id": i,
+                        "box": (x1, y1, x2, y2),
+                        "deepfake_prob": float(fake_prob),
+                        "real_prob": float(real_prob),
+                        "is_fake": bool(is_fake),
+                        "detailed_info": region_info
+                    })
+        # Afficher un résumé global
+        if results:
+            print("===== RÉSUMÉ DE LA DÉTECTION DE DEEPFAKE =====")
+            for info in detailed_info:
+                print(info)
+            fake_regions = sum(1 for r in results if r.get("is_fake", False))
+            print(f"Total des régions analysées: {len(results)}")
+            print(f"Régions fausses détectées: {fake_regions}")
+            print(f"Régions réelles détectées: {len(results) - fake_regions}")
+        else:
+            print("Aucune région n'a été analysée avec succès pour les deepfakes")
+        return results
+    except Exception as e:
+        print(f"Erreur dans la détection de deepfake: {e}")
+        traceback.print_exc()
+        return []
+def visualize_results(image, damage_outputs, deepfake_results, damage_threshold):
+    """Create visualization of results"""
+    try:
+        img_copy = image.copy()
+        # Draw damage detection
+        if damage_outputs is not None and DETECTRON2_AVAILABLE:
+            try:
+                v = Visualizer(img_copy[:, :, ::-1], scale=1.0, instance_mode=ColorMode.IMAGE_BW)
+                v = v.draw_instance_predictions(damage_outputs["instances"].to("cpu"))
+                result_img = v.get_image()[:, :, ::-1]
+                result_img = np.array(result_img, dtype=np.uint8)
+            except Exception as e:
+                print(f"Error visualizing damage: {e}")
+                result_img = img_copy
+        else:
+            result_img = img_copy
+        # Add deepfake results with enhanced information
+        for result in deepfake_results:
+            try:
+                if "box" in result:
+                    x1, y1, x2, y2 = result["box"]
+                    fake_prob = result["deepfake_prob"]
+                    real_prob = result.get("real_prob", 1.0 - fake_prob)  # Calculate real_prob if not present
+                    is_fake = result["is_fake"]
+                    region_id = result.get("region_id", 0)
+                    # Enhanced status text with both probabilities
+                    text = f"R{region_id}: {'FAKE' if is_fake else 'REAL'}"
+                    prob_text = f"F:{fake_prob*100:.1f}% R:{real_prob*100:.1f}%"
+                    # Red for fake, green for real
+                    color = (0, 0, 255) if is_fake else (0, 255, 0)
+                    # Ensure standard numpy array
+                    if not isinstance(result_img, np.ndarray):
+                        result_img = np.array(result_img, dtype=np.uint8)
+                    # Draw rectangle and text
+                    cv2.rectangle(result_img, (x1, y1), (x2, y2), color, 2)
+                    cv2.putText(result_img, text, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.8, color, 2)
+                    cv2.putText(result_img, prob_text, (x1, y1+20), cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
+                elif "region" in result and result["region"] == "full_image":
+                    fake_prob = result["deepfake_prob"]
+                    real_prob = result.get("real_prob", 1.0 - fake_prob)
+                    is_fake = result["is_fake"]
+                    text = f"Image: {'FAKE' if is_fake else 'REAL'}"
+                    prob_text = f"FAKE: {fake_prob*100:.1f}%, REAL: {real_prob*100:.1f}%"
+                    color = (0, 0, 255) if is_fake else (0, 255, 0)
+                    if not isinstance(result_img, np.ndarray):
+                        result_img = np.array(result_img, dtype=np.uint8)
+                    cv2.putText(result_img, text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, color, 2)
+                    cv2.putText(result_img, prob_text, (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
+            except Exception as e:
+                print(f"Error drawing result: {e}")
+        return result_img
+    except Exception as e:
+        print(f"Error in visualization: {e}")
+        traceback.print_exc()
+        return np.array(image, dtype=np.uint8)
 def process_image(input_image, damage_threshold, deepfake_threshold, skip_damage, device_str, usage_count, recipient_email):
     """Process an image through the detection pipeline and send results by email"""
     # Handle empty or None usage_count
     return analysis_text + f"\n\n📧 {email_message}", usage_count, email_message
 def create_gradio_interface():
     # Define a theme
     theme = gr.themes.Soft(
             outputs=[usage_display]
         )
+    return app
+if __name__ == "__main__":
+    # Create and launch the Gradio app
+    app = create_gradio_interface()
+    app.launch(share=False)