ZSL Basic in Pytorch

https://medium.com/@cetinsamet/zero-shot-learning-53080995d45f

import VGGnet
import gzip
import _pickle as cPickle
import torch.nn as nn
import numpy as np
import torch
import time

from sklearn.preprocessing import LabelEncoder, normalize
from sklearn.neighbors import KDTree

# define train & zsl classes

global train_classes
global zsl_classes

train_classes = ['arm', 'boy', 'bread', 'chicken', 'child', 'computer', 'ear', 'house', 'leg', 'sandwich', 'television',
'truck', 'vehicle','watch', 'woman']
zsl_classes = ['car', 'food', 'hand', 'man', 'neck']

# define PATH

WORD2VECPATH="./data/class_vectors.npy"
DATAPATH= "./data/zeroshot_data.pkl"

def to_one_hot(list_,num):
one_hot=[0 for i in range(num)]
one_hot[list_[0]]=1

return one_hot

def un_one_hot(list_,num):
for i in range(num):
if list_[i]==1:
temp=[]
temp.append(i)
return temp

def custom_kernel_init(shape):
class_vectors = np.load(WORD2VECPATH)
training_vectors = sorted([(label, vec) for (label, vec) in class_vectors if label in train_classes], key=lambda x: x[0]) classnames, vectors = zip(*training_vectors)
print((classnames))
vectors = np.asarray(vectors, dtype=np.float)
vectors = vectors
return vectors

class Word2Vec_Model(nn.Module):
def __init__(self):
super(Word2Vec_Model, self).__init__()
self.classifier = nn.Sequential(
nn.Linear(4096, 1024),
nn.ReLU(True),
nn.Dropout(),
nn.Linear(1024, 512),
nn.ReLU(True),
nn.Dropout(),
nn.Linear(512, 256),
nn.ReLU(True),
nn.Linear(256,300),
) self.LastLinear=nn.Linear(300,15)
self.Softmax=nn.Softmax()
self.LastLinear.weight=nn.Parameter(torch.from_numpy(custom_kernel_init(1)),requires_grad=False)

def forward(self, x):
x=self.classifier(x)
# print('after sequential:{}'.format(x.size()))
x=self.LastLinear(x)
# print('after Last Linear:{}'.format(x.size()))
return x

def load_data():
with gzip.GzipFile(DATAPATH, 'rb') as infile:
data=cPickle.load(infile)

label_encoder = LabelEncoder()
label_encoder.fit(train_classes)

training_data=[instance for instance in data if instance[0] in train_classes]
zero_shot_data=[instance for instance in data if instance[0] not in train_classes]

#SHUFFLE Training data
np.random.shuffle(training_data)

# SPLIT DATA FOR TRAINING
# 300 data per class -> total 4500( 300 * 15, class)
train_size=300
train_data=list()
valid_data=list()
# print(len(training_data))
for class_label in train_classes:
ct=0 for instance in training_data:
if ct< train_size:
train_data.append(instance)
ct+=1
continue
valid_data.append(instance)

#shuffle training data and validation data
np.random.shuffle(training_data)
np.random.shuffle(valid_data)
print(instance[0]) print(instance[1]) print(label_encoder.transform([instance[0]]))

print(to_one_hot([7],15))

#make data (img, class[onehot])

train_data = [(instance[1], to_one_hot(label_encoder.transform([instance[0]]), len(train_classes)))for instance in train_data]
valid_data = [(instance[1], to_one_hot(label_encoder.transform([instance[0]]), len(train_classes)))for instance in valid_data]

x_train, y_train = zip(*train_data)
x_train, y_train = np.squeeze(np.asarray(x_train)), np.squeeze(np.asarray(y_train))

x_train = normalize(x_train, norm='l2')

# FORM X_VALID AND Y_VALID
x_valid, y_valid = zip(*valid_data)
x_valid, y_valid = np.squeeze(np.asarray(x_valid)), np.squeeze(np.asarray(y_valid))

# L2 NORMALIZE X_VALID
x_valid = normalize(x_valid, norm='l2')

# FORM X_ZSL AND Y_ZSL
y_zsl, x_zsl = zip(*zero_shot_data)
x_zsl, y_zsl = np.squeeze(np.asarray(x_zsl)), np.squeeze(np.asarray(y_zsl))
# L2 NORMALIZE X_ZSL
x_zsl = normalize(x_zsl, norm='l2')

print("-> data loading is completed.")

return (x_train, x_valid, x_zsl), (y_train, y_valid, y_zsl)

