Демонстрация работы архитектуры

Создание датасета:

# зададим какие преобразования необходимо сделать с каждым изображением

batch_size=32 #кол-во изображений загружаемых за раз

transform = transforms.Compose(

[transforms.Resize((224,224)), #изменим размер изображений

transforms.ToTensor(), #переведем в формат который необходим нейронной сети - тензор

transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5])]) # проведем нормализацию изображения

trainset = torchvision.datasets.ImageFolder(train_dir, transform=transform)

trainloader = torch.utils.data.DataLoader(trainset, batch_size=batch_size, num_workers=2, shuffle=True)

testset = torchvision.datasets.ImageFolder(test_dir, transform=transform)

testloader = torch.utils.data.DataLoader(testset, batch_size=batch_size, num_workers=2, shuffle=True)

classes = list(set(list(csv_file["Class"])))

def imshow(img):

img = img / 2 + 0.5

npimg = img.numpy()

plt.imshow(np.transpose(npimg, (1, 2, 0)))

plt.show()

dataiter = iter(trainloader)

images, labels = next(dataiter)

imshow(torchvision.utils.make_grid(images))

print(' '.join(f'{classes[labels[j]]}' for j in range(batch_size)))

Класс собственной архитектуры:

class Net4(nn.Module):

def __init__(self):

super(Net4,self).__init__()

self.relu = nn.ReLU()

self.conv1 = nn.Conv2d(3, 8, kernel_size =3, stride =1, padding = 0)

self.conv2 = nn.Conv2d(8, 16, kernel_size =3, stride =1, padding= 1)

self.conv3 = nn.Conv2d(16, 32, kernel_size =3, stride =1, padding= 2)

self.conv4 = nn.Conv2d(32, 64, kernel_size =3, stride =1, padding= 3)

self.maxpool = nn.MaxPool2d(2, 2)

self.fc1 = nn.Linear(16384, len(classes))

# Это forward функция, которая определяет структуру сети.

# Здесь мы принимаем только один вход, но можно использовать больше.

def forward(self, x):

x = self.conv1(x)

x = self.relu(x)

x = self.maxpool(x)

x = self.conv2(x)

x = self.relu(x)

x = self.maxpool(x)

x = self.conv3(x)

x = self.relu(x)

x = self.maxpool(x)

x = self.conv4(x)

x = self.maxpool(x)

x = x.view(x.size(0),-1)

x = self.fc1(x)

return torch.log_softmax(x, dim=1)

Мы хотели сделать легковесную модель на основе VGG11.

Обучение с помощью cnn

Мы пытались протестировать различные архитектуры на основе свёрточных нейронных сетей (CNN), но, к сожалению, они давали крайне неудовлетворительный результат.

Мы предполагали, что присваивание меток должно начинаться не с 1, а с 0, вследствие чего мы поменяли значения в CSV-файле. Однако данный способ также не помог увеличить результат. В итоге, мы пришли к выводу, что использование DataLoader без ImageFolder для выборок увеличивает точность модели.

Почему мы используем функцию потерь именно “CrossEntropyLoss”, и оптимизатор “Adam”? С функцией оптимизации “Adam” мы знакомы по практике и по лабораторным работам, а “CrossEntropyLoss” мы используем потому, что она использует как активацию softmax, так и отрицательную логарифмическую вероятность потери.

Код обучающего класса:

from torch.autograd import Variable

# Модифицируем (видимо, не вперый раз) вспомогательный класс под текущую задачу

class Optimization:

"""Optimization is a helper class that allows training, validation, prediction.

Optimization is a helper class that takes model, loss function, optimizer function

learning scheduler (optional), early stopping (optional) as inputs. In return, it

provides a framework to train and validate the models, and to predict future values

based on the models.

Attributes:

model (nn.Module): Model class

loss_fn (torch.nn.modules.Loss): Loss function to calculate the losses

optimizer (torch.optim.Optimizer): Optimizer function to optimize the loss function

train_losses (list[float]): The loss values from the training

val_losses (list[float]): The loss values from the validation

"""

