import torch import webvtt import os import cv2 from minigpt4.common.eval_utils import prepare_texts, init_model, eval_parser, eval_bleu,eval_cider,chat_gpt_eval from minigpt4.conversation.conversation import CONV_VISION from torchvision import transforms import json from tqdm import tqdm import soundfile as sf import argparse import moviepy.editor as mp import gradio as gr from pytubefix import YouTube import shutil from PIL import Image from moviepy.editor import VideoFileClip from theme import minigptlv_style, custom_css,text_css from huggingface_hub import login, hf_hub_download hf_token = os.environ.get('HF_TKN') login(token=hf_token) hf_hub_download( repo_id='Vision-CAIR/MiniGPT4-Video', filename='video_llama_checkpoint_last.pth', local_dir='checkpoints', local_dir_use_symlinks=False, ) import spaces def create_video_grid(images, rows, cols,save_path): image_width, image_height = images[0].size grid_width = cols * image_width grid_height = rows * image_height new_image = Image.new("RGB", (grid_width, grid_height)) for i in range(rows): for j in range(cols): index = i * cols + j if index < len(images): image = images[index] x_offset = j * image_width y_offset = i * image_height new_image.paste(image, (x_offset, y_offset)) # new_image.save(save_path) return new_image def prepare_input(vis_processor,video_path,subtitle_path,instruction): cap = cv2.VideoCapture(video_path) if subtitle_path is not None: # Load the VTT subtitle file vtt_file = webvtt.read(subtitle_path) print("subtitle loaded successfully") clip = VideoFileClip(video_path) total_num_frames = int(clip.duration * clip.fps) # print("Video duration = ",clip.duration) clip.close() else : # calculate the total number of frames in the video using opencv total_num_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) if "mistral" in args.ckpt : max_images_length=90 max_sub_len = 800 else: max_images_length = 45 max_sub_len = 400 images = [] frame_count = 0 sampling_interval = int(total_num_frames / max_images_length) if sampling_interval == 0: sampling_interval = 1 img_placeholder = "" subtitle_text_in_interval = "" history_subtitles = {} # raw_frames=[] number_of_words=0 transform=transforms.Compose([ transforms.ToPILImage(), ]) while cap.isOpened(): ret, frame = cap.read() if not ret: break # Find the corresponding subtitle for the frame and combine the interval subtitles into one subtitle # we choose 1 frame for every 2 seconds,so we need to combine the subtitles in the interval of 2 seconds if subtitle_path is not None: for subtitle in vtt_file: sub=subtitle.text.replace('\n',' ') if (subtitle.start_in_seconds <= (frame_count / int(clip.fps)) <= subtitle.end_in_seconds) and sub not in subtitle_text_in_interval: if not history_subtitles.get(sub,False): subtitle_text_in_interval+=sub+" " history_subtitles[sub]=True break if frame_count % sampling_interval == 0: # raw_frames.append(Image.fromarray(cv2.cvtColor(frame.copy(), cv2.COLOR_BGR2RGB))) frame = transform(frame[:,:,::-1]) # convert to RGB frame = vis_processor(frame) images.append(frame) img_placeholder += '' if subtitle_path is not None and subtitle_text_in_interval != "" and number_of_words< max_sub_len: img_placeholder+=f'{subtitle_text_in_interval}' number_of_words+=len(subtitle_text_in_interval.split(' ')) subtitle_text_in_interval = "" frame_count += 1 if len(images) >= max_images_length: break cap.release() cv2.destroyAllWindows() if len(images) == 0: # skip the video if no frame is extracted return None,None # video_grid_image=create_video_grid(raw_frames,8,len(raw_frames)//8,"concatenated.jpg") images = torch.stack(images) instruction = img_placeholder + '\n' + instruction return images,instruction def extract_audio(video_path, audio_path): video_clip = mp.VideoFileClip(video_path) audio_clip = video_clip.audio audio_clip.write_audiofile(audio_path, codec="libmp3lame", bitrate="320k") def generate_subtitles(video_path): video_id=video_path.split('/')[-1].split('.')[0] audio_path = f"workspace/inference_subtitles/mp3/{video_id}"+'.mp3' os.makedirs("workspace/inference_subtitles/mp3",exist_ok=True) if existed_subtitles.get(video_id,False): return f"workspace/inference_subtitles/{video_id}"+'.vtt' try: extract_audio(video_path,audio_path) print("successfully extracted") os.system(f"whisper {audio_path} --language English --model large --output_format vtt --output_dir workspace/inference_subtitles") # remove the audio file os.system(f"rm {audio_path}") print("subtitle successfully generated") return f"workspace/inference_subtitles/{video_id}"+'.vtt' except Exception as e: print("error",e) print("error",video_path) return None @spaces.GPU() def run (video_path,instruction,model,vis_processor,gen_subtitles=True): if gen_subtitles: subtitle_path=generate_subtitles(video_path) else : subtitle_path=None prepared_images,prepared_instruction=prepare_input(vis_processor,video_path,subtitle_path,instruction) if prepared_images is None: return "Video cann't be open ,check the video path again" length=len(prepared_images) prepared_images=prepared_images.unsqueeze(0) conv = CONV_VISION.copy() conv.system = "" # if you want to make conversation comment the 2 lines above and make the conv is global variable conv.append_message(conv.roles[0], prepared_instruction) conv.append_message(conv.roles[1], None) prompt = [conv.get_prompt()] answers = model.generate(prepared_images, prompt, max_new_tokens=args.max_new_tokens, do_sample=True, lengths=[length],num_beams=2) # remove the subtitle file and the video file if subtitle_path: os.system(f"rm {subtitle_path}") #if video_path.split('.')[-1] == 'mp4' or video_path.split('.')[-1] == 'mkv' or video_path.split('.')[-1] == 'avi': # os.system(f"rm {video_path}") return answers[0] def run_single_image (image_path,instruction,model,vis_processor): image=Image.open(image_path) image = vis_processor(image) prepared_images=torch.stack([image]) prepared_instruction=''+instruction length=len(prepared_images) prepared_images=prepared_images.unsqueeze(0) conv = CONV_VISION.copy() conv.system = "" # if you want to make conversation comment the 2 lines above and make the conv is global variable conv.append_message(conv.roles[0], prepared_instruction) conv.append_message(conv.roles[1], None) prompt = [conv.get_prompt()] answers = model.generate(prepared_images, prompt, max_new_tokens=args.max_new_tokens, do_sample=False, lengths=[length],num_beams=1) return answers[0] def download_video(youtube_url, download_finish): video_id=youtube_url.split('v=')[-1].split('&')[0] # Create a YouTube object youtube = YouTube(youtube_url) # Get the best available video stream video_stream = youtube.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().first() # if has_subtitles: # Download the video to the workspace folder print('Downloading video') video_stream.download(output_path="workspace",filename=f"{video_id}.mp4") print('Video downloaded successfully') processed_video_path= f"workspace/{video_id}.mp4" download_finish = gr.State(value=True) return processed_video_path, download_finish def get_video_url(url,has_subtitles): # get video id from url video_id=url.split('v=')[-1].split('&')[0] # Create a YouTube object youtube = YouTube(url) # Get the best available video stream video_stream = youtube.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().first() # if has_subtitles: # Download the video to the workspace folder print('Downloading video') video_stream.download(output_path="workspace",filename=f"{video_id}.mp4") print('Video downloaded successfully') return f"workspace/{video_id}.mp4" # else: # return video_stream.url def get_arguments(): parser = argparse.ArgumentParser(description="Inference parameters") parser.add_argument("--cfg-path", help="path to configuration file.",default="test_configs/llama2_test_config.yaml") parser.add_argument("--ckpt", type=str,default='checkpoints/video_llama_checkpoint_last.pth', help="path to checkpoint") parser.add_argument("--max_new_tokens", type=int, default=512, help="max number of generated tokens") parser.add_argument("--lora_r", type=int, default=64, help="lora rank of the model") parser.add_argument("--lora_alpha", type=int, default=16, help="lora alpha") parser.add_argument( "--options", nargs="+", help="override some settings in the used config, the key-value pair " "in xxx=yyy format will be merged into config file (deprecate), " "change to --cfg-options instead.", ) return parser.parse_args() args=get_arguments() model, vis_processor = init_model(args) conv = CONV_VISION.copy() conv.system = "" inference_subtitles_folder="workspace/inference_subtitles" os.makedirs(inference_subtitles_folder,exist_ok=True) existed_subtitles={} for sub in os.listdir(inference_subtitles_folder): existed_subtitles[sub.split('.')[0]]=True def gradio_demo_local(video_path,has_sub,instruction): pred=run(video_path,instruction,model,vis_processor,gen_subtitles=has_sub) return pred def gradio_demo_youtube(youtube_url,has_sub,instruction): video_path=get_video_url(youtube_url,has_sub) pred=run(video_path,instruction,model,vis_processor,gen_subtitles=has_sub) return pred def use_example(url,has_sub_1,q): # set the youtube link and the question with the example values youtube_link.value=url has_subtitles.value=has_sub_1 question.value=q title = """

