Spaces:
Runtime error
Runtime error
| import torch | |
| import clip | |
| import cv2, youtube_dl | |
| from PIL import Image,ImageDraw, ImageFont | |
| import os | |
| from functools import partial | |
| from multiprocessing.pool import Pool | |
| import shutil | |
| from pathlib import Path | |
| import numpy as np | |
| import datetime | |
| import gradio as gr | |
| # load model and preprocess | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| model, preprocess = clip.load("ViT-B/32") | |
| def select_video_format(url, format_note='480p', ext='mp4'): | |
| defaults = ['480p', '360p','240p','144p'] | |
| ydl_opts = {} | |
| ydl = youtube_dl.YoutubeDL(ydl_opts) | |
| info_dict = ydl.extract_info(url, download=False) | |
| formats = info_dict.get('formats', None) | |
| available_format_notes = set([f['format_note'] for f in formats]) | |
| if format_note not in available_format_notes: | |
| format_note = [d for d in defaults if d in available_format_notes][0] | |
| formats = [f for f in formats if f['format_note'] == format_note and f['ext'] == ext and f['vcodec'].split('.')[0] != 'av01'] | |
| format = formats[0] | |
| format_id = format.get('format_id', None) | |
| fps = format.get('fps', None) | |
| print(f'format selected: {format}') | |
| return(format_id, fps) | |
| def download_video(url,format_id): | |
| # testing | |
| print(f"testing...all the files in local directory: {os.listdir('.')}") | |
| ydl_opts = { | |
| 'format':format_id, | |
| 'outtmpl': "%(id)s.%(ext)s"} | |
| with youtube_dl.YoutubeDL(ydl_opts) as ydl: | |
| try: | |
| ydl.cache.remove() | |
| meta = ydl.extract_info(url) | |
| save_location = meta['id'] + '.' + meta['ext'] | |
| except youtube_dl.DownloadError as error: | |
| print(f'error with download_video function: {error}') | |
| return(save_location) | |
| def process_video_parallel(video, skip_frames, dest_path, num_processes, process_number): | |
| cap = cv2.VideoCapture(video) | |
| frames_per_process = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) // (num_processes) | |
| count = frames_per_process * process_number | |
| print(f"worker: {process_number}, process frames {count} ~ {frames_per_process * (process_number + 1)} \n total number of frames: {cap.get(cv2.CAP_PROP_FRAME_COUNT)} \n video: {video}; isOpen? : {cap.isOpened()}") | |
| while count < frames_per_process * (process_number + 1) : | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| count += 1 | |
| if (count - frames_per_process * process_number) % skip_frames ==0: | |
| filename =f"{dest_path}/{count}.jpg" | |
| cv2.imwrite(filename, frame) | |
| #print(f"saved {filename}") | |
| cap.release() | |
| def vid2frames(url, sampling_interval=1, ext='mp4'): | |
| # create folder for extracted frames - if folder exists, delete and create a new one | |
| dest_path = Path('frames') | |
| try: | |
| dest_path.mkdir(parents=True) | |
| except FileExistsError: | |
| shutil.rmtree(dest_path) | |
| dest_path.mkdir(parents=True) | |
| # figure out the format for download, | |
| # by default select 480p, if not available, choose the best format available | |
| # mp4 | |
| format_id, fps = select_video_format(url, format_note='480p', ext='mp4') | |
| # download the video | |
| video = download_video(url,format_id) | |
| # calculate skip_frames | |
| try: | |
| skip_frames = int(fps * sampling_interval) | |
| except: | |
| skip_frames = int(30 * sampling_interval) | |
| print(f'video saved at: {video}, fps:{fps}, skip_frames: {skip_frames}') | |
| # extract video frames at given sampling interval with multiprocessing - | |
| print('extracting frames...') | |
| n_workers = min(os.cpu_count(), 1) | |
| # testing.. | |
| cap = cv2.VideoCapture(video) | |
| print(f'video: {video}; isOpen? : {cap.isOpened()}') | |
| print(f'n_workers: {n_workers}') | |
| with Pool(n_workers) as pool: | |
| pool.map(partial(process_video_parallel, video, skip_frames, dest_path, n_workers), range(n_workers)) | |
| # read frames | |
| original_images = [] | |
| images = [] | |
| filenames = sorted(dest_path.glob('*.jpg'),key=lambda p: int(p.stem)) | |
| print(f"extracted {len(filenames)} frames") | |
| for filename in filenames: | |
| image = Image.open(filename).convert("RGB") | |
| original_images.append(image) | |
| images.append(preprocess(image)) | |
| return original_images, images | |
| def captioned_strip(images, caption=None, times=None, rows=1): | |
| increased_h = 0 if caption is None else 30 | |
| w, h = images[0].size[0], images[0].size[1] | |
| img = Image.new("RGB", (len(images) * w // rows, h * rows + increased_h)) | |
| for i, img_ in enumerate(images): | |
| img.paste(img_, (i // rows * w, increased_h + (i % rows) * h)) | |
| if caption is not None: | |
| draw = ImageDraw.Draw(img) | |
| #font = ImageFont.load_default() | |
| #font_small = ImageFont.truetype("arial.pil", 12) | |
| font = ImageFont.truetype( | |
| "/usr/share/fonts/truetype/liberation2/LiberationMono-Bold.ttf", 16 | |
| ) | |
| font_small = ImageFont.truetype("/usr/share/fonts/truetype/liberation2/LiberationMono-Bold.ttf", 12) | |
| draw.text((60, 3), caption, (255, 255, 255), font=font) | |
| for i,ts in enumerate(times): | |
| draw.text(( | |
| (i % rows) * w + 40 , #column poistion | |
| i // rows * h + 33) # row position | |
| , ts, | |
| (255, 255, 255), font=font_small) | |
| return img | |
| def run_inference(url, sampling_interval, search_query): | |
| original_images, images = vid2frames(url,sampling_interval) | |
| image_input = torch.tensor(np.stack(images)).to(device) | |
| print("testing.. created image_input") | |
| with torch.no_grad(): | |
| image_features = model.encode_image(image_input) | |
| text_features = model.encode_text(clip.tokenize(search_query).to(device)) | |
| image_features /= image_features.norm(dim=-1, keepdim=True) | |
| text_features /= text_features.norm(dim=-1, keepdim=True) | |
| similarity = (100.0 * image_features @ text_features.T) | |
| values, indices = similarity.topk(4, dim=0) | |
| print("testing.. selected best frames") | |
| best_frames = [original_images[ind] for ind in indices] | |
| times = [f'{datetime.timedelta(seconds = ind[0].item() * sampling_interval)}' for ind in indices] | |
| print("testing... before captioned_strip func") | |
| image_output = captioned_strip(best_frames,search_query, times,2) | |
| title = search_query | |
| print("testing... after captioned_strip func") | |
| return(title, image_output) | |
| inputs = [gr.inputs.Textbox(label="Give us the link to your youtube video!"), | |
| gr.Number(5,label='sampling interval (seconds)'), | |
| gr.inputs.Textbox(label="What do you want to search?")] | |
| outputs = [ | |
| gr.outputs.HTML(label=""), # To be used as title | |
| gr.outputs.Image(label=""), | |
| ] | |
| gr.Interface( | |
| run_inference, | |
| inputs=inputs, | |
| outputs=outputs, | |
| title="It Happened One Frame", | |
| description='A CLIP-based app that search video frame based on text', | |
| examples=[ | |
| ['https://youtu.be/v1rkzUIL8oc', 1, "James Cagney dancing down the stairs"], | |
| ['https://youtu.be/k4R5wZs8cxI', 1, "James Cagney smashes a grapefruit into Mae Clarke's face"], | |
| ['https://youtu.be/0diCvgWv_ng', 1, "little Deborah practicing her ballet while wearing a tutu in empty restaurant"] | |
| ] | |
| ).launch(debug=True,enable_queue=True) | |