Create app.py

app.py

@@ -0,0 +1,113 @@
+import cv2
+import torch
+from PIL import Image
+import numpy as np
+import os
+import shutil
+import gradio as gr
+from transformers import LlavaNextVideoProcessor, LlavaNextVideoForConditionalGeneration, BitsAndBytesConfig
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+model_id = "llava-hf/LLaVA-NeXT-Video-7B-hf"
+
+quantization_config = BitsAndBytesConfig(
+    load_in_4bit=True,
+    bnb_4bit_compute_dtype=torch.float16,
+    bnb_4bit_use_double_quant=True,
+    bnb_4bit_quant_type="nf4"
+)
+
+model = LlavaNextVideoForConditionalGeneration.from_pretrained(
+    model_id,
+    quantization_config=quantization_config,
+    low_cpu_mem_usage=True,
+    device_map="auto"
+)
+
+processor = LlavaNextVideoProcessor.from_pretrained(model_id)
+
+def sample_frames(video_path, num_frames):
+    output_dir = "/tmp/processed_frames"
+    
+    if os.path.exists(output_dir):
+        shutil.rmtree(output_dir)
+    os.makedirs(output_dir)
+    
+    video = cv2.VideoCapture(video_path)
+    
+    if not video.isOpened():
+        raise ValueError(f"Could not open video file: {video_path}")
+    
+    total_frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
+    interval = max(1, total_frames // num_frames)
+    frames = []
+    frame_count = 0
+    
+    for i in range(total_frames):
+        ret, frame = video.read()
+        if not ret:
+            continue
+        if i % interval == 0 and len(frames) < num_frames:
+            cv2.imwrite(f"{output_dir}/frame_{frame_count:03d}.jpg", frame)
+            pil_img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+            frames.append(pil_img)
+            frame_count += 1
+    
+    video.release()
+    
+    frame_paths = [f"{output_dir}/frame_{i:03d}.jpg" for i in range(frame_count)]
+    
+    return frames, frame_paths
+
+def analyze_video(video_path):
+    conversation = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "Analyze this gas pipe quality control video. Answer these two questions with True/False: 1) DIPPED IN WATER: Was the pipe dipped in water for testing? Look for pipe being submerged in water container. 2) BUBBLES AFTER IMMERSION: After the pipe was fully immersed (ignore initial displacement bubbles), were there any bubbles indicating a leak? Format: DIPPED IN WATER: True/False, BUBBLES AFTER IMMERSION: True/False"},
+                {"type": "video"},
+            ],
+        },
+    ]
+    
+    prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
+    
+    video_frames, frame_paths = sample_frames(video_path, 20)
+    
+    inputs = processor(text=prompt, videos=video_frames, padding=True)
+    inputs = {k: v.to(model.device) for k, v in inputs.items()}
+    
+    output = model.generate(
+        **inputs,
+        max_new_tokens=100,
+        do_sample=True,
+        temperature=0.3,
+        top_p=0.9,
+        top_k=50,
+        repetition_penalty=1.1,
+        pad_token_id=processor.tokenizer.eos_token_id
+    )
+    
+    result = processor.decode(output[0][2:], skip_special_tokens=True)
+    
+    return frame_paths, result
+
+examples = [
+    ["07.mp4"],
+    ["09.mp4"], 
+    ["29.mp4"]
+]
+
+iface = gr.Interface(
+    fn=analyze_video,
+    inputs=gr.Video(),
+    outputs=[
+        gr.Gallery(label="Processed Frames"),
+        gr.Textbox(label="LLM Analysis", lines=10)
+    ],
+    title="Gas Pipe Quality Control Analyzer",
+    examples=examples,
+    cache_examples=False
+)
+
+iface.launch(share=True)