| --- |
| license: apache-2.0 |
| --- |
| |
| This repository contains the weights for the **VideoSAVi (Self-Aligned Video Language Model)** introduced in the paper [VideoSAVi: Self-Aligned Video Language Models without Human Supervision](https://arxiv.org/abs/2412.00624). |
| - **Project Page:** [https://people-robots.github.io/VideoSAVi/](https://people-robots.github.io/VideoSAVi/) |
|
|
| ## Usage Instructions |
| We provide sample inference code below. |
|
|
| ```python |
| #!/usr/bin/env python3 |
| |
| import argparse |
| import json |
| import numpy as np |
| import torch |
| import torchvision.transforms as T |
| from decord import VideoReader, cpu |
| from PIL import Image |
| from torchvision.transforms.functional import InterpolationMode |
| from transformers import AutoModel, AutoTokenizer |
| |
| IMAGENET_MEAN = (0.485, 0.456, 0.406) |
| IMAGENET_STD = (0.229, 0.224, 0.225) |
| |
| |
| def build_transform(input_size): |
| transform = T.Compose( |
| [T.Lambda(lambda img: img.convert("RGB") if img.mode != "RGB" else img), T.Resize((input_size, input_size), interpolation=InterpolationMode.BICUBIC), T.ToTensor(), T.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD)] |
| ) |
| return transform |
| |
| |
| def get_index(bound, fps, max_frame, first_idx=0, num_segments=32): |
| if bound: |
| start, end = bound[0], bound[1] |
| else: |
| start, end = -100000, 100000 |
| start_idx = max(first_idx, round(start * fps)) |
| end_idx = min(round(end * fps), max_frame) |
| seg_size = float(end_idx - start_idx) / num_segments |
| frame_indices = np.array([int(start_idx + (seg_size / 2) + np.round(seg_size * idx)) for idx in range(num_segments)]) |
| return frame_indices |
| |
| |
| def dynamic_preprocess(image, min_num=1, max_num=12, image_size=448, use_thumbnail=False): |
| orig_width, orig_height = image.size |
| aspect_ratio = orig_width / orig_height |
| |
| # calculate the existing image aspect ratio |
| target_ratios = set((i, j) for n in range(min_num, max_num + 1) for i in range(1, n + 1) for j in range(1, n + 1) if i * j <= max_num and i * j >= min_num) |
| target_ratios = sorted(target_ratios, key=lambda x: x[0] * x[1]) |
| |
| # find the closest aspect ratio to the target |
| best_ratio_diff = float("inf") |
| best_ratio = (1, 1) |
| area = orig_width * orig_height |
| for ratio in target_ratios: |
| target_aspect_ratio = ratio[0] / ratio[1] |
| ratio_diff = abs(aspect_ratio - target_aspect_ratio) |
| if ratio_diff < best_ratio_diff: |
| best_ratio_diff = ratio_diff |
| best_ratio = ratio |
| elif ratio_diff == best_ratio_diff: |
| if area > 0.5 * image_size * image_size * ratio[0] * ratio[1]: |
| best_ratio = ratio |
| |
| # calculate the target width and height |
| target_width = image_size * best_ratio[0] |
| target_height = image_size * best_ratio[1] |
| blocks = best_ratio[0] * best_ratio[1] |
| |
| # resize the image |
| resized_img = image.resize((target_width, target_height)) |
| processed_images = [] |
| for i in range(blocks): |
| box = ((i % (target_width // image_size)) * image_size, (i // (target_width // image_size)) * image_size, ((i % (target_width // image_size)) + 1) * image_size, ((i // (target_width // image_size)) + 1) * image_size) |
| # split the image |
| split_img = resized_img.crop(box) |
| processed_images.append(split_img) |
| assert len(processed_images) == blocks |
| if use_thumbnail and len(processed_images) != 1: |
| thumbnail_img = image.resize((image_size, image_size)) |
| processed_images.append(thumbnail_img) |
| return processed_images |
| |
| |
| def load_video(video_path, bound=None, input_size=448, max_num=12, num_segments=8): |
| vr = VideoReader(video_path, ctx=cpu(0), num_threads=1) |
| max_frame = len(vr) - 1 |
| fps = float(vr.get_avg_fps()) |
| |
| pixel_values_list, num_patches_list = [], [] |
| transform = build_transform(input_size=input_size) |
| frame_indices = get_index(bound, fps, max_frame, first_idx=0, num_segments=num_segments) |
| |
| for frame_index in frame_indices: |
| img = Image.fromarray(vr[frame_index].asnumpy()).convert("RGB") |
| img = dynamic_preprocess(img, image_size=input_size, use_thumbnail=True, max_num=max_num) |
| pixel_values = [transform(tile) for tile in img] |
| pixel_values = torch.stack(pixel_values) |
| num_patches_list.append(pixel_values.shape[0]) |
| pixel_values_list.append(pixel_values) |
| |
| pixel_values = torch.cat(pixel_values_list) |
| return pixel_values, num_patches_list |
| |
| |
| def parse_args(): |
| parser = argparse.ArgumentParser(description="Inference Script") |
| |
| parser.add_argument("--video_path", type=str, required=True, help="Path to the input video file") |
| parser.add_argument("--model_path", type=str, default="yogkul2000/VideoSAVi", help="Path to the VideoSAVi model") |
| |
| parser.add_argument("--num_segments", type=int, default=8, help="Number of video segments to sample (default: 8)") |
| |
| parser.add_argument("--max_patches", type=int, default=12, help="Maximum patches per frame (default: 12)") |
| |
| parser.add_argument("--input_size", type=int, default=448, help="Input image size (default: 448)") |
| |
| parser.add_argument("--max_new_tokens", type=int, default=1024, help="Maximum number of tokens to generate (default: 1024)") |
| |
| parser.add_argument("--do_sample", action="store_true", default=False, help="Whether to use sampling for generation") |
| |
| parser.add_argument("--temperature", type=float, default=0, help="Sampling temperature)") |
| |
| parser.add_argument("--top_p", type=float, default=1.0, help="Top-p sampling parameter (default: 1.0)") |
| |
| parser.add_argument("--question", type=str, default="What is happening in this video?", help="Question to ask about the video") |
| |
| parser.add_argument("--output_file", type=str, default=None, help="Optional output file to save results") |
| |
| parser.add_argument("--device", type=str, default="cuda", help="Device to use for inference (default: cuda)") |
| |
| parser.add_argument("--torch_dtype", type=str, default="bfloat16", choices=["float16", "bfloat16", "float32"], help="Torch dtype for model (default: bfloat16)") |
| |
| parser.add_argument("--verbose", action="store_true", help="Enable verbose output") |
| |
| parser.add_argument("--no_follow_up", action="store_true", help="Skip follow-up question") |
| |
| return parser.parse_args() |
| |
| |
| def main(): |
| args = parse_args() |
| |
| if args.verbose: |
| print(f"Loading model from: {args.model_path}") |
| print(f"Processing video: {args.video_path}") |
| print(f"Video segments: {args.num_segments}") |
| print(f"Max patches per frame: {args.max_patches}") |
| |
| torch_dtype_map = {"float16": torch.float16, "bfloat16": torch.bfloat16, "float32": torch.float32} |
| torch_dtype = torch_dtype_map[args.torch_dtype] |
| |
| try: |
| model = AutoModel.from_pretrained(args.model_path, torch_dtype=torch_dtype, low_cpu_mem_usage=True, use_flash_attn=True, trust_remote_code=True).eval() |
| |
| if args.device == "cuda" and torch.cuda.is_available(): |
| model = model.cuda() |
| |
| tokenizer = AutoTokenizer.from_pretrained(args.model_path, trust_remote_code=True, use_fast=False) |
| |
| if args.verbose: |
| print("Model and tokenizer loaded successfully!") |
| |
| except Exception as e: |
| print(f"Error loading model: {e}") |
| return |
| |
| try: |
| if args.verbose: |
| print("Loading and processing video...") |
| |
| pixel_values, num_patches_list = load_video(args.video_path, num_segments=args.num_segments, max_num=args.max_patches, input_size=args.input_size) |
| |
| pixel_values = pixel_values.to(torch_dtype) |
| if args.device == "cuda" and torch.cuda.is_available(): |
| pixel_values = pixel_values.cuda() |
| |
| if args.verbose: |
| print(f"Video processed: {len(num_patches_list)} segments, {pixel_values.shape[0]} total patches") |
| |
| except Exception as e: |
| print(f"Error processing video: {e}") |
| return |
| |
| # Create video prefix for frames |
| video_prefix = "".join([f"Frame{i+1}: <image>\n" for i in range(len(num_patches_list))]) |
| |
| # Generation config |
| generation_config = {"max_new_tokens": args.max_new_tokens, "do_sample": args.do_sample, "temperature": args.temperature, "top_p": args.top_p} |
| |
| results = {} |
| |
| try: |
| question = video_prefix + args.question |
| |
| if args.verbose: |
| print(f"\nAsking question: {args.question}") |
| |
| response, history = model.chat(tokenizer, pixel_values, question, generation_config, num_patches_list=num_patches_list, history=None, return_history=True) |
| |
| print(f"\nUser: {args.question}") |
| print(f"VideoSAVi: {response}") |
| |
| results["question_1"] = {"question": args.question, "response": response} |
| |
| # Clear GPU cache |
| if args.device == "cuda" and torch.cuda.is_available(): |
| torch.cuda.empty_cache() |
| |
| except Exception as e: |
| print(f"Error during first inference: {e}") |
| return |
| |
| # Save results if output file specified |
| if args.output_file: |
| try: |
| results["video_path"] = args.video_path |
| results["model_path"] = args.model_path |
| results["config"] = {"num_segments": args.num_segments, "max_patches": args.max_patches, "input_size": args.input_size, "generation_config": generation_config} |
| |
| with open(args.output_file, "w") as f: |
| json.dump(results, f, indent=2) |
| |
| print(f"\nResults saved to: {args.output_file}") |
| |
| except Exception as e: |
| print(f"Error saving results: {e}") |
| |
| if args.verbose: |
| print("\nInference completed successfully!") |
| |
| |
| if __name__ == "__main__": |
| main() |
| ``` |
|
|
|
|
