google

app.py

@@ -3,334 +3,149 @@ os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
 os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
 
 import numpy as np
-from PIL import Image
 import tensorflow as tf
-from tensorflow.keras.models import load_model, Model
-from tensorflow.keras.layers import Input
+from tensorflow.keras.applications import EfficientNetB7, InceptionResNetV2
+from tensorflow.keras.layers import (
+    Input, GlobalAveragePooling2D,
+    Flatten, Concatenate
+)
+from tensorflow.keras.models import Model
+from vit_keras import vit
 import joblib
 import gradio as gr
 import cv2
-import h5py
-
-from custom_objects import get_custom_objects
 
 # ======================================================
 # CONFIG
 # ======================================================
 IMG_SIZE = 224
+WEIGHTS_PATH = "hybrid_model_weights.h5"
+CLASSIFIER_PATH = "gbdt_model.pkl"
 
-# ======================================================
-# DEBUG HYBRID MODEL
-# ======================================================
-def debug_hybrid_model():
-    """Debug the hybrid_model.keras file"""
-    print("\n🔍 Debugging hybrid_model_weights.h5...")
-    
-    try:
-        # Method 1: Inspect the file directly
-        print("Method 1: Inspecting HDF5 structure...")
-        with h5py.File('hybrid_model_weights.h5', 'r') as f:
-            print("Keys in file:", list(f.keys()))
-            if 'model_weights' in f:
-                print("Model weights groups:", list(f['model_weights'].keys()))
-    except Exception as e:
-        print(f"HDF5 inspection failed: {e}")
-    
-    # Method 2: Try to load with different approaches
-    print("\nMethod 2: Trying different loading strategies...")
-    
-    # Strategy A: Load without custom objects first
-    try:
-        model = tf.keras.models.load_model('hybrid_model_weights.h5', compile=False)
-        print("✓ Loaded without custom objects")
-        return model
-    except Exception as e:
-        print(f"✗ Strategy A failed: {e}")
-    
-    # Strategy B: Try to rebuild from config
-    try:
-        print("\nTrying to rebuild from JSON config...")
-        # Check if there's a JSON config
-        with h5py.File('hybrid_model_weights.h5', 'r') as f:
-            if 'model_config' in f:
-                config = f['model_config'][()]
-                config_str = config.decode('utf-8') if isinstance(config, bytes) else config
-                
-                # Try to load from JSON
-                import json
-                model_config = json.loads(config_str)
-                
-                # Try to create model from config
-                model = tf.keras.models.model_from_json(
-                    config_str,
-                    custom_objects=get_custom_objects()
-                )
-                
-                # Try to load weights
-                model.load_weights('hybrid_model_weights.h5', by_name=True, skip_mismatch=True)
-                print("✓ Rebuilt from config with custom objects")
-                return model
-    except Exception as e:
-        print(f"✗ Strategy B failed: {e}")
-    
-    # Strategy C: Extract just the feature extraction part
-    try:
-        print("\nTrying to extract feature extractor submodel...")
-        # Load the full model first
-        full_model = tf.keras.models.load_model(
-            'hybrid_model_weights.h5',
-            custom_objects=get_custom_objects(),
-            compile=False
-        )
-        
-        # Try to find the feature extractor layer
-        # Common patterns for feature extractors
-        layer_names = [layer.name for layer in full_model.layers]
-        print(f"Available layers: {layer_names}")
-        
-        # Look for feature/dense/flatten layers
-        feature_layer_names = []
-        for name in layer_names:
-            if 'feature' in name.lower() or 'dense' in name or 'flatten' in name or 'global' in name:
-                feature_layer_names.append(name)
-        
-        if feature_layer_names:
-            print(f"Potential feature layers: {feature_layer_names}")
-            # Use the last dense layer before classification
-            for layer_name in reversed(feature_layer_names):
-                try:
-                    extractor = Model(
-                        inputs=full_model.input,
-                        outputs=full_model.get_layer(layer_name).output
-                    )
-                    print(f"✓ Created extractor from layer: {layer_name}")
-                    return extractor
-                except:
-                    continue
-        
-        # If no specific layer found, try to remove classification layers
-        # Assuming the model ends with Dense layers for classification
-        for i, layer in enumerate(reversed(full_model.layers)):
-            if isinstance(layer, tf.keras.layers.Dense) and layer.units <= 2:  # Classification layer
-                # Get output from layer before classification
-                extractor = Model(
-                    inputs=full_model.input,
-                    outputs=full_model.layers[-i-2].output
-                )
-                print(f"✓ Created extractor by removing last {i+1} classification layers")
-                return extractor
-                
-    except Exception as e:
-        print(f"✗ Strategy C failed: {e}")
-    
-    return None
+extractor = None
+classifier = None
 
 # ======================================================
-# FALLBACK EXTRACTOR
+# BUILD HYBRID EXTRACTOR (MUST MATCH TRAINING)
 # ======================================================
