Пайплайн по решению задачи классификации изображений на Keras

0. Импорт библиотек

Для работы с облаком в Google Colab:

from google.colab import drive

drive.mount('/content/gdrive')

Для работы с ОС и взаимодействия с локальными файлами:

import os

import shutil

Для математических операций, удобнoй работы с матрицами и массивами:

import numpy as np

Для работы с таблицами:

import pandas as pd

Для визуализации данных и вывода статистики по обучению

import matplotlib.pyplot as plt

Самое важное из TF:

import keras_tuner # !pip install keras_tuner

from tensorflow.keras.models import Sequential, save_model, load_model

from tensorflow.keras.layers import Dense, Embedding, Conv2D, MaxPooling2D, Flatten, BatchNormalization, Dropout, Activation

from tensorflow.keras.utils import load_img

from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard

from tensorflow.keras.preprocessing import image, image_dataset_from_directory

from tensorflow.keras.preprocessing.image import ImageDataGenerator

from keras.preprocessing.image import ImageDataGenerator

from tensorflow.keras.optimizers.schedules import ExponentialDecay

from tensorflow.keras.optimizers import Adam, SGD

Предобученные нейросети:

from tensorflow.keras.applications.densenet import DenseNet121, DenseNet169

from tensorflow.keras.applications.resnet import ResNet101, ResNet50

from tensorflow.keras.applications.efficientnet import EfficientNetB0, EfficientNetB1, EfficientNetB2

1. Работа с файловой системой

Изменение структуры распределения данных (каталог с подкаталогами в каталог с подкаталогами):

final_classes = data['classes'].unique() # инициализация конечного набора классов, в соответствии с которым должны быть распределены данные

start_classes = os.listdir(path) # инициализация стартового набора классов, в соответствии с которым распределены данные изначально

os.mkdir(final_path) # создание главного каталога

for final_class in final_classes: # пробег по всем конечным классам

-os.mkdir(final_path + '/' + final_class) # создание подкаталога для данных одного из конечных классов

-for start_class in start_classes: # пробег по всем начальным классам

--if final_class in start_class: # здесь может быть любое условие, по которому определяется, будут ли данные из выбранного начального подкаталога перенесены в выбранный финальный подкаталог

---files = os.listdir(start_path + '/' + final_class) # получение списка с названиями всех файлов стартового подкаталога

---for file_name in files:

----shutil.move(start_path + '/' + final_class + '/' + file_name, final_path + '/' + final_class + '/' + final_class + '_' + file_name) # перенос файла

Сброс всех данных из подкаталогов в единый каталог (каталог с подкаталогами в каталог):

os.mkdir(final_catalog) # создание конечного каталога

paths = os.listdir(start_catalog) # пути к подкаталогам начального каталога

counter = 0 # счётчик необязателен, если все элементы данных называются по-разному

for path in paths:

-local_paths = os.listdir(start_catalog + '/' + path) # создание списка с элементами подкаталога

-for local_path in local_paths:

--shutil.move(start_catalog + '/' + path + '/' + local_path,  final_catalog + '/' + str(counter) + '.jpg')

--counter += 1

Распределение данных из одного каталога по множеству каталогов (каталог в каталог с подкаталогами):

os.mkdir(final_catalog) # создание конечного каталога

paths = os.listdir(start_catalog) # пути к подкаталогам начального каталога

for path in paths:

-cls = path[0] # определение класса объекта, зависит от условия задачи, может быть любым

-try:

--os.mkdir(final_catalog + '/' + cls) # если подкаталог этого класса еще не создан, создаём

-except FileExistsError:

--pass

-shutil.move(start_catalog + '/' + path,  final_catalog + '/' + cls + '/' + path) # перемещаем элемент

Разбиение датасета на тренировочную и валидационную/тестовую выборки:

paths = os.listdir(train_path)

os.mkdir(valid_path)

split_coef = 0.8 * 10 # коэффициент можно менять, но важно, чтоб оставался только один знак после запятой

counter = 0

for path in paths:

os.mkdir(valid_path + '/' + path)

elems = os.listdir(train_path + '/' + path)

for elem in elems:

if counter % split_coef == 0:

shutil.move(train_path + '/' + path + '/' + elem, valid_path + '/' + path + '/' + elem)

counter += 1

2. Первичная работа с данными

Загрузка таблицы с путями и классами файлов:

data = pd.read_csv(path)