def __init__(self, model, loss_fn, optimizer):

"""

Args:

model (nn.Module): Model class

loss_fn (torch.nn.modules.Loss): Loss function to calculate the losses

optimizer (torch.optim.Optimizer): Optimizer function to optimize the loss function

"""

self.model = model

self.loss_fn = loss_fn

self.optimizer = optimizer

self.train_losses = []

self.val_losses = []

self.err_rates = []

def train_step(self, x, y) -> "loss":

"""The method train_step completes one step of training.

Given the features (x) and the target values (y) tensors, the method completes

one step of the training. First, it activates the train mode to enable back prop.

After generating predicted values (yhat) by doing forward propagation, it calculates

the losses by using the loss function. Then, it computes the gradients by doing

back propagation and updates the weights by calling step() function.

Args:

x (torch.Tensor): Tensor for features to train one step

y (torch.Tensor): Tensor for target values to calculate losses

"""

# Sets model to train mode

self.model.train() # dropout is here, so call it

# Makes predictions

yhat = self.model(x)

# Computes loss

loss = self.loss_fn(yhat.logits, y) # change y hat.logits to y.hat

# Computes gradients

loss.backward()

# Updates parameters and zeroes gradients

self.optimizer.step()

self.optimizer.zero_grad()

# Returns the loss

return loss.item()

def train(self, train_loader, val_loader, n_epochs, test_loader = None) -> None:

"""The method train performs the model training

The method takes DataLoaders for training and validation datasets, batch size for

mini-batch training, number of epochs to train.

Then, it carries out the training by iteratively calling the method train_step for

n_epochs times. If early stopping is enabled, then it checks the stopping condition

to decide whether the training needs to halt before n_epochs steps. Finally, it saves

the model in a designated file path.

Args:

train_loader (torch.utils.data.DataLoader): DataLoader that stores training data

val_loader (torch.utils.data.DataLoader): DataLoader that stores validation data

n_epochs (int): Number of epochs, i.e., train steps, to train

test_loader (torch.utils.data.DataLoader): set it if need error_rate after one epoch

"""

model_path = f'model_{datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S")}'

start_time_gl = time.time()

for epoch in range(1, n_epochs + 1):

start_time_ep = time.time()

gi, N = 0, len(train_loader)

batch_losses = []

print(f"({epoch}/{n_epochs}) {N}: ", end="")

start_time_bt = time.time()

for x_batch, y_batch in train_loader:

x_batch = x_batch.to(device)

y_batch = y_batch.to(device)

loss = self.train_step(x_batch, y_batch)

batch_losses.append(loss)

gi+=1

print(f"_{gi}", end="")

print(f" ({round(time.time() - start_time_bt, 2)} s)")

training_loss = np.mean(batch_losses)

self.train_losses.append(training_loss)

with torch.no_grad():

batch_val_losses = []

for x_val, y_val in val_loader:

x_val = x_val.to(device)

y_val = y_val.to(device)

self.model.eval()

yhat = self.model(x_val)

val_loss = self.loss_fn(yhat, y_val).item()

batch_val_losses.append(val_loss)

validation_loss = np.mean(batch_val_losses)

self.val_losses.append(validation_loss)

print(f"({epoch}/{n_epochs}) Training loss: {training_loss:.4f}\t Validation loss: {validation_loss:.4f}\t", end ="")

print(f" ({round(time.time() - start_time_ep, 2)} s)")

print(f"Train time: ({round(time.time() - start_time_gl, 2)} s)")

torch.save(self.model.state_dict(), model_path)

def evaluate(self, test_loader) -> "(predicted, values)":

