logging added

app.py

@@ -1,15 +1,29 @@
 import gradio as gr
 import utils
 from config import KINETICS_600_LABELS, MODEL
+from logger import logging
+
 
 def get_predictions(video_path):
- video, frame_list = utils.preprocess_video(video_path)
+
+ logging.info(f">>> Getting predictions for video file : {video_path}")
+
+ video, _ = utils.preprocess_video(video_path)
  model = MODEL
  probs = model(video)
  labels = utils.get_top_k(probs, label_map=KINETICS_600_LABELS)
+
+ logging.info(f"Getting predictions successful : {labels}")
+
  return labels
 
+
 label = gr.components.Label(num_top_classes=5)
 vd = gr.components.Video()
+
+logging.info(">>> Launching the gradio app... ")
+
 iface = gr.Interface(fn=get_predictions, inputs=vd, outputs=label)
-iface.launch(debug=True)
+iface.launch(share=True)
+
+logging.info(">>> Launched successfully.")

config.py

@@ -1,20 +1,31 @@
 import tensorflow as tf
 import tensorflow_hub as hub
-from keras.models import load_model
 from pathlib import Path
 import numpy as np
 import config
 import os
+from logger import logging
 
 
 FRAME_HT = 224
 FRAME_WD = 224
 FRAME_NUM = 8
+
+# tensorflow urls to download the model and lables
 TENSORFLOW_HUB_URL_LABELS = "https://raw.githubusercontent.com/tensorflow/models/f8af2291cced43fc9f1d9b41ddbf772ae7b0d7d2/official/projects/movinet/files/kinetics_600_labels.txt"
 TENSORFLOW_HUB_URL_MODEL = "https://tfhub.dev/tensorflow/movinet/a2/base/kinetics-600/classification/3"
+
 MODEL_PATH = os.path.join(os.getcwd(), 'models', 'Activity_recognition.h5')
 
-def get_labels():
+
+def get_labels() :
+ """
+ Downloads and saves the labels for tensorflow 'movienet' model.
+
+ Returns the path of the file 'labels.txt' where the labels are saved.
+ """
+
+ logging.info(">>> Downloading the labels 'movienet' model... ")
 
  labels_path = tf.keras.utils.get_file(
  fname=os.path.join(os.getcwd(), 'labels.txt'),
@@ -26,10 +37,20 @@ def get_labels():
  lines = labels_path.read_text().splitlines()
  KINETICS_600_LABELS = np.array([line.strip() for line in lines])
 
+ logging.info("Labels retrieved successfully.")
+
  return KINETICS_600_LABELS
 
 
-def get_model():
+def get_model() -> tf.keras.models.Model :
+ """
+ Downloads the tensorflow 'movienet' model. 
+
+ Returns tensorflow.keras.models.Model object instance.
+ """
+
+ logging.info(">>> Downloading the 'movienet' model from tensorflow...")
+
  encoder = hub.KerasLayer(TENSORFLOW_HUB_URL_MODEL, trainable=True)
 
  inputs = tf.keras.layers.Input(
@@ -38,12 +59,14 @@ def get_model():
  name='image'
  )
 
- # [batch_size, 600] 
  outputs = encoder(dict(image=inputs))
 
  model = tf.keras.Model(inputs, outputs, name='movinet')
 
+ logging.info("Model downloaded successfully.")
+
  return model
 
-KINETICS_600_LABELS = get_labels()
+
 MODEL = get_model()
+KINETICS_600_LABELS = get_labels()

logger.py

@@ -0,0 +1,7 @@
+import logging
+
+logging.basicConfig(
+ format="[ %(asctime)s ] %(lineno)d %(name)s - %(levelname)s %(message)s",
+ level=logging.INFO
+)
+

utils.py

@@ -1,50 +1,67 @@
 import tensorflow as tf
 import cv2
-import os
 import numpy as np
-from pathlib import Path
 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
 
-def preprocess_video(video_path : str) :
- # load the video
- video_capture = cv2.VideoCapture(video_path)
+ returns :
 
- # the number of frames in the original video
- original_number_of_frames = video_capture.get(cv2.CAP_PROP_FRAME_COUNT)
+ Returns tuple (input_tensor, frame_list)
 
- # gap between two consecutive frames to capture
- frame_interval = int(original_number_of_frames / config.FRAME_NUM)
+ input_tensor : video with required number of frames and size
+ frame_list : list of required number of frames 
+ """
 
- 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()
+ logging.info(">>> Preprocessing the video....")
 
- if not success :
- print("video loading failed")
- 
- 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)
+ # 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)
 
- new_video_array = np.asarray(new_video)
+ input_tensor = tf.expand_dims(new_video_array, axis=0)
 
- input_tensor = tf.expand_dims(new_video_array, axis=0)
 
+ video_capture.release()
 
- video_capture.release()
+ logging.info("Video processing successful.")
 
- return input_tensor, frame_list
+ return input_tensor, frame_list
 
 
 # Get top_k labels and probabilities
@@ -72,5 +89,4 @@ def get_top_k(probs, label_map,k=5 ):
  output = dict()
  for label, prob in zip(labels, top_probs):
  output[label] = float(prob) / 100
- print(output)
  return output