<jupyter_start><jupyter_text>Lab 01 : Video Classification - exerciseThe goal is to implement three architectures (late fusion 2D-CNN, early fusion 2D-CNN, 3D-CNN) that classifies video sequences.<jupyter_code># For Google Colaboratory
import sys, os
if 'google.colab' in sys.modules:
    # mount google drive
    from google.colab import drive
    drive.mount('/content/gdrive')
    path_to_file = '/content/gdrive/My Drive/CS4243_codes/codes/labs_lecture14'
    print(path_to_file)
    # move to Google Drive directory
    os.chdir(path_to_file)
    !pwd
import torch
import torch.nn as nn
import torch.optim as optim
import utils
import time
#device= torch.device("cuda")
device= torch.device("cpu")
print(device)<jupyter_output><empty_output><jupyter_text>Generate the dataset<jupyter_code># Libraries
#from torchvision.transforms import ToTensor
#from PIL import Image
%matplotlib notebook 
import matplotlib.pyplot as plt
plt.rcParams.update({'figure.max_open_warning': 0})

# Global constants
# im_size = image size  
# nb_action_classes = number of action classes
# nb_steps = number of time steps
# batch_size = nb_action_classes * nb_steps
# radius_actions = radius of the circular movement 
im_size = 28 
nb_action_classes = 3
nb_steps = 10
batch_size = nb_action_classes
radius_actions = 6

