pytorch_video_classification2.py

# -*- coding: utf-8 -*-
"""Pytorch_Video_Classification2.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1NWoWsh9lA0hYHaiSIdhkSzpo6vp9MZ8g
"""

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import os
from tqdm import tqdm
from PIL import Image
import matplotlib.pyplot as plt
import torch.nn.functional as F
from sklearn.preprocessing import LabelBinarizer
import cv2
import time
import joblib



dirs="/content/drive/MyDrive/dataset/Video_classification2/"

device=torch.device("cuda" if torch.cuda.is_available() else "cpu")

folders=["Driving","Tennis","football"]

data=pd.DataFrame()
counter=0
label=[]
images_folder=["jpg","JPG","PNG","png"]
for folder in folders:
  path_folders=os.path.join(dirs,folder)
  for imgs in os.listdir(path_folders):
    if imgs.split(".")[-1] in images_folder:
      full_path=os.path.join(path_folders,imgs)
      data.loc[counter,"image_folder"]=full_path
      data.loc[counter,"target"]=folder
      counter+=1
      label.append(folder)

data["target"].unique()

label = np.array(label)
lb=LabelBinarizer()
labels=lb.fit_transform(label)
labels

class_mapping={idx:value for value,idx in enumerate(np.unique(data["target"]))}
data["target"]=data["target"].map(class_mapping)

data["target"]=data["target"].astype(float)

data.head()

data.target.unique()

data = data.sample(frac=1).reset_index(drop=True)

data.to_csv(dirs+"data.csv",index=False)

joblib.dump(lb, dirs+'lb.pkl')

class CustomeCnn(nn.Module):
  def __init__(self):
    super().__init__()
    self.conv1=nn.Conv2d(3,16,5)
    self.conv2=nn.Conv2d(16,32,5)
    self.conv3=nn.Conv2d(32,64,3)
    self.conv4=nn.Conv2d(64,128,5)
    self.pool=nn.MaxPool2d(2,2)
    self.fc=nn.Linear(128,256)
    self.fc2=nn.Linear(256,len(class_mapping))
  
  def forward(self,x):
    x=F.relu(self.pool(self.conv1(x)))
    x=F.relu(self.pool(self.conv2(x)))
    x=F.relu(self.pool(self.conv3(x)))
    x=F.relu(self.pool(self.conv4(x)))
    bs,_,_,_=x.shape
    x=F.adaptive_avg_pool2d(x,1).reshape(bs,-1)
    x=F.relu(self.fc(x))
    x=self.fc2(x)
    return x

model=CustomeCnn()
model=model.to(device)

import torchvision.transforms as transforms
import cv2
from torch.utils.data import DataLoader,Dataset,random_split
from sklearn.model_selection import train_test_split
import albumentations

lr=1e-3
batch_size=32
device=torch.device("cuda" if torch.cuda.is_available() else "cpu")
device

data=pd.read_csv(dirs+"data.csv")
data.head()

data.image_folder[0],data.image_folder[1],data.image_folder[3]

X=data.image_folder.values
Y=data.target.values

X_train,x_test,Y_train,y_test=train_test_split(X,Y,test_size=0.10,random_state=42)

# custom dataset
class ImageDataset(Dataset):
    def __init__(self, images, labels=None, tfms=None):
        self.X = images
        self.y = labels

        # apply augmentations
        if tfms == 0: # if validating
            self.aug = albumentations.Compose([
                albumentations.Resize(224, 224, always_apply=True),
            ])
        else: # if training
            self.aug = albumentations.Compose([
                albumentations.Resize(224, 224, always_apply=True),
                albumentations.HorizontalFlip(p=0.5),
                albumentations.ShiftScaleRotate(
                    shift_limit=0.3,
                    scale_limit=0.3,
                    rotate_limit=15,
                    p=0.5
                ),
            ])
         
    def __len__(self):
        return (len(self.X))
    
    def __getitem__(self, i):
        image = Image.open(self.X[i])
        image = image.convert('RGB')
        image = self.aug(image=np.array(image))['image']
        image = np.transpose(image, (2, 0, 1)).astype(np.float32)
        label = self.y[i]
        return (torch.tensor(image, dtype=torch.float), torch.tensor(label, dtype=torch.long))

