# -*- coding: utf-8 -*-
__version__ = '1.0.7'
__author__ = "Avinash Kak (kak@purdue.edu)"
__date__ = '2020-March-8'
__url__ = 'https://engineering.purdue.edu/kak/distDLS/DLStudio-1.0.7.html'
__copyright__ = "(C) 2020 Avinash Kak. Python Software Foundation."
import sys,os,os.path
import torch
import torch.nn as nn
import torch.nn.functional as F
import torchvision
import torchvision.transforms as tvt
import torch.optim as optim
from torchsummary import summary
import numpy as np
import re
import math
import random
import copy
import matplotlib.pyplot as plt
import gzip
import pickle
#______________________________ DLStudio Class Definition ________________________________
class DLStudio(object):
def __init__(self, *args, **kwargs ):
if args:
raise ValueError(
'''DLStudio constructor can only be called with keyword arguments for
the following keywords: epochs, learning_rate, batch_size, momentum,
convo_layers_config, image_size, dataroot, path_saved_model, classes,
image_size, convo_layers_config, fc_layers_config, debug_train, use_gpu, and
debug_test''')
learning_rate = epochs = batch_size = convo_layers_config = momentum = None
image_size = fc_layers_config = dataroot = path_saved_model = classes = use_gpu = None
debug_train = debug_test = None
if 'dataroot' in kwargs : dataroot = kwargs.pop('dataroot')
if 'learning_rate' in kwargs : learning_rate = kwargs.pop('learning_rate')
if 'momentum' in kwargs : momentum = kwargs.pop('momentum')
if 'epochs' in kwargs : epochs = kwargs.pop('epochs')
if 'batch_size' in kwargs : batch_size = kwargs.pop('batch_size')
if 'convo_layers_config' in kwargs : convo_layers_config = kwargs.pop('convo_layers_config')
if 'image_size' in kwargs : image_size = kwargs.pop('image_size')
if 'fc_layers_config' in kwargs : fc_layers_config = kwargs.pop('fc_layers_config')
if 'path_saved_model' in kwargs : path_saved_model = kwargs.pop('path_saved_model')
if 'classes' in kwargs : classes = kwargs.pop('classes')
if 'use_gpu' in kwargs : use_gpu = kwargs.pop('use_gpu')
if 'debug_train' in kwargs : debug_train = kwargs.pop('debug_train')
if 'debug_test' in kwargs : debug_test = kwargs.pop('debug_test')
if len(kwargs) != 0: raise ValueError('''You have provided unrecognizable keyword args''')
if dataroot:
self.dataroot = dataroot
if convo_layers_config:
self.convo_layers_config = convo_layers_config
if image_size:
self.image_size = image_size
if fc_layers_config:
self.fc_layers_config = fc_layers_config
if fc_layers_config[0] is not -1:
raise Exception("""\n\n\nYour 'fc_layers_config' construction option is not correct. """
"""The first element of the list of nodes in the fc layer must be -1 """
"""because the input to fc will be set automatically to the size of """
"""the final activation volume of the convolutional part of the network""")
if path_saved_model:
self.path_saved_model = path_saved_model
if classes:
self.class_labels = classes
if learning_rate:
self.learning_rate = learning_rate
else:
self.learning_rate = 1e-6
if momentum:
self.momentum = momentum
if epochs:
self.epochs = epochs
if batch_size:
self.batch_size = batch_size
if use_gpu is not None:
self.use_gpu = use_gpu
if use_gpu is True:
if torch.cuda.is_available():
self.device = torch.device("cuda:0")
else:
raise Exception("You requested GPU support, but there's no GPU on this machine")
else:
self.device = torch.device("cpu")
if debug_train:
self.debug_train = debug_train
else:
self.debug_train = 0
if debug_test:
self.debug_test = debug_test
else:
self.debug_test = 0
self.debug_config = 0
# self.device = torch.device("cuda:0" if torch.cuda.is_available() and self.use_gpu is False else "cpu")
def parse_config_string_for_convo_layers(self):
'''
Each collection of 'n' otherwise identical layers in a convolutional network is
specified by a string that looks like:
"nx[a,b,c,d]-MaxPool(k)"
where
n = num of this type of convo layer
a = number of out_channels [in_channels determined by prev layer]
b,c = kernel for this layer is of size (b,c) [b along height, c along width]
d = stride for convolutions
k = maxpooling over kxk patches with stride of k
Example:
"n1x[a1,b1,c1,d1]-MaxPool(k1) n2x[a2,b2,c2,d2]-MaxPool(k2)"
'''
configuration = self.convo_layers_config
configs = configuration.split()
all_convo_layers = []
image_size_after_layer = self.image_size
for k,config in enumerate(configs):
two_parts = config.split('-')
how_many_conv_layers_with_this_config = int(two_parts[0][:config.index('x')])
if self.debug_config:
print("\n\nhow many convo layers with this config: %d" % how_many_conv_layers_with_this_config)
maxpooling_size = int(re.findall(r'\d+', two_parts[1])[0])
if self.debug_config:
print("\nmax pooling size for all convo layers with this config: %d" % maxpooling_size)
for conv_layer in range(how_many_conv_layers_with_this_config):
convo_layer = {'out_channels':None,
'kernel_size':None,
'convo_stride':None,
'maxpool_size':None,
'maxpool_stride': None}
kernel_params = two_parts[0][config.index('x')+1:][1:-1].split(',')
if self.debug_config:
print("\nkernel_params: %s" % str(kernel_params))
convo_layer['out_channels'] = int(kernel_params[0])
convo_layer['kernel_size'] = (int(kernel_params[1]), int(kernel_params[2]))
convo_layer['convo_stride'] = int(kernel_params[3])
image_size_after_layer = [x // convo_layer['convo_stride'] for x in image_size_after_layer]
convo_layer['maxpool_size'] = maxpooling_size
convo_layer['maxpool_stride'] = maxpooling_size
image_size_after_layer = [x // convo_layer['maxpool_size'] for x in image_size_after_layer]
all_convo_layers.append(convo_layer)
configs_for_all_convo_layers = {i : all_convo_layers[i] for i in range(len(all_convo_layers))}
if self.debug_config:
print("\n\nAll convo layers: %s" % str(configs_for_all_convo_layers))
last_convo_layer = configs_for_all_convo_layers[len(all_convo_layers)-1]
out_nodes_final_layer = image_size_after_layer[0] * image_size_after_layer[1] * \
last_convo_layer['out_channels']
self.fc_layers_config[0] = out_nodes_final_layer
self.configs_for_all_convo_layers = configs_for_all_convo_layers
return configs_for_all_convo_layers
def build_convo_layers(self, configs_for_all_convo_layers):
conv_layers = nn.ModuleList()
in_channels_for_next_layer = None
for layer_index in configs_for_all_convo_layers:
if self.debug_config:
print("\n\n\nLayer index: %d" % layer_index)
in_channels = 3 if layer_index == 0 else in_channels_for_next_layer
out_channels = configs_for_all_convo_layers[layer_index]['out_channels']
kernel_size = configs_for_all_convo_layers[layer_index]['kernel_size']
padding = tuple((k-1) // 2 for k in kernel_size)
stride = configs_for_all_convo_layers[layer_index]['convo_stride']
maxpool_size = configs_for_all_convo_layers[layer_index]['maxpool_size']
if self.debug_config:
print("\n in_channels=%d out_channels=%d kernel_size=%s stride=%s \
maxpool_size=%s" % (in_channels, out_channels, str(kernel_size), str(stride),
str(maxpool_size)))
conv_layers.append( nn.Conv2d( in_channels,out_channels,kernel_size,stride=stride,padding=padding) )
conv_layers.append( nn.MaxPool2d( maxpool_size ) )
conv_layers.append( nn.ReLU() ),
in_channels_for_next_layer = out_channels
return conv_layers
def build_fc_layers(self):
fc_layers = nn.ModuleList()
for layer_index in range(len(self.fc_layers_config) - 1):
fc_layers.append( nn.Linear( self.fc_layers_config[layer_index],
self.fc_layers_config[layer_index+1] ) )
return fc_layers
def load_cifar_10_dataset(self):
'''
We make sure that the transformation applied to the image end the images being normalized.
Consider this call to normalize: "Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))". The three
numbers in the first tuple affect the means in the three color channels and the three
numbers in the second tuple affect the standard deviations. In this case, we want the
image value in each channel to be changed to:
image_channel_val = (image_channel_val - mean) / std
So with mean and std both set 0.5 for all three channels, if the image tensor originally
was between 0 and 1.0, after this normalization, the tensor will be between -1.0 and +1.0.
If needed we can do inverse normalization by
image_channel_val = (image_channel_val * std) + mean
'''
## The call to ToTensor() converts the usual int range 0-255 for pixel values to 0-1.0 float vals
## But then the call to Normalize() changes the range to -1.0-1.0 float vals.
transform = tvt.Compose([tvt.ToTensor(),
tvt.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))]) ## accuracy: 51%
## Define where the training and the test datasets are located:
train_data_loc = torchvision.datasets.CIFAR10(root=self.dataroot, train=True,
download=True, transform=transform)
test_data_loc = torchvision.datasets.CIFAR10(root=self.dataroot, train=False,
download=True, transform=transform)
## Now create the data loaders:
self.train_data_loader = torch.utils.data.DataLoader(train_data_loc,batch_size=self.batch_size,
shuffle=True, num_workers=2)
self.test_data_loader = torch.utils.data.DataLoader(test_data_loc,batch_size=self.batch_size,
shuffle=False, num_workers=2)
def load_cifar_10_dataset_with_augmentation(self):
'''
In general, we want to do data augmentation for training:
'''
transform_train = tvt.Compose([
tvt.RandomCrop(32, padding=4),
tvt.RandomHorizontalFlip(),
tvt.ToTensor(),
# tvt.Normalize((0.20, 0.20, 0.20), (0.20, 0.20, 0.20))])
tvt.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])
## Don't need any augmentation for the test data:
transform_test = tvt.Compose([
tvt.ToTensor(),
# tvt.Normalize((0.20, 0.20, 0.20), (0.20, 0.20, 0.20))])
tvt.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])
## Define where the training and the test datasets are located
train_data_loc = torchvision.datasets.CIFAR10(
root=self.dataroot, train=True, download=True, transform=transform_train)
test_data_loc = torchvision.datasets.CIFAR10(
root=self.dataroot, train=False, download=True, transform=transform_test)
## Now create the data loaders:
self.train_data_loader = torch.utils.data.DataLoader(train_data_loc, batch_size=self.batch_size,
shuffle=True, num_workers=2)
self.test_data_loader = torch.utils.data.DataLoader(test_data_loc, batch_size=self.batch_size,
shuffle=False, num_workers=2)
def imshow(self, img):
'''
called by display_tensor_as_image() for displaying the image
'''
img = img / 2 + 0.5 # unnormalize
npimg = img.numpy()
plt.imshow(np.transpose(npimg, (1, 2, 0)))
plt.show()
class Net(nn.Module):
def __init__(self, convo_layers, fc_layers):
super(DLStudio.Net, self).__init__()
self.my_modules_convo = convo_layers
self.my_modules_fc = fc_layers
def forward(self, x):
for m in self.my_modules_convo:
x = m(x)
x = x.view(x.size(0), -1)
for m in self.my_modules_fc:
x = m(x)
return x
def show_network_summary(self, net):
print("\n\n\nprinting out the model:")
print(net)
print("\n\n\na summary of input/output for the model:")
summary(net, (3,self.image_size[0],self.image_size[1]),-1, device='cpu')
def run_code_for_training(self, net):
filename_for_out = "performance_numbers_" + str(self.epochs) + ".txt"
FILE = open(filename_for_out, 'w')
net = copy.deepcopy(net)
net = net.to(self.device)
'''
We will use torch.nn.CrossEntropyLoss for the loss function. Assume that the vector
x corresponds to the values at the 10 output nodes. We will interpret normalized versions
of these values as probabilities --- the normalization being as shown inside the square
brackets below. Let 'class' be the true class for the input --- remember 'class' in an
integer index in range(10). If our classification was absolutely correct, the NORMALIZED
value for x[class], with normalization being carried out by the ratio inside the square
brackets, would be 1 and x would be zero at the other nine positions in the vector.
In this case, the ratio inside the brackets shown below would be 1.0 and the log of
that would be 0. That is, when a correct classification decision is made, the value for
CrossEntropyLoss would be zero. On other hand, when an incorrect decision is made
and we examine the value of the same element x[class], it will DEFINITELY be less
than 1 and possibly even 0. The closer x[class] is to zero, the larger the value for
CrossEntropyLoss shown below.
_ _
| exp( x[class] ) |
CrossEntropyLoss(x, class) = - log | --------------------- |
|_ \sum_j exp( x[j] ) _|
Note that "exp( x[class])" is always positive and, by normalizing it with the
summation in the denominator, the quantity inside the square brackets is guaranteed
to be in the range [0,1.0]. Since the log of a fraction is always negative, the
value calculated for the CrossEntropyLoss when the label assigned to an input is
'class' will always be a positive number in the range [0, +inf). In summary, the loss
is zero when the output classification is correct and some large positive number when
the classification is wrong.
'''
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(net.parameters(), lr=self.learning_rate, momentum=self.momentum)
for epoch in range(self.epochs):
## We will use running_loss to accumulate the losses over 2000 batches in order
## to present an averaged (over 2000) loss to the user.
print("\n")
running_loss = 0.0
for i, data in enumerate(self.train_data_loader):
inputs, labels = data
if self.debug_train and i % 2000 == 1999:
print("\n\n[iter=%d:] Ground Truth: " % (i+1) +
' '.join('%5s' % self.class_labels[labels[j]] for j in range(self.batch_size)))
inputs = inputs.to(self.device)
labels = labels.to(self.device)
## Since PyTorch likes to construct dynamic computational graphs, we need to
## zero out the previously calculated gradients for the learnable parameters:
optimizer.zero_grad()
# Make the predictions with the model:
outputs = net(inputs)
loss = criterion(outputs, labels)
if self.debug_train and i % 2000 == 1999:
_, predicted = torch.max(outputs.data, 1)
print("[iter=%d:] Predicted Labels: " % (i+1) +
' '.join('%5s' % self.class_labels[predicted[j]] for j in range(self.batch_size)))
self.display_tensor_as_image(torchvision.utils.make_grid(inputs, normalize=True),
"see terminal for TRAINING results at iter=%d" % (i+1))
loss.backward()
optimizer.step()
## Present to the average value of the loss over the past 2000 batches:
running_loss += loss.item()
if i % 2000 == 1999:
# print("[epoch:%d, batch:%5d] loss: %.3f" % (epoch + 1, i + 1, running_loss / float(2000)))
avg_loss = running_loss / float(2000)
print("[epoch:%d, batch:%5d] loss: %.3f" % (epoch + 1, i + 1, avg_loss))
FILE.write("%.3f\n" % avg_loss)
FILE.flush()
running_loss = 0.0
print("\nFinished Training\n")
self.save_model(net)
def display_tensor_as_image(self, tensor, title=""):
'''
This method converts the argument tensor into a photo image that you can display
in your terminal screen. It can convert tensors of three different shapes
into images: (3,H,W), (1,H,W), and (H,W), where H, for height, stands for the
number of pixels in the vertical direction and W, for width, for the same
along the horizontal direction. When the first element of the shape is 3,
that means that the tensor represents a color image in which each pixel in
the (H,W) plane has three values for the three color channels. On the other
hand, when the first element is 1, that stands for a tensor that will be
shown as a grayscale image. And when the shape is just (H,W), that is
automatically taken to be for a grayscale image.
'''
tensor_range = (torch.min(tensor).item(), torch.max(tensor).item())
if tensor_range == (-1.0,1.0):
## The tensors must be between 0.0 and 1.0 for the display:
print("\n\n\nimage un-normalization called")
tensor = tensor/2.0 + 0.5 # unnormalize
plt.figure(title)
### The call to plt.imshow() shown below needs a numpy array. We must also
### transpose the array so that the number of channels (the same thing as the
### number of color planes) is in the last element. For a tensor, it would be in
### the first element.
if tensor.shape[0] == 3 and len(tensor.shape) == 3:
# plt.imshow( tensor.numpy().transpose(1,2,0) )
plt.imshow( tensor.numpy().transpose(1,2,0) )
### If the grayscale image was produced by calling torchvision.transform's
### ".ToPILImage()", and the result converted to a tensor, the tensor shape will
### again have three elements in it, however the first element that stands for
### the number of channels will now be 1
elif tensor.shape[0] == 1 and len(tensor.shape) == 3:
tensor = tensor[0,:,:]
plt.imshow( tensor.numpy(), cmap = 'gray' )
### For any one color channel extracted from the tensor representation of a color
### image, the shape of the tensor will be (W,H):
elif len(tensor.shape) == 2:
plt.imshow( tensor.numpy(), cmap = 'gray' )
else:
sys.exit("\n\n\ntensor for image is ill formed -- aborting")
plt.show()
def check_a_sampling_of_images(self):
'''
Displays the first batch_size number of images in your dataset.
'''
dataiter = iter(self.train_data_loader)
images, labels = dataiter.next()
# Since negative pixel values make no sense for display, setting the 'normalize'
# option to True will change the range back from (-1.0,1.0) to (0.0,1.0):
self.display_tensor_as_image(torchvision.utils.make_grid(images, normalize=True))
# Print class labels for the images shown:
print(' '.join('%5s' % self.class_labels[labels[j]] for j in range(self.batch_size)))
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.path_saved_model)
def run_code_for_testing(self, net):
net.load_state_dict(torch.load(self.path_saved_model))
## In what follows, in addition to determining the predicted label for each test
## image, we will also compute some stats to measure the overall performance of
## the trained network. This we will do in two different ways: For each class,
## we will measure how frequently the network predicts the correct labels. In
## we will compute the confusion matrix for the predictions.
correct = 0
total = 0
confusion_matrix = torch.zeros(len(self.class_labels), len(self.class_labels))
class_correct = [0] * len(self.class_labels)
class_total = [0] * len(self.class_labels)
with torch.no_grad():
for i,data in enumerate(self.test_data_loader):
## data is set to the images and the labels for one batch at a time:
images, labels = data
if self.debug_test and i % 1000 == 0:
print("\n\n[i=%d:] Ground Truth: " %i + ' '.join('%5s' % self.class_labels[labels[j]]
for j in range(self.batch_size)))
outputs = net(images)
## max() returns two things: the max value and its index in the 10 element
## output vector. We are only interested in the index --- since that is
## essentially the predicted class label:
_, predicted = torch.max(outputs.data, 1)
if self.debug_test and i % 1000 == 0:
print("[i=%d:] Predicted Labels: " %i + ' '.join('%5s' % self.class_labels[predicted[j]]
for j in range(self.batch_size)))
self.display_tensor_as_image(torchvision.utils.make_grid(images, normalize=True),
"see terminal for test results at i=%d" % i)
for label,prediction in zip(labels,predicted):
confusion_matrix[label][prediction] += 1
total += labels.size(0)
correct += (predicted == labels).sum().item()
## comp is a list of size batch_size of "True" and "False" vals
comp = predicted == labels
for j in range(self.batch_size):
label = labels[j]
## The following works because, in a numeric context, the boolean value
## "False" is the same as number 0 and the boolean value True is the
## same as number 1. For that reason "4 + True" will evaluate to 5 and
## "4 + False" will evaluate to 4. Also, "1 == True" evaluates to "True"
## "1 == False" evaluates to "False". However, note that "1 is True"
## evaluates to "False" because the operator "is" does not provide a
## numeric context for "True". And so on. In the statement that follows,
## while c[j].item() will either return "False" or "True", for the
## addition operator, Python will use the values 0 and 1 instead.
class_correct[label] += comp[j].item()
class_total[label] += 1
for j in range(len(self.class_labels)):
print('Prediction accuracy for %5s : %2d %%' % (
self.class_labels[j], 100 * class_correct[j] / class_total[j]))
print("\n\n\nOverall accuracy of the network on the 10000 test images: %d %%" %
(100 * correct / float(total)))
print("\n\nDisplaying the confusion matrix:\n")
out_str = " "
for j in range(len(self.class_labels)): out_str += "%7s" % self.class_labels[j]
print(out_str + "\n")
for i,label in enumerate(self.class_labels):
out_percents = [100 * confusion_matrix[i,j] / float(class_total[i])
for j in range(len(self.class_labels))]
out_percents = ["%.2f" % item.item() for item in out_percents]
out_str = "%6s: " % self.class_labels[i]
for j in range(len(self.class_labels)): out_str += "%7s" % out_percents[j]
print(out_str)
################## Start Definition of Inner Class ExperimentsWithSequential ##############
class ExperimentsWithSequential(nn.Module):
"""
Demonstrates how to use the torch.nn.Sequential container class
"""
def __init__(self, dl_studio ):
super(DLStudio.ExperimentsWithSequential, self).__init__()
self.dl_studio = dl_studio
def load_cifar_10_dataset(self):
self.dl_studio.load_cifar_10_dataset()
def load_cifar_10_dataset_with_augmentation(self):
self.dl_studio.load_cifar_10_dataset_with_augmentation()
class Net(nn.Module):
"""
To see if the DLStudio class would work with any network that a user may want
to experiment with, I copy-and-pasted the the network shown below from the following
page by Zhenye at GitHub:
https://zhenye-na.github.io/2018/09/28/pytorch-cnn-cifar10.html
"""
def __init__(self):
super(DLStudio.ExperimentsWithSequential.Net, self).__init__()
self.conv_seqn = nn.Sequential(
# Conv Layer block 1:
nn.Conv2d(in_channels=3, out_channels=32, kernel_size=3, padding=1),
nn.BatchNorm2d(32),
nn.ReLU(inplace=True),
nn.Conv2d(in_channels=32, out_channels=64, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2),
# Conv Layer block 2:
nn.Conv2d(in_channels=64, out_channels=128, kernel_size=3, padding=1),
nn.BatchNorm2d(128),
nn.ReLU(inplace=True),
nn.Conv2d(in_channels=128, out_channels=128, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2),
nn.Dropout2d(p=0.05),
# Conv Layer block 3:
nn.Conv2d(in_channels=128, out_channels=256, kernel_size=3, padding=1),
nn.BatchNorm2d(256),
nn.ReLU(inplace=True),
nn.Conv2d(in_channels=256, out_channels=256, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2),
)
self.fc_seqn = nn.Sequential(
nn.Dropout(p=0.1),
nn.Linear(4096, 1024),
nn.ReLU(inplace=True),
nn.Linear(1024, 512),
nn.ReLU(inplace=True),
nn.Dropout(p=0.1),
nn.Linear(512, 10)
)
def forward(self, x):
x = self.conv_seqn(x)
# flatten
x = x.view(x.size(0), -1)
x = self.fc_seqn(x)
return x
def run_code_for_training(self, net):
self.dl_studio.run_code_for_training(net)
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.dl_studio.path_saved_model)
def run_code_for_testing(self, model):
self.dl_studio.run_code_for_testing(model)
################## Start Definition of Inner Class ExperimentsWithCIFAR ##############
class ExperimentsWithCIFAR(nn.Module):
def __init__(self, dl_studio ):
super(DLStudio.ExperimentsWithCIFAR, self).__init__()
self.dl_studio = dl_studio
def load_cifar_10_dataset(self):
self.dl_studio.load_cifar_10_dataset()
def load_cifar_10_dataset_with_augmentation(self):
self.dl_studio.load_cifar_10_dataset_with_augmentation()
## You can instantiate two different types when experimenting with the inner class
## ExperimentsWithCIFAR. The network shown below is from the PyTorch tutorial
##
## https://pytorch.org/tutorials/beginner/blitz/cifar10_tutorial.html
##
class Net(nn.Module):
def __init__(self):
super(DLStudio.ExperimentsWithCIFAR.Net, self).__init__()
self.conv1 = nn.Conv2d(3, 6, 5)
self.pool = nn.MaxPool2d(2, 2)
self.conv2 = nn.Conv2d(6, 16, 5)
self.fc1 = nn.Linear(16 * 5 * 5, 120)
self.fc2 = nn.Linear(120, 84)
self.fc3 = nn.Linear(84, 10)
def forward(self, x):
x = self.pool(F.relu(self.conv1(x)))
x = self.pool(F.relu(self.conv2(x)))
x = x.view(-1, 16 * 5 * 5)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
x = self.fc3(x)
return x
## Instead of using the network shown above, you can also use the network shown below.
## if you are playing with the ExperimentsWithCIFAR inner class. If that's what you
## want to do, in the script "playing_with_cifar10.py" in the Examples directory,
## you will need to replace the statement
## model = exp_cifar.Net()
## by the statement
## model = exp_cifar.Net2()
##
class Net2(nn.Module):
def __init__(self):
"""
I created this network class just to see if it was possible to simply calculate
the size of the first of the fully connected layers from strides in the convo
layers up to that point and from the out_channels used in the top-most convo
layer. In what you see below, I am keeping track of all the strides by pushing
them into the array 'strides'. Subsequently, in the formula shown in line (A),
I use the product of all strides and the number of out_channels for the topmost
layer to compute the size of the first fully-connected layer.
"""
super(DLStudio.ExperimentsWithCIFAR.Net2, self).__init__()
self.relu = nn.ReLU()
strides = []
patch_size = 2
## conv1:
out_ch, ker_size, conv_stride, pool_stride = 128,5,1,2
self.conv1 = nn.Conv2d(3, out_ch, (ker_size,ker_size), padding=(ker_size-1)//2)
self.pool1 = nn.MaxPool2d(patch_size, pool_stride)
strides += (conv_stride, pool_stride)
## conv2:
in_ch = out_ch
out_ch, ker_size, conv_stride, pool_stride = 128,3,1,2
self.conv2 = nn.Conv2d(in_ch, out_ch, ker_size, padding=(ker_size-1)//2)
self.pool2 = nn.MaxPool2d(patch_size, pool_stride)
strides += (conv_stride, pool_stride)
## conv3:
## meant for repeated invocation, must have same in_ch, out_ch and strides of 1
in_ch = out_ch
out_ch, ker_size, conv_stride, pool_stride = in_ch,2,1,1
self.conv3 = nn.Conv2d(in_ch, out_ch, ker_size, padding=1)
self.pool3 = nn.MaxPool2d(patch_size, pool_stride)
# strides += (conv_stride, pool_stride)
## figure out the number of nodes needed for entry into fc:
in_size_for_fc = out_ch * (32 // np.prod(strides)) ** 2 ## (A)
self.in_size_for_fc = in_size_for_fc
self.fc1 = nn.Linear(in_size_for_fc, 150)
self.fc2 = nn.Linear(150, 100)
self.fc3 = nn.Linear(100, 10)
def forward(self, x):
## We know that forward() begins its with work x shaped as (4,3,32,32) where
## 4 is the batch size, 3 in_channels, and where the input image size is 32x32.
x = self.relu(self.conv1(x))
x = self.pool1(x)
x = self.relu(self.conv2(x))
x = self.pool2(x)
for _ in range(5):
x = self.pool3(self.relu(self.conv3(x)))
x = x.view(-1, self.in_size_for_fc)
x = self.relu(self.fc1( x ))
x = self.relu(self.fc2( x ))
x = self.fc3(x)
return x
def run_code_for_training(self, net):
self.dl_studio.run_code_for_training(net)
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.dl_studio.path_saved_model)
def run_code_for_testing(self, model):
self.dl_studio.run_code_for_testing(model)
################# Start Definition of Inner Class AutogradCustomization #############
class AutogradCustomization(nn.Module):
"""
This class illustrates how you can add additional functionality of Autograd by
following the instructions posted at
https://pytorch.org/docs/stable/notes/extending.html
"""
def __init__(self, dl_studio, num_samples_per_class):
super(DLStudio.AutogradCustomization, self).__init__()
self.dl_studio = dl_studio
self.num_samples_per_class = num_samples_per_class
class DoSillyWithTensor(torch.autograd.Function):
"""
Extending Autograd requires that you define a new verb class, as I have with
the class DoSillyWithTensor shown below, with definitions for two static
methods, "forward()" and "backward()". An instance constructed from this
class is callable. So when, in the "forward()" of the network, you pass a
training sample through an instance of DoSillyWithTensor, it is subject to
the code shown below in the "forward()" of this class.
"""
@staticmethod
def forward(ctx, input):
"""
The parameter 'input' is set to the training sample that is being
processed by an instance of DoSillyWithTensor in the "forward()" of a
network. We first make a deep copy of this tensor (which should be a
32-bit float) and then we subject the copy to a conversion to a one-byte
integer, which should cause a significant loss of information. We
calculate the difference between the original 32-bit float and the 8-bit
version and store it away in the context variable "ctx".
"""
input_orig = input.clone().double()
input = input.to(torch.uint8).double()
diff = input_orig.sub(input)
ctx.save_for_backward(diff)
return input
@staticmethod
def backward(ctx, grad_output):
"""
Whatever was stored in the context variable "ctx" during the forward pass
can be retrieved in the backward pass as shown below.
"""
diff, = ctx.saved_tensors
grad_input = grad_output.clone()
grad_input = grad_input + diff
return grad_input
def gen_training_data(self):
mean1,mean2 = [3.0,3.0], [5.0,5.0]
covar1,covar2 = [[1.0,0.0], [0.0,1.0]], [[1.0,0.0], [0.0,1.0]]
data1 = [(list(x),1) for x in np.random.multivariate_normal(mean1, covar1,
self.num_samples_per_class)]
data2 = [(list(x),2) for x in np.random.multivariate_normal(mean2, covar2,
self.num_samples_per_class)]
training_data = data1 + data2
random.shuffle( training_data )
self.training_data = training_data
def train_with_straight_autograd(self):
dtype = torch.float
D_in,H,D_out = 2,10,2
# w1 = torch.randn(D_in, H, device="cpu", dtype=dtype, requires_grad=True)
# w2 = torch.randn(H, D_out, device="cpu", dtype=dtype, requires_grad=True)
w1 = torch.randn(D_in, H, device="cpu", dtype=dtype)
w2 = torch.randn(H, D_out, device="cpu", dtype=dtype)
w1 = w1.to(self.dl_studio.device)
w2 = w2.to(self.dl_studio.device)
w1.requires_grad_()
w2.requires_grad_()
Loss = []
for epoch in range(self.dl_studio.epochs):
for i,data in enumerate(self.training_data):
input, label = data
x,y = torch.as_tensor(np.array(input)), torch.as_tensor(np.array(label))
x,y = x.float(), y.float()
if self.dl_studio.use_gpu is True:
x,y = x.to(self.dl_studio.device), y.to(self.dl_studio.device)
y_pred = x.view(1,-1).mm(w1).clamp(min=0).mm(w2)
loss = (y_pred - y).pow(2).sum()
if i % 200 == 199:
Loss.append(loss.item())
print("epoch=%d i=%d" % (epoch,i), loss.item())
# w1.retain_grad()
# w2.retain_grad()
loss.backward()
with torch.no_grad():
w1 -= self.dl_studio.learning_rate * w1.grad
w2 -= self.dl_studio.learning_rate * w2.grad
w1.grad.zero_()
w2.grad.zero_()
print("\n\n\nLoss: %s" % str(Loss))
import matplotlib.pyplot as plt
plt.figure("Loss vs training (straight autograd)")
plt.plot(Loss)
plt.show()
def train_with_extended_autograd(self):
dtype = torch.float
D_in,H,D_out = 2,10,2
# w1 = torch.randn(D_in, H, device="cpu", dtype=dtype, requires_grad=True)
# w2 = torch.randn(H, D_out, device="cpu", dtype=dtype, requires_grad=True)
w1 = torch.randn(D_in, H, device="cpu", dtype=dtype)
w2 = torch.randn(H, D_out, device="cpu", dtype=dtype)
w1 = w1.to(self.dl_studio.device)
w2 = w2.to(self.dl_studio.device)
w1.requires_grad_()
w2.requires_grad_()
Loss = []
for epoch in range(self.dl_studio.epochs):
for i,data in enumerate(self.training_data):
## Constructing an instance of DoSillyWithTensor. It is callable.
do_silly = DLStudio.AutogradCustomization.DoSillyWithTensor.apply
input, label = data
x,y = torch.as_tensor(np.array(input)), torch.as_tensor(np.array(label))
## Now process the training instance with the "do_silly" instance:
x = do_silly(x)
x,y = x.float(), y.float()
x,y = x.to(self.dl_studio.device), y.to(self.dl_studio.device)
y_pred = x.view(1,-1).mm(w1).clamp(min=0).mm(w2)
loss = (y_pred - y).pow(2).sum()
if i % 200 == 199:
Loss.append(loss.item())
print("epoch=%d i=%d" % (epoch,i), loss.item())
# w1.retain_grad()
# w2.retain_grad()
loss.backward()
with torch.no_grad():
w1 -= self.dl_studio.learning_rate * w1.grad
w2 -= self.dl_studio.learning_rate * w2.grad
w1.grad.zero_()
w2.grad.zero_()
print("\n\n\nLoss: %s" % str(Loss))
import matplotlib.pyplot as plt
plt.figure("loss vs training (extended autograd)")
plt.plot(Loss)
plt.show()
############### Start Definition of Inner Class SkipConnections ##############
class SkipConnections(nn.Module):
"""
This educational class is meant for illustrating the concepts related to the
use of skip connections in neural network. It is now well known that deep
networks are difficult to train because of the vanishing gradients problem.
What that means is that as the depth of network increases, the loss gradients
calculated for the early layers become more and more muted, which suppresses
the learning of the parameters in those layers. An important mitigation
strategy for addressing this problem consists of creating a CNN using blocks
with skip connections.
With the code shown in this inner class of the module, you can now experiment
with skip connections in a CNN to see how a deep network with this feature
might improve the classification results. As you will see in the code shown
below, the network that allows you to construct a CNN with skip connections
is named BMEnet. As shown in the script playing_with_skip_connections.py in
the Examples directory of the distribution, you can easily create a CNN with
arbitrary depth just by using the "depth" constructor option for the BMEnet
class. The basic block of the network constructed by BMEnet is called
SkipBlock which, very much like the BasicBlock in ResNet-18, has a couple of
convolutional layers whose output is combined with the input to the block.
Note that the value given to the the "depth" constructor option for the
BMEnet class does NOT translate directly into the actual depth of the
CNN. [Again, see the script playing_with_skip_connections.py in the Examples
directory for how to use this option.] The value of "depth" is translated
into how many instances of SkipBlock to use for constructing the CNN.
"""
def load_cifar_10_dataset(self):
self.dl_studio.load_cifar_10_dataset()
def load_cifar_10_dataset_with_augmentation(self):
self.dl_studio.load_cifar_10_dataset_with_augmentation()
def __init__(self, dl_studio):
super(DLStudio.SkipConnections, self).__init__()
self.dl_studio = dl_studio
class SkipBlock(nn.Module):
def __init__(self, in_ch, out_ch, downsample=False, skip_connections=True):
super(DLStudio.SkipConnections.SkipBlock, self).__init__()
self.downsample = downsample
self.skip_connections = skip_connections
self.in_ch = in_ch
self.out_ch = out_ch
self.convo = nn.Conv2d(in_ch, out_ch, 3, stride=1, padding=1)
norm_layer = nn.BatchNorm2d
self.bn = norm_layer(out_ch)
if downsample:
self.downsampler = nn.Conv2d(in_ch, out_ch, 1, stride=2)
def forward(self, x):
identity = x
out = self.convo(x)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.in_ch == self.out_ch:
out = self.convo(out)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.downsample:
out = self.downsampler(out)
identity = self.downsampler(identity)
if self.skip_connections:
if self.in_ch == self.out_ch:
out += identity
else:
out[:,:self.in_ch,:,:] += identity
out[:,self.in_ch:,:,:] += identity
return out
class BMEnet(nn.Module):
def __init__(self, skip_connections=True, depth=32):
super(DLStudio.SkipConnections.BMEnet, self).__init__()
self.pool_count = 3
self.depth = depth // 2
self.conv = nn.Conv2d(3, 64, 3, padding=1)
self.pool = nn.MaxPool2d(2, 2)
self.skip64 = DLStudio.SkipConnections.SkipBlock(64, 64, skip_connections=skip_connections)
self.skip64ds = DLStudio.SkipConnections.SkipBlock(64, 64,
downsample=True, skip_connections=skip_connections)
self.skip64to128 = DLStudio.SkipConnections.SkipBlock(64, 128,
skip_connections=skip_connections )
self.skip128 = DLStudio.SkipConnections.SkipBlock(128, 128, skip_connections=skip_connections)
self.skip128ds = DLStudio.SkipConnections.SkipBlock(128,128,
downsample=True, skip_connections=skip_connections)
self.fc1 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc2 = nn.Linear(1000, 10)
def forward(self, x):
x = self.pool(torch.nn.functional.relu(self.conv(x)))
for _ in range(self.depth // 4):
x = self.skip64(x)
x = self.skip64ds(x)
for _ in range(self.depth // 4):
x = self.skip64(x)
x = self.skip64to128(x)
for _ in range(self.depth // 4):
x = self.skip128(x)
x = self.skip128ds(x)
for _ in range(self.depth // 4):
x = self.skip128(x)
x = x.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x = torch.nn.functional.relu(self.fc1(x))
x = self.fc2(x)
return x
def run_code_for_training(self, net):
self.dl_studio.run_code_for_training(net)
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.dl_studio.path_saved_model)
def run_code_for_testing(self, model):
self.dl_studio.run_code_for_testing(model)
############# Start Definition of Inner Class CustomDataLoading ############
class CustomDataLoading(nn.Module):
"""
This is a testbed for experimenting with a completely grounds-up attempt at
designing a custom data loader. Ordinarily, if the basic format of how the
dataset is stored is similar to one of the datasets that the Torchvision
module knows about, you can go ahead and use that for your own dataset. At
worst, you may need to carry out some light customizations depending on the
number of classes involved, etc.
However, if the underlying dataset is stored in a manner that does not look
like anything in Torchvision, you have no choice but to supply yourself all
of the data loading infrastructure. That is what this inner class of the
DLStudio module is all about.
Note that the dataloader developed in this inner class is not yet a finished
product --- in the sense that it does not yet have multi-worker
support. Also, at the moment, it uncompresses the data archive whenever it is
called afresh. A faster implementation would uncompress it the first time
the archive is accessed and, for its subsequent usage, store the uncomressed
version locally in a disk-based hash table of some sort so that any future
loadings from the dataset would be much faster. I hope to do that in a
future version of DLStudio.
The custom data loading exercise here is related to a dataset called
PurdueShapes5 that contains 32x32 images of binary shapes belonging to the
following five classes:
1. rectangle
2. triangle
3. disk
4. oval
5. star
The dataset was generated by randomizing the sizes and the orientations
of these five patterns. Since the patterns are rotated with a very simple
non-interpolating transform, just the act of random rotations can introduce
boundary and even interior noise in the patterns.
Each 32x32 image is stored in the dataset as the following list:
[R, G, B, Bbox, Label]
where
R : is a 1024 element list of the values for the red component
of the color at all the pixels
B : the same as above but for the green component of the color
G : the same as above but for the blue component of the color
Bbox : a list like [x1,y1,x2,y2] that defines the bounding box
for the object in the image
Label : the shape of the object
I serialize the dataset with Python's pickle module and then compress it with
the gzip module.
You will find the following dataset directories in the "data" subdirectory
of Examples in the DLStudio distro:
PurdueShapes5-10000-train.gz
PurdueShapes5-1000-test.gz
PurdueShapes5-20-train.gz
PurdueShapes5-20-test.gz
The number that follows the main name string "PurdueShapes5-" is for the
number of images in the dataset.
You will find the last two datasets, with 20 images each, useful for debugging
your logic for object detection and bounding-box regression.
"""
def __init__(self, dl_studio, dataserver_train=None, dataserver_test=None, dataset_file_train=None, dataset_file_test=None):
super(DLStudio.CustomDataLoading, self).__init__()
self.dl_studio = dl_studio
self.dataserver_train = dataserver_train
self.dataserver_test = dataserver_test
class PurdueShapes5Dataset(torch.utils.data.Dataset):
def __init__(self, dl_studio, dataset_file, transform=None):
super(DLStudio.CustomDataLoading.PurdueShapes5Dataset, self).__init__()
root_dir = dl_studio.dataroot
f = gzip.open(root_dir + dataset_file, 'rb')
dataset = f.read()
self.dataset, self.label_map = pickle.loads(dataset)
# reverse the key-value pairs in the label dictionary:
self.class_labels = dict(map(reversed, self.label_map.items()))
self.transform = transform
def __len__(self):
return len(self.dataset)
def __getitem__(self, idx):
r = np.array( self.dataset[idx][0] )
g = np.array( self.dataset[idx][1] )
b = np.array( self.dataset[idx][2] )
R,G,B = r.reshape(32,32), g.reshape(32,32), b.reshape(32,32)
im_tensor = torch.zeros(3,32,32, dtype=torch.float)
im_tensor[0,:,:] = torch.from_numpy(R)
im_tensor[1,:,:] = torch.from_numpy(G)
im_tensor[2,:,:] = torch.from_numpy(B)
sample = {'image' : im_tensor,
'bbox' : self.dataset[idx][3],
'label' : self.dataset[idx][4] }
if self.transform:
sample = self.transform(sample)
return sample
def load_PurdueShapes5_dataset(self, dataserver_train, dataserver_test ):
transform = tvt.Compose([tvt.ToTensor(),
tvt.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])
self.train_dataloader = torch.utils.data.DataLoader(dataserver_train,
batch_size=self.dl_studio.batch_size,shuffle=True, num_workers=0)
self.test_dataloader = torch.utils.data.DataLoader(dataserver_test,
batch_size=self.dl_studio.batch_size,shuffle=False, num_workers=0)
class SkipBlock(nn.Module):
def __init__(self, in_ch, out_ch, downsample=False, skip_connections=True):
super(DLStudio.CustomDataLoading.SkipBlock, self).__init__()
self.downsample = downsample
self.skip_connections = skip_connections
self.in_ch = in_ch
self.out_ch = out_ch
self.convo = nn.Conv2d(in_ch, out_ch, 3, stride=1, padding=1)
norm_layer = nn.BatchNorm2d
self.bn = norm_layer(out_ch)
if downsample:
self.downsampler = nn.Conv2d(in_ch, out_ch, 1, stride=2)
def forward(self, x):
identity = x
out = self.convo(x)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.in_ch == self.out_ch:
out = self.convo(out)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.downsample:
out = self.downsampler(out)
identity = self.downsampler(identity)
if self.skip_connections:
if self.in_ch == self.out_ch:
out += identity
else:
out[:,:self.in_ch,:,:] += identity
out[:,self.in_ch:,:,:] += identity
return out
class BMEnet(nn.Module):
def __init__(self, skip_connections=True, depth=32):
super(DLStudio.CustomDataLoading.BMEnet, self).__init__()
self.pool_count = 3
self.depth = depth // 2
self.conv = nn.Conv2d(3, 64, 3, padding=1)
self.pool = nn.MaxPool2d(2, 2)
self.skip64 = DLStudio.SkipConnections.SkipBlock(64, 64,
skip_connections=skip_connections)
self.skip64ds = DLStudio.SkipConnections.SkipBlock(64, 64,
downsample=True, skip_connections=skip_connections)
self.skip64to128 = DLStudio.SkipConnections.SkipBlock(64, 128,
skip_connections=skip_connections )
self.skip128 = DLStudio.SkipConnections.SkipBlock(128, 128,
skip_connections=skip_connections)
self.skip128ds = DLStudio.SkipConnections.SkipBlock(128,128,
downsample=True, skip_connections=skip_connections)
self.fc1 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc2 = nn.Linear(1000, 10)
def forward(self, x):
x = self.pool(torch.nn.functional.relu(self.conv(x)))
for _ in range(self.depth // 4):
x = self.skip64(x)
x = self.skip64ds(x)
for _ in range(self.depth // 4):
x = self.skip64(x)
x = self.skip64to128(x)
for _ in range(self.depth // 4):
x = self.skip128(x)
x = self.skip128ds(x)
for _ in range(self.depth // 4):
x = self.skip128(x)
x = x.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x = torch.nn.functional.relu(self.fc1(x))
x = self.fc2(x)
return x
def run_code_for_training_with_custom_loading(self, net):
filename_for_out = "performance_numbers_" + str(self.dl_studio.epochs) + ".txt"
FILE = open(filename_for_out, 'w')
net = copy.deepcopy(net)
net = net.to(self.dl_studio.device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(net.parameters(),
lr=self.dl_studio.learning_rate, momentum=self.dl_studio.momentum)
for epoch in range(self.dl_studio.epochs):
# print("\n")
running_loss = 0.0
for i, data in enumerate(self.train_dataloader):
inputs, bounding_box, labels = data['image'], data['bbox'], data['label']
if self.dl_studio.debug_train and i % 1000 == 999:
print("\n\n\nlabels: %s" % str(labels))
print("\n\n\ntype of labels: %s" % type(labels))
print("\n\n[iter=%d:] Ground Truth: " % (i+1) +
' '.join('%5s' % self.dataserver_train.class_labels[labels[j].item()] for j in range(self.dl_studio.batch_size)))
inputs = inputs.to(self.dl_studio.device)
labels = labels.to(self.dl_studio.device)
optimizer.zero_grad()
outputs = net(inputs)
loss = criterion(outputs, labels)
if self.dl_studio.debug_train and i % 1000 == 999:
_, predicted = torch.max(outputs.data, 1)
print("[iter=%d:] Predicted Labels: " % (i+1) +
' '.join('%5s' % self.dataserver.class_labels[predicted[j]]
for j in range(self.dl_studio.batch_size)))
self.dl_studio.display_tensor_as_image(torchvision.utils.make_grid(
inputs, normalize=True), "see terminal for TRAINING results at iter=%d" % (i+1))
loss.backward()
optimizer.step()
running_loss += loss.item()
if i % 1000 == 999:
avg_loss = running_loss / float(1000)
print("[epoch:%d, batch:%5d] loss: %.3f" % (epoch + 1, i + 1, avg_loss))
FILE.write("%.3f\n" % avg_loss)
FILE.flush()
running_loss = 0.0
print("\nFinished Training\n")
self.save_model(net)
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.dl_studio.path_saved_model)
def run_code_for_testing_with_custom_loading(self, net):
net.load_state_dict(torch.load(self.dl_studio.path_saved_model))
correct = 0
total = 0
confusion_matrix = torch.zeros(len(self.dataserver_train.class_labels),
len(self.dataserver_train.class_labels))
class_correct = [0] * len(self.dataserver_train.class_labels)
class_total = [0] * len(self.dataserver_train.class_labels)
with torch.no_grad():
for i, data in enumerate(self.test_dataloader):
images, bounding_box, labels = data['image'], data['bbox'], data['label']
labels = labels.tolist()
if self.dl_studio.debug_test and i % 1000 == 0:
print("\n\n[i=%d:] Ground Truth: " %i + ' '.join('%10s' %
self.dataserver_train.class_labels[labels[j]] for j in range(self.dl_studio.batch_size)))
outputs = net(images)
## max() returns two things: the max value and its index in the 10 element
## output vector. We are only interested in the index --- since that is
## essentially the predicted class label:
_, predicted = torch.max(outputs.data, 1)
predicted = predicted.tolist()
if self.dl_studio.debug_test and i % 1000 == 0:
print("[i=%d:] Predicted Labels: " %i + ' '.join('%10s' %
self.dataserver_train.class_labels[predicted[j]] for j in range(self.dl_studio.batch_size)))
self.dl_studio.display_tensor_as_image(
torchvision.utils.make_grid(images, normalize=True),
"see terminal for test results at i=%d" % i)
for label,prediction in zip(labels,predicted):
confusion_matrix[label][prediction] += 1
total += len(labels)
correct += [predicted[ele] == labels[ele] for ele in range(len(predicted))].count(True)
comp = [predicted[ele] == labels[ele] for ele in range(len(predicted))]
for j in range(self.dl_studio.batch_size):
label = labels[j]
class_correct[label] += comp[j]
class_total[label] += 1
print("\n")
for j in range(len(self.dataserver_train.class_labels)):
print('Prediction accuracy for %5s : %2d %%' % (
self.dataserver_train.class_labels[j], 100 * class_correct[j] / class_total[j]))
print("\n\n\nOverall accuracy of the network on the 10000 test images: %d %%" %
(100 * correct / float(total)))
print("\n\nDisplaying the confusion matrix:\n")
out_str = " "
for j in range(len(self.dataserver_train.class_labels)):
out_str += "%15s" % self.dataserver_train.class_labels[j]
print(out_str + "\n")
for i,label in enumerate(self.dataserver_train.class_labels):
out_percents = [100 * confusion_matrix[i,j] / float(class_total[i])
for j in range(len(self.dataserver_train.class_labels))]
out_percents = ["%.2f" % item.item() for item in out_percents]
out_str = "%12s: " % self.dataserver_train.class_labels[i]
for j in range(len(self.dataserver_train.class_labels)):
out_str += "%15s" % out_percents[j]
print(out_str)
#################### Start Definition of Inner Class DetectAndLocalize ##################
class DetectAndLocalize(nn.Module):
"""The purpose of this inner class is to focus on object detection in images --- as
opposed to image classification. Most people would say that object detection
is a more challenging problem than image classification because, in general,
the former also requires localization. The simplest interpretation of what
is meant by localization is that the code that carries out object detection
must also output a bounding-box rectangle for the object that was detected.
You will find in this inner class some examples of LOADnet classes meant
for solving the object detection and localization problem. The acronym
"LOAD" in "LOADnet" stands for
"LOcalization And Detection"
The different network examples included here are LOADnet1, LOADnet2, and
LOADnet3. For now, only pay attention to LOADnet2 since that's the class I
have worked with the most for the 1.0.7 distribution.
"""
def __init__(self, dl_studio, dataserver_train=None, dataserver_test=None, dataset_file_train=None, dataset_file_test=None):
super(DLStudio.DetectAndLocalize, self).__init__()
self.dl_studio = dl_studio
self.dataserver_train = dataserver_train
self.dataserver_test = dataserver_test
class PurdueShapes5Dataset(torch.utils.data.Dataset):
def __init__(self, dl_studio, dataset_file, transform=None):
super(DLStudio.DetectAndLocalize.PurdueShapes5Dataset, self).__init__()
root_dir = dl_studio.dataroot
f = gzip.open(root_dir + dataset_file, 'rb')
dataset = f.read()
self.dataset, self.label_map = pickle.loads(dataset)
# reverse the key-value pairs in the label dictionary:
self.class_labels = dict(map(reversed, self.label_map.items()))
self.transform = transform
def __len__(self):
return len(self.dataset)
def __getitem__(self, idx):
r = np.array( self.dataset[idx][0] )
g = np.array( self.dataset[idx][1] )
b = np.array( self.dataset[idx][2] )
R,G,B = r.reshape(32,32), g.reshape(32,32), b.reshape(32,32)
im_tensor = torch.zeros(3,32,32, dtype=torch.float)
im_tensor[0,:,:] = torch.from_numpy(R)
im_tensor[1,:,:] = torch.from_numpy(G)
im_tensor[2,:,:] = torch.from_numpy(B)
bb_tensor = torch.tensor(self.dataset[idx][3], dtype=torch.float)
sample = {'image' : im_tensor,
'bbox' : bb_tensor,
'label' : self.dataset[idx][4] }
if self.transform:
sample = self.transform(sample)
return sample
def load_PurdueShapes5_dataset(self, dataserver_train, dataserver_test ):
# transform = tvt.Compose([tvt.ToTensor(),
# tvt.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])
self.train_dataloader = torch.utils.data.DataLoader(dataserver_train,
batch_size=self.dl_studio.batch_size,shuffle=True, num_workers=0)
self.test_dataloader = torch.utils.data.DataLoader(dataserver_test,
batch_size=self.dl_studio.batch_size,shuffle=False, num_workers=0)
class SkipBlock(nn.Module):
def __init__(self, in_ch, out_ch, downsample=False, skip_connections=True):
super(DLStudio.DetectAndLocalize.SkipBlock, self).__init__()
self.downsample = downsample
self.skip_connections = skip_connections
self.in_ch = in_ch
self.out_ch = out_ch
self.convo = nn.Conv2d(in_ch, out_ch, 3, stride=1, padding=1)
norm_layer = nn.BatchNorm2d
self.bn = norm_layer(out_ch)
if downsample:
self.downsampler = nn.Conv2d(in_ch, out_ch, 1, stride=2)
def forward(self, x):
identity = x
out = self.convo(x)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.in_ch == self.out_ch:
out = self.convo(out)
out = self.bn(out)
out = torch.nn.functional.relu(out)
if self.downsample:
out = self.downsampler(out)
identity = self.downsampler(identity)
if self.skip_connections:
if self.in_ch == self.out_ch:
out += identity
else:
out[:,:self.in_ch,:,:] += identity
out[:,self.in_ch:,:,:] += identity
return out
class LOADnet1(nn.Module):
"""
The acronym 'LOAD' stands for 'LOcalization And Detection'.
LOADnet1 only uses fully-connected layers for the regression
"""
def __init__(self, skip_connections=True, depth=32):
super(DLStudio.DetectAndLocalize.LOADnet1, self).__init__()
self.pool_count = 3
self.depth = depth // 2
self.conv = nn.Conv2d(3, 64, 3, padding=1)
self.pool = nn.MaxPool2d(2, 2)
self.skip64 = DLStudio.SkipConnections.SkipBlock(64, 64,
skip_connections=skip_connections)
self.skip64ds = DLStudio.SkipConnections.SkipBlock(64, 64,
downsample=True, skip_connections=skip_connections)
self.skip64to128 = DLStudio.SkipConnections.SkipBlock(64, 128,
skip_connections=skip_connections )
self.skip128 = DLStudio.SkipConnections.SkipBlock(128, 128,
skip_connections=skip_connections)
self.skip128ds = DLStudio.SkipConnections.SkipBlock(128,128,
downsample=True, skip_connections=skip_connections)
self.fc1 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc2 = nn.Linear(1000, 5)
self.fc3 = nn.Linear(32768, 1000)
self.fc4 = nn.Linear(1000, 4)
def forward(self, x):
x = self.pool(torch.nn.functional.relu(self.conv(x)))
## The labeling section:
for _ in range(self.depth // 4):
x1 = self.skip64(x)
x1 = self.skip64ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip64(x1)
x1 = self.skip64to128(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = self.skip128ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = x1.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x1 = torch.nn.functional.relu(self.fc1(x1))
x1 = self.fc2(x1)
## The Bounding Box regression:
x2 = x.view(-1, 32768 )
x2 = torch.nn.functional.relu(self.fc3(x2))
x2 = self.fc4(x2)
return x1,x2
class LOADnet2(nn.Module):
"""
The acronym 'LOAD' stands for 'LOcalization And Detection'.
LOADnet2 uses both convo and linear layers for regression
"""
def __init__(self, skip_connections=True, depth=32):
super(DLStudio.DetectAndLocalize.LOADnet2, self).__init__()
self.pool_count = 3
self.depth = depth // 2
self.conv = nn.Conv2d(3, 64, 3, padding=1)
self.pool = nn.MaxPool2d(2, 2)
self.skip64 = DLStudio.SkipConnections.SkipBlock(64, 64,
skip_connections=skip_connections)
self.skip64ds = DLStudio.SkipConnections.SkipBlock(64, 64,
downsample=True, skip_connections=skip_connections)
self.skip64to128 = DLStudio.SkipConnections.SkipBlock(64, 128,
skip_connections=skip_connections )
self.skip128 = DLStudio.SkipConnections.SkipBlock(128, 128,
skip_connections=skip_connections)
self.skip128ds = DLStudio.SkipConnections.SkipBlock(128,128,
downsample=True, skip_connections=skip_connections)
self.fc1 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc2 = nn.Linear(1000, 5)
## for regression
self.conv_seqn = nn.Sequential(
nn.Conv2d(in_channels=64, out_channels=64, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.Conv2d(in_channels=64, out_channels=64, kernel_size=3, padding=1),
nn.ReLU(inplace=True)
)
self.fc_seqn = nn.Sequential(
nn.Linear(16384, 1024),
nn.ReLU(inplace=True),
nn.Linear(1024, 512),
nn.ReLU(inplace=True),
nn.Linear(512, 4)
)
def forward(self, x):
x = self.pool(torch.nn.functional.relu(self.conv(x)))
## The labeling section:
x1 = x.clone()
for _ in range(self.depth // 4):
x1 = self.skip64(x1)
x1 = self.skip64ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip64(x1)
x1 = self.skip64to128(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = self.skip128ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = x1.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x1 = torch.nn.functional.relu(self.fc1(x1))
x1 = self.fc2(x1)
## The Bounding Box regression:
x2 = self.conv_seqn(x)
x2 = self.conv_seqn(x2)
# flatten
x2 = x2.view(x.size(0), -1)
x2 = self.fc_seqn(x2)
return x1,x2
class LOADnet3(nn.Module):
"""
The acronym 'LOAD' stands for 'LOcalization And Detection'.
LOADnet2 uses both convo and linear layers for regression
"""
def __init__(self, skip_connections=True, depth=32):
super(DLStudio.DetectAndLocalize.LOADnet2, self).__init__()
self.pool_count = 3
self.depth = depth // 2
self.conv = nn.Conv2d(3, 64, 3, padding=1)
self.pool = nn.MaxPool2d(2, 2)
self.skip64 = DLStudio.SkipConnections.SkipBlock(64, 64,
skip_connections=skip_connections)
self.skip64ds = DLStudio.SkipConnections.SkipBlock(64, 64,
downsample=True, skip_connections=skip_connections)
self.skip64to128 = DLStudio.SkipConnections.SkipBlock(64, 128,
skip_connections=skip_connections )
self.skip128 = DLStudio.SkipConnections.SkipBlock(128, 128,
skip_connections=skip_connections)
self.skip128ds = DLStudio.SkipConnections.SkipBlock(128,128,
downsample=True, skip_connections=skip_connections)
self.fc1 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc2 = nn.Linear(1000, 5)
self.fc3 = nn.Linear(128 * (32 // 2**self.pool_count)**2, 1000)
self.fc4 = nn.Linear(2048, 1024)
def forward(self, x):
x = self.pool(torch.nn.functional.relu(self.conv(x)))
## The labeling section:
for _ in range(self.depth // 4):
x1 = self.skip64(x)
x1 = self.skip64ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip64(x1)
x1 = self.skip64to128(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = self.skip128ds(x1)
for _ in range(self.depth // 4):
x1 = self.skip128(x1)
x1 = x1.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x1 = torch.nn.functional.relu(self.fc1(x1))
x1 = self.fc2(x1)
## The Bounding Box regression:
for _ in range(4):
x2 = self.skip64(x)
x2 = self.skip64to128(x2)
for _ in range(4):
x2 = self.skip128(x2)
x2 = x.view(-1, 128 * (32 // 2**self.pool_count)**2 )
x2 = torch.nn.functional.relu(self.fc3(x2))
x2 = self.fc4(x2)
return x1,x2
class IOULoss(nn.Module):
def __init__(self, batch_size):
super(DLStudio.DetectAndLocalize.IOULoss, self).__init__()
self.batch_size = batch_size
def forward(self, input, target):
composite_loss = []
for idx in range(self.batch_size):
union = intersection = 0.0
for i in range(32):
for j in range(32):
inp = input[idx,i,j]
tap = target[idx,i,j]
if (inp == tap) and (inp==1):
intersection += 1
union += 1
elif (inp != tap) and ((inp==1) or (tap==1)):
union += 1
if union == 0.0:
raise Exception("something_wrong")
batch_sample_iou = intersection / float(union)
composite_loss.append(batch_sample_iou)
total_iou_for_batch = sum(composite_loss)
return 1 - torch.tensor([total_iou_for_batch / self.batch_size])
def run_code_for_training_with_CrossEntropy_and_BCE_Losses(self, net):
"""
BCE stands for the Binary Cross Entropy Loss that is used for
the regression loss in this training method.
"""
filename_for_out1 = "performance_numbers_" + str(self.dl_studio.epochs) + "label.txt"
filename_for_out2 = "performance_numbers_" + str(self.dl_studio.epochs) + "regres.txt"
FILE1 = open(filename_for_out1, 'w')
FILE2 = open(filename_for_out2, 'w')
net = copy.deepcopy(net)
net = net.to(self.dl_studio.device)
criterion1 = nn.CrossEntropyLoss()
# criterion2 = self.dl_studio.DetectAndLocalize.IOULoss(self.dl_studio.batch_size)
criterion2 = nn.BCELoss()
optimizer = optim.SGD(net.parameters(),
lr=self.dl_studio.learning_rate, momentum=self.dl_studio.momentum)
for epoch in range(self.dl_studio.epochs):
running_loss_labeling = 0.0
running_loss_regression = 0.0
for i, data in enumerate(self.train_dataloader):
gt_too_small = False
inputs, bbox_gt, labels = data['image'], data['bbox'], data['label']
if self.dl_studio.debug_train and i % 1000 == 999:
print("\n\n[iter=%d:] Ground Truth: " % (i+1) +
' '.join('%5s' % self.dataserver_train.class_labels[labels[j].item()] for j in range(self.dl_studio.batch_size)))
inputs = inputs.to(self.dl_studio.device)
labels = labels.to(self.dl_studio.device)
bbox_gt = bbox_gt.to(self.dl_studio.device)
optimizer.zero_grad()
outputs = net(inputs)
outputs_label = outputs[0]
bbox_pred = outputs[1]
if self.dl_studio.debug_train and i % 500 == 499:
inputs_copy = inputs.detach().clone()
inputs_copy = inputs_copy.cpu()
bbox_pc = bbox_pred.detach().clone()
bbox_pc[bbox_pc<0] = 0
bbox_pc[bbox_pc>31] = 31
_, predicted = torch.max(outputs_label.data, 1)
print("[iter=%d:] Predicted Labels: " % (i+1) +
' '.join('%10s' % self.dataserver_train.class_labels[predicted[j].item()]
for j in range(self.dl_studio.batch_size)))
for idx in range(self.dl_studio.batch_size):
i1 = int(bbox_gt[idx][1])
i2 = int(bbox_gt[idx][3])
j1 = int(bbox_gt[idx][0])
j2 = int(bbox_gt[idx][2])
k1 = int(bbox_pc[idx][1])
k2 = int(bbox_pc[idx][3])
l1 = int(bbox_pc[idx][0])
l2 = int(bbox_pc[idx][2])
print(" gt_bb: [%d,%d,%d,%d]"%(j1,i1,j2,i2))
print(" pred_bb: [%d,%d,%d,%d]"%(l1,k1,l2,k2))
inputs_copy[idx,0,i1:i2,j1] = 255
inputs_copy[idx,0,i1:i2,j2] = 255
inputs_copy[idx,0,i1,j1:j2] = 255
inputs_copy[idx,0,i2,j1:j2] = 255
inputs_copy[idx,2,k1:k2,l1] = 255
inputs_copy[idx,2,k1:k2,l2] = 255
inputs_copy[idx,2,k1,l1:l2] = 255
inputs_copy[idx,2,k2,l1:l2] = 255
self.dl_studio.display_tensor_as_image(
torchvision.utils.make_grid(inputs_copy, normalize=True),
"see terminal for TRAINING results at iter=%d" % (i+1))
mask_regress = torch.zeros(self.dl_studio.batch_size,32,32,requires_grad=False)
mask_gt = torch.zeros(self.dl_studio.batch_size, 32,32)
for k,out_regres in enumerate(bbox_pred):
x1,y1,x2,y2 = bbox_pred[k].tolist()
x1_gt,y1_gt,x2_gt,y2_gt = bbox_gt[k].tolist()
x1,y1,x2,y2 = [int(item) if item >0 else 0 for item in (x1,y1,x2,y2)]
x1_gt,y1_gt,x2_gt,y2_gt = [int(item) if item>0 else 0 for item in (x1_gt,y1_gt,x2_gt,y2_gt)]
if abs(x1_gt - x2_gt)<5 or abs(y1_gt-y2_gt) < 5: gt_too_small = True
mask_regress_np = np.zeros((32,32), dtype=bool)
mask_gt_np = np.zeros((32,32), dtype=bool)
mask_regress_np[y1:y2,x1:x2] = 1
mask_gt_np[y1_gt:y2_gt, x1_gt:x2_gt] = 1
mask_regress[k,:,:] = torch.from_numpy(mask_regress_np)
mask_regress.reqiures_grad=True
mask_gt[k,:,:] = torch.from_numpy(mask_gt_np)
mask_gt.reqiures_grad=True
loss_labeling = criterion1(outputs_label, labels)
loss_labeling.backward(retain_graph=True)
loss_regression = criterion2(mask_regress, mask_gt)
loss_regression.requires_grad = True
loss_regression.backward()
optimizer.step()
running_loss_labeling += loss_labeling.item()
running_loss_regression += loss_regression.item()
if i % 1000 == 999:
avg_loss_labeling = running_loss_labeling / float(1000)
avg_loss_regression = running_loss_regression / float(1000)
print("[epoch:%d, batch:%5d] loss_labeling: %.3f loss_regression: %.3f " % (epoch + 1, i + 1, avg_loss_labeling, avg_loss_regression))
FILE1.write("%.3f\n" % avg_loss_labeling)
FILE1.flush()
FILE2.write("%.3f\n" % avg_loss_regression)
FILE2.flush()
running_loss_labeling = 0.0
running_loss_regression = 0.0
print("\nFinished Training\n")
self.save_model(net)
def run_code_for_training_with_CrossEntropy_and_MSE_Losses(self, net):
filename_for_out1 = "performance_numbers_" + str(self.dl_studio.epochs) + "label.txt"
filename_for_out2 = "performance_numbers_" + str(self.dl_studio.epochs) + "regres.txt"
FILE1 = open(filename_for_out1, 'w')
FILE2 = open(filename_for_out2, 'w')
net = copy.deepcopy(net)
net = net.to(self.dl_studio.device)
criterion1 = nn.CrossEntropyLoss()
criterion2 = nn.MSELoss()
optimizer = optim.SGD(net.parameters(),
lr=self.dl_studio.learning_rate, momentum=self.dl_studio.momentum)
for epoch in range(self.dl_studio.epochs):
running_loss_labeling = 0.0
running_loss_regression = 0.0
for i, data in enumerate(self.train_dataloader):
gt_too_small = False
inputs, bbox_gt, labels = data['image'], data['bbox'], data['label']
if self.dl_studio.debug_train and i % 500 == 499:
# if self.dl_studio.debug_train and ((epoch==0 and (i==0 or i==9 or i==99)) or i%500==499):
print("\n\n[epoch=%d iter=%d:] Ground Truth: " % (epoch+1, i+1) +
' '.join('%10s' % self.dataserver_train.class_labels[labels[j].item()] for j in range(self.dl_studio.batch_size)))
inputs = inputs.to(self.dl_studio.device)
labels = labels.to(self.dl_studio.device)
bbox_gt = bbox_gt.to(self.dl_studio.device)
optimizer.zero_grad()
outputs = net(inputs)
outputs_label = outputs[0]
bbox_pred = outputs[1]
if self.dl_studio.debug_train and i % 500 == 499:
# if self.dl_studio.debug_train and ((epoch==0 and (i==0 or i==9 or i==99)) or i%500==499):
inputs_copy = inputs.detach().clone()
inputs_copy = inputs_copy.cpu()
bbox_pc = bbox_pred.detach().clone()
bbox_pc[bbox_pc<0] = 0
bbox_pc[bbox_pc>31] = 31
_, predicted = torch.max(outputs_label.data, 1)
print("[epoch=%d iter=%d:] Predicted Labels: " % (epoch+1, i+1) +
' '.join('%10s' % self.dataserver_train.class_labels[predicted[j].item()]
for j in range(self.dl_studio.batch_size)))
for idx in range(self.dl_studio.batch_size):
i1 = int(bbox_gt[idx][1])
i2 = int(bbox_gt[idx][3])
j1 = int(bbox_gt[idx][0])
j2 = int(bbox_gt[idx][2])
k1 = int(bbox_pc[idx][1])
k2 = int(bbox_pc[idx][3])
l1 = int(bbox_pc[idx][0])
l2 = int(bbox_pc[idx][2])
print(" gt_bb: [%d,%d,%d,%d]"%(j1,i1,j2,i2))
print(" pred_bb: [%d,%d,%d,%d]"%(l1,k1,l2,k2))
inputs_copy[idx,0,i1:i2,j1] = 255
inputs_copy[idx,0,i1:i2,j2] = 255
inputs_copy[idx,0,i1,j1:j2] = 255
inputs_copy[idx,0,i2,j1:j2] = 255
inputs_copy[idx,2,k1:k2,l1] = 255
inputs_copy[idx,2,k1:k2,l2] = 255
inputs_copy[idx,2,k1,l1:l2] = 255
inputs_copy[idx,2,k2,l1:l2] = 255
# self.dl_studio.display_tensor_as_image(
# torchvision.utils.make_grid(inputs_copy, normalize=True),
# "see terminal for TRAINING results at iter=%d" % (i+1))
loss_labeling = criterion1(outputs_label, labels)
loss_labeling.backward(retain_graph=True)
loss_regression = criterion2(bbox_pred, bbox_gt)
loss_regression.backward()
optimizer.step()
running_loss_labeling += loss_labeling.item()
running_loss_regression += loss_regression.item()
if i % 500 == 499:
avg_loss_labeling = running_loss_labeling / float(500)
avg_loss_regression = running_loss_regression / float(500)
print("\n[epoch:%d, iteration:%5d] loss_labeling: %.3f loss_regression: %.3f " % (epoch + 1, i + 1, avg_loss_labeling, avg_loss_regression))
FILE1.write("%.3f\n" % avg_loss_labeling)
FILE1.flush()
FILE2.write("%.3f\n" % avg_loss_regression)
FILE2.flush()
running_loss_labeling = 0.0
running_loss_regression = 0.0
if self.dl_studio.debug_train and i%500==499:
# if self.dl_studio.debug_train and ((epoch==0 and (i==0 or i==9 or i==99)) or i%500==499):
self.dl_studio.display_tensor_as_image(
torchvision.utils.make_grid(inputs_copy, normalize=True),
"see terminal for TRAINING results at iter=%d" % (i+1))
print("\nFinished Training\n")
self.save_model(net)
def save_model(self, model):
'''
Save the trained model to a disk file
'''
torch.save(model.state_dict(), self.dl_studio.path_saved_model)
def run_code_for_testing_detection_and_localization(self, net):
net.load_state_dict(torch.load(self.dl_studio.path_saved_model))
correct = 0
total = 0
confusion_matrix = torch.zeros(len(self.dataserver_train.class_labels),
len(self.dataserver_train.class_labels))
class_correct = [0] * len(self.dataserver_train.class_labels)
class_total = [0] * len(self.dataserver_train.class_labels)
with torch.no_grad():
for i, data in enumerate(self.test_dataloader):
images, bounding_box, labels = data['image'], data['bbox'], data['label']
labels = labels.tolist()
if self.dl_studio.debug_test and i % 50 == 0:
print("\n\n[i=%d:] Ground Truth: " %i + ' '.join('%10s' %
self.dataserver_train.class_labels[labels[j]] for j in range(self.dl_studio.batch_size)))
outputs = net(images)
outputs_label = outputs[0]
outputs_regression = outputs[1]
outputs_regression[outputs_regression < 0] = 0
outputs_regression[outputs_regression > 31] = 31
output_bb = outputs_regression.tolist()
_, predicted = torch.max(outputs_label.data, 1)
predicted = predicted.tolist()
if self.dl_studio.debug_test and i % 50 == 0:
print("[i=%d:] Predicted Labels: " %i + ' '.join('%10s' %
self.dataserver_train.class_labels[predicted[j]] for j in range(self.dl_studio.batch_size)))
for idx in range(self.dl_studio.batch_size):
i1 = int(bounding_box[idx][1])
i2 = int(bounding_box[idx][3])
j1 = int(bounding_box[idx][0])
j2 = int(bounding_box[idx][2])
k1 = int(output_bb[idx][1])
k2 = int(output_bb[idx][3])
l1 = int(output_bb[idx][0])
l2 = int(output_bb[idx][2])
print(" gt_bb: [%d,%d,%d,%d]"%(j1,i1,j2,i2))
print(" pred_bb: [%d,%d,%d,%d]"%(l1,k1,l2,k2))
images[idx,0,i1:i2,j1] = 255
images[idx,0,i1:i2,j2] = 255
images[idx,0,i1,j1:j2] = 255
images[idx,0,i2,j1:j2] = 255
images[idx,2,k1:k2,l1] = 255
images[idx,2,k1:k2,l2] = 255
images[idx,2,k1,l1:l2] = 255
images[idx,2,k2,l1:l2] = 255
self.dl_studio.display_tensor_as_image(
torchvision.utils.make_grid(images, normalize=True),
"see terminal for test results at i=%d" % i)
# sys.exit("delib")
for label,prediction in zip(labels,predicted):
confusion_matrix[label][prediction] += 1
total += len(labels)
correct += [predicted[ele] == labels[ele] for ele in range(len(predicted))].count(True)
comp = [predicted[ele] == labels[ele] for ele in range(len(predicted))]
for j in range(self.dl_studio.batch_size):
label = labels[j]
class_correct[label] += comp[j]
class_total[label] += 1
print("\n")
for j in range(len(self.dataserver_train.class_labels)):
print('Prediction accuracy for %5s : %2d %%' % (
self.dataserver_train.class_labels[j], 100 * class_correct[j] / class_total[j]))
print("\n\n\nOverall accuracy of the network on the 1000 test images: %d %%" %
(100 * correct / float(total)))
print("\n\nDisplaying the confusion matrix:\n")
out_str = " "
for j in range(len(self.dataserver_train.class_labels)):
out_str += "%15s" % self.dataserver_train.class_labels[j]
print(out_str + "\n")
for i,label in enumerate(self.dataserver_train.class_labels):
out_percents = [100 * confusion_matrix[i,j] / float(class_total[i])
for j in range(len(self.dataserver_train.class_labels))]
out_percents = ["%.2f" % item.item() for item in out_percents]
out_str = "%12s: " % self.dataserver_train.class_labels[i]
for j in range(len(self.dataserver_train.class_labels)):
out_str += "%15s" % out_percents[j]
print(out_str)
def plot_loss(self):
plt.figure()
plt.plot(self.LOSS)
plt.show()
#_________________________ End of DLStudio Class Definition ___________________________
#______________________________ Test code follows _________________________________
if __name__ == '__main__':
pass