ensemble_inference.py

import cv2
import numpy as np
import tensorflow as tf
import mediapipe as mp
import json
import os
import time

# Load ensemble metadata
if os.path.exists('ensemble_meta.json'):
    with open('ensemble_meta.json', 'r') as f:
        ensemble_meta = json.load(f)
    model_paths = ensemble_meta['models']
    class_names = ensemble_meta['class_names']
    img_size = tuple(ensemble_meta['img_size'])
else:
    # Fallback to single model
    model_paths = ['asl_model_best.keras']
    if os.path.exists('asl_class_names.json'):
        with open('asl_class_names.json', 'r') as f:
            class_names = json.load(f)
    else:
        class_names = ["A", "B", "C", "D", "Del", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "Nothing",
                      "O", "P", "Q", "R", "S", "Space", "T", "U", "V", "W", "X", "Y", "Z"]
    # Try to infer image size from model
    temp_model = tf.keras.models.load_model(model_paths[0])
    img_size = temp_model.input_shape[1:3]

# Load all models
print(f"Loading {len(model_paths)} models for ensemble prediction...")
models = []
for path in model_paths:
    if os.path.exists(path):
        model = tf.keras.models.load_model(path)
        models.append(model)
        print(f"Loaded model from {path}")
    else:
        print(f"Warning: Model {path} not found")

if not models:
    print("Error: No models could be loaded")
    exit()

print(f"Using classes: {class_names}")
print(f"Input image size: {img_size}")

# Initialize MediaPipe Hands with improved settings
mp_hands = mp.solutions.hands
mp_drawing = mp.solutions.drawing_utils
mp_drawing_styles = mp.solutions.drawing_styles

hands = mp_hands.Hands(
    static_image_mode=False,
    max_num_hands=1,
    min_detection_confidence=0.6,
    min_tracking_confidence=0.6
)

# Enhanced hand preprocessing (same as in main.py)
def preprocess_hand_roi(hand_roi, target_size):
    """
    Preprocess hand ROI for better recognition:
    - Apply contrast enhancement
    - Reduce noise
    - Normalize
    """
    if hand_roi.shape[0] == 0 or hand_roi.shape[1] == 0:
        return np.zeros((*target_size, 3))
        
    # Convert to grayscale for processing
    gray = cv2.cvtColor(hand_roi, cv2.COLOR_BGR2GRAY)
    
    # Apply adaptive histogram equalization for better contrast
    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
    gray = clahe.apply(gray)
    
    # Apply slight Gaussian blur to reduce noise
    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
    
    # Convert back to RGB
    processed = cv2.cvtColor(blurred, cv2.COLOR_GRAY2RGB)
    
    # Resize to target size
    resized = cv2.resize(processed, target_size)
    
    # Normalize
    normalized = resized / 255.0
    
    return normalized

# Ensemble prediction function
def ensemble_predict(models, input_tensor):
    """
    Make prediction using ensemble of models
    """
    predictions = []
    for model in models:
        pred = model.predict(input_tensor, verbose=0)
        predictions.append(pred)
    
    # Average predictions
    avg_pred = np.mean(predictions, axis=0)
    return avg_pred

# Initialize video capture
cap = cv2.VideoCapture(0)
if not cap.isOpened():
    print("Error: Could not open webcam")
    exit()

# For calculating prediction confidence
prediction_queue = []
MAX_QUEUE_SIZE = 15
CONFIDENCE_THRESHOLD = 0.65
TIME_DECAY_FACTOR = 0.9

# For tracking prediction stability
stable_prediction = None
stability_counter = 0
STABILITY_THRESHOLD = 5

# For FPS calculation
prev_frame_time = 0
new_frame_time = 0
fps_values = []