Загрузка и вывод одного изображения:

image_file_name = 'path'

img = image.load_img(image_file_name, target_size=(150, 150))

plt.imshow(img)

Инициализация генератора изображений с расширением набора данных:

train_datagen = ImageDataGenerator(rescale=1. / 255,

rotation_range=40,

width_shift_range=0.2,

height_shift_range=0.2,

zoom_range=0.2,

horizontal_flip=True,

fill_mode='nearest')

Инициализация генератора изображений без расширения набора данных:

test_datagen = ImageDataGenerator(rescale=1. / 255)

Вывод нескольких примеров работы генератора:

image_file_name = '/content/gdrive/MyDrive/kards1/valid_final/0/0.jpg'

img = image.load_img(image_file_name, target_size=(150, 150))

x = np.array([image.img_to_array(img)])

i = 0

for batch in train_datagen.flow(x, batch_size=1):

-plt.figure(i)

-imgplot = plt.imshow(image.array_to_img(batch[0]))

-i += 1

-if i % 4 == 0:

--break

plt.show()

Создание тренировочного набора данных на генераторе:

train_generator = train_datagen.flow_from_directory(

train_path,

target_size=(224, 224),

batch_size=64,

class_mode='categorical') # 'binary' при бинарной классификации

Создание тренировочного статичного набора данных:

static_train = image_dataset_from_directory(

train_path,

color_mode="rgb",

label_mode='categorical', # кодировка лейблов, "binary" при бинарной классификации

batch_size=64,

image_size=(224, 224),

subset='training', # "validation" при создании валидационного набора данных

seed=54,

validation_split=0.2,

labels="inferred") # называть лейблы в соответствии с названиями подкаталогов

3. Аномалии в изображениях

https://machinelearningmastery.ru/anomaly-detection-in-images-777534980aeb/

4. Создание сети

Создание сети через keras_tuner:

def call_existing_code(activation, backbone, optimizator):

if optimizator == 'adam':

-optim = Adam(lr=0.1)

elif optimizator == 'sgdm':

-optim = SGD(lr=0.1, momentum=0.9, nesterov=True)

if backbone == "densenet":

-back = DenseNet121(weights='imagenet', include_top=False, input_shape=(224, 224, 3))

elif backbone == "resnet":

-back = ResNet101(weights='imagenet', include_top=False, input_shape=(224, 224, 3))

elif backbone == "effnet":

-back = EfficientNetB0(weights='imagenet', include_top=False, input_shape=(224, 224, 3))

back.trainable = False # если хотим не обучать backbone

model = Sequential()

model.add(BatchNormalization())

model.add(back)

model.add(Flatten())

model.add(Dense(1000))

model.add(BatchNormalization())

model.add(Activation(activation))

model.add(Dropout(0.4))

model.add(Dense(500))

model.add(BatchNormalization())

model.add(Activation(activation))

model.add(Dropout(0.3))

model.add(Dense(100))

model.add(BatchNormalization())

model.add(Activation(activation))

model.add(Dropout(0.2))

model.add(Dense(52))

model.add(BatchNormalization())

model.add(Activation("softmax"))

model.compile(

optimizer=optim,

loss="categorical_crossentropy",

metrics=["accuracy"],

)

return model

def build_model(hp):

activation = hp.Choice("activation", ["relu", "tanh"])

backbone = hp.Choice("backbone", ["densenet", "resnet", "effnet"])

optimizator = hp.Choice("optimizator", ["adam", "sgdm"])

#dropout = hp.Float("units", min_value=0.2, max_value=0.4, step=0.1)

#lr = hp.Float("lr", min_value=0.01, max_value=0.2, step=0.05)

# call existing model-building code with the hyperparameter values.

model = call_existing_code(

activation=activation, backbone=backbone, optimizator=optimizator #dropout=dropout, lr=lr

)

return model

build_model(keras_tuner.HyperParameters())

Инициализация тюнера:

tuner = keras_tuner.Hyperband(

hypermodel=build_model,

objective="val_accuracy",

max_epochs=5,

factor=3,

hyperband_iterations=2

)

Поиск гиперпараметров (может занимать очень много времени):

tuner.search(static_train, epochs=5, validation_data=static_valid)

Загрузка свёрточной части сети на примере densenet:

densenet_conv = DenseNet121(weights='imagenet', include_top=False, input_shape=(224, 224, 3))