MiniGPT4-video 🎞️🍿

""" description = """
This is the demo of MiniGPT4-video Model.
""" project_page = """

""" code_link="""

""" paper_link="""

""" #video_path="" with gr.Blocks(title="MiniGPT4-video 🎞️🍿",css=text_css ) as demo : # with gr.Row(): # with gr.Column(scale=2): gr.Markdown(title) gr.Markdown(description) # gr.Image("repo_imgs/Designer_2_new.jpeg",scale=1,show_download_button=False,show_label=False) # with gr.Row(): # gr.Markdown(project_page) # gr.Markdown(code_link) # gr.Markdown(paper_link) with gr.Tab("Local videos"): # local_interface=gr.Interface( # fn=gradio_demo_local, # inputs=[gr.Video(sources=["upload"]),gr.Checkbox(label='Use subtitles'),gr.Textbox(label="Write any Question")], # outputs=["text", # ], # # title="

Local videos

", # description="Upload your videos with length from one to two minutes", # examples=[ # ["example_videos/sample_demo_1.mp4", True, "Why is this video funny"], # ["example_videos/sample_demo_2.mp4", False, "Generate a creative advertisement for this product."], # ["example_videos/sample_demo_3.mp4", False, "Write a poem inspired by this video."], # ], # css=custom_css, # Apply custom CSS # allow_flagging='auto' # ) with gr.Row(): with gr.Column(): video_player_local = gr.Video(sources=["upload"]) question_local = gr.Textbox(label="Your Question", placeholder="Default: What's this video talking about?") has_subtitles_local = gr.Checkbox(label="Use subtitles", value=True) process_button_local = gr.Button("Answer the Question (QA)") with gr.Column(): answer_local=gr.Text("Answer will be here",label="MiniGPT4-video Answer") process_button_local.click(fn=gradio_demo_local, inputs=[video_player_local, has_subtitles_local, question_local], outputs=[answer_local]) with gr.Tab("Youtube videos"): # youtube_interface=gr.Interface( # fn=gradio_demo_youtube, # inputs=[gr.Textbox(label="Enter the youtube link"),gr.Checkbox(label='Use subtitles'),gr.Textbox(label="Write any Question")], # outputs=["text", # ], # # title="

YouTube videos

", # description="Videos length should be from one to two minutes", # examples=[ # ["https://www.youtube.com/watch?v=8kyg5u6o21k", True, "What happens in this video?"], # ["https://www.youtube.com/watch?v=zWfX5jeF6k4", True, "what is the main idea in this video?"], # ["https://www.youtube.com/watch?v=W5PRZuaQ3VM", True, "Inspired by this video content suggest a creative advertisement about the same content."], # ["https://www.youtube.com/watch?v=W8jcenQDXYg", True, "Describe what happens in this video."], # ["https://www.youtube.com/watch?v=u3ybWiEUaUU", True, "what is creative in this video ?"], # ["https://www.youtube.com/watch?v=nEwfSZfz7pw", True, "What Monica did in this video ?"], # ], # css=custom_css, # Apply custom CSS # allow_flagging='auto', # ) with gr.Row(): with gr.Column(): youtube_link = gr.Textbox(label="Enter the youtube link", placeholder="Paste YouTube URL here") video_player = gr.Video(autoplay=False) download_finish = gr.State(value=False) youtube_link.change( fn=download_video, inputs=[youtube_link, download_finish], outputs=[video_player, download_finish] ) question = gr.Textbox(label="Your Question", placeholder="Default: What's this video talking about?") has_subtitles = gr.Checkbox(label="Use subtitles", value=True) process_button = gr.Button("Answer the Question (QA)") with gr.Column(): answer=gr.Text("Answer will be here",label="MiniGPT4-video Answer") process_button.click(fn=gradio_demo_youtube, inputs=[youtube_link, has_subtitles, question], outputs=[answer]) ## Add examples to make the demo more interactive and user-friendly # with gr.Row(): # url_1=gr.Text("https://www.youtube.com/watch?v=8kyg5u6o21k") # has_sub_1=True # q_1=gr.Text("What happens in this video?") # # add button to change the youtube link and the question with the example values # use_example_1_btn=gr.Button("Use this example") # use_example_1_btn.click(use_example,inputs=[url_1,has_sub_1,q_1]) if __name__ == "__main__": demo.queue(max_size=10).launch(share=False,show_error=True, show_api=False)