# Function that generate a batch of training data
def generate_batch_data(im_size, batch_size, nb_action_classes, radius_actions, nb_steps):
    batch_videos = torch.zeros(nb_action_classes,nb_steps,im_size,im_size)
    batch_labels = torch.zeros(nb_action_classes)
    
    offset = radius_actions
    coord_objects = torch.LongTensor(2).random_(-offset,offset)
    
    # class action 0
    action_class = 0
    video = torch.zeros(nb_steps,im_size,im_size) 
    for k in range(nb_steps):        
        y = im_size//2 + (radius_actions* torch.sin(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[1]
        x = im_size//2 + (radius_actions* torch.cos(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[0]
        video[k,y,x] = 1.0
    batch_videos[action_class,:,:,:] = video
    batch_labels[action_class] = action_class
    # class action 1
    action_class = 1
    video = torch.zeros(nb_steps,im_size,im_size) 
    for k in range(nb_steps//2+1):        
        y = im_size//2 + (radius_actions* torch.sin(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[1]
        x = im_size//2 + (radius_actions* torch.cos(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[0]
        video[k,y,x] = 1.0
    for k in range(nb_steps-1,nb_steps//2,-1):        
        y = im_size//2 + (radius_actions* torch.sin(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[1]
        x = im_size//2 + (radius_actions* torch.cos(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[0]
        video[nb_steps//2+nb_steps-k,y,x] = 1.0
    batch_videos[action_class,:,:,:] = video
    batch_labels[action_class] = action_class   
    # class action 2
    action_class = 2
    video = torch.zeros(nb_steps,im_size,im_size) 
    for k in range(nb_steps//2,-1,-1):  
        y = im_size//2 + (radius_actions* torch.sin(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[1]
        x = im_size//2 + (radius_actions* torch.cos(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[0]
        video[nb_steps//2-k,y,x] = 1.0
    for k in range(nb_steps//2+1,nb_steps):        
        y = im_size//2 + (radius_actions* torch.sin(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[1]
        x = im_size//2 + (radius_actions* torch.cos(torch.tensor(k*2*3.1415/nb_steps))).long() + coord_objects[0]
        video[k,y,x] = 1.0
    batch_videos[action_class,:,:,:] = video
    batch_labels[action_class] = action_class  
    return batch_videos, batch_labels

# Plot a mini-batch of images
batch_videos, batch_labels = generate_batch_data(im_size, batch_size, nb_action_classes, radius_actions, nb_steps)
print(batch_videos.size())
print(batch_labels.size())
for a in range(nb_action_classes):
    fig = plt.figure(1,figsize=(7,7))
    plt.show(block=False)
    for k in range(nb_steps):
        plt.imshow(batch_videos[a,k,:,:], cmap='gray')
        plt.title('Action class : ' + str(a) )
        fig.canvas.draw()
        time.sleep(0.15)
    plt.close()
fig = plt.figure(1,figsize=(7,7))
plt.show(block=False)
sum_images = torch.zeros(im_size,im_size)
for t in range(nb_steps):
    sum_images = sum_images + batch_videos[a,t,:,:]
plt.imshow(sum_images, cmap='gray')
plt.title('Sum of temporal images')
fig.canvas.draw()

# Define the video classification network architecture
class videoNN(nn.Module):
    
    def __init__(self, type_videocnn):
        super(videoNN, self).__init__()
        # Hyper-parameters
        self.type_videocnn = type_videocnn
        hidden_dim = 50
        # Single-frame/late fusion 2D-CNN
        if type_videocnn == 'late_fusion':
            # conv layer
            # COMPLETE HERE
            
            # Classification layer from flattened conv features
            # COMPLETE HERE
            
        # Early fusion 2D-CNN
        if type_videocnn == 'early_fusion':
            # conv layer
            # COMPLETE HERE
            
            # Classification layer from flattened conv features
            # COMPLETE HERE
            
        if type_videocnn == '3d_cnn':
            # conv layer
            # COMPLETE HERE
            
            # Classification layer from flattened conv features
            # COMPLETE HERE
        
    def forward(self, x): 
        # Single-frame/late fusion 2D-CNN
        if self.type_videocnn == 'late_fusion':
            # Conv2D for each time frame independently
            # COMPLETE HERE
            
            # Classification layer
            
        # Early fusion 2D-CNN
        if self.type_videocnn == 'early_fusion':
            # Conv2D for concatenated frames
            # COMPLETE HERE
            
            # Classification layer
            
        # 3D-CNN
        if self.type_videocnn == '3d_cnn':
            # Conv3D for volume time x space
            # COMPLETE HERE
            
            # Classification layer
            
        return scores_action_class
    
# Instantiate the network
type_videocnn = 'late_fusion'
#type_videocnn = 'early_fusion'
#type_videocnn = '3d_cnn'
net = videoNN(type_videocnn)
net = net.to(device)
print(net)
utils.display_num_param(net) 

# Test the forward pass, backward pass and gradient update with a single batch
init_lr = 0.001
optimizer = torch.optim.Adam(net.parameters(), lr=init_lr)
batch_videos, batch_labels = generate_batch_data(im_size, batch_size, nb_action_classes, radius_actions, nb_steps)
print(batch_videos.size())
print(batch_labels.size())
optimizer.zero_grad()
scores_action_class = net(batch_videos) # [batch_size, nb_action_classes] = [3, 3]
batch_labels = batch_labels.long() # [batch_size] = [3]
# loss
loss = nn.CrossEntropyLoss()(scores_action_class, batch_labels)
loss.backward()
optimizer.step()

# Training loop
type_videocnn = 'late_fusion'
#type_videocnn = 'early_fusion'
#type_videocnn = '3d_cnn'
print('Architecture : ',type_videocnn)
net = videoNN(type_videocnn)
net = net.to(device)
utils.display_num_param(net) 

# Optimizer
init_lr = 0.001
optimizer = torch.optim.Adam(net.parameters(), lr=init_lr)

# Number of mini-batches per epoch
nb_batch = 25

start=time.time()
for epoch in range(10):

    running_loss = 0.0
    num_batches = 0
    
    for _ in range(nb_batch):
        
        # FORWARD AND BACKWARD PASS
        batch_videos, batch_labels = generate_batch_data(im_size, batch_size, nb_action_classes, radius_actions, nb_steps)
        optimizer.zero_grad()
        scores_action_class = net(batch_videos) # [batch_size, nb_action_classes] = [3, 3]
        batch_labels = batch_labels.long() # [batch_size] = [3]
        # loss
        loss = nn.CrossEntropyLoss()(scores_action_class, batch_labels)
        loss.backward()
        optimizer.step()

        # COMPUTE STATS
        running_loss += loss.detach().item()
        num_batches += 1        
    
    # AVERAGE STATS THEN DISPLAY
    total_loss = running_loss/num_batches
    elapsed = (time.time()-start)/60
    print('epoch=',epoch, '\t time=', elapsed,'min', '\t lr=', init_lr  ,'\t loss=', total_loss )
    
# Test time

# generate the batch of 3 actions
batch_videos, batch_labels = generate_batch_data(im_size, batch_size, nb_action_classes, radius_actions, nb_steps)

# forward pass
scores_action_class = net(batch_videos) # [batch_size, nb_action_classes] = [3, 3]

# class prediction
pred_action_class = torch.argmax(scores_action_class, dim=1) # [batch_size] = [3]

print('Predicted action class : ',pred_action_class)
print('Label action class     : ',batch_labels.long())<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11424365942028986, 'fraction_numerical': 0.020267210144927536, 'mean_word_length': 3.2102001906577695, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '93245', 'license': 'MIT', 'n_tokens_mistral': 3182, 'n_tokens_neox': 3028, 'n_words': 620, 'path': 'codes/labs_lecture14/lecture14_video_classification_exercise.ipynb', 'repo_name': 'johanesrafael/CS4243_2022', 'size': '8832'}