def train_model(model,train_data,valid_data):
device='cuda'
training_epochs=80
criterion = torch.nn.CrossEntropyLoss().to(device) # Softmax is internally computed.
optimizer = torch.optim.Adam(model.parameters(), lr=0.0005)

# train my model
model.train() # set the model to train mode (dropout=True)
model.float()
print('Learning started. It takes sometime.')
for epoch in range(training_epochs):
avg_cost = 0
for X, Y in zip(*train_data):
Y=un_one_hot(Y,15)
Y=np.asarray(Y)
X = torch.from_numpy(X).to(device)
Y = torch.from_numpy(Y).to(device)
optimizer.zero_grad()
hypothesis = model(X.unsqueeze(0))
cost = criterion(hypothesis, Y)
cost.backward()
optimizer.step()
print('Epoch:{} is completed . time:{}'.format(epoch+1,time.time()-start))

print('Learning Finished!')

def main():
device='cuda'
global start
start=time.time()

VGGmodel = VGGnet.vgg16(pretrained=True)
VGGmodel.classifier = nn.Sequential(*[VGGmodel.classifier[i] for i in range(4)])

# ---------------------------------------------------------------------------------------------------------------- #
# ---------------------------------------------------------------------------------------------------------------- #
# SET HYPERPARAMETERS

global NUM_CLASS, NUM_ATTR, EPOCH, BATCH_SIZE
NUM_CLASS = 15
NUM_ATTR = 300
BATCH_SIZE = 128
EPOCH = 30

# ---------------------------------------------------------------------------------------------------------------- #
# ---------------------------------------------------------------------------------------------------------------- #
# TRAINING PHASE

(x_train, x_valid, x_zsl), (y_train, y_valid, y_zsl) = load_data()
word2vec_model=Word2Vec_Model().to(device)

print((x_train,y_train))

train_model(word2vec_model, (x_train, y_train), (x_valid, y_valid))

print(word2vec_model)

print("-------------------------------------")

ZSL_layers=list(word2vec_model.children())[:-2]
ZSL_model=nn.Sequential(*ZSL_layers).cuda()

print(ZSL_model)

# EVALUATION OF ZERO-SHOT LEARNING PERFORMANCE

class_vectors = sorted(np.load(WORD2VECPATH), key=lambda x: x[0]) classnames, vectors = zip(*class_vectors)
classnames = list(classnames)
vectors = np.asarray(vectors, dtype=np.float)

print(vectors[0].shape)

tree=KDTree(vectors)

print(tree)
print(x_zsl.shape)
x_zsl=torch.from_numpy(x_zsl).to(device)
pred_zsl=ZSL_model(x_zsl.unsqueeze(0))

print(pred_zsl.shape)

pred_zsl=pred_zsl.squeeze(0)

print(pred_zsl.shape)

top5, top3, top1 = 0, 0, 0

with torch.no_grad():
for i, pred in enumerate(pred_zsl):
pred=pred.to('cpu')
pred=pred.detach().numpy()
pred = np.expand_dims(pred, axis=0)
dist_5, index_5 = tree.query(pred, k=5)
pred_labels = [classnames[index] for index in index_5[0]]
true_label = y_zsl[i]
if true_label in pred_labels:
top5 += 1
if true_label in pred_labels[:3]: top3 += 1
if true_label in pred_labels[0]: top1 += 1

print()
print("ZERO SHOT LEARNING SCORE")
print("-> Top-5 Accuracy: %.2f" % (top5 / float(len(x_zsl))))
print("-> Top-3 Accuracy: %.2f" % (top3 / float(len(x_zsl))))
print("-> Top-1 Accuracy: %.2f" % (top1 / float(len(x_zsl))))

if __name__ == '__main__':
main()

-------------

# VGGnet.py

import torch.nn as nn
import torch.utils.model_zoo as zoo

__all__ = [
'VGG', 'vgg11', 'vgg11_bn', 'vgg13', 'vgg13_bn', 'vgg16', 'vgg16_bn',
'vgg19_bn', 'vgg19',
]

model_urls = {
'vgg11': 'https://download.pytorch.org/models/vgg11-bbd30ac9.pth', 'vgg13': 'https://download.pytorch.org/models/vgg13-c768596a.pth', 'vgg16': 'https://download.pytorch.org/models/vgg16-397923af.pth', 'vgg19': 'https://download.pytorch.org/models/vgg19-dcbb9e9d.pth', 'vgg11_bn': 'https://download.pytorch.org/models/vgg11_bn-6002323d.pth', 'vgg13_bn': 'https://download.pytorch.org/models/vgg13_bn-abd245e5.pth', 'vgg16_bn': 'https://download.pytorch.org/models/vgg16_bn-6c64b313.pth', 'vgg19_bn': 'https://download.pytorch.org/models/vgg19_bn-c79401a0.pth', }

