""" **Aim:** This is the final code of video blur along with UI **Author:** Shalu Singh **Starting Date:** 12/9/23 **Ending Date:** 14/1/24 """ # import libraries import tensorflow as tf import tensorflow_hub as hub import numpy as np from PIL import Image import cv2 import os import pandas as pd import keras import gradio from concurrent.futures import ThreadPoolExecutor from moviepy.editor import VideoFileClip, concatenate_videoclips # path to ouput video out_video_path = 'blured_op_video.mp4' # class label coco_classes = { 0: 'unlabeled', 1: 'person', 2: 'bicycle', 3: 'car', 4: 'motorcycle', 5: "airplane", 6: "bus", 7: "train", 8: "truck", 9: "boat", 10:" traffic light", 11: "fire hydrant", 12: "street sign", 13: "stop |sign", 14: "parking meter", 15: "bench", 16: "bird", 17: "cat", 18: "dog", 19: "horse", 20: "sheep", 21: "cow", 22: "elephant", 23:" bear", 24: "zebra", 25: "giraffe", 26: "hat", 27: "backpack", 28: "umbrella", 29: "shoe", 30: "eye glasses", 31: "handbag", 32:" tie", 33: "suitcase", 34:" frisbee", 35: "skis", 36: "snowboard", 37: "sports ball", 38: "kite", 39: "baseball bat", 40: "baseball glove", 41: "skateboard", 42: "surfboard", 43: "tennis racket", 44: "bottle", 45: "plate", 46: "wine glass", 47: "cup", 48: "fork", 49: "knife", 50: "spoon", 51: "bowl", 52: "banana", 53:"apple", 54:"sandwich", 55:" orange", 56: "broccoli", 57: "carrot", 58: "hot dog", 59:' pizza', 60: "donut", 61: 'cake', 62: "chair", 63: "couch", 64: "potted plant", 65: "bed", 66: "mirror", 67: "dining table", 68: "window", 69: "desk", 70: "toilet", 71: "door", 72: "tv", 73:" laptop", 74: "mouse", 75: "remote", 76:" keyboard", 77: "cell phone", 78: "microwave", 79: "oven", 80: "toaster", 81: "sink", 82: "refrigerator", 83: "blender", 84: "book", 85:"clock", 86: "vase", 87: "scissors", 88: "teddy bear", 89: "hair drier", 90: "toothbrush", } coco_encode = {value:key for key,value in coco_classes.items()} coco_labels = list(coco_classes.values()) # function: blur the image def blur_image(image = None,coordinates = None,blur_value = 3): #print('*********** INSIDE [blur_image()] *********]') img = image.copy() # copy the image to work on new image if (coordinates is not None): #print('Performing image blur operation...') for coord in (coordinates): ymin,xmin,ymax,xmax = coord #print('Image shape:',img.shape) # Extract region of intrest Y_min,X_min,Y_max,X_max = int(ymin*img.shape[0]),int(xmin*img.shape[1]),int(ymax*img.shape[0]),int(xmax*img.shape[1]) #print('Y_min,Y_max',Y_min,Y_max) #print('X_min,X_max',X_min,X_max) roi = img[Y_min:Y_max,X_min:X_max] #show_img(roi,'Original_roi') # blur the extracted img using Gausian blur try: roi = cv2.GaussianBlur(roi,ksize = (blur_value,blur_value),sigmaX = 0) #show_img(roi,title='blured roi') # replace the original roi with blured_roi img[Y_min:Y_max, X_min:X_max] = roi except: pass return img # function: filter detection boxs def filter_detection(detector_output,select_classes,thr = 0.6): # print('********* INSIDE [filter_detection()] **********') detection_boxs = detector_output['detection_boxes'] detection_class = detector_output['detection_classes'] detection_scores = detector_output['detection_scores'] # get the masking to select classes which user choosed masked_classes = np.isin(detection_class,select_classes) # select only selected classes detection_class = detection_class[masked_classes] detection_boxs = detection_boxs[masked_classes] detection_scores = detection_scores[masked_classes] # filter the detection boxses based on threshold selected_scores = detection_scores[detection_scores >= thr] selected_class = detection_class[detection_scores >= thr] selected_boxs = detection_boxs[detection_scores >= thr].numpy() return selected_boxs,selected_class,selected_scores # get the input video # load video from local disk def load_input(ip_path): #print('******* INSIDE [load_input] ********') try: cap = cv2.VideoCapture(ip_path) print('Video loaded successfully!') return cap except: print("Failed! to load video") #function: get video property like frame_width,frame_heigh,frame_per_second(fps),codecc def out_video(cap): #print('******** INSIDE [out_video] ***********') frame_width = int(cap.get(3)) # width of the fames in the video frame_height = int(cap.get(4)) # height of the frame in the video fps = int(cap.get(5)) # frame per second total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) video_duration = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))/ fps codecc = cv2.VideoWriter_fourcc(*'mp4V') # codecc for output video ( h264 codecc) # video property info print('Frame Width:',frame_width) print('Frame height:',frame_height) print('Frame Per Second:',fps) print('Total frames:',total_frames) print('video_duration: {} minutes'.format(round(video_duration/60),2)) # VideoWriter object to save blured video out = cv2.VideoWriter(out_video_path,codecc,fps,(frame_width,frame_height)) return out,fps,total_frames,video_duration # function: to get time range to perfrom blur def time_range(start_time,end_time): #print('*********** INSIDE [time_range()] ************') start_time,end_time = start_time,end_time # change to second(s) format return start_time,end_time # function: to check if time range is valid or not def is_valid_time_range(start_time,end_time,video_duration): #print('********** INSIDE [valid_time_range()] *************') return (0 <= start_time < end_time <= video_duration) # load model object_detection_model = hub.load("https://www.kaggle.com/models/tensorflow/efficientdet/frameworks/TensorFlow2/variations/d2/versions/1") def blur_video(input_video_path, u_classes, start_time, end_time): print('STARTING OF PROCESSING...') print("u_classes:",u_classes,type(u_classes)) label_encode = np.array([coco_encode[i] for i in u_classes], dtype='float16') print('label_encode:',label_encode,type(label_encode)) cap = load_input(ip_path=input_video_path) out, fps, total_frames, video_duration = out_video(cap) start_time, end_time = time_range(start_time, end_time) if is_valid_time_range(start_time, end_time, video_duration): start_frame = int(start_time * fps) end_frame = int(end_time * fps) print('Start Frame:', start_frame) print('End Frame:', end_frame) with ThreadPoolExecutor(max_workers=4) as executor: # Adjust max_workers as needed futures = [] for i in range(total_frames): ret, frame = cap.read() if ret: frame = tf.expand_dims(frame, axis=0) else: break if start_frame <= i <= end_frame: print('Blured_frame:',i) future = executor.submit(blur_process, frame, label_encode) futures.append(future) else: out.write(frame[0].numpy()) for future in futures: blured_img = future.result() out.write(blured_img) cap.release() out.release() return out_video_path def blur_process(frame,l_encoder,blur_value): print('label_encode',l_encoder) frame = np.expand_dims(frame,axis = 0) detector_output = object_detection_model(frame) boxes,classes,scores = filter_detection(detector_output,l_encoder) blured_img = blur_image(frame[0],boxes,blur_value) return blured_img def process_and_concat_video(input_video_path,u_classes,blur_value,start_time, end_time): label_encode = np.array([coco_encode[i] for i in u_classes],dtype = 'float16') # Load the full video clip full_video_clip = VideoFileClip(input_video_path) # Process the specified part of the video processed_clip = full_video_clip.subclip(start_time, end_time).set_duration(end_time - start_time) processed_clip = processed_clip.fl_image(lambda frame: blur_process(frame,label_encode,blur_value)) print('final clip fps:',full_video_clip.fps) print('processed_clip fps:',processed_clip.fps) # Concatenate the processed and unprocessed parts final_clip = concatenate_videoclips([full_video_clip.subclip(0, start_time), processed_clip, full_video_clip.subclip(end_time, None)]) final_clip.set_fps = 25 # Assuming desired FPS is 25 # Write the final video to an output file with the specified fps out_video_path = "output_video.mp4" final_clip.write_videofile(out_video_path, codec="h264", audio_codec="aac",fps = 25) return out_video_path if __name__ == "__main__": import gradio as gr iface = gr.Interface( fn=process_and_concat_video, inputs=[ gr.Video(label="Upload Video"), gr.CheckboxGroup(choices=coco_labels[1:], label="Select Classes"), gr.Slider(label = "blur intensity",minimum = 3,maximum = 90, step = 3), gr.Number(label="Start Time (seconds)"), gr.Number(label="End Time (seconds)"), ], outputs= "video", title = 'BlurVista 👓' ) iface.launch(debug = True,inline = False)