im_history.py

# -*- coding: utf-8 -*-"
'''
Created on 2025
@author: Dr. Rukiye Karakis
'''
# import folders
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_utils import *
from Data_Generator import *
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau, ModelCheckpoint,LearningRateScheduler,TensorBoard # Erken durdurma için

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 =20 # 
height=512 #
width=512 #
input_size=(height, width)
n_channels=3
input_shape=(height, width, 3) 
color_mode='rgb'
model_tf_selection=1 
model_selection=4 # 1: VIT, 2: SE-RESNET50, 3: SE-DENSENET121, 4: SE-INCEPTIONV3 
transformer_type=2 # 1-b16, 2-b32, 3-l16, 4-l32
qf=75 #  QF=75,85,95 
tf_bool=2 #1: pretrained, 2: None           
lr=0.0002 # learning rate
                                                                          
# Define path
weights_file = r'D:\Imhistory\ImageHistory\Imhistory_weights.hdf5' 
weights_file2 = r'D:\Imhistory\ImageHistory\Imhistory_weights.hdf5' 
path_file=r'D:\Imhistory\ImageHistory\Results' 
dataset_path=r'D:\Imhistory\ImageHistory\Datasets\UCID'

# Compile parameters
optsecim=1
if optsecim==1: # SGD
    optimizer= SGD(lr=lr,  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 # Dropout

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

# -----------------------------------------------------------
for ii in range(cc):    
    # 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(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=True)
    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) 

    # Construct CNN model
    weights_file=path_file+"\\Imhistory_weights.hdf5"
    callbacks = [EarlyStopping(monitor='val_categorical_accuracy',
                           patience=50,
                           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)]
   
    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()
    
    fit = True #
    if fit == True: 
        print("TRAINING STARTED.................")
        if valbool==1: 
           history = model.fit(train_generator, validation_data=val_generator, validation_steps=step_size_val, callbacks=callbacks, 
                                steps_per_epoch=step_size_train, epochs=epochs,  use_multiprocessing=False,shuffle=True) 
        else:
            history = model.fit(train_generator, steps_per_epoch=step_size_train, 
                                epochs=epochs, callbacks=callbacks, use_multiprocessing=False,shuffle=True)
            
        # Save the model
        fnm=path_file+"\\model"+str(ii)+".json"
        model_json = model.to_json()
        with open(fnm, "w") as json_file:
            json_file.write(model_json)
        # serialize weights to HDF5
        fnm1=path_file+"\\sonuc"+str(ii)+".h5"
        model.save_weights(fnm1)
        print("Saved model to disk")
        print("TRAINING FINISHED.................")
    else:
        # LOAD MODEL
        json_file = path_file+'\\model0.json'
        weights_file = path_file+'\\sonuc0.h5'
        model_json = open(json_file, 'r')
        loaded_model_json = model_json.read()
        model_json.close()
        model = model_from_json(loaded_model_json)
        model.load_weights(weights_file)
        model.compile(loss=loss, optimizer=optimizer, metrics=[metric])
        history=0 
    ## ------------------------------   
   

    train_score,test_score,test_pred=[],[],[]
    train_score = model.evaluate(train_generator, steps=step_size_train, verbose=1)    
    print("EĞİTİM DEĞERLENDİRME BİTTİ")        
    test_score = model.evaluate(test_generator, steps=step_size_test, verbose=1)
    print("TEST DEĞERLENDİRME BİTTİ") 
    test_pred= model.predict(test_generator,steps=step_size_test,verbose=1)#,workers=0, max_queue_size=0, use_multiprocessing=False) # generator kapattim-bakalim ne olacak
    print("TEST TAHMİNİ BİTTİ")
    y_test=keras.utils.to_categorical(yprediction_list, num_classes)
    test_pred=test_pred[:len(y_test),:]
    print("TEST TAHMİNİ BİTTİ")    
    val_score=0    
    if valbool==1:
        val_score = model.evaluate(val_generator, steps=step_size_val, verbose=1)   
     
# =====================ROC degerleri======================================     
    # Test the model with last weights
    test_pred2=np.argmax(test_pred,axis=1)
    test_pred2 = np.array(test_pred2, 'float32')
    test_pred3 = keras.utils.to_categorical(test_pred2, num_classes)   
    
    test_result=np.zeros((num_classes, 14))
    test_result,confusion=metric_all(path_file,0,train_score, test_score, valbool,val_score,history,num_classes,y_test,test_pred3) 

    # save test results
    r_dict = {
              'ytest': y_test,
              'pred_test':test_pred3,
              'prediction_list':prediction_list,
              'test_result':test_result,
              'confusion':confusion
              }
    savemat(path_file+"\\imhistorytestresult1.mat", r_dict)
    
    # Test the model with best weights
    mod_selection=1
    if mod_selection==1 & valbool==1:
        drop_bool=0
        batch_bool=0
        model=get_Network(tf_bool,tf_type,num_classes,input_shape,loss, optimizer, metric,height, width,dropbool,model_selection,tf_mode,transformer_type)
        model.summary()  
        weights_file=path_file+"\\Imhistory_weights.hdf5"
        model.load_weights(weights_file)
        train_score,test_score,test_pred=[],[],[]
        #train_score = model.evaluate(train_generator, steps=step_size_train, verbose=1)    
        print("EĞİTİM DEĞERLENDİRME BİTTİ")        
        #test_score = model.evaluate(test_generator, steps=step_size_test, verbose=1)
        print("TEST DEĞERLENDİRME BİTTİ")
        test_pred= model.predict(test_generator,steps=step_size_test,verbose=1)#,workers=0, max_queue_size=0, use_multiprocessing=False) # generator kapattim-bakalim ne olacak
        print("TEST TAHMİNİ BİTTİ")
        y_test=keras.utils.to_categorical(yprediction_list, num_classes)
        test_pred=test_pred[:len(y_test),:]
        print("TEST TAHMİNİ BİTTİ")  
        #val_score = model.evaluate(val_generator, steps=step_size_val, verbose=1) 
    
        print("2nd PREDICTION EVALUATION FINISHED---------------------------------------------")         
        test_pred2=np.argmax(test_pred,axis=1)
        test_pred2 = np.array(test_pred2, 'float32')
        test_pred3 = keras.utils.to_categorical(test_pred2, num_classes)   
    
        test_result=np.zeros((num_classes, 14))
        val_score=0
        test_result,confusion=metric_all(path_file,1,train_score, test_score, valbool,val_score,history,num_classes,y_test,test_pred3) 

        hist_df = pd.DataFrame(history.history)
        hist_df2=hist_df.to_numpy()
        # CSV file 
        hist_csv_file = path_file+'\\history1.csv'
        with open(hist_csv_file, mode='w') as f:
            hist_df.to_csv(f)
        # save test results
        r_dict = {
              'history': hist_df,
              'history2': hist_df2, 
              'ytest': y_test,
              'pred_test':test_pred3,
              'prediction_list':prediction_list,
              'test_result':test_result,
              'confusion':confusion
              }
        savemat(path_file+"\\imhistorytestresult2.mat", r_dict)
        
        #del model
 #---------------------------------------------------------------------------