utils.py

import tensorflow as tf
import cv2
import numpy as np
import config
from logger import logging



def preprocess_video(video_path : str) -> tuple[tf.Tensor, list] :
 """
 Preprocess the video by keeping the required number of frames, 
 resizing the frames and normalizing the frames.

 params : 
 video_path : path of the video file

 returns :

 Returns tuple (input_tensor, frame_list)

 input_tensor : video with required number of frames and size
 frame_list : list of required number of frames 
 """

 logging.info(">>> Preprocessing the video....")

 # load the video
 video_capture = cv2.VideoCapture(video_path)

 # the number of frames in the original video
 original_number_of_frames = video_capture.get(cv2.CAP_PROP_FRAME_COUNT)

 # gap between two consecutive frames to capture
 frame_interval = int(original_number_of_frames / config.FRAME_NUM)

 new_video , frame_list = [] , []
 for i in range(0, config.FRAME_NUM ):
 video_capture.set(cv2.CAP_PROP_POS_FRAMES, i*frame_interval)
 success, frame = video_capture.read()

 if not success :
 logging.info("video loading failed")
 break
 
 frame_list.append(frame)
 # Resize the Frame to fixed height and width.
 resized_frame = cv2.resize(frame, (config.FRAME_HT, config.FRAME_WD))
 
 # Normalize the resized frame by dividing it with 255 so that each pixel value then lies between 0 and 1
 normalized_frame = resized_frame / 255
 
 # Append the normalized frame into the frames list
 new_video.append(normalized_frame)

 new_video_array = np.asarray(new_video)

 input_tensor = tf.expand_dims(new_video_array, axis=0)


 video_capture.release()

 logging.info("Video processing successful.")

 return input_tensor, frame_list


# Get top_k labels and probabilities
def get_top_k(probs, label_map,k=5 ):
 """Outputs the top k model labels and probabilities on the given video.

 Args:
 probs: probability tensor of shape (num_frames, num_classes) that represents
 the probability of each class on each frame.
 k: the number of top predictions to select.
 label_map: a list of labels to map logit indices to label strings.

 Returns:
 a tuple of the top-k labels and probabilities.
 """
 # Sort predictions to find top_k
 indices = tf.argsort(probs, direction='DESCENDING').numpy()[0][:k]
 # collect the labels of top_k predictions
 labels = tf.gather(label_map, indices).numpy()
 # decode lablels
 labels = [label.decode('utf8') for label in labels]
 # top_k probabilities of the predictions
 top_probs = tf.gather(probs[0], indices).numpy()
 
 output = dict()
 for label, prob in zip(labels, top_probs):
 output[label] = float(prob) / 100
 return output