Использование моделей EfficientNet для классификации изображений

# Рисует график обучения def tr_plot(tr_data, start_epoch):     #Plot the training and validation data     tacc = tr_data.history['accuracy']     tloss = tr_data.history['loss']     vacc = tr_data.history['val_accuracy']     vloss = tr_data.history['val_loss']     Epoch_count = len(tacc) + start_epoch     Epochs = []     for i in range (start_epoch, Epoch_count):         Epochs.append(i + 1)     index_loss  = np.argmin(vloss)#  this is the epoch with the lowest validation loss     val_lowest  = vloss[index_loss]     index_acc   = np.argmax(vacc)     acc_highest = vacc[index_acc]     plt.style.use('fivethirtyeight')     sc_label = 'Лучшая эпоха= '+ str(index_loss + 1 + start_epoch)     vc_label = 'Лучшая эпоха= '+ str(index_acc  + 1 + start_epoch)     fig,axes=plt.subplots(nrows=1, ncols = 2, figsize = (20,8))     axes[0].plot (Epochs,tloss, 'r', label = 'Потери при обучении')     axes[0].plot (Epochs,vloss,'g',label='Потери при валидации' )     axes[0].scatter (index_loss + 1 + start_epoch,val_lowest, s = 150, c = 'blue', label = sc_label)     axes[0].set_title('Потери при валидации и обучении')     axes[0].set_xlabel('Эпохи')     axes[0].set_ylabel('Потери')     axes[0].legend()     axes[1].plot (Epochs,tacc,'r', label = 'Точность при обучении')     axes[1].plot (Epochs,vacc,'g', label = 'Точность при валидации')     axes[1].scatter(index_acc + 1 + start_epoch, acc_highest, s = 150, c = 'blue', label = vc_label)     axes[1].set_title  ('Точность при валидации и обучении')     axes[1].set_xlabel ('Эпохи')     axes[1].set_ylabel ('Точность')     axes[1].legend()     plt.tight_layout     #plt.style.use('fivethirtyeight')     plt.show() # Создаем data_frame, в котором название папки является меткой класса изображений (коты и собаки в нашем случае) # Каждому элементу data_frame будет присвоена метка (DX) и название файла (image) # Функция возвращает объект frame def load_data_frame(sdir):     classlist = os.listdir(sdir)     filepaths = []     labels = []     for klass in classlist:         classpath=os.path.join(sdir, klass)         flist=os.listdir(classpath)         for f in flist:             fpath = os.path.join(classpath, f)             filepaths.append( fpath.replace('\\', '/') )             labels.append(klass)      Fseries=pd.Series( filepaths, name = 'image' )     Lseries=pd.Series(labels, name = 'dx')     return pd.concat([Fseries, Lseries], axis=1)