densenet_conv.trainable = False # если хотим не обучать сверточные слои

Добавление базовой полносвязной архитектуры:

model = Sequential()

model.add(densenet_conv)

model.add(Flatten())

model.add(Dense(1000))

model.add(BatchNormalization())

model.add(Activation('relu'))

model.add(Dropout(0.4))

model.add(Dense(500))

model.add(BatchNormalization())

model.add(Activation('relu'))

model.add(Dropout(0.3))

model.add(Dense(100))

model.add(BatchNormalization())

model.add(Activation('relu'))

model.add(Dropout(0.2))

model.add(Dense(13)) # количество классов может меняться в зависимости от задачи

model.add(BatchNormalization())

model.add(Activation('softmax'))

5. Подготовка сети к обучению и обучение

Создание шедулера:

initial_learning_rate = 0.1

lr_schedule = ExponentialDecay(

initial_learning_rate,

decay_steps=100000,

decay_rate=0.96,

staircase=True)

Инициализация оптимизаторов через шедулер:

sgd = SGD(learning_rate=lr_schedule)

adam = Adam(learning_rate=lr_schedule)

Инициализация оптимизаторов без шедулера:

sgd = SGD(lr=0.1, decay=1e-6, momentum=0.9, nesterov=True)

adam = Adam(lr=0.1, decay=1e-6)

Объявление колбеков:

callbacks = [

EarlyStopping(monitor='val_acc', patience=3),

ModelCheckpoint(

filepath_val,

save_weights_only=True,

monitor='val_acc',

mode='max',

save_best_only=True

),

ModelCheckpoint(

filepath_loss,

save_weights_only=True,

monitor='val_loss',

mode='min',

save_best_only=True

),

TensorBoard(log_dir='tb_log', histogram_freq=1) # может ждать озу, тогда удалить

]

Компиляция сети:

model.compile(loss='categorical_crossentropy',

optimizer=sgd,

metrics=['accuracy'])

Обучение сети на генераторах:

history = model.fit(

train_generator,

steps_per_epoch=60,

epochs=10,

callbacks=callbacks,

validation_data=valid_generator,

validation_steps=25)

Обучение сети на статических данных (берет весь датасет за эпоху):

history = model.fit(

static_train,

epochs=10,

verbose="auto",

callbacks=callbacks,

validation_data=static_valid)

6. Анализ качества обучения сети

Вывод графика изменений accuracy:

plt.plot(history.history['accuracy'])

plt.plot(history.history['val_accuracy'])

plt.title('model accuracy')

plt.ylabel('accuracy')

plt.xlabel('epoch')

plt.legend(['train', 'test'], loc='upper left')

plt.show()

Вывод графика изменений loss:

plt.plot(history.history['loss'])

plt.plot(history.history['val_loss'])

plt.title('model loss')

plt.ylabel('loss')

plt.xlabel('epoch')

plt.legend(['train', 'test'], loc='upper left')

plt.show()

Запуск на тестовом наборе размеченных данных для оценки модели:

model.evaluate(test_generator) # тестовый набор создается также, как и валидационный

7. Подготовка данных для предикта и предикт

Загрузка данных в array (если хватает озу):

path = '/content/gdrive/MyDrive/kards1/valid_al'

files = os.listdir(path)

data = []

for file in files:

-elem = path + '/' + file

-img = image.load_img(elem, target_size=(224, 224))

-data.append(image.img_to_array(img))

data = np.array(data)

Декодировка предикта из array (при многоклассовой классификации):

preds = model.predict(data)

decoded_preds = [] # здесь будут лежать декодированные предикты

for i in preds:

-decoded_preds.append(list(i).index(max(i)))

Декодировка предикта из array (при бинарной классификации) (не проверено):

preds = model.predict(data)

decoded_preds = [] # здесь будут лежать декодированные предикты

for i in preds: # если i является float значением от 0 до 1

-if i > 0.5:

--decoded_preds.append(1)

-else:

--decoded_preds.append(0)

Поэлементный предикт при многоклассовой классификации (если не хватает озу):

files = os.listdir(path)

preds = []

for file in files:

-elem = path + '/' + file

-img = image.load_img(elem, target_size=(224, 224))

-x = np.array([image.img_to_array(img)]) # с нормированием я чет не раздуплился, без кажется лучше

-pred = model.predict(x)[0]

-decoded_pred = list(pred).index(max(pred))

-preds.append(decoded_pred)