import gradio as gr import ffmpeg from pathlib import Path import os import ast import json import base64 import requests import moviepy.editor as mp from PIL import Image, ImageSequence import cv2 API_URL = "https://api-inference.huggingface.co/models/facebook/wav2vec2-base-960h" HF_TOKEN = os.environ["HF_TOKEN"] headers = {"Authorization": f"Bearer {HF_TOKEN}"} video_list = [] def generate_transcripts(in_video): #generate_gifs(in_video, gif_transcript): print("********* Inside generate_transcripts() **********") #convert video to audio print(f" input video is : {in_video}") #sample #video_path = Path("./ShiaLaBeouf.mp4") audio_memory, _ = ffmpeg.input(in_video).output('-', format="wav", ac=1, ar='16k').overwrite_output().global_args('-loglevel', 'quiet').run(capture_stdout=True) #audio_memory, _ = ffmpeg.input(video_path).output('-', format="wav", ac=1, ar='16k').overwrite_output().global_args('-loglevel', 'quiet').run(capture_stdout=True) #Getting transcripts using wav2Vec2 huggingface hosted accelerated inference #sending audio file in request along with stride and chunk length information model_response = query_api(audio_memory) #model response has both - transcripts as well as character timestamps or chunks print(f"model_response is : {model_response}") transcription = model_response["text"].lower() chnk = model_response["chunks"] #creating lists from chunks to consume downstream easily timestamps = [[chunk["text"].lower(), chunk["timestamp"][0], chunk["timestamp"][1]] for chunk in chnk] #getting words and word timestamps words, words_timestamp = get_word_timestamps(timestamps) print(f"Total words in the audio transcript is:{len(words)}, transcript word list is :{words}, type of words is :{type(words)} ") print(f"Total Word timestamps derived fromcharacter timestamp are :{len(words_timestamp)}, Word timestamps are :{words_timestamp}") return transcription, words, words_timestamp def generate_gifs(in_video, gif_transcript, words, words_timestamp, vid_speed): print("********* Inside generate_gifs() **********") #creating list from input gif transcript #gif = "don't let your dreams be dreams" gif = gif_transcript #gif = gif_transcript giflist = gif.split() #getting gif indexes from the generator # Converting string to list words = ast.literal_eval(words) words_timestamp = ast.literal_eval(words_timestamp) print(f"words is :{words}") print(f"type of words is :{type(words)}") print(f"length of words is :{len(words)}") print(f"giflist is :{giflist}") giflist_indxs = list(list(get_gif_word_indexes(words, giflist))[0]) print(f"giflist_indxs is : {giflist_indxs}") #getting start and end timestamps for a gif video start_seconds, end_seconds = get_gif_timestamps(giflist_indxs, words_timestamp) print(f"start_seconds, end_seconds are : ({start_seconds}, {end_seconds})") #generated .gif image #gif_out, vid_out = gen_moviepy_gif(in_video, start_seconds, end_seconds) print(f"vid_speed from SLider is : {vid_speed}") speededit_vids_list, concat_vid = gen_moviepy_gif(in_video, start_seconds, end_seconds, float(vid_speed), video_list) return concat_vid #speededit_vids_list #calling the hosted model def query_api(audio_bytes: bytes): """ Query for Huggingface Inference API for Automatic Speech Recognition task """ print("********* Inside query_api() **********") payload = json.dumps({ "inputs": base64.b64encode(audio_bytes).decode("utf-8"), "parameters": { "return_timestamps": "char", "chunk_length_s": 10, "stride_length_s": [4, 2] }, "options": {"use_gpu": False} }).encode("utf-8") response = requests.request( "POST", API_URL, headers=headers, data=payload) json_reponse = json.loads(response.content.decode("utf-8")) print(f"json_reponse is :{json_reponse}") return json_reponse #getting word timestamps from character timestamps def get_word_timestamps(timestamps): words, word = [], [] letter_timestamp, word_timestamp, words_timestamp = [], [], [] for idx,entry in enumerate(timestamps): word.append(entry[0]) letter_timestamp.append(entry[1]) if entry[0] == ' ': words.append(''.join(word)) word_timestamp.append(letter_timestamp[0]) word_timestamp.append(timestamps[idx-1][2]) words_timestamp.append(word_timestamp) word, word_timestamp, letter_timestamp = [], [], [] words = [word.strip() for word in words] return words, words_timestamp #getting index of gif words in main transcript def get_gif_word_indexes(total_words_list, gif_words_list): if not gif_words_list: return # just optimization COUNT=0 lengthgif_words_list = len(gif_words_list) firstgif_words_list = gif_words_list[0] print(f"total_words_list is :{total_words_list}") print(f"length of total_words_list is :{len(total_words_list)}") print(f"gif_words_list is :{gif_words_list}") print(f"length of gif_words_list is :{len(gif_words_list)}") for idx, item in enumerate(total_words_list): COUNT+=1 if item == firstgif_words_list: if total_words_list[idx:idx+lengthgif_words_list] == gif_words_list: print(f"value of tuple is : {tuple(range(idx, idx+lengthgif_words_list))}") yield tuple(range(idx, idx+lengthgif_words_list)) #getting start and end timestamps for gif transcript def get_gif_timestamps(giflist_indxs, words_timestamp): print(f"******** Inside get_gif_timestamps() **********") min_idx = min(giflist_indxs) max_idx = max(giflist_indxs) print(f"min_idx is :{min_idx}") print(f"max_idx is :{max_idx}") gif_words_timestamp = words_timestamp[min_idx : max_idx+1] print(f"words_timestamp is :{words_timestamp}") print(f"gif_words_timestamp is :{gif_words_timestamp}") start_seconds, end_seconds = gif_words_timestamp[0][0], gif_words_timestamp[-1][-1] print(f"start_seconds, end_seconds are :{start_seconds},{end_seconds}") return start_seconds, end_seconds #extracting the video and building and serving a .gif image def gen_moviepy_gif(in_video, start_seconds, end_seconds, vid_speed, vid_list): print("******** inside moviepy_gif () ***************") #sample #video_path = "./ShiaLaBeouf.mp4" video = mp.VideoFileClip(in_video) #video = mp.VideoFileClip(video_path) leftover_clip_start = video.subclip(0, int(start_seconds) + float("{:.2f}".format(1-start_seconds%1))).without_audio() #float("{:.2f}".format(1-a%1)) final_clip = video.subclip(start_seconds, end_seconds) leftover_clip_end = video.subclip(int(end_seconds) + float("{:.2f}".format(1-end_seconds%1)) ).without_audio() #end=None #slowmo print(f"vid_speed from calling function is : {vid_speed}") speededit_clip = final_clip.fx(mp.vfx.speedx, vid_speed) speededit_clip = speededit_clip.without_audio() #concat concatenated_clip = mp.concatenate_videoclips([leftover_clip_start, speededit_clip, leftover_clip_end]) concatenated_clip.write_videofile("concat.mp4") filename = f"speededit{len(vid_list)}" print("filename is :",filename) speededit_clip.write_videofile("speededit.mp4") #(filename) vid_list.append("speededit.mp4") #(filename) if len(vid_list) == 1: speededit_clip.write_videofile("slomo.mp4") elif len(vid_list) == 2: speededit_clip.write_videofile("timelapse.mp4") #writing to RAM - gif and smaller clip #final_clip.write_gif("gifimage.gif") #, program='ffmpeg', tempfiles=True, fps=15, fuzz=3) #final_clip.write_videofile("gifimage.mp4") final_clip.close() #reading in a variable #gif_img = mp.VideoFileClip("gifimage.gif") #gif_vid = mp.VideoFileClip("gifimage.mp4") #im = Image.open("gifimage.gif") #vid_cap = cv2.VideoCapture('gifimage.mp4') return vid_list, "concat.mp4" #"slomo.mp4", "timelapse.mp4", #"gifimage.gif", "gifimage.mp4" #im, gif_img, gif_vid, vid_cap, #"gifimage.mp4" sample_video = ["olympic100m.mp4"] #[['./ShiaLaBeouf.mp4']] sample_vid = gr.Video(label='Video file') #for displaying the example examples = gr.components.Dataset(components=[sample_vid], samples=[sample_video], type='values') demo = gr.Blocks() with demo: gr.Markdown("""# **Watch your video in SloMo or in Timelapse!** """) gr.Markdown(""" ### **This is still work under progres......** Editing your video using ASR pipeline.. A Space by [Yuvraj Sharma](https://huggingface.co/ysharma). **Background:** In this Gradio BLocks Party Space, I am trying to - - Provide a capability to slow down your video - Timelapse your video **How To Use:** 1. Upload a video or simply click on the sample provided here. 2. Then click on 'Generate transcripts' button and first textbox will display the extract Transcript from the audio associated with your sample. 3. Clip the text from transcript or type transcripts manually in the second Textbox provided. 4. A slowed down or timelapsed version of your video will get generated on the right hand side ! Hope you have fun using this 😀 """) with gr.Row(): #for incoming video input_video = gr.Video(label="Upload a Video", visible=True) #to generate and display transcriptions for input video text_transcript = gr.Textbox(label="Transcripts", lines = 10, interactive = True ) #Just to move data between function hence keeping visible false text_words = gr.Textbox(visible=False) text_wordstimestamps = gr.Textbox(visible=False) with gr.Row(): button_transcript = gr.Button("Generate transcripts") #For SlowMo with gr.Row(): #to copy paste required gif transcript / or to populate by itself on pressing the button text_slomo_transcript = gr.Textbox(label="Transcripts", placeholder="Copy paste transcripts here to create SlowMo Video" , lines = 5, interactive = True ) def load_slomo_text(text): print("****** inside load_slomo_text() ******") print("text for slomo video is : ", text) return text text_transcript.change(load_slomo_text, text_transcript, text_slomo_transcript ) #out_gif = gr.Image(label="Generated GIF image") out_slomo_vid = gr.Video(label="Generated SlowMo Video") with gr.Row(): #button_transcript = gr.Button("Generate transcripts") vid_speed_slomo = gr.Slider(0.1,0.9, step=0.1) button_slomo = gr.Button("Create SloMo") #For TimeLapse with gr.Row(): #to copy paste required gif transcript / or to populate by itself on pressing the button text_timelapse_transcript = gr.Textbox(label="Transcripts", placeholder="Copy paste transcripts here to create GIF image" , lines = 5) #, interactive = True ) def load_timelapse_text(text): print("****** inside load_timelapse_text() ******") print("text for timelapse video is : ", text) return text text_transcript.change(load_timelapse_text, text_transcript, text_timelapse_transcript ) #out_gif = gr.Image(label="Generated GIF image") out_timelapse_vid = gr.Video(label="Generated TimeLapse Video") with gr.Row(): #button_transcript = gr.Button("Generate transcripts") vid_speed_timelapse = gr.Slider(1,2, step=0.25) button_timelapse = gr.Button("Create TimeLapse") with gr.Row(): #to render video example on mouse hover/click examples.render() #to load sample video into input_video upon clicking on it def load_examples(video): print("****** inside load_example() ******") print("in_video is : ", video[0]) return video[0] examples.click(load_examples, examples, input_video) #vid_speed = gr.Slider(0.1,0.9, step=0.1) button_transcript.click(generate_transcripts, input_video, [text_transcript, text_words, text_wordstimestamps ]) button_slomo.click(generate_gifs, [input_video, text_slomo_transcript, text_words, text_wordstimestamps, vid_speed_slomo], out_slomo_vid ) button_timelapse.click(generate_gifs, [out_slomo_vid, text_timelapse_transcript, text_words, text_wordstimestamps, vid_speed_timelapse], out_timelapse_vid ) demo.launch(debug=True)