<jupyter_start><jupyter_text>Getting the dataset and installing necessary librariesDataset link : https://data.vision.ee.ethz.ch/cvl/gfanelli/head_pose/head_forest.htmldb<jupyter_code>!wget http://data.vision.ee.ethz.ch/cvl/gfanelli/kinect_head_pose_db.tgz<jupyter_output>--2019-12-12 16:03:11--  http://data.vision.ee.ethz.ch/cvl/gfanelli/kinect_head_pose_db.tgz
Resolving data.vision.ee.ethz.ch (data.vision.ee.ethz.ch)... 172.16.58.3
Connecting to data.vision.ee.ethz.ch (data.vision.ee.ethz.ch)|172.16.58.3|:80... connected.
HTTP request sent, awaiting response... 302 Found
Location: https://data.vision.ee.ethz.ch/cvl/gfanelli/kinect_head_pose_db.tgz [following]
--2019-12-12 16:03:11--  https://data.vision.ee.ethz.ch/cvl/gfanelli/kinect_head_pose_db.tgz
Connecting to data.vision.ee.ethz.ch (data.vision.ee.ethz.ch)|172.16.58.3|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 6014398431 (5.6G) [application/x-gzip]
Saving to: ‘kinect_head_pose_db.tgz’

kinect_head_pose_db 100%[===================>]   5.60G  12.8MB/s    in 6m 33s  

2019-12-12 16:09:44 (14.6 MB/s) - ‘kinect_head_pose_db.tgz’ saved [6014398431/6014398431]<jupyter_text>Extracting the .tgz file<jupyter_code>!tar -xf '/content/kinect_head_pose_db.tgz'
import os
import cv2
from math import atan2, asin 
import numpy as np
import pandas as pd
import dlib
import math
import matplotlib.pyplot as plt
from tqdm import tqdm_notebook as tqdm
from torch.utils.data import DataLoader, Dataset, sampler
from torch.optim.lr_scheduler import ReduceLROnPlateau
from albumentations.pytorch import ToTensor
from albumentations import (HorizontalFlip, ShiftScaleRotate, Normalize, Resize, Compose, GaussNoise)
import torch
from torchvision import transforms
import torch.nn as nn
from torch.nn import functional as F
import torch.optim as optim
import torch.backends.cudnn as cudnn
import torchvision.models as models
import time
!wget http://dlib.net/files/mmod_human_face_detector.dat.bz2
!7z x "mmod_human_face_detector.dat.bz2"
detector = dlib.cnn_face_detection_model_v1("mmod_human_face_detector.dat")
!rm -r "/content/cropped_Images"<jupyter_output><empty_output><jupyter_text>Useful functions Function to get the path<jupyter_code>def PathList(path):
  Img_Path = []
  for root, dirs, files in os.walk(path, topdown=False):
      for filename in files:
        if(filename[-3:] == 'png'):
          Img_Path.append(root+'/'+filename)
  return Img_Path<jupyter_output><empty_output><jupyter_text>Here ImagesPath is a list that contains all the .png files from the dataset<jupyter_code>ImagesPath = PathList("hpdb/")<jupyter_output><empty_output><jupyter_text>This function is used to convert the transformation matrix from all the .txt files to Yaw, Pitch and Roll<jupyter_code>def getAngles(path):
  rot_mat = []
  with open(path, "r") as f:
    l = f.read()
    l = l.split("\n")
    for i in range(0, 3): 
      rot_mat.append(l[i].split(" ")[:3])
    rot_mat = np.asarray(rot_mat)
    rot_mat = rot_mat.astype(float) 
    # roll = atan2(rot_mat[2][1], rot_mat[2][2])*(180/math.pi);
    # pitch = asin(rot_mat[2][0])*(180/math.pi);
    # yaw = -atan2(rot_mat[1][0], rot_mat[0][0])*(180/math.pi);
    roll = -np.arctan2(rot_mat[1][0], rot_mat[0][0]) * 180 / np.pi
    yaw = -np.arctan2(-rot_mat[2][0], np.sqrt(rot_mat[2][1] ** 2 + rot_mat[2][2] ** 2)) * 180 / np.pi
    pitch = np.arctan2(rot_mat[2][1], rot_mat[2][2]) * 180 / np.pi
  return yaw, pitch, roll 
!mkdir cropped_Images
for i in range(1,25):
  path = "cropped_Images/"+"{:02d}".format(i)
  if not os.path.exists(path):
    os.mkdir(path)