# -*- coding: utf-8 -*- """ Created on Tue Dec 20 02:45:02 2022  @author: Heigrast """ import tensorflow as tf from tensorflow import keras  import numpy as np import pandas as pd  import matplotlib.pyplot as plt import os import seaborn as sns from sklearn.metrics import confusion_matrix, classification_report  import time   # Выводит информацию о процессе обучения def print_info( test_gen, preds, print_code, save_dir, subject ):     class_dict = test_gen.class_indices     labels = test_gen.labels     file_names = test_gen.filenames     error_list = []     true_class = []     pred_class = []     prob_list  = []     new_dict   = {}     error_indices = []     y_pred = []     for key,value in class_dict.items():         new_dict[value] = key             # dictionary {integer of class number: string of class name}     # store new_dict as a text fine in the save_dir     classes = list(new_dict.values())     # list of string of class names     errors = 0     for i, p in enumerate(preds):         pred_index = np.argmax(p)         true_index = labels[i]  # labels are integer values         if pred_index != true_index: # a misclassification has occurred             error_list.append(file_names[i])             true_class.append(new_dict[true_index])             pred_class.append(new_dict[pred_index])             prob_list.append(p[pred_index])             error_indices.append(true_index)             errors = errors + 1         y_pred.append(pred_index)     if print_code != 0:         if errors > 0:             if print_code>errors:                 r = errors             else:                 r = print_code             msg ='{0:^28s}{1:^28s}{2:^28s}{3:^16s}'.format('Filename', 'Predicted Class' , 'True Class', 'Probability')             print_in_color(msg, (0,255,0),(55,65,80))             for i in range(r):                 split1 = os.path.split(error_list[i])                 split2 = os.path.split(split1[0])                 fname  = split2[1] + '/' + split1[1]                 msg    = '{0:^28s}{1:^28s}{2:^28s}{3:4s}{4:^6.4f}'.format(fname, pred_class[i],true_class[i], ' ', prob_list[i])                 print_in_color(msg, (255,255,255), (55,65,60))                 #print(error_list[i]  , pred_class[i], true_class[i], prob_list[i])         else:             msg = 'With accuracy of 100 % there are no errors to print'             print_in_color(msg, (0,255,0),(55,65,80))     if errors > 0:         plot_bar   = []         plot_class = []         for  key, value in new_dict.items():             count = error_indices.count(key)             if count != 0:                 plot_bar.append(count) # list containg how many times a class c had an error                 plot_class.append(value)   # stores the class         fig=plt.figure()         fig.set_figheight(len(plot_class)/3)         fig.set_figwidth(10)         plt.style.use('fivethirtyeight')         for i in range(0, len(plot_class)):             c = plot_class[i]             x = plot_bar[i]             plt.barh(c, x, )             plt.title( ' Errors by Class on Test Set')     y_true = np.array(labels)     y_pred = np.array(y_pred)     if len(classes)<= 30:         # create a confusion matrix         cm = confusion_matrix(y_true, y_pred )         length = len(classes)         if length < 8:             fig_width  = 8             fig_height = 8         else:             fig_width = int(length * .5)             fig_height = int(length * .5)         plt.figure(figsize = (fig_width, fig_height))         sns.heatmap(cm, annot = True, vmin = 0, fmt = 'g', cmap = 'Blues', cbar = False)         plt.xticks(np.arange(length) + .5, classes, rotation = 90)         plt.yticks(np.arange(length) + .5, classes, rotation =0)         plt.xlabel("Predicted")         plt.ylabel("Actual")         plt.title("Confusion Matrix")         plt.show()     clr = classification_report(y_true, y_pred, target_names=classes)     print("Classification Report:\n----------------------\n", clr)  # Сохраняет результаты обучения в файл def saver(save_path, model, model_name, subject, accuracy, img_size, scalar, generator):     # first save the model     save_id = str (model_name +  '-' + subject +'-'+ str(accuracy)[:str(accuracy).rfind('.')+3] + '.h5')     model_save_loc = os.path.join(save_path, save_id)     model.save(model_save_loc)     print_in_color ('model was saved as ' + model_save_loc, (0,255,0),(55,65,80))     # now create the class_df and convert to csv file     class_dict = generator.class_indices     height = []     width = []     scale = []     for i in range(len(class_dict)):         height.append(img_size[0])         width.append(img_size[1])         scale.append(scalar)     Index_series  = pd.Series(list(class_dict.values()), name='class_index')     Class_series  = pd.Series(list(class_dict.keys()), name='class')     Height_series = pd.Series(height, name='height')     Width_series  = pd.Series(width, name='width')     Scale_series  = pd.Series(scale, name='scale by')      class_df = pd.concat([Index_series, Class_series, Height_series, Width_series, Scale_series], axis=1)     csv_name ='class_dict.csv'     csv_save_loc = os.path.join(save_path, csv_name)     class_df.to_csv(csv_save_loc, index=False)     print_in_color ('class csv file was saved as ' + csv_save_loc, (0,255,0),(55,65,80))      return model_save_loc, csv_save_loc  def print_in_color(txt_msg, fore_tupple, back_tupple,):     #prints the text_msg in the foreground color specified by fore_tupple with the background specified by back_tupple     #text_msg is the text, fore_tupple is foreground color tupple (r,g,b), back_tupple is background tupple (r,g,b)     rf,gf,bf = fore_tupple     rb,gb,bb = back_tupple     msg = '{0}' + txt_msg     mat = '\33[38;2;' + str(rf) +';' + str(gf) + ';' + str(bf) + ';48;2;' + str(rb) + ';' +str(gb) + ';' + str(bb) +'m'     print(msg.format(mat), flush = True)     print('\33[0m', flush = True) # returns default print color to back to black     return  # Класс колбеков, он дает возможность выводить доп информацию во время обучения через переопределение методов родителя class LRA(keras.callbacks.Callback):     def __init__(self,model, base_model, patience, stop_patience, threshold, factor, dwell, batches, initial_epoch, epochs, ask_epoch):         super(LRA, self).__init__()         self.model=model         self.base_model = base_model         self.patience = patience # specifies how many epochs without improvement before learning rate is adjusted         self.stop_patience = stop_patience # specifies how many times to adjust lr without improvement to stop training         self.threshold = threshold # specifies training accuracy threshold when lr will be adjusted based on validation loss         self.factor = factor # factor by which to reduce the learning rate         self.dwell = dwell         self.batches = batches # number of training batch to run per epoch         self.initial_epoch = initial_epoch         self.epochs = epochs         self.ask_epoch = ask_epoch         self.ask_epoch_initial = ask_epoch # save this value to restore if restarting training         # callback variables         self.count = 0 # how many times lr has been reduced without improvement         self.stop_count = 0         self.best_epoch = 1   # epoch with the lowest loss         self.initial_lr = float(tf.keras.backend.get_value(model.optimizer.lr)) # get the initial learning rate and save it         self.highest_tracc = 0.0 # set highest training accuracy to 0 initially         self.lowest_vloss = np.inf # set lowest validation loss to infinity initially         self.best_weights = self.model.get_weights() # set best weights to model's initial weights         self.initial_weights = self.model.get_weights()   # save initial weights if they have to get restored      def on_train_begin(self, logs = None):         if self.base_model != None:             status=self.base_model.trainable             if status:                 msg = 'initializing callback starting train with base_model trainable'             else:                 msg = 'initializing callback starting training with base_model not trainable'         else:             msg = 'initialing callback and starting training'         print_in_color (msg, (244, 252, 3), (55,65,80))         msg = '{0:^8s}{1:^10s}{2:^9s}{3:^9s}{4:^9s}{5:^9s}{6:^9s}{7:^10s}{8:^8s}'.format('Epoch', 'Loss', 'Accuracy',                                                                                               'V_loss','V_acc', 'LR', 'Next LR', 'Monitor', 'Duration')         print_in_color(msg, (244,252,3), (55,65,80))         self.start_time = time.time()      def on_train_end(self, logs = None):         stop_time = time.time()         tr_duration = stop_time- self.start_time         hours = tr_duration // 3600         minutes = (tr_duration - (hours * 3600)) // 60         seconds = tr_duration - ((hours * 3600) + (minutes * 60))          self.model.set_weights(self.best_weights) # set the weights of the model to the best weights         msg=f'Training is completed - model is set with weights from epoch {self.best_epoch} '         print_in_color(msg, (0,255,0), (55,65,80))         msg = f'training elapsed time was {str(hours)} hours, {minutes:4.1f} minutes, {seconds:4.2f} seconds)'         print_in_color(msg, (0,255,0), (55,65,80))      def on_train_batch_end(self, batch, logs=None):         acc = logs.get('accuracy')* 100  # get training accuracy         loss = logs.get('loss')         msg = '{0:20s}processing batch {1:4s} of {2:5s} accuracy= {3:8.3f}  loss: {4:8.5f}'.format(' ', str(batch), str(self.batches), acc, loss)         print(msg, '\r', end='') # prints over on the same line to show running batch count      def on_epoch_begin(self,epoch, logs = None):         self.now = time.time()      def on_epoch_end(self, epoch, logs = None):  # method runs on the end of each epoch         later = time.time()         duration = later-self.now         lr=float(tf.keras.backend.get_value(self.model.optimizer.lr)) # get the current learning rate         current_lr = lr         v_loss = logs.get('val_loss')  # get the validation loss for this epoch         acc = logs.get('accuracy')  # get training accuracy         v_acc = logs.get('val_accuracy')         loss = logs.get('loss')         # if training accuracy is below threshold adjust lr based on training accuracy         if acc < self.threshold:             monitor = 'accuracy'             if acc > self.highest_tracc: # training accuracy improved in the epoch                 self.highest_tracc = acc # set new highest training accuracy                 self.best_weights = self.model.get_weights() # traing accuracy improved so save the weights                 self.count = 0 # set count to 0 since training accuracy improved                 self.stop_count = 0 # set stop counter to 0                 if v_loss < self.lowest_vloss:                     self.lowest_vloss = v_loss                 color = (0,255,0)                 self.best_epoch=epoch + 1  # set the value of best epoch for this epoch             else:                 # training accuracy did not improve check if this has happened for patience number of epochs                 # if so adjust learning rate                 if self.count >= self.patience -1: # lr should be adjusted                     color = (245, 170, 66)                     lr = lr * self.factor # adjust the learning by factor                     tf.keras.backend.set_value(self.model.optimizer.lr, lr) # set the learning rate in the optimizer                     self.count = 0 # reset the count to 0                     self.stop_count = self.stop_count + 1 # count the number of consecutive lr adjustments                     self.count = 0 # reset counter                     if self.dwell:                         self.model.set_weights(self.best_weights) # return to better point in N space                     else:                         if v_loss < self.lowest_vloss:                             self.lowest_vloss = v_loss                 else:                     self.count=self.count +1 # increment patience counter         else: # training accuracy is above threshold so adjust learning rate based on validation loss             monitor = 'val_loss'             if v_loss < self.lowest_vloss: # check if the validation loss improved                 self.lowest_vloss = v_loss # replace lowest validation loss with new validation loss                 self.best_weights = self.model.get_weights() # validation loss improved so save the weights                 self.count = 0 # reset count since validation loss improved                 self.stop_count = 0                 color=(0,255,0)                 self.best_epoch = epoch + 1 # set the value of the best epoch to this epoch             else: # validation loss did not improve                 if self.count >= self.patience-1: # need to adjust lr                     color = (245, 170, 66)                     lr = lr * self.factor # adjust the learning rate                     self.stop_count = self.stop_count + 1 # increment stop counter because lr was adjusted                     self.count = 0 # reset counter                     tf.keras.backend.set_value(self.model.optimizer.lr, lr) # set the learning rate in the optimizer                     if self.dwell:                         self.model.set_weights(self.best_weights) # return to better point in N space                 else:                     self.count = self.count + 1 # increment the patience counter                 if acc > self.highest_tracc:                     self.highest_tracc = acc         msg=f'{str(epoch+1):^3s}/{str(self.epochs):4s} {loss:^9.3f}{acc*100:^9.3f}{v_loss:^9.5f}{v_acc*100:^9.3f}{current_lr:^9.5f}{lr:^9.5f}{monitor:^11s}{duration:^8.2f}'         print_in_color (msg,color, (55,65,80))          with open('epoch statistics.txt', 'a') as epoch_stats:             epoch_stats.write(msg + '\n')          if self.stop_count > self.stop_patience - 1: # check if learning rate has been adjusted stop_count times with no improvement             msg = f' training has been halted at epoch {epoch + 1} after {self.stop_patience} adjustments of learning rate with no improvement'             print_in_color(msg, (0,255,255), (55,65,80))             self.model.stop_training = True # stop training         else:             if self.ask_epoch != None:                 if epoch + 1 >= self.ask_epoch:                     msg ='enter H to halt ,F to fine tune model, or an integer for number of epochs to run then ask again'                     print_in_color(msg, (0,255,255), (55,65,80))                     ans=input('')                     if ans =='H' or ans=='h':                         msg =f'training has been halted at epoch {epoch + 1} due to user input'                         print_in_color(msg, (0,255,255), (55,65,80))                         self.model.stop_training = True # stop training                     elif ans == 'F' or ans=='f':                         msg ='setting base_model as trainable for fine tuning of model'                         self.base_model.trainable = True                         print_in_color(msg, (0, 255,255), (55,65,80))                         msg='{0:^8s}{1:^10s}{2:^9s}{3:^9s}{4:^9s}{5:^9s}{6:^9s}{7:^10s}{8:^8s}'.format('Epoch', 'Loss', 'Accuracy',                                                                                               'V_loss','V_acc', 'LR', 'Next LR', 'Monitor', 'Duration')                         print_in_color(msg, (244,252,3), (55,65,80))                         self.count = 0                         self.stop_count = 0                         self.ask_epoch = epoch + 1 + self.ask_epoch_initial                      else:                         ans=int(ans)                         self.ask_epoch += ans                         msg = f' training will continue until epoch ' + str(self.ask_epoch)                         print_in_color(msg, (0, 255,255), (55,65,80))                         msg = '{0:^8s}{1:^10s}{2:^9s}{3:^9s}{4:^9s}{5:^9s}{6:^9s}{7:^10s}{8:^8s}'.format('Epoch', 'Loss', 'Accuracy',                                                                                               'V_loss','V_acc', 'LR', 'Next LR', 'Monitor', 'Duration')                         print_in_color(msg, (244,252,3), (55,65,80))