imhistory_ensemble_genetic.py

import sys 
sys.path.append(r"D:\Imhistory\ImageHistory\utils")
import tensorflow as tf
from tensorflow import keras
from data_handling import *
from Network import *
from metric_utils3 import *
from Data_Generator import *
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau, ModelCheckpoint,LearningRateScheduler,TensorBoard 
import multiprocessing
import math
from scipy.io import savemat
from functools import partial

# load and save model
from tensorflow.keras.models import Sequential, Model, load_model
from tensorflow.keras.models import model_from_json
import json

# =============================================================================
# Define variables
cc=1 
num_classes = 5  #  ["Adjustment", "Denoising","High-pass","Low-pass","Uniform"] 
batch_size = 16 

epochs =100  
height=512 
width=512 
input_size=(height, width)
n_channels=3
input_shape=(height, width, 3) 
color_mode='rgb'
model_tf_selection=1 
tf_bool=2 #1: pretrained, 2: None 
model_selection=4  # 1: VIT, 2: SE-RESNET50, 3: SE-DENSENET121, 4: SE-INCEPTIONV3 
qf=75  # QF=75,85,95 

transformer_type=2 # 1-b16, 2-b32, 3-l16, 4-l32
lr=0.0002 # learning rate

dataset='UCID'
weights_file = r'D:\Imhistory\ImageHistory\Results\Imhistory_weights.hdf5' 
weights_file2 = r'D:\Imhistory\ImageHistory\Results\Imhistory_weights.hdf5' 
base_path = r'D:\Imhistory\ImageHistory\Results'
path_file = r'D:\Imhistory\ImageHistory\Results'
dataset_path=r'D:\Imhistory\ImageHistory\Datasets\UCID'

# Compile parametreleri
optsecim=1
if optsecim==1: # SGD
    optimizer= SGD(lr=lr, decay=1e-6, momentum=0.9, nesterov=True) # lr=1e-4
elif optsecim==2: #RMSprop
    optimizer=RMSprop(lr=lr, rho=0.9)
elif optsecim==3: # Adam
    optimizer=Adam(lr=lr, beta_1=0.9, beta_2=0.999, amsgrad=False)
elif optsecim==4: # Adam
    optimizer=Adadelta(lr=lr, rho=0.95, epsilon=1e-07)

metric=["categorical_accuracy"]
loss="categorical_crossentropy"
dropbool=1 

valbool=1 # 1 
trainrate=0.9 # training rate
testrate=1.0  # testing rate
valrate=0.1   # validation rate
# ------------------------------

# initialize trainset and test set
x_train, y_train, x_test, y_test, x_val, y_val= [], [], [], [], [], []    
    # transfer train and test set data
imhistclasses = ["Adjustment", "Denoising","Highpass","Lowpass","Uniform"]  
        
training_list, prediction_list, validation_list=[],[],[]
ytraining_list, yprediction_list, yvalidation_list=[],[],[]
for tclass in imhistclasses:
    training, prediction, validation = get_files_ensemble(dataset_path,tclass,trainrate,testrate,valrate,valbool,qf)
    for item in training:
            training_list.append(item)            
            ytraining_list.append(imhistclasses.index(tclass))  
    for item in prediction: 
            prediction_list.append(item)  
            yprediction_list.append(imhistclasses.index(tclass))
    for item in validation:
            validation_list.append(item)
            yvalidation_list.append(imhistclasses.index(tclass))     
    
# Adjust class weights
from sklearn.utils import class_weight
class_weight = class_weight.compute_class_weight('balanced'
                                                   ,classes=np.unique(ytraining_list)
                                                   ,y=ytraining_list)
class_weight_dict = dict(zip(np.unique(ytraining_list), class_weight))
    
print("Eğitim verisi örnek sayısı=",len(ytraining_list))  
print("Test verisi örnek sayısı=",len(yprediction_list)) 
print("Validation örnek sayısı",len(yvalidation_list))  
          
train_generator = Data_Generator(training_list, ytraining_list, batch_size,input_size,n_channels,num_classes,shuffle=False)
step_size_train = train_generator.batchlen #(train_generator.n/ train_generator.batch_size)
       
if valbool==1:
    val_generator = Data_Generator(validation_list, yvalidation_list, batch_size,input_size,n_channels,num_classes,shuffle=False)
    step_size_val = val_generator.batchlen #np.ceil(val_generator.n/ val_generator.batch_size)               
               
