| import cv2 |
| import base64 |
| import os |
| import argparse |
| import sys |
| from openai import OpenAI |
|
|
| def extract_frames(video_path, num_frames=5): |
| """Extract a few evenly spaced frames from a video and encode to base64.""" |
| cap = cv2.VideoCapture(video_path) |
| total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) |
| frames = [] |
| |
| if total_frames == 0: |
| return frames |
| |
| step = max(1, total_frames // num_frames) |
| |
| for i in range(num_frames): |
| cap.set(cv2.CAP_PROP_POS_FRAMES, min(i * step, total_frames - 1)) |
| ret, frame = cap.read() |
| if ret: |
| |
| _, buffer = cv2.imencode('.jpg', frame) |
| b64_img = base64.b64encode(buffer).decode('utf-8') |
| frames.append(f"data:image/jpeg;base64,{b64_img}") |
| |
| cap.release() |
| return frames |
|
|
| def analyze_video(video_source, prompt, api_key=None, model="MiniMaxAI/MiniMax-M3", base_url="https://api.featherless.ai/v1"): |
| api_key = api_key or os.getenv("FEATHERLESS_API_KEY") |
| if not api_key: |
| raise ValueError("API key must be provided or set in FEATHERLESS_API_KEY env var") |
| |
| client = OpenAI(base_url=base_url, api_key=api_key) |
| |
| print(f"[+] Extracting frames from {video_source}...") |
| frames = extract_frames(video_source, num_frames=5) |
| |
| if not frames: |
| raise RuntimeError("Failed to extract any frames from the video.") |
| |
| content = [{"type": "text", "text": prompt}] |
| for frame in frames: |
| content.append({"type": "image_url", "image_url": {"url": frame}}) |
| |
| messages = [ |
| { |
| "role": "user", |
| "content": content |
| } |
| ] |
| |
| print(f"[+] Sending {len(frames)} frames to vision model...") |
| response = client.chat.completions.create( |
| model=model, |
| messages=messages |
| ) |
| return response.choices[0].message.content |
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="Analyze video using Featherless AI vision models.") |
| parser.add_argument("--video-url", required=True, help="Video file path or URL.") |
| parser.add_argument("--prompt", required=True, help="Prompt for analysis.") |
| parser.add_argument("--api-key", help="Featherless API key (overrides FEATHERLESS_API_KEY env var).") |
| parser.add_argument("--model", default="MiniMaxAI/MiniMax-M3", help="Model name.") |
| parser.add_argument("--base-url", default="https://api.featherless.ai/v1", help="Base API URL.") |
| args = parser.parse_args() |
|
|
| try: |
| description = analyze_video( |
| video_source=args.video_url, |
| prompt=args.prompt, |
| api_key=args.api_key, |
| model=args.model, |
| base_url=args.base_url, |
| ) |
| print("\n--- Video Analysis ---\n") |
| print(description) |
| print("\n----------------------\n") |
| except Exception as e: |
| print(f"Error: {e}", file=sys.stderr) |
| sys.exit(1) |
|
|
| if __name__ == "__main__": |
| main() |
|
|