a = np.load("/content/drive/My Drive/CropSizeList.npy")
# def SaveImages(CropSizeList, ImagesPath):
#   for i in tqdm(range(len(ImagesPath))):
#      PersonNumber = int(ImagesPath[i][5:7])
#      top, bottom, left, right = CropSizeList[PersonNumber - 1] 
#      img = cv2.imread(ImagesPath[i])[top:bottom, left:right]
#      cv2.imwrite("cropped_Images/" + ImagesPath[i][5:], img)
# SaveImages(a, ImagesPath)
CroppedImagesPath = PathList("cropped_Images/")<jupyter_output><empty_output><jupyter_text>This function is used to crop out unnecessary background from the .png files which will later be used for the face extraction<jupyter_code>def crop():
  j = 1
  k = 1
  rows = 24
  columns = 2
  fig=plt.figure(figsize=(20, 100))
  for i in tqdm(range(1, rows*columns + 1)):
    if i%2 == 0:
        img = plt.imread("hpdb/" + "{:02d}".format(j) + "/frame_00005_rgb.png")
        j += 1
    else:
        img = plt.imread("hpdb/" + "{:02d}".format(k) + "/frame_00005_rgb.png")[65:-55,200:540]
        k += 1
    fig.add_subplot(rows, columns, i)
    plt.imshow(img,cmap = 'gray')
  plt.show()
crop()<jupyter_output><empty_output><jupyter_text>Making a seperate directory to store the faces This function will get the faces and save them in the directory created above<jupyter_code>def ExtractAndSaveFaces(detector,ImagesPath):
    jpg_crop_path = []
    count = 0
    count_fail = 0
    for i in tqdm(range(len(ImagesPath))):
        image = cv2.imread(ImagesPath[i])[100:-75,200:540]
        dets = detector(image, 0)
        if len(dets) == 1: 
            crop_img = image[dets[0].rect.top()-30:dets[0].rect.bottom()+30,dets[0].rect.left()-30:dets[0].rect.right()+30] # cropping the image with 15 extra pixels on all side
            if crop_img.size == 0:
                count_fail+=1
                continue
            cv2.imwrite("cropped_Images/" + ImagesPath[i][5:] , crop_img) # writing the image to directory
            jpg_crop_path.append("cropped_Images/" + ImagesPath[i][5:])
            count = count+1
    print(str(count) + " image written")
    print(str(count_fail) + " image failed")
    return jpg_crop_path<jupyter_output><empty_output><jupyter_text>Getting the path of final images that will be fed to the model<jupyter_code>CroppedImagesPath = ExtractAndSaveFaces(detector,ImagesPath)<jupyter_output><empty_output><jupyter_text>This function will return a dataframe which contains the image id along with its yaw pitch and roll<jupyter_code>def getDataframe(CroppedImagesPath):
    df_train = pd.DataFrame(columns = ['Person','image','pitch','yaw','roll'])
    df_val = pd.DataFrame(columns = ['Person','image','pitch','yaw','roll'])
    # df_test = pd.DataFrame(columns = ['Person','image','pitch','yaw','roll'])
    for i in tqdm(range(len(CroppedImagesPath))):
        yaw, pitch, roll = getAngles("hpdb/"+CroppedImagesPath[i][15:30]+"pose.txt")
        if(int(CroppedImagesPath[i][15:17]) < 17):
          df_train = df_train.append({'Person':CroppedImagesPath[i][15:17],'image':CroppedImagesPath[i][18:29],'pitch':pitch,'yaw':yaw,'roll':roll},ignore_index = True)
        # elif(int(CroppedImagesPath[i][15:17]) >= 17 and int(CroppedImagesPath[i][15:17]) <= 20):
        #   df_val = df_val.append({'Person':CroppedImagesPath[i][15:17],'image':CroppedImagesPath[i][18:29],'pitch':pitch,'yaw':yaw,'roll':roll},ignore_index = True)  
        else:
          df_val = df_val.append({'Person':CroppedImagesPath[i][15:17],'image':CroppedImagesPath[i][18:29],'pitch':pitch,'yaw':yaw,'roll':roll},ignore_index = True)  
    return df_train, df_val<jupyter_output><empty_output><jupyter_text>Dataset Class<jupyter_code>class BIWIDataset(Dataset):
    def __init__(self, df, path, phase):
        self.df = df
        self.phase = phase
        self.transform = ToTensor()
        self.path = path
        self.train = [path[i] for i in range(len(path)) if int(path[i][15:17]) <= 16]
        self.val = [path[i] for i in range(len(path)) if int(path[i][15:17]) > 16]
        self.fnames = self.df.index.tolist()

    def __getitem__(self, idx):
        if(self.phase == 'train'):
          Image = cv2.imread(self.train[idx])
          Angle = np.array(self.df.iloc[idx,2:])
        elif(self.phase == 'val'):
          Image = cv2.imread(self.val[idx])
          Angle = np.array(self.df.iloc[idx,2:])
        Angle = Angle.astype('float64')
        Image = cv2.resize(Image,(120,120))
        Image = Image.transpose(2,0,1)  # Making Channel First 
        Image = torch.from_numpy(Image).type(torch.FloatTensor)
        Angle = torch.from_numpy(Angle).type(torch.FloatTensor)  
        return Image, Angle

    def __len__(self):
        return len(self.fnames)