test_generator =Data_Generator(prediction_list, yprediction_list, batch_size,input_size,n_channels,num_classes,shuffle=False)
step_size_test=test_generator.batchlen # np.ceil(test_generator.n/ test_generator.batch_size) 
       
# CNN MODEL OLUSTURULUYOR-construct CNN structure
callbacks = [EarlyStopping(monitor='val_categorical_accuracy',
                           patience=15,
                           verbose=1,
                           min_delta=1e-5,
                           mode='max'),             
               ReduceLROnPlateau(monitor='val_loss',
                                 factor=0.005,
                                 patience=10,
                                 verbose=1,
                                 min_delta=1e-5,
                                 mode='auto'),
             ModelCheckpoint(monitor='val_categorical_accuracy',
                             filepath=weights_file,
                             verbose=1,
                             save_best_only=True,
                             save_weights_only=True,
                             mode='max',
                             period=1)]

# Construct CNN structures
model=get_Network(tf_bool,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,transformer_type)
model.compile(loss=loss, optimizer=optimizer, metrics=[metric])
model.summary()
    
model_input = Input(shape=input_shape)        
        
# SEDenseNet121
model_selection=3 
weights_file1 =r"D:\Imhistory\ImageHistory\Ensemble_Models\UCID\Ucid_75_Imhistory_weights1_SEDensenet.hdf5"     
model1=get_Network(tf_bool,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,transformer_type)
model1.load_weights(weights_file1)
model1.compile(loss=loss, optimizer=optimizer, metrics=[metric])
for f, layer in enumerate(model1.layers):
            # layer.name = 'layer_' + str(i)    <-- old way
            layer._name = layer.name + 'sedense121'+str(f)
model1.summary()
      
# SEINCEPTIONV3
model_selection=4   
weights_file1 =r"D:\Imhistory\ImageHistory\Ensemble_Models\UCID\Ucid_75_Imhistory_weights1_SEInception.hdf5"    
model2=get_Network(tf_bool,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,transformer_type)
model2.load_weights(weights_file1)
model2.compile(loss=loss, optimizer=optimizer, metrics=[metric])        
for f, layer in enumerate(model2.layers):
    # layer.name = 'layer_' + str(i)    <-- old way
    layer._name = layer.name + 'seinceptionv3'+str(f)   
        
# SERESNET50
model_selection=2  
weights_file1 =r"D:\Imhistory\ImageHistory\Ensemble_Models\UCID\Ucid_75_Imhistory_weights1_SEResnet.hdf5" 
model3=get_Network(tf_bool,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,transformer_type)
model3.load_weights(weights_file1)
model3.compile(loss=loss, optimizer=optimizer, metrics=[metric])
for f, layer in enumerate(model3.layers):
    # layer.name = 'layer_' + str(i)    <-- old way
    layer._name = layer.name + 'seresnet50'+str(f)       
                   
# TRANSFORMER 
model_selection=1 
weights_file1 =r"D:\Imhistory\ImageHistory\Ensemble_Models\UCID\Ucid_75_Imhistory_weights1_Transformer.hdf5"   
model4=get_Network(tf_bool,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,transformer_type)
model4.load_weights(weights_file1)
model4.compile(loss=loss, optimizer=optimizer, metrics=[metric])
for f, layer in enumerate(model4.layers):
    # layer.name = 'layer_' + str(i)    <-- old way
    layer._name = layer.name + 'transformer'+str(f)     
 

models = [model1, model2, model3, model4]  # Model Lists

############## Train_data
train_pred = [model.predict(train_generator, steps=step_size_train, verbose=1) for model in models]
train_pred = np.array(train_pred)  #(num_models, num_samples, num_classes)
y_train=np.array(ytraining_list)
num_samples_train= len(y_train)
# (4, num_samples, num_classes)
train_pred2 = np.zeros((len(models), num_samples_train, num_classes))
train_pred2 = train_pred[:, :num_samples_train, :]
####################

################# Val_data ###################################################
val_pred = [model.predict(val_generator, steps=step_size_val, verbose=1) for model in models]
val_pred = np.array(val_pred)  # Şekil: (num_models, num_samples, num_classes)
y_val=np.array(yvalidation_list)
num_samples= len(y_val)
val_pred2 = np.zeros((len(models), num_samples, num_classes))
val_pred2 = val_pred[:, :num_samples, :]