train_data=ImageDataset(X_train,Y_train,tfms=1)
test_data=ImageDataset(x_test,y_test,tfms=0)

len(train_data),len(test_data)

train_loader=DataLoader(train_data,batch_size=batch_size,shuffle=True)
test_loader=DataLoader(test_data,batch_size=batch_size,shuffle=False)

for i,j in train_loader:
  print(i.shape)
  break

total_params=(p.numel() for p in model.parameters())
print("f {total_params}")
total_trainable_params=sum(p.numel() for p in model.parameters() if p.requires_grad)

print(f"{total_trainable_params} training_parameters")

optimizer=torch.optim.SGD(model.parameters(),lr=0.01)
criterion=nn.CrossEntropyLoss()

def fit(model,train_loader):
  model.train()
  train_running_loss=0
  train_running_correct=0
  for i,data in enumerate(train_loader):
    data,target=data[0].to(device),data[1].to(device)
    optimizer.zero_grad()
    outputs=model(data)
    loss=criterion(outputs,target)
    train_running_loss+=loss.item()
    _,preds=torch.max(outputs.data,1)
    train_running_correct+=(preds == target).sum().item()
    loss.backward()
    optimizer.step()

  train_loss=train_running_loss / len(train_loader.dataset)0
  train_accuracy=100*train_running_correct / len(train_loader.dataset)

  print(f"Train Loss: {train_loss:.4f}, Train Acc: {train_accuracy:.2f}")

  return train_loss,train_accuracy

def validate(model, test_dataloader):
    print('Validating')
    model.eval()
    val_running_loss = 0.0
    val_running_correct = 0
    with torch.no_grad():
        for i, data in enumerate(test_dataloader):
            data, target = data[0].to(device), data[1].to(device)
            outputs = model(data)
            loss = criterion(outputs, target)
            
            val_running_loss += loss.item()
            _, preds = torch.max(outputs.data, 1)
            val_running_correct += (preds == target).sum().item()
        
        val_loss = val_running_loss/len(test_dataloader.dataset)
        val_accuracy = 100. * val_running_correct/len(test_dataloader.dataset)
        print(f'Val Loss: {val_loss:.4f}, Val Acc: {val_accuracy:.2f}')

        
        
        return val_loss, val_accuracy

train_loss,train_accuracy=[],[]
val_loss,val_accuracy=[],[]
start=time.time()
for epoch in range(150):
  print(f"Epoch {epoch+1}")
  train_epoch_loss,train_epoch_accuracy=fit(model,train_loader)
  test_epoch_loss,test_epoch_accuracy=validate(model,test_loader)
  train_loss.append(train_epoch_loss)
  train_accuracy.append(train_epoch_accuracy)
  val_loss.append(test_epoch_loss)
  val_accuracy.append(test_epoch_accuracy)
  torch.save(model.state_dict(),dirs+"images_data.pt")

end = time.time()
print(f"{(end-start)/60:.3f} minutes")