-def create_fallback_extractor():
-    """Create fallback extractor if hybrid model fails"""
-    print("\nCreating fallback MobileNetV2 extractor...")
-    
-    base_model = tf.keras.applications.MobileNetV2(
-        input_shape=(IMG_SIZE, IMG_SIZE, 3),
+def build_hybrid_extractor():
+    inputs = Input(shape=(IMG_SIZE, IMG_SIZE, 3), name="input_image")
+
+    eff = EfficientNetB7(
         include_top=False,
-        weights="imagenet",
-        pooling="avg"
+        weights=None,
+        input_tensor=inputs,
+        name="EfficientNetB7_backbone"
     )
-    base_model.trainable = False
-    
-    inputs = Input(shape=(IMG_SIZE, IMG_SIZE, 3))
-    x = tf.keras.applications.mobilenet_v2.preprocess_input(inputs)
-    features = base_model(x, training=False)
-    
-    # Add similar architecture to your hybrid model
-    x = tf.keras.layers.Dense(512, activation="relu")(features)
-    x = tf.keras.layers.Dropout(0.3)(x)
-    x = tf.keras.layers.Dense(256, activation="relu")(x)
-    x = tf.keras.layers.Dense(128, activation="relu")(x)
-    
-    model = Model(inputs, x, name="fallback_extractor")
-    print(f"✓ Fallback extractor created. Output shape: {model.output_shape}")
-    return model
+
+    inc = InceptionResNetV2(
+        include_top=False,
+        weights=None,
+        input_tensor=inputs,
+        name="Inception_backbone"
+    )
+
+    vit_model = vit.vit_b16(
+        image_size=IMG_SIZE,
+        pretrained=False,
+        include_top=False
+    )
+    vit_out = vit_model(inputs)
+
+    f1 = GlobalAveragePooling2D(name="eff_gap")(eff.output)
+    f2 = GlobalAveragePooling2D(name="inc_gap")(inc.output)
+    f3 = Flatten(name="vit_flat")(vit_out)
+
+    features = Concatenate(name="merged_features")([f1, f2, f3])
+
+    return Model(inputs, features, name="HybridExtractor")
 
 # ======================================================
 # LOAD MODELS
 # ======================================================
-extractor, classifier = None, None
-
 def load_models():
     global extractor, classifier
-    
-    print("\n" + "="*50)
-    print("LOADING HYBRID MODEL")
-    print("="*50)
-    
-    # 1. Try to load hybrid model with debugging
-    extractor = debug_hybrid_model()
-    
-    if extractor is None:
-        print("\n❌ Could not load hybrid_model.keras")
-        print("Creating fallback extractor...")
-        extractor = create_fallback_extractor()
-    else:
-        print(f"\n✅ Hybrid model loaded successfully!")
-        print(f"   Input shape: {extractor.input_shape}")
-        print(f"   Output shape: {extractor.output_shape}")
-        print(f"   Number of layers: {len(extractor.layers)}")
-        
-        # Test the extractor
-        print("\n🧪 Testing extractor with random input...")
-        test_input = np.random.randn(1, IMG_SIZE, IMG_SIZE, 3).astype(np.float32)
-        test_output = extractor.predict(test_input, verbose=0)
-        print(f"   Test output shape: {test_output.shape}")
-    
-    # 2. Load classifier
-    print("\n" + "="*50)
-    print("LOADING CLASSIFIER")
-    print("="*50)
-    
-    try:
-        classifier_files = ["gbdt_model.pkl", "classifier.pkl", "rf_model.pkl"]
-        
-        for cf in classifier_files:
-            if os.path.exists(cf):
-                classifier = joblib.load(cf)
-                print(f"✓ Loaded classifier: {cf}")
-                print(f"  Type: {type(classifier).__name__}")
-                
-                # Check if it's a pipeline
-                if hasattr(classifier, 'steps'):
-                    print(f"  Pipeline steps: {[name for name, _ in classifier.steps]}")
-                
-                # Test classifier
-                if extractor is not None:
-                    output_dim = extractor.output_shape[-1]
-                    test_features = np.random.randn(1, output_dim)
-                    test_pred = classifier.predict(test_features)
-                    print(f"  Test prediction: {test_pred[0]}")
-                break
-    except Exception as e:
-        print(f"✗ Classifier loading failed: {e}")
-        
-        # Create simple fallback
-        from sklearn.ensemble import RandomForestClassifier
-        output_dim = extractor.output_shape[-1] if extractor else 128
-        classifier = RandomForestClassifier(n_estimators=50, random_state=42)
-        dummy_features = np.random.randn(100, output_dim)
-        dummy_labels = np.random.randint(0, 2, 100)
-        classifier.fit(dummy_features, dummy_labels)
-        print("✓ Created fallback classifier")
-    
-    print("\n" + "="*50)
-    print("MODELS READY FOR INFERENCE")
-    print("="*50)
+
+    print("\n🚀 Loading Hybrid Extractor...")
+
+    if not os.path.exists(WEIGHTS_PATH):
+        raise FileNotFoundError(f"{WEIGHTS_PATH} not found")
+
+    extractor = build_hybrid_extractor()
+    extractor.load_weights(WEIGHTS_PATH, by_name=True)
+
+    # Sanity check
+    dummy = tf.zeros((1, IMG_SIZE, IMG_SIZE, 3))
+    feat = extractor(dummy)
+    print("✓ Extractor loaded | Feature dim:", feat.shape[-1])
+
+    print("\n🚀 Loading Classifier...")
+
+    if not os.path.exists(CLASSIFIER_PATH):
+        raise FileNotFoundError(f"{CLASSIFIER_PATH} not found")
+
+    classifier = joblib.load(CLASSIFIER_PATH)
+
+    # Verify compatibility
+    test_feat = np.random.randn(1, feat.shape[-1])
+    classifier.predict(test_feat)
+
+    print("✓ Classifier loaded")
 
 # ======================================================
-# PREPROCESSING FOR HYBRID MODEL
+# PREPROCESS IMAGE
 # ======================================================
 def preprocess_image(img):
-    """Preprocess image for the hybrid model"""
-    # Convert to numpy
-    if isinstance(img, Image.Image):
-        img = np.array(img)
-    
-    # Handle different formats
-    if len(img.shape) == 2:  # Grayscale
-        img = np.stack([img] * 3, axis=-1)
-    elif img.shape[2] == 4:  # RGBA
+    if img is None:
+        raise ValueError("No image provided")
+
+    if img.shape[-1] == 4:
         img = img[:, :, :3]
-    
-    # Convert to RGB if needed
-    if img.shape[2] == 3:
-        # Check if BGR (OpenCV)
-        if img[0, 0, 0] > img[0, 0, 2]:  # Blue > Red
-            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
-    
-    # Resize
+
     img = cv2.resize(img, (IMG_SIZE, IMG_SIZE))
-    
-    # Normalize to [0, 1] - common for custom models
     img = img.astype(np.float32) / 255.0
-    
     return img
 
 # ======================================================
 # PREDICTION
 # ======================================================
 def predict(img):
-    """Make prediction using hybrid model"""
+    img = preprocess_image(img)
+    img = np.expand_dims(img, axis=0)
+
+    features = extractor.predict(img, verbose=0)
+    features = features.reshape(features.shape[0], -1)
+
+    pred = classifier.predict(features)[0]
+
     try:
-        # Preprocess
-        img_processed = preprocess_image(img)
-        img_batch = np.expand_dims(img_processed, axis=0)
-        
-        # Extract features
-        features = extractor.predict(img_batch, verbose=0)
-        
-        # Flatten if needed
-        if len(features.shape) > 2:
-            features = features.reshape(features.shape[0], -1)
-        
-        # Classify
-        pred = classifier.predict(features)[0]
-        
-        # Get confidence
-        try:
-            proba = classifier.predict_proba(features)[0]
-            confidence = proba[pred] * 100
-        except:
-            confidence = 80.0  # Default confidence
-        
-        # Return results
-        label = "Real" if pred == 0 else "Fake"
-        return {
-            "Real": confidence if label == "Real" else 100 - confidence,
-            "Fake": confidence if label == "Fake" else 100 - confidence
-        }
-        
-    except Exception as e:
-        print(f"Prediction error: {e}")
-        return {"Real": 50.0, "Fake": 50.0}
+        proba = classifier.predict_proba(features)[0]
+        confidence = float(np.max(proba)) * 100
+    except:
+        confidence = 80.0
+
+    label = "Fake" if pred == 1 else "Real"
+
+    return {
+        "Real": confidence if label == "Real" else 100 - confidence,
+        "Fake": confidence if label == "Fake" else 100 - confidence
+    }
 
 # ======================================================
-# CREATE INTERFACE
+# GRADIO APP
 # ======================================================
-def create_interface():
-    """Create Gradio interface"""
-    # Load models first
+def main():
     load_models()
-    
-    # Create interface
+
     iface = gr.Interface(
         fn=predict,
-        inputs=gr.Image(
-            type="pil",
-            label="Upload Image",
-            image_mode="RGB"
-        ),
-        outputs=gr.Label(
-            num_top_classes=2,
-            label="Prediction"
-        ),
-        title="Hybrid Model Fake Image Detector",
-        description="Using hybrid_model.keras + GBDT classifier",
-        theme=gr.themes.Soft()
+        inputs=gr.Image(type="numpy", label="Upload Image"),
+        outputs=gr.Label(num_top_classes=2),
+        title="Hybrid Fake Image Detector",
+        description="EfficientNet + Inception + ViT + GBDT"
     )
-    
-    return iface
 
-# ======================================================
-# MAIN
-# ======================================================
+    iface.launch(server_name="0.0.0.0", server_port=7860)
+
 if __name__ == "__main__":
-    print("\n🚀 Starting Hybrid Model Detector...")
-    
-    # Create and launch
-    interface = create_interface()
-    
-    interface.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False
-    )
+    main()