################# Test_data ###################################################
test_pred = [model.predict(test_generator, steps=step_size_test, verbose=1) for model in models]
test_pred = np.array(test_pred)  # Şekil: (num_models, num_samples, num_classes)
y_test=np.array(yprediction_list)
num_samples= len(y_test)
test_pred2 = np.zeros((len(models), num_samples, num_classes))
test_pred2 = test_pred[:, :num_samples, :]

########################### GENETIC ALGORITHM #################################
import numpy as np
from deap import base, creator, tools, algorithms
from tensorflow.keras.utils import to_categorical
from scipy.io import savemat
from sklearn.metrics import confusion_matrix
from scipy.stats import pearsonr
import random
import time  
from sklearn.metrics import confusion_matrix, accuracy_score, precision_score, recall_score, f1_score, classification_report
from scipy.stats import pearsonr
import numpy as np

creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)

toolbox = base.Toolbox()
toolbox.register("attr_float", np.random.uniform, 0, 1) 
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_float, 
                 n=len(test_pred2))  
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
toolbox.register("mate", tools.cxSimulatedBinaryBounded, low=0.0, up=1.0, eta=20.0)
toolbox.register("mutate", tools.mutPolynomialBounded, low=0.0, up=1.0, eta=20.0, indpb=1.0/5)
toolbox.register("select", tools.selTournament, tournsize=3)

def normalize(individual):
    for i in range(len(individual)):
        if isinstance(individual[i], complex):
            individual[i] = float(individual[i].real)
    return individual,

def normalize_weights(individual):
    total = sum(individual)
    if total > 0:
        individual = [gene / total for gene in individual]
    return individual
toolbox.register("normalize", normalize_weights)

def evaluate(weights):
    
    weights = np.array(weights) / np.sum(weights)

    weighted_predictions_train = np.tensordot(weights, train_pred2, axes=1)
    y_pred_train = np.argmax(weighted_predictions_train, axis=1)
    accuracy_train = accuracy_score(y_train, y_pred_train)

    weighted_predictions_val = np.tensordot(weights, val_pred2, axes=1)
    y_pred_val = np.argmax(weighted_predictions_val, axis=1)
    accuracy_val = accuracy_score(y_val, y_pred_val)

    individual_accuracies = []
    for i, model_predictions in enumerate(test_pred2):
        y_pred_model = np.argmax(model_predictions, axis=1)
        model_accuracy = accuracy_score(y_test, y_pred_model)
        individual_accuracies.append(model_accuracy)
        print(f"Model {i+1} Accuracy: {model_accuracy:.4f}")

    mean_accuracy = np.mean(individual_accuracies)

    penalties = []
    for i, acc in enumerate(individual_accuracies):
        if acc < mean_accuracy:  
            penalties.append(weights[i] * 0.25) 
        else:
            penalties.append(0)  

    total_penalty = sum(penalties)

    fitness = 0.9 * accuracy_train + 0.1 * accuracy_val - total_penalty
    return fitness,

def evaluate(weights):   
    weights = np.array(weights)
    weights = np.maximum(weights, 0.1) 
    weights = weights / np.sum(weights)  

    weighted_predictions_train = np.tensordot(weights, train_pred2, axes=1)
    y_pred_train = np.argmax(weighted_predictions_train, axis=1)
    accuracy_train = accuracy_score(y_train, y_pred_train)

    weighted_predictions_val = np.tensordot(weights, val_pred2, axes=1)
    y_pred_val = np.argmax(weighted_predictions_val, axis=1)
    accuracy_val = accuracy_score(y_val, y_pred_val)

    fitness = 0.9 * accuracy_train + 0.1 * accuracy_val
    return fitness,

toolbox.register("evaluate", evaluate)

population = toolbox.population(n=100)  
NGEN = 100  
CXPB, MUTPB = 0.5, 0.3  

start_time = time.time()