# Main loop
while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        print("Error: Can't receive frame")
        break
        
    # Calculate FPS
    new_frame_time = time.time()
    fps = 1/(new_frame_time - prev_frame_time) if prev_frame_time > 0 else 0
    prev_frame_time = new_frame_time
    fps_values.append(fps)
    if len(fps_values) > 10:
        fps_values.pop(0)
    avg_fps = sum(fps_values) / len(fps_values)
        
    # Flip the frame horizontally for a more intuitive mirror view
    frame = cv2.flip(frame, 1)
    
    # Convert to RGB for MediaPipe
    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    
    # Process the frame to detect hands
    results = hands.process(rgb_frame)
    
    # Draw a blank prediction if no hands detected
    if not results.multi_hand_landmarks:
        cv2.putText(frame, "No hand detected", (50, 50), 
                   cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
        stability_counter = 0  # Reset stability counter
    else:
        # Process each detected hand
        for hand_landmarks in results.multi_hand_landmarks:
            # Draw hand landmarks
            mp_drawing.draw_landmarks(
                frame, 
                hand_landmarks, 
                mp_hands.HAND_CONNECTIONS,
                mp_drawing_styles.get_default_hand_landmarks_style(),
                mp_drawing_styles.get_default_hand_connections_style()
            )
            
            # Calculate bounding box with adaptive padding
            h, w, c = frame.shape
            x_min, y_min = w, h
            x_max, y_max = 0, 0
            
            for landmark in hand_landmarks.landmark:
                x, y = int(landmark.x * w), int(landmark.y * h)
                x_min, y_min = min(x, x_min), min(y, y_min)
                x_max, y_max = max(x_max, x), max(y_max, y)
            
            # Add adaptive padding to bounding box (relative to hand size)
            hand_width = x_max - x_min
            hand_height = y_max - y_min
            
            # Ensure consistent aspect ratio
            aspect_ratio = 1.0  # Square bounding box
            center_x = (x_min + x_max) // 2
            center_y = (y_min + y_max) // 2
            
            # Take the larger dimension
            half_size = max(hand_width, hand_height) // 2
            half_size = int(half_size * 1.3)  # Add some extra padding
            
            # Calculate new bounding box
            x_min = max(center_x - half_size, 0)
            y_min = max(center_y - half_size, 0)
            x_max = min(center_x + half_size, w)
            y_max = min(center_y + half_size, h)
            
            # Draw bounding box
            cv2.rectangle(frame, (x_min, y_min), (x_max, y_max), (0, 255, 0), 2)
            
            # Extract hand ROI
            hand_roi = frame[y_min:y_max, x_min:x_max]
            
            if hand_roi.shape[0] > 0 and hand_roi.shape[1] > 0:
                # Preprocess the hand image
                processed_roi = preprocess_hand_roi(hand_roi, img_size[:2])
                input_tensor = np.expand_dims(processed_roi, axis=0)
                
                # Get ensemble prediction
                prediction = ensemble_predict(models, input_tensor)
                predicted_class_idx = np.argmax(prediction[0])
                confidence = float(prediction[0][predicted_class_idx])
                
                # Get the top 3 predictions for display
                top_indices = np.argsort(prediction[0])[-3:][::-1]
                top_confidences = prediction[0][top_indices]
                
                # Get the predicted label
                if predicted_class_idx < len(class_names):
                    label = class_names[predicted_class_idx]
                else:
                    label = f"Unknown ({predicted_class_idx})"
                
                # Prediction smoothing using a weighted queue
                prediction_queue.append((label, confidence, time.time()))
                if len(prediction_queue) > MAX_QUEUE_SIZE:
                    prediction_queue.pop(0)
                
                # Weighted voting with recency bias
                label_scores = {}
                current_time = time.time()
                
                for l, c, t in prediction_queue:
                    if c >= CONFIDENCE_THRESHOLD:
                        # Apply time decay - newer predictions count more
                        time_weight = TIME_DECAY_FACTOR ** (current_time - t)
                        label_scores[l] = label_scores.get(l, 0) + (c * time_weight)
                
                # Get most common label with sufficient confidence
                if label_scores:
                    most_common_label = max(label_scores, key=label_scores.get)
                    score = label_scores[most_common_label]
                    
                    # Check if prediction is stable
                    if stable_prediction == most_common_label:
                        stability_counter += 1
                    else:
                        stability_counter = 0
                        stable_prediction = most_common_label
                    
                    # Display with confidence color coding
                    color = (0, int(255 * min(score / 2, 1.0)), 0)  # Greener means more confident
                    
                    # Show counts in the queue
                    counts = {}
                    for l, _, _ in prediction_queue:
                        counts[l] = counts.get(l, 0) + 1
                    
                    # Final prediction text
                    if stability_counter >= STABILITY_THRESHOLD:
                        # Highlight stable predictions
                        box_text = f"STABLE: {most_common_label}"
                        cv2.putText(frame, box_text, 
                                   (x_min, y_min - 10), 
                                   cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
                    else:
                        # Show regular prediction
                        box_text = f"{most_common_label} ({counts.get(most_common_label, 0)}/{len(prediction_queue)})"
                        cv2.putText(frame, box_text, 
                                   (x_min, y_min - 10), 
                                   cv2.FONT_HERSHEY_SIMPLEX, 1, color, 2)
                    
                    # Show confidence
                    conf_text = f"Confidence: {confidence:.2f}"
                    cv2.putText(frame, conf_text, 
                               (x_min, y_max + 25), 
                               cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 0, 0), 2)
                    
                    # Show ensemble info
                    cv2.putText(frame, f"Ensemble ({len(models)} models)", 
                               (10, frame.shape[0] - 10), 
                               cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
                    
                    # Show top 3 predictions on the side
                    for i, (idx, conf) in enumerate(zip(top_indices, top_confidences)):
                        top_label = class_names[idx] if idx < len(class_names) else f"Unknown ({idx})"
                        side_text = f"{i+1}. {top_label}: {conf:.2f}"
                        cv2.putText(frame, side_text, 
                                   (10, 30 + i * 30), 
                                   cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 0), 2)
                else:
                    # If no prediction has sufficient confidence
                    cv2.putText(frame, "Uncertain", 
                               (x_min, y_min - 10), 
                               cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
    
    # Display FPS
    cv2.putText(frame, f"FPS: {avg_fps:.1f}", (frame.shape[1] - 120, 30),
               cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2)
    
    # Display the resulting frame
    cv2.imshow('Ensemble Sign Language Recognition', frame)
    
    # Break the loop if 'q' is pressed or window is closed
    if cv2.waitKey(1) & 0xFF == ord('q') or cv2.getWindowProperty('Ensemble Sign Language Recognition', cv2.WND_PROP_VISIBLE) < 1:
        break

# Release resources
cap.release()
cv2.destroyAllWindows()