class VGG(nn.Module):

def __init__(self, features, num_classes=1000, init_weights=True):
super(VGG, self).__init__()
self.features = features
self.avgpool = nn.AdaptiveAvgPool2d((7, 7))
self.classifier = nn.Sequential(
nn.Linear(512 * 7 * 7, 4096),
nn.ReLU(True),
nn.Dropout(),
nn.Linear(4096, 4096),
nn.ReLU(True),
nn.Dropout(),
nn.Linear(4096, num_classes)
) if init_weights:
self._initialize_weights()

def forward(self, x):
x = self.features(x)
x = self.avgpool(x)
x = x.view(x.size(0), -1)
x = self.classifier(x)
return x

def _initialize_weights(self):
for m in self.modules():
if isinstance(m, nn.Conv2d):
nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
if m.bias is not None:
nn.init.constant_(m.bias, 0)
elif isinstance(m, nn.BatchNorm2d):
nn.init.constant_(m.weight, 1)
nn.init.constant_(m.bias, 0)
elif isinstance(m, nn.Linear):
nn.init.normal_(m.weight, 0, 0.01)
nn.init.constant_(m.bias, 0)

def make_layers(cfg, batch_norm=False):
layers = []
in_channels = 3
for v in cfg:
if v == 'M':
layers += [nn.MaxPool2d(kernel_size=2, stride=2)]
else:
conv2d = nn.Conv2d(in_channels, v, kernel_size=3, padding=1)
if batch_norm:
layers += [conv2d, nn.BatchNorm2d(v), nn.ReLU(inplace=True)]
else:
layers += [conv2d, nn.ReLU(inplace=True)]
in_channels = v
return nn.Sequential(*layers)

cfgs = {
'A': [64, 'M', 128, 'M', 256, 256, 'M', 512, 512, 'M', 512, 512, 'M'], 'B': [64, 64, 'M', 128, 128, 'M', 256, 256, 'M', 512, 512, 'M', 512, 512, 'M'], 'D': [64, 64, 'M', 128, 128, 'M', 256, 256, 256, 'M', 512, 512, 512, 'M', 512, 512, 512, 'M'], 'E': [64, 64, 'M', 128, 128, 'M', 256, 256, 256, 256, 'M', 512, 512, 512, 512, 'M', 512, 512, 512, 512, 'M'], }

def _vgg(arch, cfg, batch_norm, pretrained, progress, **kwargs):
if pretrained:
kwargs['init_weights'] = False
model = VGG(make_layers(cfgs[cfg], batch_norm=batch_norm), **kwargs)
print(model_urls[arch]) if pretrained:
state_dict = zoo.load_url(model_urls[arch],progress=progress)
model.load_state_dict(state_dict)
return model

def vgg11(pretrained=False, progress=True, **kwargs):
"""VGG 11-layer model (configuration "A")

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg11', 'A', False, pretrained, progress, **kwargs)

def vgg11_bn(pretrained=False, progress=True, **kwargs):
"""VGG 11-layer model (configuration "A") with batch normalization

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg11_bn', 'A', True, pretrained, progress, **kwargs)

def vgg13(pretrained=False, progress=True, **kwargs):
"""VGG 13-layer model (configuration "B")

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg13', 'B', False, pretrained, progress, **kwargs)

def vgg13_bn(pretrained=False, progress=True, **kwargs):
"""VGG 13-layer model (configuration "B") with batch normalization

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg13_bn', 'B', True, pretrained, progress, **kwargs)

def vgg16(pretrained=False, progress=True, **kwargs):
"""VGG 16-layer model (configuration "D")

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg16', 'D', False, pretrained, progress, **kwargs)

def vgg16_bn(pretrained=False, progress=True, **kwargs):
"""VGG 16-layer model (configuration "D") with batch normalization

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg16_bn', 'D', True, pretrained, progress, **kwargs)

def vgg19(pretrained=False, progress=True, **kwargs):
"""VGG 19-layer model (configuration "E")

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg19', 'E', False, pretrained, progress, **kwargs)

def vgg19_bn(pretrained=False, progress=True, **kwargs):
"""VGG 19-layer model (configuration 'E') with batch normalization

Args:
pretrained (bool): If True, returns a model pre-trained on ImageNet
progress (bool): If True, displays a progress bar of the download to stderr
"""
return _vgg('vgg19_bn', 'E', True, pretrained, progress, **kwargs)