!pip install git+https://github.com/netbrainml/nbml.git
from nbml.pytorch import *
Load csv with pandas
import pandas as pd
def get_data(full=False, top=10):
df = pd.read_csv("../input/labels.csv")
if full: return df['id'], df['breed']
top_breeds = sorted(list(df['breed'].value_counts().head(top).index))
df_top = df[df['breed'].isin(top_breeds)]
return df_top['id'], df_top['breed']
imloc, label = get_data(full=False, top=20)
Use this to visualize our dataset
import matplotlib.pyplot as plt
from IPython.display import clear_output
import time
import cv2
from fastai.vision import Path
p = Path("../input/")
ptrain = p/"train"
for data in zip(imloc[:10],label[:10]):
plt.imshow(cv2.resize(cv2.imread(str(ptrain/(data[0]+".jpg"))),
(224,224)))
plt.title(data[1])
plt.show()
time.sleep(0.5)
clear_output(wait=True)
import gc
gc.collect()
Save in images and its label in numpy array
import numpy as np
from tqdm import tqdm
x = []
y = []
classes = list(set(label))
nc = len(classes)
for data in tqdm(zip(imloc,label),total=len(imloc)):
x.append(cv2.resize(cv2.imread(str(ptrain/(data[0]+".jpg"))), (224,224)))
y.append(classes.index(data[1]))
x = np.array(x)
y = np.array(y)
x.shape, y.shape
Convert to torch tensors
def getHist(train, valid, nc=120):
from collections import Counter
import operator
plt.figure(figsize=(8, 4) if nc<21 else (50, 25))
train_c = dict(Counter(train)); valid_c = dict(Counter(valid))
for cl in range(nc):
if cl not in train_c: train_c[cl] = 0
if cl not in valid_c: valid_c[cl] = 0
tlabels, tvalues = [list(m) for m in zip(*sorted(train_c.items(), key=operator.itemgetter(0)))]
vlabels, vvalues = [list(m) for m in zip(*sorted(valid_c.items(), key=operator.itemgetter(0)))]
width = 0.5; indices = np.arange(len(tlabels))
plt.bar(indices, tvalues, width=width,
color='b', label='Training')
plt.bar([i+0.25*width for i in indices], vvalues,
width=0.5*width, color='r', alpha=0.5, label='Validation')
plt.xticks(indices+width/2.,
['{}'.format(i) for i in range(len(tvalues))] )
plt.xticks(rotation=0 if nc<11 else 90)
plt.legend()
plt.show()
import torch
from torch.utils.data import DataLoader, TensorDataset
from fastai.vision import imagenet_stats, normalize
def nptotorch(x,y, split=0.6, subset=1000, verbose=True, nc=120):
x_t = torch.tensor(x).transpose(-1,1)
y_t = torch.tensor(y)
idx = get_idx(x_t)[:subset]
loc = int(split * subset)
if verbose: getHist(y[idx[:loc]], y[idx[loc:]], nc=nc)
tdl = DataLoader(TensorDataset(normalize(x_t[idx[:loc]].float(),torch.tensor(imagenet_stats[0]),torch.tensor(imagenet_stats[1]))
,y_t[idx[:loc]]
), batch_size = 40)
vdl = DataLoader(TensorDataset(normalize(x_t[idx[loc:]].float(),torch.tensor(imagenet_stats[0]),torch.tensor(imagenet_stats[1]))
,y_t[idx[loc:]]
), batch_size = 40)
return tdl,vdl
tdl,vdl = nptotorch(x,y, subset=x.shape[0], nc=nc)
Load in pretrained ResNet
from torchvision.models import resnet34
resnet34()
Freeze our head with torch.no_grad()
Has the option to apply discrimitative learning rate on transfer learning models.
It applies a different learning rate on the body and the head.
from progressbar import progressbar
class TransferTrainableClassifier(BasicTrainableClassifier):
def fit(self, train_ds, valid_ds, cbs=False,
epochs=1, learning_rate=1e-3, lr_seg = None):
self.train()
if lr_seg is not None:
op = [self.opt(self.body.parameters(), lr=lr_seg[0]), self.opt(self.head.parameters(), lr=lr_seg[1])]
else: op = self.opt(self.parameters(), lr=learning_rate)
for e in range(epochs):
torch.cuda.empty_cache() if torch.cuda.device_count() else None
for data in progressbar(train_ds):
op.zero_grad() if lr_seg is None else [o.zero_grad() for o in op]
pred = self(cc(data[0]))
loss = self.crit(pred, cc(data[1], self.dtype))
loss.backward(retain_graph = self.rg)
op.step() if lr_seg is None else [o.step() for o in op]
self.cbs_(e, train_ds, valid_ds=valid_ds) if cbs else None
This prevents the pretrained model from learning by only aggregating gradients on the head, or the untrained part.
Applies L2 regularization, which prevents the weights from being too large.
This is done by introducing the squared sum of all the weights to the loss function.
Thus, the optimizer tries to minimize the actual loss function in addition to all the weights.
from functools import partial
from fastai.torch_core import Module
def init_Linear(m):
if getattr(m, 'bias', None) is not None: nn.init.constant_(m.bias, 0)
if isinstance(m, (nn.Linear)): nn.init.kaiming_normal_(m.weight)
for l in m.children(): init_cnn(l)
class noop(Module):
def __call__(self, x): return x
class resCD(TransferTrainableClassifier):
def __init__(self, nc, frozen=True, **kwargs):
super().__init__(opt=partial(torch.optim.Adam, weight_decay=1e-2))
self.body = resnet34(**kwargs); self.body.fc = noop()
self.head = nn.Sequential(nn.Linear(512,256),
nn.Linear(256,128),
nn.Linear(128,nc))
self.frozen = frozen
init_Linear(self)
def forward(self,x):
if not self.frozen: return self.head(self.body(x))
with torch.no_grad(): fm = self.body(x)
return self.head(fm)
@property
def unfreeze(self):
self.frozen=False
@property
def freeze(self):
self.frozen=True
pt = resCD(nc, pretrained=True).cuda()
print(pt.frozen); time.sleep(1)
pt.fit(tdl,vdl,cbs=True,
epochs=3,learning_rate=1e-3)
pt.unfreeze; print(pt.frozen); time.sleep(1)
pt.fit(tdl,vdl,cbs=True,
epochs=7,lr_seg=[1e-6,1e-4])
npt = resCD(nc, pretrained=False).cuda()
print(npt.frozen); time.sleep(1)
npt.fit(tdl,vdl,cbs=True,
epochs=3,learning_rate=1e-3)
npt.unfreeze; print(npt.frozen); time.sleep(1)
npt.fit(tdl,vdl,cbs=True,
epochs=7,lr_seg=[1e-6,1e-4])
pt.plot;npt.plot