for gen in range(NGEN):
    offspring = algorithms.varAnd(population, toolbox,cxpb=CXPB, mutpb=MUTPB)

    for ind in offspring:
        ind[:] = normalize_weights(ind)
    
    fits = toolbox.map(toolbox.evaluate, offspring)
    
    for fit, ind in zip(fits, offspring):
        ind.fitness.values = fit

    population = toolbox.select(offspring, k=len(population))
    top_ind = tools.selBest(population, k=1)[0]
   
    print(f"Generation {gen}, Best Fitness: {top_ind.fitness.values[0]}")
    print(f"Best weights: {top_ind[:]}")
    print(f"Sum of best weights: {sum(top_ind)}")  
   
end_time = time.time()
total_time = end_time - start_time    

print("\nOptimization completed in {:.2f} seconds".format(total_time))

best_weights = top_ind[:]
print(f"Final Best weights: {best_weights}")
print(f"Sum of Final Best weights: {sum(best_weights)}")

best_weights = tools.selBest(population, k=1)[0]
print("Best weights:", best_weights)

def ensemble_with_weights(weights, test_predictions):
    weights = np.array(weights) / np.sum(weights)    
    weighted_sum = np.tensordot(weights, test_predictions, axes=1)    
    result = np.argmax(weighted_sum, axis=1)
    return result

ensemble_predictions_train= ensemble_with_weights(best_weights, train_pred2)

ensemble_predictions_test= ensemble_with_weights(best_weights, test_pred2)

best_weights = best_weights / np.sum(best_weights)

print("Normalized Best Weights:", best_weights)
print("Sum of Weights:", np.sum(best_weights))

### SAVE RESULTS ##############################################################
from sklearn.metrics import confusion_matrix, accuracy_score, precision_score, recall_score, f1_score, classification_report
from scipy.stats import pearsonr
import numpy as np

accuracy = accuracy_score(y_test, ensemble_predictions_test)
recall = recall_score(y_test, ensemble_predictions_test, average='macro')  
precision = precision_score(y_test, ensemble_predictions_test, average='macro')  
f1 = f1_score(y_test, ensemble_predictions_test, average='macro')  

correlation, _ = pearsonr(y_test, ensemble_predictions_test)
conf_matrix = confusion_matrix(y_test, ensemble_predictions_test)

specificity_per_class = []
for i in range(conf_matrix.shape[0]):
    true_negatives = np.sum(conf_matrix) - (np.sum(conf_matrix[i, :]) + np.sum(conf_matrix[:, i]) - conf_matrix[i, i])
    false_positives = np.sum(conf_matrix[:, i]) - conf_matrix[i, i]
    specificity = true_negatives / (true_negatives + false_positives)
    specificity_per_class.append(specificity)
specificity_macro = np.mean(specificity_per_class) 

print("\nModel Evaluation Metrics:")
print(f"Accuracy: {accuracy:.4f}")
print(f"Recall (Macro): {recall:.4f}")
print(f"Precision (Macro): {precision:.4f}")
print(f"F1 Score (Macro): {f1:.4f}")
print(f"Correlation Coefficient: {correlation:.4f}")
print(f"Specificity (Macro): {specificity_macro:.4f}")

print("\nConfusion Matrix:")
print(conf_matrix)

report = classification_report(y_test, ensemble_predictions_test, target_names=imhistclasses)
print("\nClassification Report:")
print(report)

y_test_categorical = to_categorical(y_test, num_classes=num_classes)
ensemble_predictions_test_categorical = to_categorical(ensemble_predictions_test, num_classes=num_classes)

y_train_categorical = to_categorical(y_train, num_classes=num_classes)
ensemble_predictions_train_categorical = to_categorical(ensemble_predictions_train, num_classes=num_classes)

r_dict = {
    'ytest': y_test,  
    'ytest_categorical': y_test_categorical,  
    'pred_test': ensemble_predictions_test,  
    'pred_test_categorical': ensemble_predictions_test_categorical,  
    'ytrain': y_train, 
    'ytrain_categorical': y_train_categorical,  
    'pred_train': ensemble_predictions_train, 
    'pred_train_categorical': ensemble_predictions_train_categorical,  
    'best_weights': best_weights,  
    'confusion_matrix': conf_matrix,  
    'metrics': {  
        'accuracy': accuracy,
        'recall_macro': recall,
        'precision_macro': precision,
        'f1_macro': f1,
        'correlation_coefficient': correlation,
        'specificity_macro': specificity_macro,
        'specificity_per_class': specificity_per_class,
    }
}

savemat(path_file+"\\ensemble_results_with_metrics_and_categorical2.mat", r_dict)