df_train, df_val = getDataframe(CroppedImagesPath)
df_train
df = df_train.append(df_val, ignore_index=True)
df.sample(frac=1).reset_index(inplace = True, drop=True)
df
df
df.to_csv("df.csv", index = False)<jupyter_output><empty_output><jupyter_text>This function provides the training and validation dataloader accordingly<jupyter_code>def provider(path, phase,df_train, df_val,  batch_size=8, num_workers=0):    
    df = df_train if phase == "train" else df_val
    image_dataset = BIWIDataset(df, path, phase)
    dataloader = DataLoader(
        image_dataset,
        batch_size=batch_size,
        num_workers=num_workers,
        pin_memory=True,
        shuffle=True,   
    )
    return dataloader<jupyter_output><empty_output><jupyter_text>Trainer Class<jupyter_code>class Trainer(object):
    '''This class takes care of training and validpathation of our model'''
    def __init__(self, model, path, df_train, df_val):
        self.num_workers = 0
        self.batch_size = {"train": 16, "val": 8}
        self.accumulation_steps = 32 // self.batch_size['train']
        self.lr = 5e-5
        self.num_epochs = 25
        self.best_loss = float("inf")
        self.phases = ["train", "val"]
        self.device = torch.device("cuda:0")
        torch.set_default_tensor_type("torch.cuda.FloatTensor")
        self.net = model
        self.df_train = df_train
        self.df_val = df_val
        self.criterion = torch.nn.L1Loss()
        self.optimizer = optim.Adam(self.net.parameters(), lr=self.lr)
        self.scheduler = ReduceLROnPlateau(self.optimizer, mode="min", patience=3, verbose=True)
        self.net = self.net.to(self.device)
        cudnn.benchmark = True
        self.dataloaders = {
            phase: provider(
                path = path,
                phase=phase,
                df_train = df_train,
                df_val = df_val,
                batch_size=self.batch_size[phase],
                num_workers=self.num_workers
            )
            for phase in self.phases
        }
        self.losses = {phase: [] for phase in self.phases}
        
    def forward(self, images, targets):
        images = images.to(self.device)
        target = targets.to(self.device)
        outputs = self.net(images)
        loss = self.criterion(outputs, target)
        return loss, outputs

    def iterate(self, epoch, phase):
        start = time.strftime("%H:%M:%S")
        print(f"Starting epoch: {epoch} | phase: {phase} | ⏰: {start}")
        batch_size = self.batch_size[phase]
        self.net.train(phase == "train")
        dataloader = self.dataloaders[phase]
        running_loss = 0.0
        total_batches = len(dataloader)
        tk0 = tqdm(dataloader, total=total_batches)
        self.optimizer.zero_grad()
        for itr, batch in enumerate(tk0): # replace `dataloader` with `tk0` for tqdm
            images, targets = batch
            loss, outputs = self.forward(images, targets)
            loss = loss / self.accumulation_steps
            if phase == "train":
                loss.backward()
                if (itr + 1 ) % self.accumulation_steps == 0:
                    self.optimizer.step()
                    self.optimizer.zero_grad()
            running_loss += loss.item()
            outputs = outputs.detach().cpu()
        epoch_loss = (running_loss * self.accumulation_steps) / total_batches
        torch.cuda.empty_cache()
        print(f'loss:{epoch_loss}')
        return epoch_loss

    def train_end(self):
        train_loss = self.losses["train"]
        val_loss = self.losses["val"]
        df_data=np.array([train_loss,val_loss]).T
        df = pd.DataFrame(df_data,columns = ['train_loss','val_loss'])
        df.to_csv("Training_log.csv")

    def start(self):
        for epoch in range(self.num_epochs):
            train_loss = self.iterate(epoch, "train")
            self.losses["train"].append(train_loss)
            state = {
                "epoch": epoch,
                "best_loss": self.best_loss,
                "state_dict": self.net.state_dict(),
                "optimizer": self.optimizer.state_dict(),
            }
            with torch.no_grad():
                val_loss = self.iterate(epoch, "val")
                self.losses["val"].append(val_loss)
                self.scheduler.step(val_loss)
            if val_loss < self.best_loss:
                print("******** New optimal found, saving state ********")
                state["best_loss"] = self.best_loss = val_loss
                torch.save(state, "./model.pth")
            print()
            self.train_end()    