"""The method evaluate performs the model evaluation

The method takes DataLoaders for the test dataset, batch size for mini-batch testing,

and number of features as inputs. Similar to the model validation, it iteratively

predicts the target values and calculates losses. Then, it returns two lists that

hold the predictions and the actual values.

Note:

This method assumes that the prediction from the previous step is available at

the time of the prediction, and only does one-step prediction into the future.

Args:

test_loader (torch.utils.data.DataLoader): DataLoader that stores test data

Returns:

list[float]: The values predicted by the model

list[float]: The actual values in the test set.

"""

with torch.no_grad():

predictions = []

values = []

for x_test, y_test in test_loader:

x_test = x_test.to(device)

y_test = y_test.to(device)

self.model.eval()

yhat = self.model(x_test)

predictions.append(yhat.to("cpu").detach().numpy())

values.append(y_test.to("cpu").detach().numpy())

return predictions, values

def check_adequacy(self, test_loader) -> None:

start_time = time.time()

# correct_pred = {classname: 0 for classname in CarsDS.get_classes_list()}

# total_pred = {classname: 0 for classname in CarsDS.get_classes_list()}

correct_pred, total_pred = 0, 0

preds, reals = self.evaluate(test_loader)

for pred_i, real_i in zip(preds, reals):

for pred_i_j, real_i_j in zip(pred_i, real_i):

pred_i_j_one = np.argmax(pred_i_j)

if(pred_i_j_one == real_i_j):

#correct_pred[CarsDS.convert_tensor_to_labels([real_i_j])[0]] += 1

correct_pred += 1

total_pred += 1

#total_pred[CarsDS.convert_tensor_to_labels([real_i_j])[0]] += 1

# accuracy_one_u = 0

# accuracy_one_d = 0

# for classname, correct_count in correct_pred.items():

# if (total_pred[classname] == 0):

# accuracy = 0

# else:

# accuracy = 100 * float(correct_count) / total_pred[classname]

# accuracy_one_u += correct_count

# accuracy_one_d += total_pred[classname]

# print(f'Accuracy for class: {classname:7d} is {accuracy:.1f} %')

#print(f'Accuracy: {100.0*accuracy_one_u/accuracy_one_d:.1f} %')

print(f'Accuracy: {100.0*correct_pred/total_pred:.1f} %')

print(f"Test time: ({round(time.time() - start_time, 2)} s)")

def find_loss(model, valid_loader, lr_min, lr_max, steps=50) -> None:

weight_decay = 0

#dataiter = iter(testloader)

#xs, ys = next(dataiter)

#xs, ys = xs.to(device), ys.to(device)

i, N = 0, steps

X, Y = [], []

print(f"{N}: ", end="")

start_time = time.time()

for i in range(N+1):

cur_lr = lr_min + (lr_max-lr_min)*(i/N)

loss_fn = nn.CrossEntropyLoss()

optimizer = optim.Adam(model.parameters(), lr=cur_lr, weight_decay=weight_decay)

opt = Optimization(model=model, loss_fn=loss_fn, optimizer=optimizer)

#loss = opt.train_step(xs, ys)

loss = 0

for x_batch, y_batch in valid_loader:

x_batch = x_batch.to(device)

y_batch = y_batch.to(device)

loss_l = opt.train_step(x_batch, y_batch)

loss += loss_l

print(f"_{i+1}", end="")

X.append(cur_lr)

Y.append(loss)

print(f" ({round(time.time() - start_time, 2)} s)")

plt.plot(X, Y)

#print(f"X={X}, Y={Y}")

df = pd.DataFrame(data={"lr": X, "loss": Y})

with pd.option_context('display.max_rows', None, 'display.max_columns', None):

print(df)

def plot_losses(self):

"""The method plots the calculated loss values for training and validation

"""

plt.plot(self.train_losses, label="Training loss")

plt.plot(self.val_losses, label="Validation loss")

if(len(self.err_rates) > 0):

plt.plot(self.err_rates, label="Error rate")

plt.legend()

plt.title("Losses")

plt.show()

plt.close()