# accuracy plots
plt.figure(figsize=(10, 7))
plt.plot(train_accuracy, color='green', label='train accuracy')
plt.plot(val_accuracy, color='blue', label='validataion accuracy')
plt.xlabel('Epochs')
plt.ylabel('Accuracy')
plt.legend()
plt.savefig(dirs+'accuracy.png')
plt.show()
# loss plots
plt.figure(figsize=(10, 7))
plt.plot(train_loss, color='orange', label='train loss')
plt.plot(val_loss, color='red', label='validataion loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.legend()
plt.savefig(dirs+'loss.png')
plt.show()
    
# serialize the model to disk
# print('Saving model...')
# torch.save(model.state_dict(), args['model'])
 
print('TRAINING COMPLETE')



from google.colab.patches import cv2_imshow

aug = albumentations.Compose([
    albumentations.Resize(224, 224),
    ])

# ls=[]

cap = cv2.VideoCapture(dirs+"Driving2.mp4")
if (cap.isOpened() == False):
    print('Error while trying to read video. Plese check again...')
# get the frame width and height
frame_width = int(cap.get(3))
frame_height = int(cap.get(4))

fourcc = cv2.VideoWriter_fourcc(*'MP4V')
out = cv2.VideoWriter(dirs+'driving_output.mp4', fourcc, 20.0, (640,480))


while(cap.isOpened()):
    # capture each frame of the video
    ret, frame = cap.read()
    if ret == True:
        model.eval()
        with torch.no_grad():
            # conver to PIL RGB format before predictions
            pil_image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
            pil_image = aug(image=np.array(pil_image))['image']
            pil_image = np.transpose(pil_image, (2, 0, 1)).astype(np.float32)
            pil_image = torch.tensor(pil_image, dtype=torch.float).to(device)
            pil_image = pil_image.unsqueeze(0)
            
            outputs = model(pil_image)
            _, preds = torch.max(outputs.data, 1)
        
        cv2.putText(frame, lb.classes_[preds], (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 200, 0), 4)
        out.write(frame)
        cv2_imshow(frame)
        # out.write(frame)
        # press `q` to exit
        if cv2.waitKey(27) & 0xFF == ord('q'):
            break
    else: 
        break
# release VideoCapture()
cap.release()
# close all frames and video windo





ls=[]

cap = cv2.VideoCapture(dirs+"tennis.mp4")
if (cap.isOpened() == False):
    print('Error while trying to read video. Plese check again...')
# get the frame width and height
frame_width = int(cap.get(3))
frame_height = int(cap.get(4))

fourcc = cv2.VideoWriter_fourcc(*'MP4V')
out = cv2.VideoWriter(dirs'tennis_output.mp4', fourcc, 20.0, (640,480))

# out = cv2.VideoWriter(str(args['output']), cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width,frame_height))


while(cap.isOpened()):
    # capture each frame of the video
    ret, frame = cap.read()
    if ret == True:
        model.eval()
        with torch.no_grad():
            # conver to PIL RGB format before predictions
            pil_image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
            pil_image = aug(image=np.array(pil_image))['image']
            pil_image = np.transpose(pil_image, (2, 0, 1)).astype(np.float32)
            pil_image = torch.tensor(pil_image, dtype=torch.float).to(device)
            pil_image = pil_image.unsqueeze(0)
            
            outputs = model(pil_image)
            _, preds = torch.max(outputs.data, 1)
        
        cv2.putText(frame, lb.classes_[preds], (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 200, 0), 4)
        out.write(frame)
        cv2_imshow(frame)
        # out.write(frame)
        # press `q` to exit
        if cv2.waitKey(27) & 0xFF == ord('q'):
            break
    else: 
        break
# release VideoCapture()
cap.release()
# close all frames and video windo







ls=[]

cap = cv2.VideoCapture(dirs+"football.mp4")

if (cap.isOpened() == False):
    print('Error while trying to read video. Plese check again...')
# get the frame width and height
frame_width = int(cap.get(3))
frame_height = int(cap.get(4))
fourcc = cv2.VideoWriter_fourcc(*'MP4V')
out = cv2.VideoWriter(dirs+'football_output.mp4', fourcc, 20.0, (640,480))


# out = cv2.VideoWriter(str(args['output']), cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width,frame_height))


while(cap.isOpened()):
    # capture each frame of the video
    ret, frame = cap.read()
    if ret == True:
        model.eval()
        with torch.no_grad():
            # conver to PIL RGB format before predictions
            pil_image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
            pil_image = aug(image=np.array(pil_image))['image']
            pil_image = np.transpose(pil_image, (2, 0, 1)).astype(np.float32)
            pil_image = torch.tensor(pil_image, dtype=torch.float).to(device)
            pil_image = pil_image.unsqueeze(0)
            
            outputs = model(pil_image)
            _, preds = torch.max(outputs.data, 1)
        
        cv2.putText(frame, lb.classes_[preds], (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 200, 0), 4)
        out.write(frame)
        cv2_imshow(frame)
        # out.write(frame)
        # press `q` to exit
        if cv2.waitKey(27) & 0xFF == ord('q'):
            break
    else: 
        break
# release VideoCapture()
cap.release()
# close all frames and video windo