# class Tester(object):
#     '''This class takes care of training and validpathation of our model'''
#     def __init__(self, model, df_test):
#         self.num_workers = 0
#         self.batch_size = 16
#         self.accumulation_steps = 32 // self.batch_size['train']
#         self.lr = 5e-5
#         self.num_epochs = 25
#         self.best_loss = float("inf")
#         self.device = torch.device("cuda:0")
#         torch.set_default_tensor_type("torch.cuda.FloatTensor")
#         self.net = model
#         self.df_test = df_test
#         self.criterion = torch.nn.L1Loss()
#         self.net = self.net.to(self.device)
#         cudnn.benchmark = True
#         self.dataloaders = {
#             phase: provider(
#                 path = path,
#                 phase='test',
#                 df_train = df_train,
#                 df_val = df_val,
#                 batch_size=self.batch_size[phase],
#                 num_workers=self.num_workers
#             )
#             for phase in self.phases
#         }
#         # self.losses = {phase: [] for phase in self.phases}
        
#     def forward(self, images, targets):
#         images = images.to(self.device)
#         target = targets.to(self.device)
#         outputs = self.net(images)
#         loss = self.criterion(outputs, target)
#         return loss, outputs

#     def iterate(self, epoch, phase):
#         start = time.strftime("%H:%M:%S")
#         print(f"Starting epoch: {epoch} | phase: {phase} | ⏰: {start}")
#         batch_size = self.batch_size[phase]
#         self.net.train(phase == "train")
#         dataloader = self.dataloaders[phase]
#         running_loss = 0.0
#         total_batches = len(dataloader)
#         tk0 = tqdm(dataloader, total=total_batches)
#         self.optimizer.zero_grad()
#         for itr, batch in enumerate(tk0): # replace `dataloader` with `tk0` for tqdm
#             images, targets = batch
#             loss, outputs = self.forward(images, targets)
#             loss = loss / self.accumulation_steps
#             # if phase == "train":
#             #     loss.backward()
#             #     if (itr + 1 ) % self.accumulation_steps == 0:
#             #         self.optimizer.step()
#             #         self.optimizer.zero_grad()
#             running_loss += loss.item()
#             outputs = outputs.detach().cpu()
#         epoch_loss = (running_loss * self.accumulation_steps) / total_batches
#         torch.cuda.empty_cache()
#         print(f'loss:{epoch_loss}')
#         return epoch_loss

#     def train_end(self):
#         test_loss = self.losses["train"]
#         val_loss = self.losses["val"]
#         df_data=np.array([train_loss,val_loss]).T
#         df = pd.DataFrame(df_data,columns = ['train_loss','val_loss'])
#         df.to_csv("Training_log.csv")

#     def start(self):
#         for epoch in range(self.num_epochs):
#             # train_loss = self.iterate(epoch, "train")
#             # self.losses["train"].append(train_loss)
#             state = {
#                 "epoch": epoch,
#                 "best_loss": self.best_loss,
#                 "state_dict": self.net.state_dict(),
#                 "optimizer": self.optimizer.state_dict(),
#             }
#             with torch.no_grad():
#                 val_loss = self.iterate(epoch, "val")
#                 self.losses["val"].append(val_loss)
#                 self.scheduler.step(val_loss)
#             if val_loss < self.best_loss:
#                 print("******** New optimal found, saving state ********")
#                 state["best_loss"] = self.best_loss = val_loss
#                 torch.save(state, "./model.pth")
#             print()
#             self.train_end()<jupyter_output><empty_output><jupyter_text>Getting the model and Training<jupyter_code>!pip install efficientnet_pytorch
from efficientnet_pytorch import EfficientNet
model = EfficientNet.from_pretrained('efficientnet-b3')
# model = models.resnet50(pretrained = True)
num_ftrs = model._fc.in_features
model._fc = nn.Linear(num_ftrs, 3)
model_trainer = Trainer(model, CroppedImagesPath, df_train, df_val)
model_trainer.start()
model.eval()<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup