essentials.py

"""
Essentials module for image processing and prediction
Optimized for CPU usage with batch processing and caching
"""
from timeit import default_timer as timer
from PIL import Image
import tensorflow as tf
import numpy as np
import os

# Configure TensorFlow for CPU optimization
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
tf.config.threading.set_intra_op_parallelism_threads(4)
tf.config.threading.set_inter_op_parallelism_threads(4)


def image_transformer(image_paths, target_size=(32, 32)):
    """
    Transform images to the required format for model prediction
    
    Args:
        image_paths: List of image file paths
        target_size: Tuple of (height, width) for resizing
        
    Returns:
        Tuple of (numpy array of processed images, list of original PIL images)
    """
    images = []
    or_images = []
    
    for image_path in image_paths:
        try:
            # Open and convert image to RGB
            img = Image.open(image_path).convert('RGB')
            or_images.append(img)
            
            # Resize using PIL for better CPU performance
            resized_img = img.resize(target_size, Image.LANCZOS)
            
            # Convert to numpy array and normalize
            img_array = np.array(resized_img, dtype=np.float32) / 255.0
            images.append(img_array)
            
        except Exception as e:
            print(f"Error processing image {image_path}: {str(e)}")
            continue
    
    return np.array(images, dtype=np.float32), or_images


def pred_and_plot_image(
    model: tf.keras.Sequential,
    class_names,
    image_paths,
    transformer=image_transformer,
    batch_size=32):
    """
    Predict classes for multiple images with optimized batch processing
    
    Args:
        model: Trained Keras model
        class_names: List of class names
        image_paths: List of image file paths
        transformer: Image transformation function
        batch_size: Batch size for prediction (optimized for CPU)
        
    Returns:
        Tuple of (probabilities, predicted classes, throughput, total time)
    """
    # Transform images
    np_images, or_images = transformer(image_paths)
    
    if len(np_images) == 0:
        return [], [], 0, 0
    
    start_time = timer()
    
    # Predict with optimized batch size
    pred_probs = model.predict(np_images, batch_size=batch_size, verbose=0)
    
    end_time = timer()
    t_time = end_time - start_time
    
    # Calculate metrics
    y_pred_label = np.argmax(pred_probs, axis=-1)
    throughput = len(or_images) / t_time if t_time > 0 else 0
    pred_classes = [class_names[i][2:] for i in y_pred_label]
    
    # Calculate confidence scores efficiently
    main_list = []
    for pred_prob in pred_probs:
        # Use numpy for faster computation
        score = np.max(pred_prob)
        main_list.append(round(float(score), 4))
    
    return main_list, pred_classes, throughput, t_time